From ab326c96eb6dac0637a980251b73c04b94826944 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Fri, 7 Aug 2026 00:25:37 +0800 Subject: [PATCH 1/2] fix(openai-ws): exclude terminal events from TTFT --- .../service/openai_ws_v2/passthrough_relay.go | 18 +----- .../passthrough_relay_internal_test.go | 27 ++++++++- .../openai_ws_v2/passthrough_relay_test.go | 60 ++++++++++++++++++- 3 files changed, 84 insertions(+), 21 deletions(-) diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay.go b/backend/internal/service/openai_ws_v2/passthrough_relay.go index b6936fe994..5487009ad5 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay.go @@ -953,23 +953,7 @@ func shouldParseUsage(eventType string) bool { } func isTokenEvent(eventType string) bool { - if eventType == "" { - return false - } - switch eventType { - case "response.created", "response.in_progress", "response.output_item.added", "response.output_item.done": - return false - } - if strings.Contains(eventType, ".delta") { - return true - } - if strings.HasPrefix(eventType, "response.output_text") { - return true - } - if strings.HasPrefix(eventType, "response.output") { - return true - } - return eventType == "response.completed" || eventType == "response.done" + return strings.HasSuffix(strings.TrimSpace(eventType), ".delta") } func minDuration(a, b time.Duration) time.Duration { diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go b/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go index 6036be7451..093db89d4f 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go @@ -246,7 +246,7 @@ func TestHelperFunctionsCoverage(t *testing.T) { require.True(t, isTokenEvent("response.output_text.delta")) require.True(t, isTokenEvent("response.output_audio.delta")) - require.True(t, isTokenEvent("response.completed")) + require.False(t, isTokenEvent("response.completed")) require.False(t, isTokenEvent("")) require.False(t, isTokenEvent("response.created")) @@ -407,8 +407,29 @@ func TestIsTokenEventCoverageBranches(t *testing.T) { require.False(t, isTokenEvent("response.in_progress")) require.False(t, isTokenEvent("response.output_item.added")) require.True(t, isTokenEvent("response.output_audio.delta")) - require.True(t, isTokenEvent("response.output")) - require.True(t, isTokenEvent("response.done")) + require.True(t, isTokenEvent("response.function_call_arguments.delta")) + require.True(t, isTokenEvent("response.reasoning_summary_text.delta")) + require.False(t, isTokenEvent("response.output")) + require.False(t, isTokenEvent("response.output_text.done")) + require.False(t, isTokenEvent("response.output_audio.done")) + require.False(t, isTokenEvent("response.output_text.annotation.added")) + require.False(t, isTokenEvent("response.done")) +} + +func TestTerminalAndTokenEventSetsAreDisjoint(t *testing.T) { + t.Parallel() + + for _, eventType := range []string{ + "response.completed", + "response.done", + "response.failed", + "response.incomplete", + "response.cancelled", + "response.canceled", + } { + require.True(t, isTerminalEvent(eventType), eventType) + require.False(t, isTokenEvent(eventType), eventType) + } } func TestShouldParseUsageTerminalEvents(t *testing.T) { diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay_test.go b/backend/internal/service/openai_ws_v2/passthrough_relay_test.go index c083943bcd..f8aeea6e5d 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay_test.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay_test.go @@ -202,7 +202,7 @@ func TestRelay_BasicRelayAndUsage(t *testing.T) { require.Equal(t, 7, result.Usage.InputTokens) require.Equal(t, 3, result.Usage.OutputTokens) require.Equal(t, 2, result.Usage.CacheReadInputTokens) - require.NotNil(t, result.FirstTokenMs) + require.Nil(t, result.FirstTokenMs) require.Equal(t, int64(1), result.ClientToUpstreamFrames) require.Equal(t, int64(1), result.UpstreamToClientFrames) require.Equal(t, int64(0), result.DroppedDownstreamFrames) @@ -865,6 +865,64 @@ func (c *errorOnWriteFrameConn) Close() error { return nil } +func TestRelay_NoDeltaTerminalSequence_FirstTokenMsNil(t *testing.T) { + t.Parallel() + + for _, terminalEvent := range []string{"response.completed", "response.done"} { + terminalEvent := terminalEvent + t.Run(terminalEvent, func(t *testing.T) { + t.Parallel() + + clientConn := newPassthroughTestFrameConn(nil, false) + upstreamConn := newPassthroughTestFrameConn([]passthroughTestFrame{ + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.created","response":{"id":"resp_no_delta"}}`), + }, + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.in_progress","response":{"id":"resp_no_delta"}}`), + }, + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.output_text.done","response_id":"resp_no_delta","text":""}`), + }, + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.content_part.done","response_id":"resp_no_delta"}`), + }, + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.output_item.done","response_id":"resp_no_delta"}`), + }, + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"` + terminalEvent + `","response":{"id":"resp_no_delta","usage":{"input_tokens":2,"output_tokens":0}}}`), + }, + }, true) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + var turn RelayTurnResult + result, relayExit := Relay( + ctx, + clientConn, + upstreamConn, + []byte(`{"type":"response.create","model":"gpt-5.3-codex","input":[]}`), + RelayOptions{OnTurnComplete: func(current RelayTurnResult) { turn = current }}, + ) + + require.Nil(t, relayExit) + require.Equal(t, terminalEvent, turn.TerminalEventType) + require.Nil(t, turn.FirstTokenMs) + require.Equal(t, terminalEvent, result.TerminalEventType) + require.Nil(t, result.FirstTokenMs) + require.Equal(t, int64(6), result.UpstreamToClientFrames) + }) + } +} + func TestRelay_OnTurnComplete_RealOpenAIStream_FirstTokenMs(t *testing.T) { t.Parallel() From e24cb99b79379c7294b552422861cc89028753ff Mon Sep 17 00:00:00 2001 From: wucm667 Date: Tue, 11 Aug 2026 16:01:34 +0800 Subject: [PATCH 2/2] fix(openai-ws): retain no-delta TTFT fallback --- .../service/openai_ws_v2/passthrough_relay.go | 5 +- .../passthrough_relay_internal_test.go | 5 +- .../openai_ws_v2/passthrough_relay_test.go | 72 ++++++++++++++++--- 3 files changed, 69 insertions(+), 13 deletions(-) diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay.go b/backend/internal/service/openai_ws_v2/passthrough_relay.go index 5487009ad5..764b86e7cb 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay.go @@ -953,7 +953,10 @@ func shouldParseUsage(eventType string) bool { } func isTokenEvent(eventType string) bool { - return strings.HasSuffix(strings.TrimSpace(eventType), ".delta") + eventType = strings.TrimSpace(eventType) + return strings.HasSuffix(eventType, ".delta") || + eventType == "response.output_text.done" || + eventType == "response.function_call_arguments.done" } func minDuration(a, b time.Duration) time.Duration { diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go b/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go index 093db89d4f..374fb8e301 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay_internal_test.go @@ -409,9 +409,12 @@ func TestIsTokenEventCoverageBranches(t *testing.T) { require.True(t, isTokenEvent("response.output_audio.delta")) require.True(t, isTokenEvent("response.function_call_arguments.delta")) require.True(t, isTokenEvent("response.reasoning_summary_text.delta")) + require.True(t, isTokenEvent("response.output_text.done")) + require.True(t, isTokenEvent("response.function_call_arguments.done")) require.False(t, isTokenEvent("response.output")) - require.False(t, isTokenEvent("response.output_text.done")) require.False(t, isTokenEvent("response.output_audio.done")) + require.False(t, isTokenEvent("response.content_part.done")) + require.False(t, isTokenEvent("response.output_item.done")) require.False(t, isTokenEvent("response.output_text.annotation.added")) require.False(t, isTokenEvent("response.done")) } diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay_test.go b/backend/internal/service/openai_ws_v2/passthrough_relay_test.go index f8aeea6e5d..c41e7d293b 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay_test.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay_test.go @@ -865,7 +865,7 @@ func (c *errorOnWriteFrameConn) Close() error { return nil } -func TestRelay_NoDeltaTerminalSequence_FirstTokenMsNil(t *testing.T) { +func TestRelay_NoSemanticOutputTerminalSequence_FirstTokenMsNil(t *testing.T) { t.Parallel() for _, terminalEvent := range []string{"response.completed", "response.done"} { @@ -877,27 +877,23 @@ func TestRelay_NoDeltaTerminalSequence_FirstTokenMsNil(t *testing.T) { upstreamConn := newPassthroughTestFrameConn([]passthroughTestFrame{ { msgType: coderws.MessageText, - payload: []byte(`{"type":"response.created","response":{"id":"resp_no_delta"}}`), + payload: []byte(`{"type":"response.created","response":{"id":"resp_no_output"}}`), }, { msgType: coderws.MessageText, - payload: []byte(`{"type":"response.in_progress","response":{"id":"resp_no_delta"}}`), + payload: []byte(`{"type":"response.in_progress","response":{"id":"resp_no_output"}}`), }, { msgType: coderws.MessageText, - payload: []byte(`{"type":"response.output_text.done","response_id":"resp_no_delta","text":""}`), + payload: []byte(`{"type":"response.content_part.done","response_id":"resp_no_output"}`), }, { msgType: coderws.MessageText, - payload: []byte(`{"type":"response.content_part.done","response_id":"resp_no_delta"}`), + payload: []byte(`{"type":"response.output_item.done","response_id":"resp_no_output"}`), }, { msgType: coderws.MessageText, - payload: []byte(`{"type":"response.output_item.done","response_id":"resp_no_delta"}`), - }, - { - msgType: coderws.MessageText, - payload: []byte(`{"type":"` + terminalEvent + `","response":{"id":"resp_no_delta","usage":{"input_tokens":2,"output_tokens":0}}}`), + payload: []byte(`{"type":"` + terminalEvent + `","response":{"id":"resp_no_output","usage":{"input_tokens":2,"output_tokens":0}}}`), }, }, true) @@ -918,7 +914,61 @@ func TestRelay_NoDeltaTerminalSequence_FirstTokenMsNil(t *testing.T) { require.Nil(t, turn.FirstTokenMs) require.Equal(t, terminalEvent, result.TerminalEventType) require.Nil(t, result.FirstTokenMs) - require.Equal(t, int64(6), result.UpstreamToClientFrames) + require.Equal(t, int64(5), result.UpstreamToClientFrames) + }) + } +} + +func TestRelay_NoDeltaOutputDoneEvent_RecordsFirstTokenBeforeTerminal(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + donePayload string + }{ + { + name: "output text done", + donePayload: `{"type":"response.output_text.done","response_id":"resp_done","text":"hello"}`, + }, + { + name: "function call arguments done", + donePayload: `{"type":"response.function_call_arguments.done","response_id":"resp_done","arguments":"{\"city\":\"Paris\"}"}`, + }, + } + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + clientConn := newPassthroughTestFrameConn(nil, false) + upstreamConn := newPassthroughTestFrameConn([]passthroughTestFrame{ + {msgType: coderws.MessageText, payload: []byte(`{"type":"response.created","response":{"id":"resp_done"}}`)}, + {msgType: coderws.MessageText, payload: []byte(tt.donePayload)}, + {msgType: coderws.MessageText, payload: []byte(`{"type":"response.completed","response":{"id":"resp_done","usage":{"input_tokens":2,"output_tokens":1}}}`)}, + }, true) + + base := time.Unix(0, 0) + var nowTick atomic.Int64 + nowFn := func() time.Time { + return base.Add(time.Duration(nowTick.Add(1)) * 10 * time.Millisecond) + } + var turn RelayTurnResult + result, relayExit := Relay( + context.Background(), + clientConn, + upstreamConn, + []byte(`{"type":"response.create","model":"gpt-5.3-codex","input":[]}`), + RelayOptions{ + Now: nowFn, + OnTurnComplete: func(current RelayTurnResult) { turn = current }, + }, + ) + + require.Nil(t, relayExit) + require.NotNil(t, turn.FirstTokenMs) + require.Less(t, int64(*turn.FirstTokenMs), turn.Duration.Milliseconds()) + require.NotNil(t, result.FirstTokenMs) + require.Less(t, int64(*result.FirstTokenMs), result.Duration.Milliseconds()) }) } }