diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 73c59c4ece..23684af4aa 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -173,7 +173,23 @@ func openAIStreamEventIsTerminal(data string) bool { if trimmed == "[DONE]" { return true } - switch gjson.Get(trimmed, "type").String() { + return openAIStreamEventTypeIsTerminal(gjson.Get(trimmed, "type").String()) +} + +// openAIStreamEventIsTerminalWithType 复用已提取的 type，避免 SSE 热路径重复扫描 JSON。 +func openAIStreamEventIsTerminalWithType(data, eventType string) bool { + trimmed := strings.TrimSpace(data) + if trimmed == "" { + return false + } + if trimmed == "[DONE]" { + return true + } + return openAIStreamEventTypeIsTerminal(eventType) +} + +func openAIStreamEventTypeIsTerminal(eventType string) bool { + switch eventType { case "response.completed", "response.done", "response.failed", "response.incomplete", "response.cancelled", "response.canceled": return true default: diff --git a/backend/internal/service/openai_gateway_response_flush_test.go b/backend/internal/service/openai_gateway_response_flush_test.go index 3494824d6c..2c35374596 100644 --- a/backend/internal/service/openai_gateway_response_flush_test.go +++ b/backend/internal/service/openai_gateway_response_flush_test.go @@ -402,6 +402,39 @@ func TestOpenAIResponseFlush_FailedAndErrorEventsFlushAtBoundaries(t *testing.T) }) } +func TestOpenAIResponseFlush_ReusedTypeKeepsSSEBytesAndTerminalSemantics(t *testing.T) { + tests := []struct { + name string + body string + flushCount int + }{ + { + name: "whitespace around done", + body: "data: \t[DONE] \n\n", + flushCount: 1, + }, + { + name: "invalid JSON before done", + body: "data: {\"type\":\n\ndata: [DONE]\n\n", + flushCount: 2, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + recorder := newOpenAIResponseFlushRecorder() + + result, err := runOpenAIResponseFlushTest(recorder, io.NopCloser(strings.NewReader(tt.body)), config.GatewayConfig{}) + + require.NoError(t, err) + require.NotNil(t, result) + gotBody, flushes := recorder.snapshot() + require.Equal(t, tt.body, gotBody) + require.Len(t, flushes, tt.flushCount) + }) + } +} + func TestOpenAIResponseFlush_ClientDisconnectStillDrainsUsage(t *testing.T) { first := "data: {\"type\":\"response.output_text.delta\",\"delta\":\"a\"}\n\n" terminal := "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":7,\"output_tokens\":5,\"input_tokens_details\":{\"cached_tokens\":2}}}}\n\n" diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 8857bdd527..2acf0a9307 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -396,10 +396,12 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. // Extract data from SSE line (supports both "data: " and "data:" formats) if data, ok := extractOpenAISSEDataLine(line); ok { dataBytes := []byte(data) - if openAIStreamEventIsTerminal(data) { + eventTypeRaw := gjson.GetBytes(dataBytes, "type").String() + eventType := strings.TrimSpace(eventTypeRaw) + // 初始上游 data 的 type 只解析一次：原始值保持终止事件的精确匹配，规范化值供后续分支复用。 + if openAIStreamEventIsTerminalWithType(data, eventTypeRaw) { sawTerminalEvent = true } - eventType := strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String()) if responseID == "" { responseID = extractOpenAIResponseIDFromJSONBytes(dataBytes) } diff --git a/backend/internal/service/openai_gateway_response_handling_type_test.go b/backend/internal/service/openai_gateway_response_handling_type_test.go new file mode 100644 index 0000000000..b783ab5c79 --- /dev/null +++ b/backend/internal/service/openai_gateway_response_handling_type_test.go @@ -0,0 +1,74 @@ +package service + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func TestOpenAIStreamEventIsTerminalWithTypeMatchesExistingSemantics(t *testing.T) { + tests := []struct { + name string + data string + want bool + }{ + {name: "empty", data: "", want: false}, + {name: "whitespace", data: " \t ", want: false}, + {name: "done", data: " [DONE] ", want: true}, + {name: "JSON outer whitespace", data: " \n\t {\"type\":\"response.completed\"} \r\n", want: true}, + {name: "completed", data: `{"type":"response.completed"}`, want: true}, + {name: "response done", data: `{"type":"response.done"}`, want: true}, + {name: "failed", data: `{"type":"response.failed"}`, want: true}, + {name: "incomplete", data: `{"type":"response.incomplete"}`, want: true}, + {name: "cancelled", data: `{"type":"response.cancelled"}`, want: true}, + {name: "canceled", data: `{"type":"response.canceled"}`, want: true}, + {name: "delta", data: `{"type":"response.output_text.delta"}`, want: false}, + {name: "invalid JSON", data: `{"type":`, want: false}, + {name: "terminal with trailing garbage", data: `{"type":"response.completed"} trailing`, want: true}, + {name: "nonterminal with trailing garbage", data: `{"type":"response.output_text.delta"} trailing`, want: false}, + {name: "type whitespace remains nonterminal", data: `{"type":" response.completed "}`, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + eventType := gjson.GetBytes([]byte(tt.data), "type").String() + got := openAIStreamEventIsTerminalWithType(tt.data, eventType) + + require.Equal(t, tt.want, got) + require.Equal(t, openAIStreamEventIsTerminal(tt.data), got) + }) + } +} + +var ( + benchmarkOpenAIResponseSSEEventTypeSink string + benchmarkOpenAIResponseSSETerminalSink bool +) + +func BenchmarkOpenAIResponseSSETypeExtraction(b *testing.B) { + data := `{"type":"response.output_text.delta","sequence_number":42,"delta":"streaming response benchmark payload"}` + dataBytes := []byte(data) + + b.Run("legacy double parse", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(len(dataBytes))) + b.ResetTimer() + for i := 0; i < b.N; i++ { + benchmarkOpenAIResponseSSETerminalSink = openAIStreamEventIsTerminal(data) + benchmarkOpenAIResponseSSEEventTypeSink = strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String()) + } + }) + + b.Run("reused single parse", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(len(dataBytes))) + b.ResetTimer() + for i := 0; i < b.N; i++ { + eventTypeRaw := gjson.GetBytes(dataBytes, "type").String() + benchmarkOpenAIResponseSSEEventTypeSink = strings.TrimSpace(eventTypeRaw) + benchmarkOpenAIResponseSSETerminalSink = openAIStreamEventIsTerminalWithType(data, eventTypeRaw) + } + }) +}