fix(#3996): drop gateway SSE keepalive frames before the genai parser

pkg/httpclient/client.go

@@ -19,6 +19,10 @@ type HTTPOptions struct {
 	Header http.Header
 	Query  url.Values
 
+	// dropSSEKeepaliveEvents enables keepalive-frame dropping in the SSE
+	// filter transport; see WithSSEKeepaliveFilter.
+	dropSSEKeepaliveEvents bool
+
 	// cagentID resolves the persistent install UUID stamped as
 	// `X-Cagent-Id` on gateway-bound requests. It defaults to
 	// [userid.Get]; tests inject their own source via
@@ -52,7 +56,10 @@ func NewHTTPClient(ctx context.Context, opts ...Opt) *http.Client {
 
 	var wrapped http.RoundTripper = &userAgentTransport{
 		httpOptions: httpOptions,
-		rt:          &sseFilterTransport{base: rt},
+		rt: &sseFilterTransport{
+			base:                rt,
+			dropKeepaliveEvents: httpOptions.dropSSEKeepaliveEvents,
+		},
 	}
 	if httpOptions.refreshAuth != nil {
 		// Outermost, so a replayed request goes through the whole chain again.
@@ -187,6 +194,19 @@ func WithQuery(query url.Values) Opt {
 	}
 }
 
+// WithSSEKeepaliveFilter makes the SSE filter transport also drop whole
+// `event: keepalive` frames whose data carries no payload (`data: {}` or
+// empty). The Docker AI Gateway emits such frames during long generations
+// (e.g. Gemini image output), and google.golang.org/genai's SSE parser
+// treats any `event:` line as a fatal invalid chunk. Only enable this for
+// clients whose SDK cannot tolerate `event:` lines — providers like
+// Anthropic rely on `event:` headers as meaningful framing.
+func WithSSEKeepaliveFilter() Opt {
+	return func(o *HTTPOptions) {
+		o.dropSSEKeepaliveEvents = true
+	}
+}
+
 // newTransport returns an HTTP transport with automatic gzip compression disabled and Docker Desktop PAC support.
 func newTransport(_ context.Context) http.RoundTripper {
 	rt := newAllowPrivateIPsTransport()

pkg/httpclient/sse_filter.go

@@ -30,8 +30,17 @@ import (
 // (comment-only events, or events bearing only `event:` / `id:` headers)
 // never reach the SDK. Well-formed events pass through verbatim, and the
 // filter is a no-op on non-SSE responses.
+//
+// dropKeepaliveEvents additionally drops whole `event: keepalive` frames
+// whose data carries no payload (`data: {}` or empty). The Docker AI
+// Gateway emits such frames during long generations; the genai SDK's SSE
+// parser hard-fails on ANY `event:` line, so they must never reach it. The
+// mode is opt-in (see WithSSEKeepaliveFilter) because other providers —
+// Anthropic in particular — use `event:` headers as meaningful framing that
+// must pass through untouched.
 type sseFilterTransport struct {
-	base http.RoundTripper
+	base                http.RoundTripper
+	dropKeepaliveEvents bool
 }
 
 func (t *sseFilterTransport) RoundTrip(req *http.Request) (*http.Response, error) {
@@ -42,7 +51,7 @@ func (t *sseFilterTransport) RoundTrip(req *http.Request) (*http.Response, error
 	// Match the prefix so charset suffixes (e.g. "text/event-stream;
 	// charset=utf-8") still trigger filtering.
 	if strings.HasPrefix(strings.ToLower(res.Header.Get("Content-Type")), "text/event-stream") {
-		res.Body = newSSEFilterReader(res.Body)
+		res.Body = newSSEFilterReader(res.Body, t.dropKeepaliveEvents)
 	}
 	return res, err
 }
@@ -58,15 +67,19 @@ type sseFilterReader struct {
 	out     bytes.Buffer // bytes ready to hand back to the caller
 	pending bytes.Buffer // accumulated lines for the current event
 	hasData bool         // saw at least one `data:` line in `pending`
+
+	dropKeepaliveEvents bool // see sseFilterTransport
+	isKeepalive         bool // current event is named `keepalive`
+	hasMeaningfulData   bool // saw a `data:` line whose payload isn't empty or `{}`
 }
 
-func newSSEFilterReader(src io.ReadCloser) *sseFilterReader {
+func newSSEFilterReader(src io.ReadCloser, dropKeepaliveEvents bool) *sseFilterReader {
 	scn := bufio.NewScanner(src)
 	// SSE events can be large (long completion tokens, image URLs, …). Match
 	// the buffer size used by openai-go's own SSE decoder so we don't trip
 	// `bufio.ErrTooLong` on payloads it would happily accept.
 	scn.Buffer(make([]byte, 0, 64*1024), bufio.MaxScanTokenSize<<9)
-	return &sseFilterReader{src: src, scn: scn}
+	return &sseFilterReader{src: src, scn: scn, dropKeepaliveEvents: dropKeepaliveEvents}
 }
 
 func (r *sseFilterReader) Read(p []byte) (int, error) {
@@ -85,24 +98,47 @@ func (r *sseFilterReader) Read(p []byte) (int, error) {
 func (r *sseFilterReader) consumeLine(line []byte) {
 	switch {
 	case len(line) == 0:
-		// Event boundary: emit the buffered event iff it had data.
-		if r.hasData {
+		// Event boundary: emit the buffered event iff it had data and is
+		// not a payload-free keepalive frame in keepalive-dropping mode.
+		if r.hasData && (!r.isKeepalive || r.hasMeaningfulData) {
 			r.out.Write(r.pending.Bytes())
 			r.out.WriteByte('\n')
 		}
 		r.pending.Reset()
 		r.hasData = false
+		r.isKeepalive = false
+		r.hasMeaningfulData = false
 	case line[0] == ':':
 		// SSE comment — drop entirely.
 	default:
 		r.pending.Write(line)
 		r.pending.WriteByte('\n')
-		if bytes.HasPrefix(line, []byte("data:")) {
+		if value, ok := fieldValue(line, "data"); ok {
 			r.hasData = true
+			if r.dropKeepaliveEvents {
+				if payload := bytes.TrimSpace(value); len(payload) > 0 && !bytes.Equal(payload, []byte("{}")) {
+					r.hasMeaningfulData = true
+				}
+			}
+		} else if r.dropKeepaliveEvents {
+			if value, ok := fieldValue(line, "event"); ok && string(bytes.TrimSpace(value)) == "keepalive" {
+				r.isKeepalive = true
+			}
 		}
 	}
 }
 
+// fieldValue returns the value of an SSE line whose field name is `name`,
+// with the single optional leading space the SSE grammar allows already
+// removed.
+func fieldValue(line []byte, name string) ([]byte, bool) {
+	value, ok := bytes.CutPrefix(line, []byte(name+":"))
+	if !ok {
+		return nil, false
+	}
+	return bytes.TrimPrefix(value, []byte(" ")), true
+}
+
 func (r *sseFilterReader) Close() error {
 	return r.src.Close()
 }

pkg/httpclient/sse_filter_test.go

@@ -153,7 +153,7 @@ func TestSSEFilter_LargeEvent(t *testing.T) {
 	t.Parallel()
 
 	largeData := "data: " + strings.Repeat("x", 256*1024) + "\n\n"
-	r := newSSEFilterReader(io.NopCloser(strings.NewReader(largeData)))
+	r := newSSEFilterReader(io.NopCloser(strings.NewReader(largeData)), false)
 
 	output, err := io.ReadAll(r)
 	require.NoError(t, err)
@@ -167,7 +167,7 @@ func TestSSEFilter_PartialReads(t *testing.T) {
 	t.Parallel()
 
 	input := "data: test1\n\ndata: test2\n\n"
-	r := newSSEFilterReader(io.NopCloser(strings.NewReader(input)))
+	r := newSSEFilterReader(io.NopCloser(strings.NewReader(input)), false)
 
 	var output []byte
 	buf := make([]byte, 5)
@@ -192,7 +192,7 @@ func TestSSEFilter_IncompleteEventAtEOF(t *testing.T) {
 	t.Parallel()
 
 	input := "data: complete\n\ndata: incomplete"
-	r := newSSEFilterReader(io.NopCloser(strings.NewReader(input)))
+	r := newSSEFilterReader(io.NopCloser(strings.NewReader(input)), false)
 
 	output, err := io.ReadAll(r)
 	require.NoError(t, err)
@@ -204,7 +204,7 @@ func TestSSEFilter_IncompleteEventAtEOF(t *testing.T) {
 func TestSSEFilter_EmptyInput(t *testing.T) {
 	t.Parallel()
 
-	r := newSSEFilterReader(io.NopCloser(strings.NewReader("")))
+	r := newSSEFilterReader(io.NopCloser(strings.NewReader("")), false)
 
 	output, err := io.ReadAll(r)
 	require.NoError(t, err)
@@ -218,7 +218,7 @@ func TestSSEFilter_OnlyComments(t *testing.T) {
 	t.Parallel()
 
 	input := ": comment1\n\n: comment2\n\n"
-	r := newSSEFilterReader(io.NopCloser(strings.NewReader(input)))
+	r := newSSEFilterReader(io.NopCloser(strings.NewReader(input)), false)
 
 	output, err := io.ReadAll(r)
 	require.NoError(t, err)
@@ -230,7 +230,7 @@ func TestSSEFilter_OnlyComments(t *testing.T) {
 func TestSSEFilter_ScannerError(t *testing.T) {
 	t.Parallel()
 
-	r := newSSEFilterReader(io.NopCloser(&errorReader{err: io.ErrUnexpectedEOF}))
+	r := newSSEFilterReader(io.NopCloser(&errorReader{err: io.ErrUnexpectedEOF}), false)
 
 	_, err := io.ReadAll(r)
 	assert.ErrorIs(t, err, io.ErrUnexpectedEOF)
@@ -256,7 +256,7 @@ func TestSSEFilter_CloseWithoutRead(t *testing.T) {
 		onClose: func() { closed = true },
 	}
 
-	r := newSSEFilterReader(tracker)
+	r := newSSEFilterReader(tracker, false)
 	require.NoError(t, r.Close())
 	assert.True(t, closed, "underlying reader should be closed")
 }
@@ -344,3 +344,154 @@ func fetchThroughFilter(t *testing.T, url string) string {
 	require.NoError(t, err)
 	return string(body)
 }
+
+// Gemini-shaped data chunks used by the keepalive tests: a text delta and a
+// media (inlineData) delta of the kind an image-output model streams.
+const (
+	geminiTextChunk  = `data: {"candidates":[{"content":{"parts":[{"text":"hi"}],"role":"model"}}]}` + "\n\n"
+	geminiMediaChunk = `data: {"candidates":[{"content":{"parts":[{"inlineData":{"mimeType":"image/png","data":"aGVsbG8="}}],"role":"model"},"finishReason":"STOP"}]}` + "\n\n"
+	keepaliveFrame   = "event: keepalive\ndata: {}\n\n"
+)
+
+// TestSSEFilter_KeepaliveMode covers the opt-in keepalive-dropping mode used
+// by the Gemini gateway client: payload-free `event: keepalive` frames are
+// removed while every other frame — including named events with meaningful
+// data — passes through verbatim.
+func TestSSEFilter_KeepaliveMode(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{
+			// The gateway scenario: keepalive frames interleaved with
+			// real text and media chunks during a long image generation.
+			name: "drops keepalive frames interleaved with data and media chunks",
+			in:   keepaliveFrame + geminiTextChunk + keepaliveFrame + keepaliveFrame + geminiMediaChunk,
+			want: geminiTextChunk + geminiMediaChunk,
+		},
+		{
+			name: "drops keepalive without a space after data:",
+			in:   "event: keepalive\ndata:{}\n\n" + geminiTextChunk,
+			want: geminiTextChunk,
+		},
+		{
+			name: "drops keepalive with an empty data payload",
+			in:   "event: keepalive\ndata:\n\n" + geminiTextChunk,
+			want: geminiTextChunk,
+		},
+		{
+			// Anthropic-style framing: a named event with meaningful data
+			// must never be touched, even in keepalive mode.
+			name: "preserves named events with meaningful data",
+			in:   "event: content_block_delta\ndata: {\"delta\":{\"text\":\"hi\"}}\n\n",
+			want: "event: content_block_delta\ndata: {\"delta\":{\"text\":\"hi\"}}\n\n",
+		},
+		{
+			// Conservative: only payload-free keepalives are dropped. A
+			// keepalive-named event carrying real data is preserved.
+			name: "preserves keepalive-named event with meaningful data",
+			in:   "event: keepalive\ndata: {\"note\":\"x\"}\n\n",
+			want: "event: keepalive\ndata: {\"note\":\"x\"}\n\n",
+		},
+		{
+			// The base filter's behavior is unchanged by keepalive mode.
+			name: "still drops comment-only and no-data event frames",
+			in:   ": ping\n\nevent: ping\nid: abc\n\n" + geminiTextChunk,
+			want: geminiTextChunk,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			r := newSSEFilterReader(io.NopCloser(strings.NewReader(tt.in)), true)
+			out, err := io.ReadAll(r)
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, string(out))
+		})
+	}
+}
+
+// TestSSEFilter_KeepaliveMode_OutputParsableByGenaiStyleParser feeds a
+// keepalive-interleaved Gemini stream through the keepalive-mode filter and
+// verifies the result against the constraint that made the fix necessary:
+// genai's iterateResponseStream (google.golang.org/genai api_client.go)
+// treats ANY non-blank line without a `data:` prefix as a fatal invalid
+// chunk. Every data payload must survive, in order.
+func TestSSEFilter_KeepaliveMode_OutputParsableByGenaiStyleParser(t *testing.T) {
+	t.Parallel()
+
+	in := keepaliveFrame + geminiTextChunk + keepaliveFrame + geminiMediaChunk + keepaliveFrame
+	r := newSSEFilterReader(io.NopCloser(strings.NewReader(in)), true)
+	out, err := io.ReadAll(r)
+	require.NoError(t, err)
+
+	var payloads []string
+	for line := range strings.Lines(string(out)) {
+		line = strings.TrimSuffix(line, "\n")
+		if line == "" {
+			continue
+		}
+		require.True(t, strings.HasPrefix(line, "data:"), "genai would reject this line as an invalid stream chunk: %q", line)
+		payloads = append(payloads, strings.TrimPrefix(line, "data: "))
+	}
+
+	require.Len(t, payloads, 2)
+	assert.Contains(t, payloads[0], `"text":"hi"`)
+	assert.Contains(t, payloads[1], `"inlineData"`)
+}
+
+// TestSSEFilter_SharedPathKeepsKeepaliveFrames pins that the shared default
+// filter (used by every other provider) does NOT gain keepalive dropping:
+// an `event: keepalive` frame has a data line, so it passes through
+// verbatim, exactly like Anthropic's meaningful named events.
+func TestSSEFilter_SharedPathKeepsKeepaliveFrames(t *testing.T) {
+	t.Parallel()
+
+	in := keepaliveFrame +
+		"event: content_block_delta\ndata: {\"delta\":{\"text\":\"hi\"}}\n\n"
+
+	assert.Equal(t, in, fetchSSE(t, in))
+}
+
+// TestNewHTTPClient_SSEKeepaliveFilterOptIn verifies the option wiring end
+// to end through NewHTTPClient: keepalive frames are dropped only when
+// WithSSEKeepaliveFilter is passed, and the default client leaves them in.
+func TestNewHTTPClient_SSEKeepaliveFilterOptIn(t *testing.T) {
+	t.Parallel()
+
+	in := keepaliveFrame + geminiTextChunk
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, in)
+	}))
+	t.Cleanup(srv.Close)
+
+	fetch := func(t *testing.T, client *http.Client) string {
+		t.Helper()
+		req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, srv.URL, http.NoBody)
+		require.NoError(t, err)
+		res, err := client.Do(req)
+		require.NoError(t, err)
+		defer func() { _ = res.Body.Close() }()
+		body, err := io.ReadAll(res.Body)
+		require.NoError(t, err)
+		return string(body)
+	}
+
+	t.Run("opted in drops keepalive frames", func(t *testing.T) {
+		t.Parallel()
+		client := NewHTTPClient(t.Context(), WithSSEKeepaliveFilter())
+		assert.Equal(t, geminiTextChunk, fetch(t, client))
+	})
+
+	t.Run("default keeps keepalive frames", func(t *testing.T) {
+		t.Parallel()
+		client := NewHTTPClient(t.Context())
+		assert.Equal(t, in, fetch(t, client))
+	})
+}

pkg/model/provider/gemini/client.go

@@ -179,6 +179,12 @@ func NewClient(ctx context.Context, cfg *latest.ModelConfig, env environment.Pro
 				}
 			}
 
+			// The gateway keeps long generations alive with `event: keepalive`
+			// + `data: {}` frames, which genai's SSE parser rejects as fatal
+			// invalid chunks. Drop them here, on the gateway path only — direct
+			// Gemini/Vertex clients never receive them.
+			httpOptions = append(httpOptions, httpclient.WithSSEKeepaliveFilter())
+
 			gatewayHTTPClient := httpclient.NewHTTPClient(ctx, httpOptions...)
 			globalOptions.WrapTransport(ctx, gatewayHTTPClient)