fix(#3996): filter gateway SSE keepalives and test image requests

pkg/httpclient/client.go

@@ -24,6 +24,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
@@ -89,7 +93,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.
@@ -235,6 +242,16 @@ func WithQuery(query url.Values) Opt {
 	}
 }
 
+// WithSSEKeepaliveFilter strips payload-free events named "keepalive".
+// The Gemini gateway emits these transport frames, but the GenAI SDK rejects
+// event-prefixed lines even when their only data is {}. Other names and
+// keepalives with meaningful payloads are deliberately left unchanged.
+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

@@ -10,7 +10,7 @@ import (
 
 // sseFilterTransport wraps a base RoundTripper and, when the response is a
 // `text/event-stream`, replaces the body with one that strips SSE events
-// containing no `data:` lines or named `keepalive`.
+// containing no `data:` lines.
 //
 // Why this exists: some upstreams (notably OpenRouter) inject comment-only
 // keep-alive frames into their streams:
@@ -28,12 +28,19 @@ import (
 //
 // The filter normalises the byte stream so events with no `data:` lines
 // (comment-only events, or events bearing only `event:` / `id:` headers)
-// never reach the SDK. Named `keepalive` events are also dropped, even when
-// they carry `data: {}`, because Gemini's SDK rejects their `event:` header.
-// Other data-bearing events pass through, and the filter is a no-op on
-// non-SSE responses.
+// 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) {
@@ -44,32 +51,35 @@ 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
 }
 
 // sseFilterReader buffers the lines of a single SSE event and only emits
 // them once it has seen the trailing blank line AND the event contained at
-// least one `data:` line and was not a keepalive. A half-built event still
-// pending at EOF is dropped silently — without the terminating blank line a
-// downstream parser would not have dispatched it anyway.
+// least one `data:` line. A half-built event still pending at EOF is
+// dropped silently — without the terminating blank line a downstream parser
+// would not have dispatched it anyway.
 type sseFilterReader struct {
-	src       io.ReadCloser
-	scn       *bufio.Scanner
-	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`
-	keepalive bool         // the last `event:` field names a keepalive
+	src     io.ReadCloser
+	scn     *bufio.Scanner
+	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) {
@@ -88,28 +98,47 @@ func (r *sseFilterReader) Read(p []byte) (int, error) {
 func (r *sseFilterReader) consumeLine(line []byte) {
 	switch {
 	case len(line) == 0:
-		// Event boundary: emit data-bearing events except keepalives.
-		if r.hasData && !r.keepalive {
+		// 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.keepalive = 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 field, value, _ := bytes.Cut(line, []byte(":")); bytes.Equal(field, []byte("event")) {
+				r.isKeepalive = bytes.Equal(bytes.TrimPrefix(value, []byte(" ")), []byte("keepalive"))
+			}
 		}
-		if field, value, _ := bytes.Cut(line, []byte(":")); bytes.Equal(field, []byte("event")) {
-			r.keepalive = bytes.Equal(bytes.TrimPrefix(value, []byte(" ")), []byte("keepalive"))
-		}
 	}
 }
 
+// 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()
 }