Documentation
¶
Overview ¶
Package sse implements the SSE parser described in ARCHITECTURE § 10.6.
The parser is dispatch-free in the wire-shape sense: a single line reader handles all three SSE shapes api-log records (OpenAI Chat data-only, OpenAI Responses event-named, Anthropic Messages event-named). Only the **terminator vocabulary** is shape-aware — the parser recognizes `data: [DONE]`, `event: response.completed`, and `event: message_stop` as clean-end markers and reports `stream_done` accordingly. It does NOT interpret the meaning of the events; that is the consumer's job.
This parser is the FINALIZE-time parser. The capture-time drainer (internal/capture) is shape-aware at a different layer (raw `\n\n` frame boundaries for timing). See ARCHITECTURE § 10.6 for the two-layer distinction.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func WriteEvent ¶
WriteEvent serializes one Event back to wire bytes — the inverse of Parse / Scanner. Used by the AFTER-hook stream dispatcher to re-emit events (mutated or otherwise) to the client.
Wire shape per RFC: optional `event: NAME\n`, then `data: ...\n` for every line of the JSON payload, then a single blank line.
One Event = one frame. Multi-line data payloads (rare; we round-trip json.RawMessage which is always single-line) split on `\n`.
Types ¶
type Event ¶
type Event struct {
// Name is the `event:` field. Empty string for data-only frames
// (OpenAI Chat Completions). The empty-string sentinel is documented
// in ARCHITECTURE § 3 sentinels table.
Name string `json:"event"`
// Data is the raw JSON of the `data:` line. Captured verbatim so
// downstream consumers see exactly what the upstream emitted.
Data json.RawMessage `json:"data"`
// TDeltaMs is the per-event arrival timing in ms from the trace's
// ts_start. nil when the trace was reparse-reconstructed (sentinel
// per ARCHITECTURE § 3) or when the upstream response was content-
// encoded (drainer can't see frame boundaries through compression).
TDeltaMs *int64 `json:"t_delta_ms,omitempty"`
// Offset is the byte offset of this event's first field line in the
// parsed input. Used at finalize to look up the chunk-arrival
// timestamp; not serialized to JSONL.
Offset int64 `json:"-"`
}
Event is one parsed SSE event. Data is captured as a raw JSON value when the `data:` line was parseable JSON, or left nil if not (the trace falls back to body_b64 in that case at the caller's discretion).
type ParseResult ¶
type ParseResult struct {
Events []Event
StreamDone bool
// ParseError is non-empty if the body could not be parsed as any
// SSE shape (e.g. random bytes that don't look like `data:` /
// `event:` lines). The caller falls back to body_b64 in that case.
ParseError string
}
ParseResult is the output of Parse over a complete SSE stream.
func Parse ¶
func Parse(r io.Reader) ParseResult
Parse reads an SSE stream from r until EOF and returns the parsed events plus stream-done status.
Each Event carries its byte Offset in the input so finalize can attach per-event t_delta_ms by looking up the chunk that contained the byte (capture.LookupChunkTime). The offset is the byte position of the event's FIRST non-blank, non-comment field line.
The parser tolerates the three terminator shapes ARCHITECTURE § 10.6 names:
- OpenAI Chat Completions: terminates on `data: [DONE]`.
- OpenAI Responses: terminates on `event: response.completed`.
- Anthropic Messages: terminates on `event: message_stop`.
Stream EOF without seeing a terminator → StreamDone=false. Pending `event:` / `data:` accumulator at EOF → emit one final event (some servers don't terminate the last frame with a blank line).
type Scanner ¶
type Scanner struct {
// contains filtered or unexported fields
}
Scanner is the incremental counterpart to Parse. It reads SSE frames one at a time from an io.Reader; callers drive a per-event loop via Next() and react to io.EOF in the usual way.
The shape recognized here mirrors Parse exactly — same field syntax, same terminator vocabulary, same offset bookkeeping. The split exists so the AFTER-hook stream dispatcher can transform-and-forward events as they arrive rather than buffering the whole response.
One Scanner is single-goroutine-safe only. Wrap the upstream io.ReadCloser in a sync layer if multiple goroutines need to consume.
func NewScanner ¶
NewScanner returns a Scanner over r. Buffer size matches Parse's 64 KiB so single-event payloads (Anthropic message_stop final text, OpenAI Responses output_text final delta) fit on the bufio path without expansion.
func (*Scanner) Next ¶
Next reads frames until one full event is buffered, then returns it. Returns io.EOF when the upstream is fully drained AND no pending event remains (the SSE spec allows the final frame to lack a trailing blank line; Next flushes that case on EOF).
The returned Event mirrors Parse's Event: Name + Data + Offset. Data is verbatim raw JSON when the data line parsed as JSON, otherwise JSON-string-escaped so the resulting Data is always valid JSON. This is the same contract Parse upholds; the AFTER dispatcher's classifier expects it.
func (*Scanner) StreamDone ¶
StreamDone reports whether a terminator vocabulary item (data:[DONE], event:response.completed, event:message_stop) has been observed. Mirrors Parse.ParseResult.StreamDone.
Source Files