Merge pull request #6299 from william-drakemond/fix/openai-raw-stream-truncation

fix(openai): fail raw CC streams truncated before any terminal chunk

backend/internal/service/openai_gateway_chat_completions_raw.go

@@ -285,6 +285,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
 	clientOutputStarted := false
 	pendingLines := make([]string, 0, 8)
 	refusalDetector := newOpenAIChatSilentRefusalDetector(requestBodyLen)
+	var terminal openAIRawStreamTerminalState
 
 	writeLine := func(line string) {
 		if clientDisconnected {
@@ -323,6 +324,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
 		refusalDetector.ObserveSSELine(line)
 		if payload, ok := extractOpenAISSEDataLine(line); ok {
 			trimmedPayload := strings.TrimSpace(payload)
+			terminal.ObserveDataLine(trimmedPayload)
 			if trimmedPayload != "[DONE]" {
 				observer.ObserveOpenAI([]byte(payload), strings.TrimSpace(gjson.Get(payload, "type").String()))
 				usageOnlyChunk := isOpenAIChatUsageOnlyStreamChunk(payload)
@@ -350,14 +352,65 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
 		}
 	}
 
-	if err := scanner.Err(); err != nil {
-		if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
-			logger.L().Warn("openai chat_completions raw: stream read error",
-				zap.Error(err),
-				zap.String("request_id", requestID),
-			)
+	resultWithUsage := func() *OpenAIForwardResult {
+		return &OpenAIForwardResult{
+			RequestID:                     requestID,
+			Usage:                         usage,
+			Model:                         originalModel,
+			BillingModel:                  billingModel,
+			UpstreamModel:                 upstreamModel,
+			UpstreamResponseModel:         observedUpstreamResponseModel(c),
+			UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c),
+			UpstreamResponseServiceTier:   observedUpstreamResponseServiceTier(c),
+			ReasoningEffort:               reasoningEffort,
+			ServiceTier:                   resolvedOpenAIUpstreamServiceTier(c, serviceTier),
+			Stream:                        true,
+			Duration:                      time.Since(startTime),
+			FirstTokenMs:                  firstTokenMs,
 		}
-	} else if !clientDisconnected && !clientOutputStarted {
+	}
+
+	scanErr := scanner.Err()
+	if scanErr != nil && !errors.Is(scanErr, context.Canceled) && !errors.Is(scanErr, context.DeadlineExceeded) {
+		logger.L().Warn("openai chat_completions raw: stream read error",
+			zap.Error(scanErr),
+			zap.String("request_id", requestID),
+		)
+	}
+
+	// 客户端取消/断开后上游读失败与上游截断不可区分（取消会连带取消上游请求），
+	// 沿用既有语义：按已收到的用量正常收尾计费，不判为上游故障。
+	clientAborted := clientDisconnected ||
+		errors.Is(scanErr, context.Canceled) ||
+		errors.Is(scanErr, context.DeadlineExceeded)
+
+	// 上游在任何终止信号之前结束：连接被 reset（scanErr != nil）或干净 EOF。
+	// 两者都不能再记成功——此前统一返回 nil error，把上游截断伪装成
+	// `HTTP 200 + usage 0/0`，客户端收到半截回答且 Ops 侧完全无感。
+	if !clientAborted && terminal.IsTruncated(clientOutputStarted) {
+		cause := scanErr
+		if cause == nil {
+			cause = ErrOpenAIUpstreamStreamTruncated
+		}
+		logger.L().Warn("openai chat_completions raw: upstream stream truncated before terminal chunk",
+			zap.Error(cause),
+			zap.String("request_id", requestID),
+			zap.Int64("account_id", account.ID),
+			zap.String("upstream_model", upstreamModel),
+			zap.Bool("saw_sse_data", terminal.sawDataLine),
+			zap.Bool("client_output_started", clientOutputStarted),
+		)
+		if !clientOutputStarted {
+			// 响应头尚未提交：可以透明换号重试，客户端不会看到半截流。
+			return nil, newOpenAIRawStreamTruncatedFailoverError(c, account, requestID, cause)
+		}
+		// 已写出语义字节：无法再 failover，改为带类型的上游错误。handler 会据此
+		// 补发 SSE error 帧并把本次请求计入 SLA 失败。
+		recordOpenAIRawStreamTruncation(c, account, requestID, cause, "http_error")
+		return resultWithUsage(), newOpenAIUpstreamStreamReadError(cause)
+	}
+
+	if scanErr == nil && !clientDisconnected && !clientOutputStarted {
 		if refusalDetector.IsSilentRefusal() {
 			return nil, newOpenAISilentRefusalFailoverError(c, account, requestID)
 		}
@@ -380,21 +433,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
 		}
 	}
 
-	return &OpenAIForwardResult{
-		RequestID:                     requestID,
-		Usage:                         usage,
-		Model:                         originalModel,
-		BillingModel:                  billingModel,
-		UpstreamModel:                 upstreamModel,
-		UpstreamResponseModel:         observedUpstreamResponseModel(c),
-		UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c),
-		UpstreamResponseServiceTier:   observedUpstreamResponseServiceTier(c),
-		ReasoningEffort:               reasoningEffort,
-		ServiceTier:                   resolvedOpenAIUpstreamServiceTier(c, serviceTier),
-		Stream:                        true,
-		Duration:                      time.Since(startTime),
-		FirstTokenMs:                  firstTokenMs,
-	}, nil
+	return resultWithUsage(), nil
 }
 
 // ensureOpenAIChatStreamUsage 确保 raw Chat Completions 流式请求会让上游返回 usage。

