From 8e27ff20afdb7af0c6231696c9e041f4e9a15915 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Fri, 5 Jun 2026 18:11:23 +0800 Subject: [PATCH] fix(openai): handle missing messages stream terminal --- .../handler/openai_gateway_handler.go | 12 ++ .../service/openai_compat_model_test.go | 177 +++++++++++++++++- .../service/openai_gateway_messages.go | 82 ++++++-- .../service/openai_gateway_service.go | 1 + 4 files changed, 250 insertions(+), 22 deletions(-) diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index f3d4caf08d..2a147f2b05 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -758,6 +758,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { if channelMappingMsg.Mapped { forwardBody = h.gatewayService.ReplaceModelInBody(body, channelMappingMsg.MappedModel) } + writerSizeBeforeForward := c.Writer.Size() result, err := func() (*service.OpenAIForwardResult, error) { defer func() { if accountReleaseFunc != nil { @@ -787,6 +788,10 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { } else { var failoverErr *service.UpstreamFailoverError if errors.As(err, &failoverErr) { + if c.Writer.Size() != writerSizeBeforeForward { + h.handleAnthropicFailoverExhausted(c, failoverErr, true) + return + } h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, false, nil) // 池模式：同账号重试 if failoverErr.RetryableOnSameAccount { @@ -827,6 +832,13 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { ) continue } + if result != nil && result.ClientDisconnect { + reqLog.Info("openai_messages.client_disconnected", + zap.Int64("account_id", account.ID), + zap.Error(err), + ) + return + } h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, false, nil) wroteFallback := h.ensureAnthropicErrorResponse(c, streamStarted) reqLog.Warn("openai_messages.forward_failed", diff --git a/backend/internal/service/openai_compat_model_test.go b/backend/internal/service/openai_compat_model_test.go index f8b9d36033..7c7ac1b94f 100644 --- a/backend/internal/service/openai_compat_model_test.go +++ b/backend/internal/service/openai_compat_model_test.go @@ -1663,7 +1663,7 @@ func TestForwardAsAnthropic_BufferedEventNamedTerminalWithoutUpstreamCloseReturn } } -func TestForwardAsAnthropic_DoneSentinelWithoutTerminalReturnsError(t *testing.T) { +func TestForwardAsAnthropic_MissingTerminalBeforeOutputReturnsFailoverAndOps(t *testing.T) { gin.SetMode(gin.TestMode) rec := httptest.NewRecorder() @@ -1675,7 +1675,62 @@ func TestForwardAsAnthropic_DoneSentinelWithoutTerminalReturnsError(t *testing.T upstreamBody := "data: [DONE]\n\n" upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_missing_terminal"}}, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{"rid_missing_terminal"}}, + 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.ForwardAsAnthropic(context.Background(), c, account, body, "", "gpt-5.1") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.True(t, errors.As(err, &failoverErr), "missing terminal before output must use failover path") + require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) + require.Contains(t, string(failoverErr.ResponseBody), "OpenAI messages stream ended before a terminal event") + require.NotNil(t, result) + require.Zero(t, result.Usage.InputTokens) + require.Zero(t, result.Usage.OutputTokens) + require.False(t, c.Writer.Written(), "no client body/header should be committed before safe failover") + require.Empty(t, rec.Body.String()) + + events := openAICompatOpsEvents(t, c) + require.Len(t, events, 1) + require.Equal(t, "failover", events[0].Kind) + require.Equal(t, http.StatusBadGateway, events[0].UpstreamStatusCode) + require.Equal(t, int64(1), events[0].AccountID) + require.Equal(t, "rid_missing_terminal", events[0].UpstreamRequestID) + require.Contains(t, events[0].Message, "terminal event") +} + +func TestForwardAsAnthropic_MissingTerminalAfterOutputRecordsOpsWithoutFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + body := []byte(`{"model":"gpt-5.4","max_tokens":16,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.4","status":"in_progress","output":[]}}`, + "", + `data: {"type":"response.output_text.delta","delta":"partial"}`, + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{"rid_partial_missing_terminal"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} @@ -1695,9 +1750,123 @@ func TestForwardAsAnthropic_DoneSentinelWithoutTerminalReturnsError(t *testing.T result, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "gpt-5.1") require.Error(t, err) require.Contains(t, err.Error(), "missing terminal event") + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr), "partial output must not be replayed through failover") require.NotNil(t, result) - require.Zero(t, result.Usage.InputTokens) - require.Zero(t, result.Usage.OutputTokens) + require.False(t, result.ClientDisconnect) + require.True(t, c.Writer.Written()) + require.Contains(t, rec.Body.String(), "event: message_start") + require.Contains(t, rec.Body.String(), "partial") + + events := openAICompatOpsEvents(t, c) + require.Len(t, events, 1) + require.Equal(t, "stream_missing_terminal", events[0].Kind) + require.Equal(t, http.StatusBadGateway, events[0].UpstreamStatusCode) + require.Equal(t, int64(1), events[0].AccountID) + require.Equal(t, "rid_partial_missing_terminal", events[0].UpstreamRequestID) + require.Contains(t, events[0].Message, "terminal event") +} + +func TestForwardAsAnthropic_MissingTerminalAfterClientDisconnectSkipsOpsAndFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Writer = &openAICompatFailingWriter{ResponseWriter: c.Writer, failAfter: 0} + body := []byte(`{"model":"gpt-5.4","max_tokens":16,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.4","status":"in_progress","output":[]}}`, + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{"rid_client_disconnect_missing_terminal"}}, + 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.ForwardAsAnthropic(context.Background(), c, account, body, "", "gpt-5.1") + require.Error(t, err) + require.Contains(t, err.Error(), "missing terminal event") + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr)) + require.NotNil(t, result) + require.True(t, result.ClientDisconnect) + require.Empty(t, rec.Body.String()) + _, ok := c.Get(OpsUpstreamErrorsKey) + require.False(t, ok, "client disconnect must not be attributed as an upstream error") +} + +func TestForwardAsAnthropic_CompleteStreamDoesNotRecordMissingTerminalOps(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + body := []byte(`{"model":"gpt-5.4","max_tokens":16,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.4","status":"in_progress","output":[]}}`, + "", + `data: {"type":"response.output_text.delta","delta":"ok"}`, + "", + `data: {"type":"response.completed","response":{"id":"resp_1","object":"response","model":"gpt-5.4","status":"completed","output":[{"type":"message","id":"msg_1","role":"assistant","status":"completed","content":[{"type":"output_text","text":"ok"}]}],"usage":{"input_tokens":9,"output_tokens":4,"total_tokens":13}}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{"rid_complete_terminal"}}, + 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.ForwardAsAnthropic(context.Background(), c, account, body, "", "gpt-5.1") + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, 9, result.Usage.InputTokens) + require.Equal(t, 4, result.Usage.OutputTokens) + require.Contains(t, rec.Body.String(), "event: message_stop") + _, ok := c.Get(OpsUpstreamErrorsKey) + require.False(t, ok) +} + +func openAICompatOpsEvents(t *testing.T, c *gin.Context) []*OpsUpstreamErrorEvent { + t.Helper() + v, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := v.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + return events } func TestForwardAsAnthropic_UpstreamRequestIgnoresClientCancel(t *testing.T) { diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 4398bd27cb..99653fec45 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -360,7 +360,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( var result *OpenAIForwardResult var handleErr error if clientStream { - result, handleErr = s.handleAnthropicStreamingResponse(resp, c, originalModel, billingModel, upstreamModel, startTime) + result, handleErr = s.handleAnthropicStreamingResponse(resp, c, account, originalModel, billingModel, upstreamModel, startTime) } else { // Client wants JSON: buffer the streaming response and assemble a JSON reply. result, handleErr = s.handleAnthropicBufferedStreamingResponse(resp, c, originalModel, billingModel, upstreamModel, startTime) @@ -476,6 +476,27 @@ func isOpenAICompatResponsesTerminalEvent(eventType string) bool { } } +func (s *OpenAIGatewayService) recordOpenAIMessagesStreamUpstreamError(c *gin.Context, account *Account, upstreamRequestID, kind, message string) { + if c == nil { + return + } + message = sanitizeUpstreamErrorMessage(message) + setOpsUpstreamError(c, http.StatusBadGateway, message, "") + event := OpsUpstreamErrorEvent{ + Platform: PlatformOpenAI, + UpstreamStatusCode: http.StatusBadGateway, + UpstreamRequestID: strings.TrimSpace(upstreamRequestID), + Kind: kind, + Message: message, + } + if account != nil { + event.Platform = account.Platform + event.AccountID = account.ID + event.AccountName = account.Name + } + appendOpsUpstreamError(c, event) +} + func isOpenAICompatDoneSentinelLine(line string) bool { payload, ok := extractOpenAISSEDataLine(line) return ok && strings.TrimSpace(payload) == "[DONE]" @@ -649,6 +670,7 @@ func (s *OpenAIGatewayService) readOpenAICompatBufferedTerminal( func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( resp *http.Response, c *gin.Context, + account *Account, originalModel string, billingModel string, upstreamModel string, @@ -656,14 +678,21 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( ) (*OpenAIForwardResult, error) { requestID := resp.Header.Get("x-request-id") - if s.responseHeaderFilter != nil { - responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) + headersWritten := false + writeStreamHeaders := func() { + if headersWritten { + return + } + headersWritten = true + if s.responseHeaderFilter != nil { + responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) + } + c.Writer.Header().Set("Content-Type", "text/event-stream") + c.Writer.Header().Set("Cache-Control", "no-cache") + c.Writer.Header().Set("Connection", "keep-alive") + c.Writer.Header().Set("X-Accel-Buffering", "no") + c.Writer.WriteHeader(http.StatusOK) } - c.Writer.Header().Set("Content-Type", "text/event-stream") - c.Writer.Header().Set("Cache-Control", "no-cache") - c.Writer.Header().Set("Connection", "keep-alive") - c.Writer.Header().Set("X-Accel-Buffering", "no") - c.Writer.WriteHeader(http.StatusOK) state := apicompat.NewResponsesEventToAnthropicState() state.Model = originalModel @@ -672,6 +701,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( var firstTokenMs *int firstChunk := true clientDisconnected := false + clientOutputStarted := false scanner := bufio.NewScanner(resp.Body) maxLineSize := defaultMaxLineSize @@ -697,15 +727,16 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( // resultWithUsage builds the final result snapshot. resultWithUsage := func() *OpenAIForwardResult { return &OpenAIForwardResult{ - RequestID: requestID, - ResponseID: responseID, - Usage: usage, - Model: originalModel, - BillingModel: billingModel, - UpstreamModel: upstreamModel, - Stream: true, - Duration: time.Since(startTime), - FirstTokenMs: firstTokenMs, + RequestID: requestID, + ResponseID: responseID, + Usage: usage, + Model: originalModel, + BillingModel: billingModel, + UpstreamModel: upstreamModel, + Stream: true, + Duration: time.Since(startTime), + FirstTokenMs: firstTokenMs, + ClientDisconnect: clientDisconnected, } } @@ -753,6 +784,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( ) continue } + writeStreamHeaders() if _, err := fmt.Fprint(c.Writer, sse); err != nil { clientDisconnected = true logger.L().Info("openai messages stream: client disconnected, continuing to drain upstream for billing", @@ -760,6 +792,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( ) break } + clientOutputStarted = true } } if len(events) > 0 && !clientDisconnected { @@ -776,6 +809,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( if err != nil { continue } + writeStreamHeaders() if _, err := fmt.Fprint(c.Writer, sse); err != nil { clientDisconnected = true logger.L().Info("openai messages stream: client disconnected during final flush", @@ -783,6 +817,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( ) break } + clientOutputStarted = true } if !clientDisconnected { c.Writer.Flush() @@ -801,7 +836,16 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( } } missingTerminalErr := func() (*OpenAIForwardResult, error) { - return resultWithUsage(), fmt.Errorf("stream usage incomplete: missing terminal event") + result := resultWithUsage() + if clientDisconnected { + return result, fmt.Errorf("stream usage incomplete: missing terminal event") + } + message := "OpenAI messages stream ended before a terminal event" + if !clientOutputStarted { + return result, s.newOpenAIStreamFailoverError(c, account, false, requestID, nil, message) + } + s.recordOpenAIMessagesStreamUpstreamError(c, account, requestID, "stream_missing_terminal", message) + return result, fmt.Errorf("stream usage incomplete: missing terminal event") } processFrame := func(frame openAICompatSSEFrame) bool { payload := openAICompatPayloadWithEventType(frame.Data, frame.EventType) @@ -943,6 +987,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( continue } // Send Anthropic-format ping event + writeStreamHeaders() if _, err := fmt.Fprint(c.Writer, "event: ping\ndata: {\"type\":\"ping\"}\n\n"); err != nil { // Client disconnected logger.L().Info("openai messages stream: client disconnected during keepalive", @@ -951,6 +996,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( clientDisconnected = true continue } + clientOutputStarted = true c.Writer.Flush() } } diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index cfe9275749..01db7080a4 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -240,6 +240,7 @@ type OpenAIForwardResult struct { ResponseHeaders http.Header Duration time.Duration FirstTokenMs *int + ClientDisconnect bool ImageCount int ImageSize string ImageInputSize string