openai_sse_guard 0.1.1

Bounded SSE stream observer for conservative replay decisions in OpenAI-compatible Dart clients.

openai_sse_guard

openai_sse_guard is a dependency-free Dart observer for OpenAI-compatible Server-Sent Events (SSE). It answers one deliberately narrow question: what evidence exists about a stream when the byte connection ends, and is an automatic replay still defensible?

It does not make HTTP requests, retry them, retain generated text, or decide whether a user was charged. Your application owns transport, cancellation, idempotency, billing, and the final replay policy.

Why observe the stream?

An HTTP 200 response is not proof that a generation completed. A connection can disappear after a visible delta, a gateway can emit an error event, or a provider can close without [DONE]/response.completed. Replaying blindly in those states can duplicate visible output or incur a second charge. This package records bounded evidence while the caller renders bytes as it chooses.

Install

dart pub add openai_sse_guard

Then import it:

import 'package:openai_sse_guard/openai_sse_guard.dart';

Run the complete example/example.dart program with:

dart run example/example.dart

It feeds a bounded Responses stream to observeSse and prints the terminal snapshot. The example uses the same package: import path as an external consumer.

Observe an HTTP byte stream

final observer = SseReplayObserver();

await for (final chunk in responseByteStream) {
  observer.add(chunk); // List<int>, including chunks split in UTF-8
  renderChunk(chunk);  // The application controls user-visible output.
}

final state = observer.finish();
if (state.termination == StreamTermination.unexpectedEof &&
    state.hasOutput) {
  // A request may already have generated content. Do not replay blindly.
  throw StateError('stream ended after partial output');
}

For a complete Dart Stream<List<int>>, use observeSse(stream) instead. Utf8Decoder.startChunkedConversion preserves code points split across HTTP chunks. SSE frames are bounded before they can grow an in-memory buffer.

State contract

StreamSnapshot is immutable and contains only bounded metadata:

Field	Meaning
protocol	chatCompletions, responses, or unknown
termination	done, incomplete, error, unexpectedEof, or open
hasOutput	A visible delta, tool call, or unknown data-bearing event was seen
sawTerminalEvent	[DONE], completed, or incomplete terminal event was seen
eventCount	Complete frames accepted within the configured event limit
malformedEventCount	Invalid JSON, UTF-8, or over-limit frames
lastEventType	A short, allow-listed event identifier (not provider prose)
errorCode	A short provider error code/type, if present

The observer never stores response text, error messages, prompts, or complete JSON payloads. Defaults are a 64 KiB frame and 10,000 events; both can be lowered for a tighter memory budget.

Protocol and retry boundaries

The framing follows the WHATWG Server-Sent Events specification. Chat Completions choices[].delta.content and tool calls count as output. Responses output events such as response.output_text.delta count as output; response.completed is done and response.incomplete is explicitly incomplete. [DONE] is recognized as the conventional Chat Completions marker. Malformed data-bearing frames fail closed by setting hasOutput, while EOF without a terminal marker becomes unexpectedEof.

The observer is not a retry policy. Combine its state with your operation's idempotency key, whether any bytes were rendered, provider billing semantics, and the HTTP status. For provider error categories, see the OpenAI error-code guide and for server retry hints the MDN Retry-After reference.

The AI-ROUTER API gateway is one possible OpenAI-compatible endpoint context. This package is provider-neutral and is not affiliated with or endorsed by OpenAI.

Related implementations and evidence

The repository's stream replay-safety guide shows a decision table and fixtures. Teams using another language can compare the maintained JavaScript package on npm, Python package on PyPI, Ruby package on RubyGems, PHP package on Packagist, Rust stream guard on crates.io, and Deno package on JSR.

Development

dart pub get
dart analyze
dart test

See CONTRIBUTING.md, the security policy, and doc/stream-replay-safety.md before changing parsing or state semantics.

MIT licensed. Maintained by AI-ROUTER contributors.