backend/internal/service/openai_gateway_chat_completions_raw_test.go

@@ -687,6 +687,326 @@ func TestForwardAsRawChatCompletions_StripsEmptyToolCallIdentity(t *testing.T) {
 	require.True(t, followUpSeen)
 }
 
+// 上游在生成中途干净 EOF（无 [DONE]/usage/finish_reason）且已向客户端写出内容：
+// 不能再记成 HTTP 200 成功，必须回带类型化的上游截断错误，由 handler 补 SSE error
+// 帧并计入 SLA 失败。
+func TestForwardAsRawChatCompletions_TruncatedStreamAfterOutputFailsRequest(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstreamBody := strings.Join([]string{
+		`data: {"id":"chatcmpl_cut","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"half an ans"},"finish_reason":null}]}`,
+		"",
+	}, "\n")
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_truncated"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamBody)),
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.Error(t, err)
+	require.NotNil(t, result, "已收字节的用量仍需带回，供 ops 记录首 token 时延")
+	var failoverErr *UpstreamFailoverError
+	require.False(t, errors.As(err, &failoverErr), "已写出语义字节后不得再 failover")
+
+	code, message, ok := OpenAIUpstreamStreamReadErrorDetails(err)
+	require.True(t, ok)
+	require.Equal(t, OpenAIUpstreamStreamTruncatedCode, code)
+	require.NotEmpty(t, message)
+	// 已写出的内容保持原样透传，客户端拿到的仍是它已经收到的那部分。
+	require.Contains(t, rec.Body.String(), `"content":"half an ans"`)
+	require.NotContains(t, rec.Body.String(), "data: [DONE]")
+}
+
+// 上游 200 但一个 SSE 字节都没发：响应头尚未提交，应换号重试而不是回 200 空流。
+func TestForwardAsRawChatCompletions_EmptyStreamBeforeOutputTriggersFailover(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_empty"}},
+		Body:       io.NopCloser(strings.NewReader("")),
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.Nil(t, result)
+	var failoverErr *UpstreamFailoverError
+	require.True(t, errors.As(err, &failoverErr))
+	require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode)
+	require.Equal(t, OpenAIUpstreamStreamTruncatedCode,
+		gjson.GetBytes(failoverErr.ResponseBody, "error.code").String())
+	require.True(t, failoverErr.ShouldRetryNextAccount())
+	require.False(t, c.Writer.Written(), "换号重试前不得提交 200 响应头")
+	require.Empty(t, rec.Body.String())
+}
+
+// 传输层错误（Cloudflare edge reset 等）在写出后同样不能记成功，且分类要区别于
+// 干净 EOF，便于 ops 分辨 reset 与静默截断。
+func TestForwardAsRawChatCompletions_StreamReadErrorAfterOutputFailsRequest(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_reset"}},
+		Body: &openAIChatStreamReadErrorCloser{
+			payload: []byte(`data: {"id":"chatcmpl_reset","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"partial"}}]}` + "\n\n"),
+			err:     errors.New("read tcp 172.18.0.4->172.65.90.23:443: read: connection reset by peer"),
+		},
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.Error(t, err)
+	require.NotNil(t, result)
+
+	code, _, ok := OpenAIUpstreamStreamReadErrorDetails(err)
+	require.True(t, ok)
+	require.Equal(t, OpenAIUpstreamStreamReadErrorCode, code)
+	require.Contains(t, rec.Body.String(), `"content":"partial"`)
+}
+
+// 边界：缺 [DONE] 但收到了 usage 帧 —— 生成已完整，只是尾巴丢失。必须继续按成功
+// 计费，否则会误伤那些跑完就直接 EOF 的兼容上游并白送 token。
+func TestForwardAsRawChatCompletions_MissingDoneWithUsageStillSucceeds(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstreamBody := strings.Join([]string{
+		`data: {"id":"chatcmpl_nodone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}`,
+		"",
+		`data: {"id":"chatcmpl_nodone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[],"usage":{"prompt_tokens":11,"completion_tokens":6,"total_tokens":17}}`,
+		"",
+	}, "\n")
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_nodone_usage"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamBody)),
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, 11, result.Usage.InputTokens)
+	require.Equal(t, 6, result.Usage.OutputTokens)
+}
+
+// 边界：缺 [DONE] 与 usage，但末帧带 finish_reason —— 生成正常结束，同样不判截断。
+func TestForwardAsRawChatCompletions_MissingDoneWithFinishReasonStillSucceeds(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstreamBody := strings.Join([]string{
+		`data: {"id":"chatcmpl_finish","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"done"},"finish_reason":null}]}`,
+		"",
+		`data: {"id":"chatcmpl_finish","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`,
+		"",
+	}, "\n")
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_nodone_finish"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamBody)),
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Contains(t, rec.Body.String(), `"finish_reason":"stop"`)
+}
+
+// openAIRawStreamDisconnectedWriter 模拟客户端已断开：raw 直转路径经
+// WriteString 写出，故两个方法都必须失败（只覆盖 Write 会被内嵌 writer 绕过）。
+type openAIRawStreamDisconnectedWriter struct {
+	gin.ResponseWriter
+}
+
+func (w *openAIRawStreamDisconnectedWriter) Write([]byte) (int, error) {
+	return 0, errors.New("write failed: client disconnected")
+}
+
+func (w *openAIRawStreamDisconnectedWriter) WriteString(string) (int, error) {
+	return 0, errors.New("write failed: client disconnected")
+}
+
+// 客户端已断开时上游随后截断：两者不可区分，沿用既有语义按已收用量正常收尾计费，
+// 不得把客户端离场记成上游故障。
+func TestForwardAsRawChatCompletions_ClientDisconnectTruncationStillBills(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Writer = &openAIRawStreamDisconnectedWriter{ResponseWriter: c.Writer}
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstreamBody := strings.Join([]string{
+		`data: {"id":"chatcmpl_gone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}`,
+		"",
+	}, "\n")
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_gone"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamBody)),
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+}
+
+// 客户端取消会连带取消上游请求，上游读因此报 context.Canceled：同样不判为上游截断。
+func TestForwardAsRawChatCompletions_ClientCancelTruncationStillBills(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_cancel"}},
+		Body: &openAIChatStreamReadErrorCloser{
+			payload: []byte(`data: {"id":"chatcmpl_cancel","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}` + "\n\n"),
+			err:     context.Canceled,
+		},
+	}}
+
+	svc := &OpenAIGatewayService{
+		cfg:          rawChatCompletionsTestConfig(),
+		httpUpstream: upstream,
+	}
+
+	result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+}
+
+func TestOpenAIRawStreamTerminalState(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name           string
+		payloads       []string
+		clientStarted  bool
+		wantTerminated bool
+		wantTruncated  bool
+	}{
+		{
+			name:           "done sentinel",
+			payloads:       []string{`{"choices":[{"delta":{"content":"a"}}]}`, "[DONE]"},
+			clientStarted:  true,
+			wantTerminated: true,
+		},
+		{
+			name:           "usage chunk",
+			payloads:       []string{`{"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":1}}`},
+			clientStarted:  true,
+			wantTerminated: true,
+		},
+		{
+			name:           "finish reason",
+			payloads:       []string{`{"choices":[{"delta":{},"finish_reason":"length"}]}`},
+			clientStarted:  true,
+			wantTerminated: true,
+		},
+		{
+			name:          "null finish reason is not terminal",
+			payloads:      []string{`{"choices":[{"delta":{"content":"a"},"finish_reason":null}]}`},
+			clientStarted: true,
+			wantTruncated: true,
+		},
+		{
+			name:          "usage null is not terminal",
+			payloads:      []string{`{"choices":[{"delta":{"content":"a"}}],"usage":null}`},
+			clientStarted: true,
+			wantTruncated: true,
+		},
+		{
+			// 上游对 stream 请求回了裸 JSON：无 data: 行，既有行为是原样透传。
+			name:          "non-sse body already forwarded",
+			payloads:      nil,
+			clientStarted: true,
+		},
+		{
+			name:          "no bytes at all",
+			payloads:      nil,
+			clientStarted: false,
+			wantTruncated: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			var state openAIRawStreamTerminalState
+			for _, payload := range tt.payloads {
+				state.ObserveDataLine(payload)
+			}
+			require.Equal(t, tt.wantTerminated, state.Terminated())
+			require.Equal(t, tt.wantTruncated, state.IsTruncated(tt.clientStarted))
+		})
+	}
+}
+
 func TestForwardAsRawChatCompletions_ClientDisconnectDrainsUsage(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 

backend/internal/service/openai_raw_stream_truncation.go

@@ -0,0 +1,161 @@
+package service
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/tidwall/gjson"
+)
+
+// openAIRawStreamTruncatedUpstreamMessage 是 raw CC 直转路径上游截断的 ops 消息。
+const openAIRawStreamTruncatedUpstreamMessage = "Upstream Chat Completions stream ended before any terminal chunk"
+
+// openAIRawStreamTerminalState 记录 raw Chat Completions SSE 流是否收到过
+// **终止信号**。
+//
+// 背景：CC 直转路径把上游 SSE 原样透传，此前只要 HTTP 状态是 200 就按成功收尾——
+// 上游中途断流（Cloudflare edge reset、后端 worker 掉线）会被伪装成
+// `HTTP 200 + usage 0/0`：客户端拿到半截回答，网关既不报错也不计入 SLA，
+// Ops 侧完全不可见。
+//
+// 三种终止信号任一出现即认为上游"讲完了"，只是尾巴可能丢失，不作截断处理：
+//
+//   - [DONE]        —— OpenAI CC 协议标准哨兵
+//   - usage chunk   —— include_usage 生效时的末尾用量帧（网关强制打开）
+//   - finish_reason —— 生成正常结束（stop/length/tool_calls/...）
+//
+// 只认 [DONE] 会误伤那些跑完最后一帧就直接 EOF 的兼容上游；只认 usage 会误伤
+// 不支持 include_usage 的上游。三者取并集，把误判压到"上游确实在生成中途被切断"。
+type openAIRawStreamTerminalState struct {
+	// sawDataLine 表示上游至少发过一行 `data:`，即响应确实是 SSE 语义流。
+	// 非 SSE 响应体（上游对 stream 请求返回裸 JSON）不参与截断判定，保持既有透传行为。
+	sawDataLine     bool
+	sawDone         bool
+	sawUsage        bool
+	sawFinishReason bool
+}
+
+// ObserveDataLine 从单行 SSE `data:` 载荷中提取终止信号。payload 需已 TrimSpace。
+func (t *openAIRawStreamTerminalState) ObserveDataLine(payload string) {
+	if t == nil {
+		return
+	}
+	t.sawDataLine = true
+	if payload == "[DONE]" {
+		t.sawDone = true
+		return
+	}
+	if usage := gjson.Get(payload, "usage"); usage.Exists() && usage.IsObject() {
+		t.sawUsage = true
+	}
+	if t.sawFinishReason {
+		return
+	}
+	for _, choice := range gjson.Get(payload, "choices").Array() {
+		// finish_reason 为 null 时 String() 返回空串，不算终止。
+		if strings.TrimSpace(choice.Get("finish_reason").String()) != "" {
+			t.sawFinishReason = true
+			return
+		}
+	}
+}
+
+// Terminated 表示上游给出过终止信号。
+func (t *openAIRawStreamTerminalState) Terminated() bool {
+	return t != nil && (t.sawDone || t.sawUsage || t.sawFinishReason)
+}
+
+// IsTruncated 判定上游是否在任何终止信号之前结束。clientOutputStarted 用于放行
+// 非 SSE 响应体：那类响应本就没有 data: 行，既有行为是原样透传，不在本次判定范围内；
+// 但"一个字节都没收到"的空 200 依然算截断。
+func (t *openAIRawStreamTerminalState) IsTruncated(clientOutputStarted bool) bool {
+	if t == nil || t.Terminated() {
+		return false
+	}
+	return t.sawDataLine || !clientOutputStarted
+}
+
+// newOpenAIRawStreamTruncatedFailoverError 处理"上游截断且尚未向客户端写出任何
+// 字节"的情况：响应头还没提交，可以透明换号重试，客户端不会看到半截流。
+func newOpenAIRawStreamTruncatedFailoverError(
+	c *gin.Context,
+	account *Account,
+	upstreamRequestID string,
+	cause error,
+) *UpstreamFailoverError {
+	recordOpenAIRawStreamTruncation(c, account, upstreamRequestID, cause, "failover")
+
+	headers := http.Header{}
+	if id := strings.TrimSpace(upstreamRequestID); id != "" {
+		headers.Set("x-request-id", id)
+	}
+	return &UpstreamFailoverError{
+		StatusCode:      http.StatusBadGateway,
+		ResponseBody:    openAIRawStreamTruncatedErrorBody(cause),
+		ResponseHeaders: headers,
+	}
+}
+
+// recordOpenAIRawStreamTruncation 把上游截断记入 ops 上下文，使其在错误日志与
+// 账号健康度中可见——这正是此前"HTTP 200 假成功"丢掉的信息。
+func recordOpenAIRawStreamTruncation(
+	c *gin.Context,
+	account *Account,
+	upstreamRequestID string,
+	cause error,
+	kind string,
+) {
+	if c == nil {
+		return
+	}
+	message := openAIRawStreamTruncatedMessage(cause)
+	platform := PlatformOpenAI
+	accountID := int64(0)
+	accountName := ""
+	if account != nil {
+		platform = account.Platform
+		accountID = account.ID
+		accountName = account.Name
+	}
+
+	setOpsUpstreamError(c, http.StatusBadGateway, message, "")
+	appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
+		Platform:           platform,
+		AccountID:          accountID,
+		AccountName:        accountName,
+		UpstreamStatusCode: http.StatusBadGateway,
+		UpstreamRequestID:  strings.TrimSpace(upstreamRequestID),
+		Kind:               kind,
+		Message:            message,
+	})
+}
+
+// openAIRawStreamTruncatedMessage 拼出 ops 消息：干净 EOF 没有底层错误可带，
+// 传输层错误（connection reset / http2 stream error）则保留原因以便定位。
+func openAIRawStreamTruncatedMessage(cause error) string {
+	if cause == nil || errors.Is(cause, ErrOpenAIUpstreamStreamTruncated) {
+		return openAIRawStreamTruncatedUpstreamMessage
+	}
+	return openAIRawStreamTruncatedUpstreamMessage + ": " + cause.Error()
+}
+
+// openAIRawStreamTruncatedErrorBody 构造 failover 错误体，code/message 与
+// 写出后走 openAIUpstreamStreamReadError 的客户端分类保持一致。
+func openAIRawStreamTruncatedErrorBody(cause error) []byte {
+	code, message := classifyOpenAIUpstreamStreamReadError(cause)
+	body, err := json.Marshal(map[string]any{
+		"error": map[string]any{
+			"type":    "upstream_error",
+			"code":    code,
+			"message": message,
+		},
+	})
+	if err != nil {
+		return []byte(`{"error":{"type":"upstream_error","code":"` + OpenAIUpstreamStreamTruncatedCode +
+			`","message":"Upstream response stream ended before completion"}}`)
+	}
+	return body
+}

