streaming

type Action string

type BufferOptions struct {
	// MaxBytes caps the buffered upstream bytes; 0 selects 4 MiB. Exceeding
	// it fails closed with error code "response_too_large".
	MaxBytes int
	// KeepAliveInterval spaces the SSE comments sent to the client while the
	// upstream is being drained; 0 selects 15s, a negative value disables
	// them.
	KeepAliveInterval time.Duration
	// KeepAliveComment is the comment text; default ": gomodel-buffering".
	KeepAliveComment string
	// OnError receives the buffer limit and finisher errors behind a
	// fail-closed replay.
	OnError func(error)
}

type Codec interface {
	// Decode classifies raw; anything not understood is KindOther. Data in
	// the returned event aliases raw.Data.
	Decode(raw RawEvent, seq int) Event
	// Track records ev as emitted to the client. Streams call it for every
	// decoded event they relay, passing the rewritten event when the text
	// was changed. A stream that emits nothing before its terminal events
	// (buffering) never calls it.
	Track(ev Event)
	// RewriteText returns a copy of ev whose delta text is replaced with
	// text; every other member of the payload is preserved.
	RewriteText(ev Event, text string) (Event, error)
	// Terminate renders the final bytes that end a cut stream, [DONE]
	// included.
	Terminate(t Termination) [][]byte
	// StripTerminal returns a copy of a text event without the members that
	// must reach the client once per choice (a chat chunk's finish_reason
	// and usage); ok reports that ev carried any. Lookbehind
	// re-segmentation emits the head of a chunk from the stripped copy and
	// the withheld tail from the original, so those members arrive once,
	// with the chunk's last text.
	StripTerminal(ev Event) (Event, bool)
	// Split divides a raw event that carries several choices into one raw
	// event per choice, so each is decoded and transformed on its own. It
	// returns nil when raw needs no splitting.
	Split(raw RawEvent) []RawEvent
	// Restate rewrites an event that repeats text already streamed (the
	// Responses *.done and response.completed events) so it carries the
	// text that was actually emitted after transformation; ok reports that
	// ev was changed. Codecs without such events return ev, false.
	Restate(ev Event) (Event, bool)
}

type Decision struct {
	Action Action
	// Text is the replacement delta text for ActionReplace. Text,
	// reasoning, and tool-call argument deltas can be replaced.
	Text string
	// Terminate describes how to end the stream for ActionTerminate; nil
	// selects the codec defaults (finish_reason "content_filter").
	Terminate *Termination
}

type Event struct {
	// Seq is the zero-based position of the event in the stream, counting
	// decoded events only (comments and blank blocks are not numbered).
	Seq  int
	Kind EventKind
	// Choice is the chat choice index; always 0 for Responses streams.
	Choice int
	// Call is the index of the tool call a tool-call delta belongs to: the
	// delta's tool_calls[].index in a chat stream, the output_index of the
	// function_call item in a Responses stream. 0 for other kinds.
	Call int
	// Text is the delta text for text and reasoning deltas, and the arguments
	// fragment carried by a tool call delta. Empty for other kinds.
	Text string
	// Overlap is the number of leading characters (runes) of Text that were
	// already shown in the previous event of this window (a choice's text,
	// or one of its tool calls' arguments). It is non-zero only for deltas
	// re-segmented under lookbehind, where consecutive windows overlap;
	// Text[Overlap:] is the new text.
	Overlap int
	// Final marks the last event of a window: its text is emitted in full
	// after the decision (the stream ended, or a delta of another kind
	// flushed the window), so nothing of it is withheld for a next event.
	Final bool
	// ClosesChoice marks a delta event whose chunk also carries the
	// finish_reason of its choice (a chat stream may end text and finish in
	// one chunk), so once it is emitted the choice needs no finish chunk
	// from Terminate. Cleared on a copy stripped of that member.
	ClosesChoice bool
	// Name is the SSE "event:" field. Empty for chat chunks.
	Name string
	// Data is the JSON payload, or the literal [DONE]. For events handed to a
	// Transformer it is only valid during the call; copy it to retain it.
	Data []byte
}

