View Source OpenaiSseGuard (openai_sse_guard v0.1.0)

A bounded, dependency-free observer for OpenAI-compatible Server-Sent Events.

The observer answers a deliberately narrow question: what evidence remains when a streaming connection ends? It keeps framing and terminal-state metadata, but never stores generated text, a complete response body, or provider prose. It does not make HTTP requests, retry, sleep, or decide billing.

The implementation is an OTP-native byte observer. It accepts binaries that may split UTF-8 code points and recognizes LF, CRLF, and CR SSE boundaries. JSON is intentionally treated as opaque data: unknown data-bearing events are conservative rather than silently classified as safe to replay.

See observe/2 for an enumerable helper and new/1, push/2, and finish/1 for incremental integration with an HTTP client.

Summary

Types

t()

Mutable-by-replacement observer state held by the caller.

Functions

finish(state)

Flushes the observer and returns an immutable snapshot.

new()

Creates an observer with bounded frame and event limits.

new(options)

observe(enumerable, opts \\ [])

Observes any enumerable of binary chunks and returns its final snapshot.

push(state, chunk)

Adds one byte chunk to an observer.

snapshot(state)

Returns an immutable snapshot without exposing the internal buffer.

Types

t()

@type t() :: %OpenaiSseGuard{
  buffer: binary(),
  error_code: String.t() | nil,
  event_count: non_neg_integer(),
  finished: boolean(),
  has_output: boolean(),
  last_event_type: String.t() | nil,
  malformed_event_count: non_neg_integer(),
  options: OpenaiSseGuard.Options.t(),
  protocol: OpenaiSseGuard.Snapshot.protocol(),
  saw_terminal_event: boolean(),
  termination: OpenaiSseGuard.Snapshot.termination()
}

Mutable-by-replacement observer state held by the caller.

Functions

finish(state)

@spec finish(t()) :: OpenaiSseGuard.Snapshot.t()

Flushes the observer and returns an immutable snapshot.

A stream that ends without a recognized terminal event is marked :unexpected_eof. An unterminated frame with visible data is treated conservatively as output evidence.

new()

Creates an observer with bounded frame and event limits.

new(options)

@spec new(keyword() | OpenaiSseGuard.Options.t()) :: t()

observe(enumerable, opts \\ [])

@spec observe(Enumerable.t(), keyword() | OpenaiSseGuard.Options.t()) ::
  OpenaiSseGuard.Snapshot.t()

Observes any enumerable of binary chunks and returns its final snapshot.

Reduction stops once the observer fails closed on a malformed or over-limit frame. A Stream remains lazy until consumed by this function.

push(state, chunk)

@spec push(t(), binary()) :: t()

Adds one byte chunk to an observer.

Chunk boundaries may split UTF-8 code points and do not need to align with SSE lines. Non-binary input raises ArgumentError so a transport bug is not mistaken for a safe replay decision.

snapshot(state)

@spec snapshot(t()) :: OpenaiSseGuard.Snapshot.t()

Returns an immutable snapshot without exposing the internal buffer.