backend/internal/service/openai_stream_read_error.go

@@ -12,8 +12,17 @@ const (
 	// when an upstream HTTP/2 response stream is reset after the request started.
 	OpenAIUpstreamHTTP2StreamErrorCode = "upstream_http2_stream_error"
 	OpenAIUpstreamStreamReadErrorCode  = "upstream_stream_read_error"
+	// OpenAIUpstreamStreamTruncatedCode is returned when an upstream SSE stream
+	// closes *cleanly* before delivering any terminal signal. A clean EOF carries
+	// no transport error, so without this classification a truncated generation is
+	// indistinguishable from a successful one.
+	OpenAIUpstreamStreamTruncatedCode = "upstream_stream_truncated"
 )
 
+// ErrOpenAIUpstreamStreamTruncated marks an upstream SSE stream that ended at
+// EOF — without a read error — before any terminal signal arrived.
+var ErrOpenAIUpstreamStreamTruncated = errors.New("upstream stream ended before any terminal chunk")
+
 type openAIUpstreamStreamReadError struct {
 	cause         error
 	clientCode    string
@@ -61,6 +70,9 @@ func OpenAIUpstreamStreamReadErrorDetails(err error) (code, message string, ok b
 
 func classifyOpenAIUpstreamStreamReadError(err error) (code, message string) {
 	if err != nil {
+		if errors.Is(err, ErrOpenAIUpstreamStreamTruncated) {
+			return OpenAIUpstreamStreamTruncatedCode, "Upstream response stream ended before completion"
+		}
 		lower := strings.ToLower(err.Error())
 		// net/http's HTTP/2 stream error is unexported. Its stable text contains
 		// "stream error: stream ID ..."; match only the transport signature and