From 900194fab2a5485fe1aa19d350e448adce347b7c Mon Sep 17 00:00:00 2001 From: Fool0ntheHill Date: Tue, 11 Aug 2026 16:31:09 +0800 Subject: [PATCH] =?UTF-8?q?fix(openai):=20=E4=BF=AE=E6=AD=A3=20Responses?= =?UTF-8?q?=20=E5=8F=AF=E8=A7=81=E8=BE=93=E5=87=BA=20TTFT?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../openai_first_output_timeout_test.go | 2 +- .../service/openai_gateway_passthrough.go | 59 +++++++- .../openai_gateway_response_handling.go | 37 +++-- .../service/openai_visible_ttft_test.go | 128 ++++++++++++++++++ 4 files changed, 210 insertions(+), 16 deletions(-) create mode 100644 backend/internal/service/openai_visible_ttft_test.go diff --git a/backend/internal/service/openai_first_output_timeout_test.go b/backend/internal/service/openai_first_output_timeout_test.go index 1cb7640a8d..799f7af12b 100644 --- a/backend/internal/service/openai_first_output_timeout_test.go +++ b/backend/internal/service/openai_first_output_timeout_test.go @@ -425,7 +425,7 @@ func TestOpenAINativeFirstOutputEOFDispatchesTerminalEventWithoutBlankLine(t *te require.NoError(t, err) require.NotNil(t, result) - require.NotNil(t, result.firstTokenMs) + require.Nil(t, result.firstTokenMs, "usage-only terminal event is not visible output") require.Equal(t, "resp_eof", result.responseID) require.Equal(t, 3, result.usage.InputTokens) require.Equal(t, 2, result.usage.OutputTokens) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 8f9c296db8..8c029f0aa8 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -811,6 +811,63 @@ func openAIStreamDataStartsClientOutput(data, eventType string) bool { return !openAIStreamEventIsPreamble(eventType) } +func openAIStreamItemHasVisibleOutput(item gjson.Result) bool { + if item.Get("arguments").String() != "" || item.Get("input").String() != "" || item.Get("result").String() != "" { + return true + } + for _, path := range []string{"content", "summary"} { + for _, part := range item.Get(path).Array() { + if part.Get("text").String() != "" || part.Get("transcript").String() != "" { + return true + } + } + } + return false +} + +// Structural progress can commit an attempt and disarm first-output failover, +// but TTFT should start only when the stream carries content a client can use. +func openAIStreamDataStartsVisibleOutput(data, eventType string) bool { + trimmed := strings.TrimSpace(data) + if trimmed == "" || trimmed == "[DONE]" || !gjson.Valid(trimmed) { + return false + } + eventType = strings.TrimSpace(eventType) + if eventType == "" { + eventType = strings.TrimSpace(gjson.Get(trimmed, "type").String()) + } + if strings.HasSuffix(eventType, ".delta") { + delta := gjson.Get(trimmed, "delta") + return delta.Exists() && delta.String() != "" + } + switch eventType { + case "response.output_text.done", + "response.reasoning_summary_text.done", + "response.reasoning_text.done", + "response.audio_transcript.done": + return gjson.Get(trimmed, "text").String() != "" + case "response.function_call_arguments.done": + return gjson.Get(trimmed, "arguments").String() != "" + case "response.custom_tool_call_input.done": + return gjson.Get(trimmed, "input").String() != "" + case "response.image_generation_call.partial_image": + return gjson.Get(trimmed, "partial_image_b64").String() != "" + case "response.content_part.added", "response.content_part.done", + "response.reasoning_summary_part.added", "response.reasoning_summary_part.done": + part := gjson.Get(trimmed, "part") + return part.Get("text").String() != "" || part.Get("transcript").String() != "" + case "response.output_item.added", "response.output_item.done": + return openAIStreamItemHasVisibleOutput(gjson.Get(trimmed, "item")) + case "response.completed", "response.done": + for _, item := range gjson.Get(trimmed, "response.output").Array() { + if openAIStreamItemHasVisibleOutput(item) { + return true + } + } + } + return false +} + // openAIStreamFailedEventErrorCode 提取流内 failed 事件的错误码（小写）， // 兼容 response.failed 的嵌套形态与裸 error 形态。 func openAIStreamFailedEventErrorCode(payload []byte) string { @@ -1321,7 +1378,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( openAIResponsesCompletedEventIsEmpty(dataBytes, usage) { return resultWithUsage(), newOpenAIResponsesEmptyCompletedFailoverError(c, account, upstreamRequestID) } - if firstTokenMs == nil && lineStartsClientOutput && trimmedData != "[DONE]" { + if firstTokenMs == nil && openAIStreamDataStartsVisibleOutput(trimmedData, eventType) { ms := int(time.Since(startTime).Milliseconds()) firstTokenMs = &ms } diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index d8a17b19f4..c2b84d5ad0 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -103,6 +103,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. maxLineSize = s.cfg.Gateway.MaxLineSize } var firstTokenMs *int + firstOutputProgressObserved := false bufferedWriter := bufio.NewWriterSize(w, 4*1024) var firstOutputStage *openAIFirstOutputStage if guardFirstOutput { @@ -114,19 +115,19 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. }() } writePendingString := func(value string) (int, error) { - if firstOutputStage != nil && firstTokenMs == nil && !firstOutputStage.closed { + if firstOutputStage != nil && !firstOutputProgressObserved && !firstOutputStage.closed { return firstOutputStage.WriteString(value) } return bufferedWriter.WriteString(value) } pendingBytes := func() int64 { - if firstOutputStage != nil && firstTokenMs == nil && !firstOutputStage.closed { + if firstOutputStage != nil && !firstOutputProgressObserved && !firstOutputStage.closed { return firstOutputStage.Buffered() } return int64(bufferedWriter.Buffered()) } flushBuffered := func() error { - if firstOutputStage != nil && firstTokenMs == nil && !firstOutputStage.closed { + if firstOutputStage != nil && !firstOutputProgressObserved && !firstOutputStage.closed { if err := firstOutputStage.CommitTo(w); err != nil { return err } @@ -235,9 +236,10 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. var streamEarlyErr error eventInProgress := false eventStartsClientOutput := false + eventStartsVisibleOutput := false eventShouldFlush := false handlePendingWriteError := func(err error) { - if firstOutputStage != nil && firstTokenMs == nil && !firstOutputStage.closed { + if firstOutputStage != nil && !firstOutputProgressObserved && !firstOutputStage.closed { message := "OpenAI first-output staging failed" if errors.Is(err, errOpenAIFirstOutputStageLimit) { message = "OpenAI first-output staging limit exceeded" @@ -253,11 +255,12 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. logger.LegacyPrintf("service.openai_gateway", "Client disconnected during streaming, continuing to drain upstream for billing") } completeGuardedEvent := func(queueDrained bool) { - completedSemanticEvent := eventStartsClientOutput + completedProgressEvent := eventStartsClientOutput + completedVisibleEvent := eventStartsVisibleOutput shouldFlush := eventShouldFlush || (queueDrained && clientOutputStarted) eventInProgress = false if !clientDisconnected { - if completedSemanticEvent { + if completedProgressEvent { applyAttemptResponseHeaders() } if shouldFlush { @@ -270,13 +273,17 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } } } - if completedSemanticEvent && firstTokenMs == nil { + if completedProgressEvent && !firstOutputProgressObserved { firstOutputScanGuard.Store(false) - ms := int(time.Since(startTime).Milliseconds()) - firstTokenMs = &ms + firstOutputProgressObserved = true stopFirstOutputTimer() } + if completedVisibleEvent && firstTokenMs == nil { + ms := int(time.Since(startTime).Milliseconds()) + firstTokenMs = &ms + } eventStartsClientOutput = false + eventStartsVisibleOutput = false eventShouldFlush = false } sendErrorEvent := func(reason string) { @@ -365,7 +372,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if scanErr == nil { return nil, nil, false } - if errors.Is(scanErr, errOpenAIFirstOutputScannerLimit) && firstTokenMs == nil { + if errors.Is(scanErr, errOpenAIFirstOutputScannerLimit) && !firstOutputProgressObserved { logger.LegacyPrintf("service.openai_gateway", "SSE token exceeded guarded first-output limit: account=%d limit=%d error=%v", account.ID, openAIFirstOutputStageMaxBytes+openAIFirstOutputScannerFramingAllowance, scanErr) failoverErr := s.newOpenAIStreamFailoverError( c, account, false, upstreamRequestID, nil, @@ -374,7 +381,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. failoverErr.SafeToFailoverAfterWrite = true return resultWithUsage(), failoverErr, true } - if errors.Is(scanErr, bufio.ErrTooLong) && guardFirstOutput && firstTokenMs == nil { + if errors.Is(scanErr, bufio.ErrTooLong) && guardFirstOutput && !firstOutputProgressObserved { logger.LegacyPrintf("service.openai_gateway", "SSE line too long before first output: account=%d max_size=%d error=%v", account.ID, maxLineSize, scanErr) failoverErr := s.newOpenAIStreamFailoverError( c, account, false, upstreamRequestID, nil, @@ -540,8 +547,10 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. line = s.replaceModelInSSELine(line, mappedModel, originalModel) } startsClientOutput := forceFlushFailedEvent || openAIStreamDataStartsClientOutput(data, eventType) + startsVisibleOutput := openAIStreamDataStartsVisibleOutput(data, eventType) if guardFirstOutput { eventStartsClientOutput = eventStartsClientOutput || startsClientOutput + eventStartsVisibleOutput = eventStartsVisibleOutput || startsVisibleOutput } if startsClientOutput && !openAIStreamEventTypeIsTerminal(eventType) { responsesSemanticOutputSeen = true @@ -562,7 +571,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. // 写入客户端（客户端断开后继续 drain 上游） if !clientDisconnected { shouldFlush := queueDrained && (clientOutputStarted || startsClientOutput) - if firstTokenMs == nil && startsClientOutput { + if firstTokenMs == nil && startsVisibleOutput { // 保证首个 token 事件尽快出站，避免影响 TTFT。 shouldFlush = true } @@ -577,7 +586,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } // Record first token time - if !guardFirstOutput && firstTokenMs == nil && startsClientOutput { + if !guardFirstOutput && firstTokenMs == nil && startsVisibleOutput { ms := int(time.Since(startTime).Milliseconds()) firstTokenMs = &ms stopFirstOutputTimer() @@ -740,7 +749,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. return resultWithUsage(), fmt.Errorf("stream data interval timeout") case <-firstOutputCh: - if firstTokenMs != nil { + if firstOutputProgressObserved { stopFirstOutputTimer() continue } diff --git a/backend/internal/service/openai_visible_ttft_test.go b/backend/internal/service/openai_visible_ttft_test.go new file mode 100644 index 0000000000..8fd243c39e --- /dev/null +++ b/backend/internal/service/openai_visible_ttft_test.go @@ -0,0 +1,128 @@ +package service + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestOpenAIVisibleOutputClassification(t *testing.T) { + tests := []struct { + name string + data string + eventType string + want bool + }{ + {name: "keepalive", data: `{"type":"keepalive"}`, want: false}, + {name: "created", data: `{"type":"response.created"}`, want: false}, + {name: "empty output item", data: `{"type":"response.output_item.added","item":{"id":"item_test","type":"reasoning","summary":[]}}`, want: false}, + {name: "empty delta", data: `{"type":"response.output_text.delta","delta":""}`, want: false}, + {name: "text delta", data: `{"type":"response.output_text.delta","delta":"test output"}`, want: true}, + {name: "tool arguments", data: `{"type":"response.function_call_arguments.delta","delta":"{}"}`, want: true}, + {name: "partial image", data: `{"type":"response.image_generation_call.partial_image","partial_image_b64":"dGVzdA=="}`, want: true}, + {name: "completed image item", data: `{"type":"response.output_item.done","item":{"id":"item_test","type":"image_generation_call","result":"dGVzdA=="}}`, want: true}, + {name: "empty completed", data: `{"type":"response.completed","response":{"id":"resp_test","output":[]}}`, want: false}, + {name: "completed with output usage only", data: `{"type":"response.completed","response":{"id":"resp_test","usage":{"input_tokens":1,"output_tokens":2}}}`, want: false}, + {name: "completed with text", data: `{"type":"response.completed","response":{"id":"resp_test","output":[{"type":"message","content":[{"type":"output_text","text":"test output"}]}]}}`, want: true}, + {name: "done marker", data: `[DONE]`, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, openAIStreamDataStartsVisibleOutput(tt.data, tt.eventType)) + }) + } +} + +func TestOpenAIResponsesTTFTStartsAtVisibleOutput(t *testing.T) { + for _, passthrough := range []bool{false, true} { + name := "native" + if passthrough { + name = "passthrough" + } + t.Run(name, func(t *testing.T) { + result := runSyntheticVisibleTTFTStream(t, passthrough, 120*time.Millisecond, 0, + `{"type":"response.output_text.delta","delta":"test output"}`) + require.NotNil(t, result.firstTokenMs) + require.GreaterOrEqual(t, *result.firstTokenMs, 100) + }) + } +} + +func TestOpenAIResponsesTTFTStartsAtCompletedImage(t *testing.T) { + for _, passthrough := range []bool{false, true} { + name := "native" + if passthrough { + name = "passthrough" + } + t.Run(name, func(t *testing.T) { + result := runSyntheticVisibleTTFTStream(t, passthrough, 120*time.Millisecond, 0, + `{"type":"response.output_item.done","item":{"id":"item_test","type":"image_generation_call","result":"dGVzdA=="}}`) + require.NotNil(t, result.firstTokenMs) + require.GreaterOrEqual(t, *result.firstTokenMs, 100) + }) + } +} + +func TestOpenAINativeProgressDisarmsTimeoutWithoutStartingTTFT(t *testing.T) { + result := runSyntheticVisibleTTFTStream(t, false, 1200*time.Millisecond, 1, + `{"type":"response.output_text.delta","delta":"test output"}`) + require.NotNil(t, result.firstTokenMs) + require.GreaterOrEqual(t, *result.firstTokenMs, 1100) +} + +func runSyntheticVisibleTTFTStream(t *testing.T, passthrough bool, visibleDelay time.Duration, timeoutSeconds int, visibleEvent string) *openaiStreamingResult { + t.Helper() + gin.SetMode(gin.TestMode) + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{ + MaxLineSize: defaultMaxLineSize, + OpenAIFirstOutputTimeoutSeconds: timeoutSeconds, + }}} + reader, writer := io.Pipe() + writerDone := make(chan struct{}) + go func() { + defer close(writerDone) + defer func() { _ = writer.Close() }() + _, _ = io.WriteString(writer, "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_test\"}}\n\n") + _, _ = io.WriteString(writer, "data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"item_test\",\"type\":\"reasoning\",\"summary\":[]}}\n\n") + time.Sleep(visibleDelay) + _, _ = io.WriteString(writer, "data: "+visibleEvent+"\n\n") + _, _ = io.WriteString(writer, "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_test\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n") + }() + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: reader} + account := &Account{ID: 1, Name: "account_test", Platform: PlatformOpenAI} + started := time.Now() + + var result *openaiStreamingResult + var err error + if passthrough { + var passthroughResult *openaiStreamingResultPassthrough + passthroughResult, err = svc.handleStreamingResponsePassthrough(context.Background(), resp, c, account, started, "test-model", "test-model") + if passthroughResult != nil { + result = &openaiStreamingResult{firstTokenMs: passthroughResult.firstTokenMs} + } + } else { + result, err = svc.handleStreamingResponse(context.Background(), resp, c, account, started, "test-model", "test-model") + } + require.NoError(t, err) + require.NotNil(t, result) + require.Contains(t, recorder.Body.String(), `"type":"response.output_item.added"`) + require.Contains(t, recorder.Body.String(), visibleEvent) + select { + case <-writerDone: + case <-time.After(time.Second): + t.Fatal("synthetic upstream writer did not exit") + } + return result +}