fix(grok): ping 帧统一改写为 SSE 注释并限制过滤缓冲

Responses 事件类型对严格客户端是闭合枚举，任何 event: ping 帧都会令
grok CLI / Codex CLI 整轮失败。原实现只精确匹配 inference-cost 标记帧
和 cost=="0" 帧，上游尾帧携带非零 cost 或格式微调即复发 #5105；且对
其它 ping 变体的保守放行同样会炸掉严格解析器。现改为：event: ping 帧
（data 声明的 type 与事件名不冲突时）一律改写为 SSE 注释 ": ping"，
所有解析器安全忽略且保留保活效果。

同时把整帧缓冲改为增量状态机：非 ping 帧首行即判定、逐行零拷贝直通，
不再累积；仅 ping 候选帧缓冲，并设 16 行 / 16KB 上限，超限回放原文
转直通，杜绝上游用永不结束的帧撑爆网关内存。

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"
 }

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 {