type EventFilter interface {
	WantsJSONEvent(raw []byte) bool
}

type EventKind string

type EventScanner struct {
	// MaxEventBytes bounds one event's body. A larger event is relayed as
	// Oversized fragments and never parsed, whether it arrived complete in
	// one chunk or is still being buffered across chunks, so the limit does
	// not depend on upstream read boundaries. Zero selects 256 KiB.
	MaxEventBytes int
	// contains filtered or unexported fields
}

type Finisher func(events []Event, raw []byte) (replay []byte, err error)

type ObservedSSEStream struct {
	io.ReadCloser
	// contains filtered or unexported fields
}

type Observer interface {
	OnJSONEvent(payload map[string]any)
	OnStreamClose()
}

type RawEvent struct {
	// Name is the "event:" field, if any.
	Name string
	// Data joins the block's "data:" lines with "\n" (per the SSE spec). Nil
	// when the block carries no data field.
	Data []byte
	// Comment marks a block without a data field (a ":" comment, an id/retry
	// only block, or an empty block). Such blocks are relayed verbatim.
	Comment bool
	// Oversized marks a fragment of an event that exceeded MaxEventBytes; the
	// fragment is relayed unparsed and never decoded.
	Oversized bool
	// Raw holds the block's original bytes including its terminating blank
	// line (when one was seen), so pass-through can be byte-identical.
	Raw []byte
}

type Renumberer interface {
	// BeginRenumber puts the codec in renumbering mode: from then on the
	// codec assigns the outgoing numbers, both to the events handed to
	// Renumber and to the ones Terminate renders.
	BeginRenumber()
	// Renumber returns a copy of ev numbered with the next outgoing number;
	// ok reports that ev was changed, and is false when the event carries no
	// number or already carries the right one.
	Renumber(ev Event) (Event, bool)
}

type StallReporter interface {
	StallError() error
}

type StreamBuffer struct {
	// contains filtered or unexported fields
}

type Termination struct {
	// FinishReason is the chat finish_reason (default "content_filter"). For
	// Responses it selects incomplete_details.reason.
	FinishReason string
	// ErrorCode, when set, ends the stream with an error (chat: an error
	// payload before the finish chunk; Responses: response.failed).
	ErrorCode    string
	ErrorMessage string
	// Text is optional final text emitted as one more delta before finishing,
	// for example a canned safe message.
	Text string
	// Usage, when set, is the provider usage object rendered into the
	// terminal event (chat: the finish chunk; Responses: response.usage), so
	// accounting observers downstream still see the tokens a cut stream
	// consumed.
	Usage any
}

type TransformOptions struct {
	// LookbehindChars withholds this many trailing characters of text per
	// choice so a pattern that spans two chunks is visible to the transformer
	// in one event. 0 disables re-segmentation. See NewTransformedSSEStream.
	LookbehindChars int
	// MinChunkChars collects the text deltas of a choice until at least this
	// many new characters (runes) are pending and presents them to the
	// transformer as one text event. 0 presents deltas as they arrive; values
	// above MaxMinChunkChars are clamped to it. See NewTransformedSSEStream.
	MinChunkChars int
	// MaxEventBytes bounds one SSE event. A larger event cannot be inspected
	// in flight, so the stream ends fail-closed with error code
	// "event_too_large" instead of relaying it past the transformer. 0
	// selects 4 MiB.
	MaxEventBytes int
	// OnError receives non-fatal problems (a replace on a non-text event, a
	// failed rewrite) and the error behind a fail-closed termination.
	OnError func(error)
}

type Transformer interface {
	// OnEvent is called for every decoded event except the [DONE] sentinel.
	// The event, including Data, is only valid during the call.
	OnEvent(ev *Event) (Decision, error)
	// OnEnd is called once after the last upstream event and before [DONE]
	// (or at upstream EOF when no [DONE] arrives). Returning a Termination
	// cuts the stream there.
	OnEnd() (*Termination, error)
}