From 277aa1411dd0f14bfdd11f4de31d130e661aabcb Mon Sep 17 00:00:00 2001 From: cxx-cod <206033780+cxx-cod@users.noreply.github.com> Date: Tue, 1 Sep 2026 19:32:24 +0000 Subject: [PATCH 1/2] fix(openai): end streaming at terminal event instead of waiting for upstream EOF The streaming forward loops treated upstream EOF as the only normal exit condition; response.completed/[DONE] only set a flag. With keep-alive / HTTP/2 connection reuse to the upstream, the upstream may hold the stream open for 8-46s after the terminal event, during which the gateway could only emit downstream keepalives, inflating tail latency. Finish the stream as soon as the terminal event frame has been fully written downstream (usage is already parsed from the terminal event), and close the upstream body to unblock the reader goroutine. Applied to the guarded/async, sync, and passthrough streaming paths. The WS HTTP bridge already behaved this way. Adds a regression test with an upstream that hangs after response.completed. --- .../service/openai_gateway_passthrough.go | 8 ++++ .../openai_gateway_response_flush_test.go | 44 +++++++++++++++++++ .../openai_gateway_response_handling.go | 15 +++++++ 3 files changed, 67 insertions(+) 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 { From 6f9fde093fd4c6e19981d9ace2d35af9e90b74cd Mon Sep 17 00:00:00 2001 From: cxx-cod <206033780+cxx-cod@users.noreply.github.com> Date: Wed, 2 Sep 2026 03:30:14 +0000 Subject: [PATCH 2/2] style(openai): apply De Morgan's law to satisfy staticcheck QF1001 --- backend/internal/service/openai_gateway_passthrough.go | 2 +- backend/internal/service/openai_gateway_response_handling.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 0705d897b1..429bc19a78 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -2191,7 +2191,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( // 空等期间只能靠 keepalive 维持，白白拉长尾延迟。usage 已在 terminal 事件中解析。 // Codex bare error 序列（error 后可能跟 response.failed 或翻盘的 completed） // 必须继续读取，不适用提前结束。 - if (sawDone || sawTerminalEvent) && line == "" && !(codexFailureTerminal && sawBareError) { + if (sawDone || sawTerminalEvent) && line == "" && (!codexFailureTerminal || !sawBareError) { break } } diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 021b060c73..6ea830b7b3 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -801,7 +801,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. // Terminal 事件完整写出后直接结束，不等上游 EOF（见下方异步循环同款说明）。 // Codex bare error 序列（error 后可能跟 response.failed 或翻盘的 completed） // 必须继续读取，不适用提前结束。 - if sawTerminalEvent && !eventInProgress && !(codexFailureTerminal && sawBareError) { + if sawTerminalEvent && !eventInProgress && (!codexFailureTerminal || !sawBareError) { return finalizeStream() } } @@ -888,7 +888,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. // 靠 keepalive 维持，白白拉长尾延迟。usage 已在 terminal 事件中解析。 // Codex bare error 序列（error 后可能跟 response.failed 或翻盘的 completed） // 必须继续读取，不适用提前结束。 - if sawTerminalEvent && !eventInProgress && !(codexFailureTerminal && sawBareError) { + if sawTerminalEvent && !eventInProgress && (!codexFailureTerminal || !sawBareError) { _ = resp.Body.Close() return finalizeStream() }