From cf10d6d01dc47ea7fa5ee00abf41ec71873c3e91 Mon Sep 17 00:00:00 2001 From: lizhaoxinlizhaoxin2-lab <288551507+lizhaoxinlizhaoxin2-lab@users.noreply.github.com> Date: Fri, 18 Sep 2026 12:38:05 +0800 Subject: [PATCH] fix(responses): emit protocol-correct stream errors once --- .../handler/openai_gateway_handler_test.go | 14 +++++ .../openai_gateway_response_handling.go | 17 ++++-- .../service/openai_gateway_service_test.go | 55 +++++++++++++++++++ 3 files changed, 80 insertions(+), 6 deletions(-) diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index 29b17d1655..916063cae5 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -381,6 +381,20 @@ func TestOpenAIEnsureForwardErrorResponse_AfterDeltaAppendsSingleValidResponseFa require.Equal(t, 1, errorEvents) } +func TestOpenAIEnsureForwardErrorResponse_PreservesCommittedStreamError(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil) + body := "event: error\ndata: {\"type\":\"error\",\"code\":\"upstream_stream_read_error\",\"message\":\"Upstream response stream was interrupted\",\"param\":null,\"sequence_number\":0}\n\n" + _, err := c.Writer.WriteString(body) + require.NoError(t, err) + service.MarkResponseCommitted(c) + h := &OpenAIGatewayHandler{} + require.False(t, h.ensureForwardErrorResponse(c, true)) + require.Equal(t, body, w.Body.String(), "preserve the service error without a second terminal event") +} + func TestOpenAIEnsureForwardErrorResponse_CompactKeepaliveOnlyWritesResponseFailed(t *testing.T) { gin.SetMode(gin.TestMode) w := httptest.NewRecorder() diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index efb9f2e0d8..8956bb58a1 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -314,17 +314,19 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. eventStartsTTFTOutput = false eventShouldFlush = false } - sendErrorEvent := func(reason string) { + sendErrorEvent := func(code, message string) { if errorEventSent || clientDisconnected || failureDelivered { return } errorEventSent = true - payload := `{"type":"error","sequence_number":0,"error":{"type":"upstream_error","message":` + strconv.Quote(reason) + `,"code":` + strconv.Quote(reason) + `}}` + // Responses error events use top-level code/message/param fields. A nested + // Chat Completions error envelope loses the classification in strict clients. + payload := `{"type":"error","sequence_number":0,"code":` + strconv.Quote(code) + `,"message":` + strconv.Quote(message) + `,"param":null}` if err := flushBuffered(); err != nil { clientDisconnected = true return } - if _, err := writePendingString("data: " + payload + "\n\n"); err != nil { + if _, err := writePendingString("event: error\ndata: " + payload + "\n\n"); err != nil { clientDisconnected = true return } @@ -334,6 +336,8 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } clientOutputStarted = true lastDownstreamWriteAt = time.Now() + // The handler must not append its generic failure after this error event. + MarkResponseCommitted(c) } needModelReplace := originalModel != mappedModel @@ -450,7 +454,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } if errors.Is(scanErr, bufio.ErrTooLong) { logger.LegacyPrintf("service.openai_gateway", "SSE line too long: account=%d max_size=%d error=%v", account.ID, maxLineSize, scanErr) - sendErrorEvent("response_too_large") + sendErrorEvent("response_too_large", "Upstream response exceeded the size limit") return resultWithUsage(), scanErr, true } if !openAIStreamClientOutputStarted(c, clientOutputStarted) && !eventShouldFlush { @@ -465,7 +469,8 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. return resultWithUsage(), fmt.Errorf("stream usage incomplete after disconnect: %w", scanErr), true } s.recordOpenAIProxyStreamDisconnect(account, scanErr, upstreamRequestID) - sendErrorEvent("stream_read_error") + code, message := classifyOpenAIUpstreamStreamReadError(scanErr) + sendErrorEvent(code, message) return resultWithUsage(), fmt.Errorf("stream read error: %w", scanErr), true } processSSELine := func(line string, queueDrained bool) { @@ -914,7 +919,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. return resultWithUsage(), grokStreamIdleFailoverError(account, streamInterval) } } - sendErrorEvent("stream_timeout") + sendErrorEvent("stream_timeout", "Upstream response stream timed out") return resultWithUsage(), fmt.Errorf("stream data interval timeout") case <-firstOutputCh: diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index cf4611ccce..c3c5474eea 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -1610,6 +1610,10 @@ func TestOpenAIStreamingTimeout(t *testing.T) { if !strings.Contains(rec.Body.String(), "\"type\":\"error\"") || !strings.Contains(rec.Body.String(), "stream_timeout") { t.Fatalf("expected OpenAI-compatible error SSE event, got %q", rec.Body.String()) } + payload := strings.TrimSpace(strings.TrimPrefix(rec.Body.String(), "event: error\ndata: ")) + require.Equal(t, "stream_timeout", gjson.Get(payload, "code").String()) + require.False(t, gjson.Get(payload, "error").Exists()) + require.True(t, IsResponseCommitted(c)) } func TestOpenAIStreamingContextCanceledReturnsIncompleteErrorWithoutInjectingErrorEvent(t *testing.T) { @@ -1674,6 +1678,53 @@ func TestOpenAIStreamingReadErrorBeforeOutputReturnsFailover(t *testing.T) { require.Empty(t, rec.Body.String()) } +func TestOpenAIStreamingReadErrorAfterOutputUsesResponsesErrorSchema(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, tc := range []struct { + name string + cause error + code string + }{ + {"reset", errors.New("read tcp 192.0.2.1:1234->192.0.2.2:443: connection reset by peer"), OpenAIUpstreamStreamReadErrorCode}, + {"http2", errors.New("stream error: stream ID 3; INTERNAL_ERROR; received from peer"), OpenAIUpstreamHTTP2StreamErrorCode}, + } { + t.Run(tc.name, func(t *testing.T) { + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}} + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: &openAIStreamReadThenErrorCloser{ + reader: strings.NewReader("event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"), + err: tc.cause, + }} + _, err := svc.handleStreamingResponse(c.Request.Context(), resp, c, &Account{ID: 1, Platform: PlatformOpenAI}, time.Now(), "model", "model") + require.ErrorIs(t, err, tc.cause) + body := rec.Body.String() + require.Contains(t, body, "partial") + require.NotContains(t, body, "192.0.2.") + require.NotContains(t, body, "stream ID") + require.NotContains(t, body, "response.completed") + require.NotContains(t, body, "[DONE]") + require.Equal(t, 1, strings.Count(body, "event: error\n")) + require.True(t, IsResponseCommitted(c), "the handler must not append another failure") + var events []gjson.Result + for _, line := range strings.Split(body, "\n") { + if strings.HasPrefix(line, "data: ") { + event := gjson.Parse(strings.TrimPrefix(line, "data: ")) + if event.Get("type").String() == "error" { + events = append(events, event) + } + } + } + require.Len(t, events, 1) + require.Equal(t, tc.code, events[0].Get("code").String()) + require.NotEmpty(t, events[0].Get("message").String()) + require.False(t, events[0].Get("error").Exists(), "Responses errors have top-level fields") + require.True(t, events[0].Get("param").Exists()) + }) + } +} + func TestOpenAIStreamingPostOutputDisconnectQuarantinesSharedProxyWithoutSameStreamFailover(t *testing.T) { gin.SetMode(gin.TestMode) proxyID := int64(4698) @@ -2791,6 +2842,10 @@ func TestOpenAIStreamingTooLong(t *testing.T) { if !strings.Contains(rec.Body.String(), "\"type\":\"error\"") || !strings.Contains(rec.Body.String(), "response_too_large") { t.Fatalf("expected OpenAI-compatible error SSE event, got %q", rec.Body.String()) } + payload := strings.TrimSpace(strings.TrimPrefix(rec.Body.String(), "event: error\ndata: ")) + require.Equal(t, "response_too_large", gjson.Get(payload, "code").String()) + require.False(t, gjson.Get(payload, "error").Exists()) + require.True(t, IsResponseCommitted(c)) } func TestOpenAINonStreamingContentTypePassThrough(t *testing.T) {