diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index ecca154781..0705d897b1 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -2186,6 +2186,14 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( responseFailedPending = false failureDelivered = true } + // Terminal 事件(response.completed / [DONE] 等)随空行完整刷出后不再等上游 + // EOF:上游在 keep-alive/HTTP2 复用连接上可能拖延关闭连接(观测到 8~46s 不等), + // 空等期间只能靠 keepalive 维持,白白拉长尾延迟。usage 已在 terminal 事件中解析。 + // Codex bare error 序列(error 后可能跟 response.failed 或翻盘的 completed) + // 必须继续读取,不适用提前结束。 + if (sawDone || sawTerminalEvent) && line == "" && !(codexFailureTerminal && sawBareError) { + break + } } ensureResponseFailedTerminal() if err := documentScanner.Err(); err != nil { diff --git a/backend/internal/service/openai_gateway_response_flush_test.go b/backend/internal/service/openai_gateway_response_flush_test.go index fe6e3ea6b2..56e071041a 100644 --- a/backend/internal/service/openai_gateway_response_flush_test.go +++ b/backend/internal/service/openai_gateway_response_flush_test.go @@ -644,3 +644,47 @@ func waitOpenAIResponseFlushSignal(t *testing.T, signal <-chan struct{}) { t.Fatal("timed out waiting for stream signal") } } + +// hangingOpenAISSEAfterTerminal 模拟上游在发完 terminal 事件后拖延关闭连接 +// (keep-alive/HTTP2 复用连接上观测到 8~46s 不 EOF)。 +type hangingOpenAISSEAfterTerminal struct { + payload []byte + sent bool + release chan struct{} + closeOnce sync.Once +} + +func (r *hangingOpenAISSEAfterTerminal) Read(data []byte) (int, error) { + if !r.sent { + r.sent = true + return copy(data, r.payload), nil + } + <-r.release + return 0, io.EOF +} + +func (r *hangingOpenAISSEAfterTerminal) Close() error { + r.closeOnce.Do(func() { close(r.release) }) + return nil +} + +func TestOpenAIResponseFlush_TerminalEventEndsStreamWithoutEOF(t *testing.T) { + body := "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":7,\"output_tokens\":5}}}\n\n" + reader := &hangingOpenAISSEAfterTerminal{payload: []byte(body), release: make(chan struct{})} + recorder := newOpenAIResponseFlushRecorder() + resultCh, errCh := runOpenAIResponseFlushTestAsync(recorder, reader, config.GatewayConfig{StreamKeepaliveInterval: 1, StreamDataIntervalTimeout: 30}) + t.Cleanup(func() { _ = reader.Close() }) + + select { + case err := <-errCh: + require.NoError(t, err) + result := <-resultCh + require.NotNil(t, result) + require.Equal(t, 7, result.usage.InputTokens) + require.Equal(t, 5, result.usage.OutputTokens) + case <-time.After(3 * time.Second): + t.Fatal("stream did not end after terminal event; still waiting for upstream EOF") + } + gotBody, _ := recorder.snapshot() + require.Equal(t, body, gotBody) +} diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 104f319dc5..021b060c73 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -798,6 +798,12 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if streamEarlyErr != nil { return resultWithUsage(), streamEarlyErr } + // Terminal 事件完整写出后直接结束,不等上游 EOF(见下方异步循环同款说明)。 + // Codex bare error 序列(error 后可能跟 response.failed 或翻盘的 completed) + // 必须继续读取,不适用提前结束。 + if sawTerminalEvent && !eventInProgress && !(codexFailureTerminal && sawBareError) { + return finalizeStream() + } } if result, err, done := handleScanErr(documentScanner.Err()); done { return result, err @@ -877,6 +883,15 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if streamEarlyErr != nil { return resultWithUsage(), streamEarlyErr } + // Terminal 事件(response.completed 等)已完整写出后不再等上游 EOF: + // 上游在 keep-alive/HTTP2 复用连接上可能拖延关闭连接,空等期间只能 + // 靠 keepalive 维持,白白拉长尾延迟。usage 已在 terminal 事件中解析。 + // Codex bare error 序列(error 后可能跟 response.failed 或翻盘的 completed) + // 必须继续读取,不适用提前结束。 + if sawTerminalEvent && !eventInProgress && !(codexFailureTerminal && sawBareError) { + _ = resp.Body.Close() + return finalizeStream() + } case <-intervalCh: if failureDelivered {