Documentation
¶
Overview ¶
Package ssefilter is the SSE keepalive guard shared by the openai-go-family adapters (the Chat Completions adapter in provider/openaichat and the Responses adapter in provider/openai).
WHY THIS EXISTS: the openai-go ssestream decoder dispatches an Event on EVERY blank line (packages/ssestream/ssestream.go) and then json.Unmarshals the frame's accumulated data with no empty-payload check. A frame that carries no data — an SSE comment keepalive, a bare extra blank line, an event:-only frame, or a data: line with an empty value (which appends a lone '\n' to the accumulator) — therefore yields an EMPTY payload, and json.Unmarshal([]byte{}, …) fails with "unexpected end of JSON input". That error is sticky: Stream.Next() latches it and returns false forever, so a SINGLE keepalive kills an otherwise healthy turn mid-stream. Once text has already streamed the turn is post-commit, so the harness's no-replay rule makes it TERMINAL — the run dies rather than retrying.
OpenCode Go does exactly this on long turns, observed on the wire as:
: ping - 2026-07-26 19:11:49.346444+00:00
arriving 496 frames into a streaming response. It is rare per-request but fatal whenever it lands, which reads to an operator as frequent unexplained failures. The mechanism was confirmed to also fire on Responses-shaped frames (event: response.output_text.delta), so this guard guards BOTH openai-go-family adapters.
WHY A MIDDLEWARE and not ssestream.RegisterDecoder: the registry is a plain package-level map written without a mutex, and mecatl re-mints these adapters per session (see WithReasoningEffort), so registering from a constructor would be a concurrent map write. Middleware is per-client, needs no global state, and is content-type agnostic (the registry lookup uses the RAW content-type header, so it silently misses "text/event-stream; charset=utf-8").
WHY WHOLE-FRAME BUFFERING: the filter buffers a complete SSE frame (every line up to the blank-line boundary) and decides at the boundary whether the WHOLE frame survives — a frame is kept iff it carries at least one data: line with a non-empty value. A data-less frame is dropped IN ITS ENTIRETY, including any event:/id:/retry:/comment lines it carried. Filtering line-at-a-time would forward those metadata lines before the boundary decision and then merge them into the NEXT frame (e.g. `event: thread.ping\n\n` becoming the event name of the following data frame — openai-go special-cases thread.* events, so that silently corrupts or drops the next chunk). Whole-frame buffering also stays byte-exact for every frame the SDK can parse and holds no more than the current frame, so SSE arrival timing (and llmresilience's idle watchdog) is preserved: the SDK's own decoder only dispatches on the blank line, so releasing a frame at its boundary is exactly when the SDK would have seen it anyway.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func New ¶
func New(body io.ReadCloser) io.ReadCloser
New returns a keepalive-filtering ReadCloser wrapping body: Read drains the filtered SSE stream; Close closes body. Use NewKeepaliveFilter instead when wiring the filter as openai-go middleware; this constructor is for direct use (and tests).
func NewKeepaliveFilter ¶
func NewKeepaliveFilter() option.Middleware
NewKeepaliveFilter returns the outermost middleware that wraps an SSE response body in the keepalive stripper, removing data-less frames before the SDK's ssestream decoder can dispatch them as empty-payload Events. It MUST be installed as the FIRST request option so it is the OUTERMOST middleware, i.e. it filters the body the SDK's decoder ultimately reads, after any inner middleware. Non-event-stream bodies (JSON responses, error bodies) pass through untouched so the SDK's error path parses them byte-for-byte.
Types ¶
This section is empty.
Source Files