From 30967d5d9a6ca6eb483b8b70bd6fcad869dc5d97 Mon Sep 17 00:00:00 2001 From: shaw Date: Fri, 31 Jul 2026 18:57:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(grok):=20ping=20=E5=B8=A7=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E6=94=B9=E5=86=99=E4=B8=BA=20SSE=20=E6=B3=A8=E9=87=8A=E5=B9=B6?= =?UTF-8?q?=E9=99=90=E5=88=B6=E8=BF=87=E6=BB=A4=E7=BC=93=E5=86=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Responses 事件类型对严格客户端是闭合枚举，任何 event: ping 帧都会令 grok CLI / Codex CLI 整轮失败。原实现只精确匹配 inference-cost 标记帧 和 cost=="0" 帧，上游尾帧携带非零 cost 或格式微调即复发 #5105；且对 其它 ping 变体的保守放行同样会炸掉严格解析器。现改为：event: ping 帧 （data 声明的 type 与事件名不冲突时）一律改写为 SSE 注释 ": ping"， 所有解析器安全忽略且保留保活效果。 同时把整帧缓冲改为增量状态机：非 ping 帧首行即判定、逐行零拷贝直通， 不再累积；仅 ping 候选帧缓冲，并设 16 行 / 16KB 上限，超限回放原文 转直通，杜绝上游用永不结束的帧撑爆网关内存。 --- .../service/openai_gateway_grok_sse_filter.go | 188 +++++++++++++----- .../openai_gateway_grok_sse_filter_test.go | 146 ++++++++------ 2 files changed, 221 insertions(+), 113 deletions(-) diff --git a/backend/internal/service/openai_gateway_grok_sse_filter.go b/backend/internal/service/openai_gateway_grok_sse_filter.go index 3d13805262..78159144d5 100644 --- a/backend/internal/service/openai_gateway_grok_sse_filter.go +++ b/backend/internal/service/openai_gateway_grok_sse_filter.go @@ -10,6 +10,21 @@ import ( "sync" ) +// OpenAI Responses SSE event types are a closed enum for strict clients +// (grok CLI, Codex CLI): an unknown `event: ping` frame aborts the whole +// turn. Vendor gateways behind Grok subscriptions inject such frames for +// billing/keepalive, so ping frames are rewritten into an SSE comment that +// every parser ignores while the connection still looks alive downstream. +var grokResponsesPingComment = []byte(": ping\n\n") + +// A vendor ping frame is one event line plus one small data line. Cap what is +// buffered while deciding, so an upstream streaming a frame that never ends +// cannot grow gateway memory; frames over the cap are passed through as-is. +const ( + grokResponsesPingFrameMaxLines = 16 + grokResponsesPingFrameMaxBytes = 16 * 1024 +) + type grokResponsesBillingPingFilterBody struct { *io.PipeReader source io.Closer @@ -61,38 +76,111 @@ func filterGrokResponsesBillingPings( } scanner.Buffer(scanBuf[:0:initialBufferSize], maxLineSize) scanner.Split(scanSSELinesPreservingEndings) - frame := make([][]byte, 0, 3) - writeFrame := func() error { - if !isGrokResponsesBillingPingFrame(frame) { - for _, line := range frame { - if _, err := destination.Write(line); err != nil { - return err - } + // Only frames opened by an `event: ping` line are buffered (pingFrame); + // every other frame streams through line by line without copying. + pingFrame := make([][]byte, 0, 3) + pingFrameBytes := 0 + inPassthroughFrame := false + + replayPingFrame := func() error { + for _, line := range pingFrame { + if _, err := destination.Write(line); err != nil { + return err } } - frame = frame[:0] + pingFrame = pingFrame[:0] + pingFrameBytes = 0 return nil } + // endPingFrame decides a complete buffered candidate: vendor ping frames + // become an SSE comment, everything else is replayed verbatim. blankLine + // is nil when the stream ends inside the frame. + endPingFrame := func(blankLine []byte) error { + if isGrokResponsesPingEventFrame(pingFrame) { + pingFrame = pingFrame[:0] + pingFrameBytes = 0 + _, err := destination.Write(grokResponsesPingComment) + return err + } + if err := replayPingFrame(); err != nil { + return err + } + if blankLine == nil { + return nil + } + _, err := destination.Write(blankLine) + return err + } + abort := func(err error) { _ = destination.CloseWithError(err) } for scanner.Scan() { - line := append([]byte(nil), scanner.Bytes()...) - frame = append(frame, line) - if len(bytes.TrimSuffix(bytes.TrimSuffix(line, []byte("\n")), []byte("\r"))) == 0 { - if err := writeFrame(); err != nil { - _ = destination.CloseWithError(err) + line := scanner.Bytes() + isBlank := len(trimSSELineEnding(line)) == 0 + + if inPassthroughFrame { + if _, err := destination.Write(line); err != nil { + abort(err) return } + if isBlank { + inPassthroughFrame = false + } + continue } + + if len(pingFrame) > 0 { + if isBlank { + if err := endPingFrame(line); err != nil { + abort(err) + return + } + continue + } + if canExtendGrokResponsesPingFrame(line) && + len(pingFrame) < grokResponsesPingFrameMaxLines && + pingFrameBytes+len(line) <= grokResponsesPingFrameMaxBytes { + pingFrame = append(pingFrame, append([]byte(nil), line...)) + pingFrameBytes += len(line) + continue + } + // Not a filterable ping frame after all (unexpected field line, + // or past the buffering caps): replay it and stream the rest of + // the frame through unchanged. + if err := replayPingFrame(); err != nil { + abort(err) + return + } + if _, err := destination.Write(line); err != nil { + abort(err) + return + } + inPassthroughFrame = true + continue + } + + // Frame start: only `event: ping` opens a buffered candidate. + if !isBlank { + if value, ok := extractOpenAISSEEventLine(string(trimSSELineEnding(line))); ok && value == "ping" { + pingFrame = append(pingFrame, append([]byte(nil), line...)) + pingFrameBytes = len(line) + continue + } + } + if _, err := destination.Write(line); err != nil { + abort(err) + return + } + inPassthroughFrame = !isBlank } - if len(frame) > 0 { - if err := writeFrame(); err != nil { - _ = destination.CloseWithError(err) + if len(pingFrame) > 0 { + if err := endPingFrame(nil); err != nil { + abort(err) return } } if err := scanner.Err(); err != nil { - _ = destination.CloseWithError(fmt.Errorf("filter Grok Responses billing ping: %w", err)) + abort(fmt.Errorf("filter Grok Responses billing ping: %w", err)) return } _ = destination.Close() @@ -119,44 +207,42 @@ func scanSSELinesPreservingEndings(data []byte, atEOF bool) (advance int, token return 0, nil, nil } -func isGrokResponsesBillingPingFrame(rawLines [][]byte) bool { - eventType := "" - data := "" - eventLines := 0 - dataLines := 0 - for _, rawLine := range rawLines { - line := strings.TrimSuffix(strings.TrimSuffix(string(rawLine), "\n"), "\r") - if line == "" { - continue - } - if value, ok := extractOpenAISSEEventLine(line); ok { - eventType = value - eventLines++ - continue +func trimSSELineEnding(line []byte) []byte { + return bytes.TrimSuffix(bytes.TrimSuffix(line, []byte("\n")), []byte("\r")) +} + +// canExtendGrokResponsesPingFrame reports whether a line may still belong to a +// vendor ping frame: only data lines and SSE comments. Any other field (a +// second event line, id, retry, ...) means the frame is not a plain ping. +func canExtendGrokResponsesPingFrame(rawLine []byte) bool { + line := trimSSELineEnding(rawLine) + if len(line) > 0 && line[0] == ':' { + return true + } + _, ok := extractOpenAISSEDataLine(string(line)) + return ok +} + +// isGrokResponsesPingEventFrame decides a buffered candidate whose first line +// is already `event: ping`. The only candidates replayed verbatim are frames +// whose data payload declares a different event type than the SSE event line; +// every other shape (billing cost, keepalive, no data, malformed JSON) would +// break strict Responses clients and is rewritten into a comment. +func isGrokResponsesPingEventFrame(rawLines [][]byte) bool { + dataParts := make([]string, 0, 1) + for _, rawLine := range rawLines[1:] { + if value, ok := extractOpenAISSEDataLine(string(trimSSELineEnding(rawLine))); ok { + dataParts = append(dataParts, value) } - if value, ok := extractOpenAISSEDataLine(line); ok { - data = value - dataLines++ - continue - } - return false } - if eventType != "ping" || eventLines != 1 || dataLines != 1 { - return false + if len(dataParts) == 0 { + return true } - var payload struct { - Type string `json:"type"` - OpenCodeType json.RawMessage `json:"x-opencode-type"` - Cost json.RawMessage `json:"cost"` - } - if err := json.Unmarshal([]byte(data), &payload); err != nil || payload.Type != "ping" { - return false + Type *string `json:"type"` } - if len(payload.OpenCodeType) > 0 { - var openCodeType string - return json.Unmarshal(payload.OpenCodeType, &openCodeType) == nil && openCodeType == "inference-cost" + if err := json.Unmarshal([]byte(strings.Join(dataParts, "\n")), &payload); err != nil || payload.Type == nil { + return true } - var cost string - return json.Unmarshal(payload.Cost, &cost) == nil && cost == "0" + return *payload.Type == "ping" } diff --git a/backend/internal/service/openai_gateway_grok_sse_filter_test.go b/backend/internal/service/openai_gateway_grok_sse_filter_test.go index 10c88e73ec..1dce65e95a 100644 --- a/backend/internal/service/openai_gateway_grok_sse_filter_test.go +++ b/backend/internal/service/openai_gateway_grok_sse_filter_test.go @@ -16,6 +16,19 @@ import ( "github.com/stretchr/testify/require" ) +func filterGrokPingTestInput(t *testing.T, input string) string { + t.Helper() + body := newGrokResponsesBillingPingFilterBody( + io.NopCloser(strings.NewReader(input)), + &Account{Platform: PlatformGrok}, + defaultMaxLineSize, + ) + output, err := io.ReadAll(body) + require.NoError(t, err) + require.NoError(t, body.Close()) + return string(output) +} + func TestGrokResponsesBillingPingFilter(t *testing.T) { input := strings.Join([]string{ ": upstream keepalive", @@ -37,18 +50,11 @@ func TestGrokResponsesBillingPingFilter(t *testing.T) { "", }, "\n") - body := newGrokResponsesBillingPingFilterBody( - io.NopCloser(strings.NewReader(input)), - &Account{Platform: PlatformGrok}, - defaultMaxLineSize, - ) - output, err := io.ReadAll(body) - require.NoError(t, err) - require.NoError(t, body.Close()) - - result := string(output) + result := filterGrokPingTestInput(t, input) + require.NotContains(t, result, "event: ping") require.NotContains(t, result, `"x-opencode-type":"inference-cost"`) require.NotContains(t, result, `{"type":"ping","cost":"0"}`) + require.Equal(t, 2, strings.Count(result, ": ping\n\n")) require.Contains(t, result, ": upstream keepalive\n\n") require.Contains(t, result, "event: response.output_text.delta") require.Contains(t, result, `{"type":"response.output_text.delta","delta":"hello"}`) @@ -58,79 +64,94 @@ func TestGrokResponsesBillingPingFilter(t *testing.T) { require.Contains(t, result, `"usage":{"input_tokens":3,"output_tokens":5}`) } -func TestGrokResponsesBillingPingFilterPreservesUnrelatedPingFrames(t *testing.T) { +// Every `event: ping` frame is outside the Responses closed event enum and +// breaks strict clients regardless of its payload shape, so all variants are +// rewritten into an SSE comment (issue #5105). +func TestGrokResponsesBillingPingFilterConvertsPingVariants(t *testing.T) { + frames := []string{ + "event: ping\ndata: {\"type\":\"ping\",\"x-opencode-type\":\"inference-cost\",\"cost\":\"0.06029240\"}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"cost\":\"0\"}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"cost\":\"0.06029240\"}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"kind\":\"keepalive\"}\n\n", + "event: ping\ndata: {\"type\":\"ping\"}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"cost\":2}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"cost\":0.0001}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"cost\":\" 0 \"}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"x-opencode-type\":\"keepalive\",\"cost\":0}\n\n", + "event: ping\ndata: {\"type\":\"ping\",\"x-opencode-type\":null,\"cost\":0}\n\n", + "event: ping\ndata: {\"cost\":\"0\"}\n\n", + "event: ping\ndata: {not-json}\n\n", + "event: ping\n\n", + "event: ping\n: vendor note\ndata: {\"type\":\"ping\"}\n\n", + } + result := filterGrokPingTestInput(t, strings.Join(frames, "")) + require.Equal(t, strings.Repeat(": ping\n\n", len(frames)), result) +} + +func TestGrokResponsesBillingPingFilterPreservesNonPingFrames(t *testing.T) { input := strings.Join([]string{ - "event: ping", - `data: {"type":"ping","kind":"keepalive"}`, - "", - "event: ping", - `data: {"type":"ping","cost":2}`, - "", - "event: ping", - `data: {"type":"ping"}`, - "", - "event: ping", - `data: {"type":"ping","cost":0.0001}`, - "", - "event: ping", - `data: {"type":"ping","cost":0}`, - "", - "event: ping", - `data: {"type":"ping","cost":" 0 "}`, - "", - "event: ping", - `data: {"type":"ping","x-opencode-type":"keepalive","cost":0}`, - "", "event: ping", `data: {"type":" ping ","cost":0}`, "", "event: ping", - `data: {"type":"ping","x-opencode-type":null,"cost":0}`, + `data: {"type":"response.completed"}`, "", "event: custom", `data: {"type":"ping","x-opencode-type":"inference-cost"}`, "", + `data: {"type":"ping","cost":"0"}`, + "", + ": keepalive comment", + "", + "retry: 1000", + "", }, "\n") - body := newGrokResponsesBillingPingFilterBody( - io.NopCloser(strings.NewReader(input)), - &Account{Platform: PlatformGrok}, - defaultMaxLineSize, - ) - output, err := io.ReadAll(body) - require.NoError(t, err) - require.NoError(t, body.Close()) - require.Equal(t, input, string(output)) + require.Equal(t, input, filterGrokPingTestInput(t, input)) } -func TestGrokResponsesBillingPingFilterPreservesFramingAndMalformedInput(t *testing.T) { +// A ping candidate that turns out to carry an unexpected SSE field is not a +// vendor billing/keepalive frame; it must be replayed byte for byte. +func TestGrokResponsesBillingPingFilterPassesThroughPingFrameWithUnknownField(t *testing.T) { + input := "event: ping\nid: 7\ndata: {\"type\":\"ping\",\"cost\":\"0\"}\n\n" + require.Equal(t, input, filterGrokPingTestInput(t, input)) +} + +// Buffering caps: a ping candidate that grows past the line or byte limit is +// streamed through unchanged instead of accumulating unbounded memory. +func TestGrokResponsesBillingPingFilterPassesThroughOversizedPingFrame(t *testing.T) { + lines := []string{"event: ping"} + for i := 0; i < grokResponsesPingFrameMaxLines; i++ { + lines = append(lines, ": filler comment") + } + lines = append(lines, `data: {"type":"ping","cost":"0"}`, "") + byLines := strings.Join(lines, "\n") + require.Equal(t, byLines, filterGrokPingTestInput(t, byLines)) + + byBytes := "event: ping\ndata: {\"type\":\"ping\",\"pad\":\"" + + strings.Repeat("x", grokResponsesPingFrameMaxBytes) + "\"}\n\n" + require.Equal(t, byBytes, filterGrokPingTestInput(t, byBytes)) +} + +func TestGrokResponsesBillingPingFilterConvertsMalformedPingFrames(t *testing.T) { input := "event: ping\r\ndata: {not-json}\r\n\r\n" + "event: ping\r\ndata: {\"type\":\"ping\",\"cost\":\"0\"} trailing\r\n\r\n" + "event: future.response.event\r\ndata: {\"type\":\"future.response.event\"}" - body := newGrokResponsesBillingPingFilterBody( - io.NopCloser(strings.NewReader(input)), - &Account{Platform: PlatformGrok}, - defaultMaxLineSize, - ) - output, err := io.ReadAll(body) - require.NoError(t, err) - require.NoError(t, body.Close()) - require.Equal(t, input, string(output)) + want := ": ping\n\n" + ": ping\n\n" + + "event: future.response.event\r\ndata: {\"type\":\"future.response.event\"}" + require.Equal(t, want, filterGrokPingTestInput(t, input)) } func TestGrokResponsesBillingPingFilterHandlesBareCRFrames(t *testing.T) { input := "event: ping\rdata: {\"type\":\"ping\",\"cost\":\"0\"}\r\r" + "event: future.event\rdata: {\"type\":\"future.event\"}\r\r" - want := "event: future.event\rdata: {\"type\":\"future.event\"}\r\r" - body := newGrokResponsesBillingPingFilterBody( - io.NopCloser(strings.NewReader(input)), - &Account{Platform: PlatformGrok}, - defaultMaxLineSize, - ) - output, err := io.ReadAll(body) - require.NoError(t, err) - require.NoError(t, body.Close()) - require.Equal(t, want, string(output)) + want := ": ping\n\n" + "event: future.event\rdata: {\"type\":\"future.event\"}\r\r" + require.Equal(t, want, filterGrokPingTestInput(t, input)) +} + +func TestGrokResponsesBillingPingFilterConvertsPartialPingFrameAtEOF(t *testing.T) { + input := "event: ping\ndata: {\"type\":\"ping\",\"cost\":\"0\"}" + require.Equal(t, ": ping\n\n", filterGrokPingTestInput(t, input)) } func TestGrokResponsesBillingPingFilterDoesNotFilterNonGrokAccounts(t *testing.T) { @@ -177,6 +198,7 @@ func TestGrokResponsesBillingPingFilterPreservesUsageAndTerminalEvent(t *testing require.Equal(t, "resp_1", result.responseID) require.Contains(t, recorder.Body.String(), "response.completed") require.NotContains(t, recorder.Body.String(), "inference-cost") + require.NotContains(t, recorder.Body.String(), "event: ping") } type grokPingFilterTestReadCloser struct {