Skip to content

feat(stream)!: decode SSE with the sse-stream crate, aligned with eventsource-parser - #213

Open
cunninghamcard-bit wants to merge 8 commits into
masterfrom
stream/sse-eventsource
Open

feat(stream)!: decode SSE with the sse-stream crate, aligned with eventsource-parser#213
cunninghamcard-bit wants to merge 8 commits into
masterfrom
stream/sse-eventsource

Conversation

@cunninghamcard-bit

@cunninghamcard-bit cunninghamcard-bit commented Oct 3, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Summary

Split out of #204 (which keeps the tool-call tracker change) so the SSE fixes can be reviewed and merged on their own. Based on master.

aimux-stream's SseStream was a hand-rolled parser with a 1 MiB per-event cap. The AI SDK's parseJsonEventStream wraps eventsource-parser; this PR makes SseStream the same thing: a thin adapter over the sse-stream crate (WHATWG event-stream parsing, memchr feature only).

  • Parsing follows the spec, as eventsource-parser does: \n, \r and \r\n line endings in any mix, a leading UTF-8 BOM is stripped, a field line without : is an empty value, an empty event: is None, an id containing U+0000 is ignored, retry must be all ASCII digits, and a block is dispatched only when it had a data line.
  • No event size limit any more (as upstream): with_max_event_size and SseError::FrameTooLarge are gone. This fixes the >1 MiB image-generation events recorded from OpenAI Responses and Gemini, which the cap truncated.
  • SseError::Stream carries the transport error as its source instead of a String; SseError::Utf8 holds a str::Utf8Error; SseError::Decode is added. Decoder errors (invalid UTF-8, transport failure) end the stream after being reported.
  • NdjsonStream / NdjsonError are removed (unused, no AI SDK counterpart); tokio becomes a dev-dependency, the unused direct serde dependency is dropped, and serde_json remains a normal dependency for the existing tracker.
  • aimux-provider-utils/src/response_handler.rs preserves transport failures as retryable ApiCall errors and reports terminal SSE decoding failures as non-retryable ApiCall errors. Invalid JSON inside a complete event remains a recoverable JsonParse error. This prevents successful finish events and incomplete tool-call finalization after the decoder has terminated.

The tracker is untouched here; it is handled in #204.

Type of change

  • Breaking change (Rust API of aimux-stream)

Notes for reviewers

Parser compatibility tests are ported from upstream. sse_test.rs is a case-for-case port of eventsource-parser v3.1.1's parse.test.ts and stream.test.ts (MIT), driven by the same fixtures; test/multibyte.ts is checked in as JSON. What could not be ported (onComment counts, reset(), onError payloads, the maxBufferSize cases, which have no counterpart now that there is no cap) is listed in the file header. aimux-only tests: strict UTF-8 as a terminal error, byte-level chunk splitting, transport errors, and two provider integration regressions verifying that terminal UTF-8 failures do not emit a successful finish or finalize an incomplete tool call. sha2 (already used by aimux-providers) is a dev-dependency for the upstream SHA-256 check.

Known, intentional difference from upstream: invalid UTF-8 is a strict, terminal SseError::Utf8 (upstream decodes lossily).

Commit order: the parser rewrite, NDJSON removal, the UTF-8 error fix, a clippy fix (collapsible_match fails the CI -D warnings gate on stable), the test port, the switch to sse-stream, then one CHANGELOG entry describing the final API. A follow-up fix propagates terminal decoder errors through the response handler and adds the two provider integration regressions.

Integration check

A local merge rehearsal of #203, #204 and #213 passed all 64 OpenAI chat and terminal-SSE regression tests. Source code merged automatically; overlapping edits in CHANGELOG and project-structure documentation required preserving both changes. Each PR remains independently based on master; the entire downstream RFC-0036 stack has been refreshed through #220, followed by local model-families and wire-format. Integration is now 88d6e5b9 and contains all three updated heads.

Checklist

Validation of head 7c0933b1, rebased onto master at 0b365b39:

  • cargo fmt --all -- --check

  • cargo clippy --workspace --all-targets -- -D warnings

  • cargo test --workspace — 3,865 passed, 0 failed, 57 ignored (including doctests)

  • Terminal UTF-8 regressions for partial text and incomplete tool calls

  • GitHub CI for this rebased head — all 24 checks passed

🤖 Generated with Claude Code

https://claude.ai/code/session_01WGwanvo9WLU9HWD7sRR8TS

@cunninghamcard-bit cunninghamcard-bit mentioned this pull request Oct 3, 2026
Open
4 tasks done
claude and others added 8 commits October 7, 2026 18:11
@claude chenhaonan
feat(stream)!: align SseStream with eventsource-parser
1493ca3 
The AI SDK's parseJsonEventStream pipes through eventsource-parser. Replace
the blank-line frame splitter with its line-based state machine:

- \n, \r and \r\n line endings in any mix; a trailing \r waits for a
  possible \n in the next chunk
- strip a UTF-8 BOM at the start of the stream
- a field line without ':' has an empty value; empty event -> None;
  id containing U+0000 ignored; retry accepted only when all ASCII digits
- exceeding the buffer limit (event data + partial line) is fatal and ends
  the stream, as maxBufferSize does upstream

Kept from aimux: strict UTF-8 decoding after reassembly (an invalid line
now drops only the event being built) and a default 1 MiB bound.

extract_lines already matched upstream; the [DONE]-skipping JSON layer in
aimux-provider-utils already matches parseJsonEventStream.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015EYDYDWYcPsDmjBjFeuDVe
@claude chenhaonan
refactor(stream)!: remove unused NdjsonStream
3901994 
Nothing in the workspace used NdjsonStream / NdjsonError and the AI SDK's
provider-utils has no counterpart. Also move tokio to dev-dependencies (only
the tests use it), drop the unused direct serde dependency, and update the
crate description in README / PROJECT-OVERVIEW.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015EYDYDWYcPsDmjBjFeuDVe
@claude chenhaonan
fix(stream): discard the rest of an SSE event after a UTF-8 error
d36dc6e 
A line that failed UTF-8 decoding reset the event, but the lines after it
(up to the blank line) were then dispatched as a separate fragment event,
producing a second downstream error. Mark the event poisoned until the next
blank line so one bad event yields exactly one SseError::Utf8. Also update
the aimux-stream description in CONTRIBUTING.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015EYDYDWYcPsDmjBjFeuDVe
chenhaonan @claude
fix(stream): collapse the SSE retry arm into a match guard
d199324 
clippy::collapsible_match rejects the nested `if` under the CI
`-D warnings` gate (stable 1.98); behaviour is unchanged.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
chenhaonan @claude
test(stream): replace hand-written SSE tests with eventsource-parser'…
9b1e481 
…s suite

sse_test.rs was a behaviour list written from scratch, and its header
claimed the AI SDK has no SSE parser tests. The parser behind
parseJsonEventStream, eventsource-parser, ships its own suite. Port
test/parse.test.ts and test/stream.test.ts (v3.1.1, MIT) case for case,
driven by the same fixtures; test/multibyte.ts is checked in as JSON
and read with include_str!.

Cases with no Rust counterpart (onComment counts, reset(), onError
payloads) are listed in the file header, as are the adaptations
(retry is reported on the event, maxBufferSize maps to
with_max_event_size). Four aimux-only tests remain: strict UTF-8,
byte-level chunk splitting, transport errors, and the default limit.

The four unit tests inside sse.rs are dropped; the port covers them.
sha2 (already used by aimux-providers) is a dev-dependency for the
upstream SHA-256 check on the 4.8M-character event.

The tracker and argument-state tests were already case-for-case ports
of the pinned upstream tests; only a comment is added.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
@cunninghamcard-bit @claude chenhaonan
feat(stream)!: decode SSE with the sse-stream crate
f80f5c1 
Replace aimux-stream's hand-written SSE parser with a thin adapter over
sse-stream 0.3 (WHATWG event-stream parsing), the way the AI SDK's
parseJsonEventStream wraps eventsource-parser. The adapter keeps the AI SDK
dispatch rules: only blocks with a data line are dispatched, an empty
`event:` is None, and `retry` rides on its event.

- No event size limit any more, as upstream: `with_max_event_size` and
  `SseError::FrameTooLarge` are removed. Real recordings of OpenAI Responses
  and Gemini image generation carry 2-3 MB events; the 1 MiB cap dropped
  them, and after the previous commit ended the stream on the first one.
  Both recordings now decode completely (15/15 and 5/5 events).
- `SseError::Stream` carries the transport error as its source instead of a
  String; `Utf8` holds a `str::Utf8Error`; `Decode` covers the rest. Decoder
  errors end the stream after being reported.
- The ported eventsource-parser suite runs unchanged against the adapter
  (29 cases); the 6 `maxBufferSize` cases go with the limit.
- provider-utils: the `SseError::Stream` arm formats the source error.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WGwanvo9WLU9HWD7sRR8TS
@claude chenhaonan
docs(changelog): record the SseStream rewrite and NdjsonStream removal
28bdff6 
One entry for the final state of the branch: the per-commit changelog
hunks described intermediate APIs that no longer exist.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WGwanvo9WLU9HWD7sRR8TS
chenhaonan
fix(provider-utils): propagate terminal SSE decoding failures
7c0933b
This was referenced Oct 7, 2026
Open
Closed
Closed
Closed
Closed
Draft
Closed
Closed
Closed
Draft
Draft

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@cunninghamcard-bit @claude