diff --git a/backend/internal/service/openai_gateway_chat_completions_raw.go b/backend/internal/service/openai_gateway_chat_completions_raw.go index 7427c6fc9c..00bbda0e51 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw.go +++ b/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。 diff --git a/backend/internal/service/openai_gateway_chat_completions_raw_test.go b/backend/internal/service/openai_gateway_chat_completions_raw_test.go index 8a83bbc7ea..2f0b8dcc08 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw_test.go +++ b/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) diff --git a/backend/internal/service/openai_raw_stream_truncation.go b/backend/internal/service/openai_raw_stream_truncation.go new file mode 100644 index 0000000000..678d5125b7 --- /dev/null +++ b/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 +} diff --git a/backend/internal/service/openai_stream_read_error.go b/backend/internal/service/openai_stream_read_error.go index 9a008d8e05..9fda5708ee 100644 --- a/backend/internal/service/openai_stream_read_error.go +++ b/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