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

CHANGELOG.md

@@ -5,6 +5,35 @@ All notable changes to aimux are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Breaking
+
+**Rust (aimux-stream)**
+
+- `SseStream` is now a thin adapter over the `sse-stream` crate (WHATWG
+  event-stream parsing), mirroring how the AI SDK's `parseJsonEventStream`
+  wraps `eventsource-parser`: `\n`, `\r` and `\r\n` line endings in any
+  mix, a leading UTF-8 BOM is stripped, a field line without `:` counts as 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. There is no event size limit any more (as
+  upstream): `with_max_event_size` and `SseError::FrameTooLarge` are gone,
+  which fixes streams carrying multi-megabyte events (OpenAI Responses and
+  Gemini image generation). `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) now end the stream after being reported.
+  The response handler propagates terminal SSE decoding failures as
+  non-retryable `ApiCall` errors, preventing a normal finish or tool-call
+  finalization after truncated input. Invalid JSON inside an otherwise valid
+  SSE event remains a recoverable `JsonParse` error.
+  `SseStream` requires the body error type to implement `std::error::Error +
+  Send + Sync + 'static`.
+- Removed `NdjsonStream` / `NdjsonError`: nothing in the workspace used them
+  and the AI SDK has no counterpart. `tokio` is now a dev-dependency only and
+  the unused direct `serde` dependency is dropped.
+
 ## [0.5.0] - 2026-09-27
 
 **Breaking release.** 13 PRs since 0.3.0: the cross-language error model

CONTRIBUTING.md

@@ -9,7 +9,7 @@ to set up a development environment, run the tests, and submit changes.
 aimux/
 ├── aimux-core/            # Core abstractions: LanguageModel / Provider / Message / StreamPart
 ├── aimux-providers/       # provider implementations + cassettes (counts: docs/api/providers.md)
-├── aimux-stream/          # SSE / NDJSON stream parsing
+├── aimux-stream/          # SSE parsing, streamed tool-call tracking
 ├── aimux-provider-utils/  # HTTP utilities: retry, backoff, error parsing, API-key loading
 ├── aimux-ffi/             # C ABI (opaque handle + JSON + push callback) for non-native bindings
 ├── bindings/              # Node, Python, Swift, Kotlin, Flutter, Go, C — share one Rust core

README.md

@@ -98,7 +98,7 @@ middleware, and telemetry per request).
 aimux/
 ├── aimux-core            # Core abstractions: LanguageModel / Provider / Message / StreamPart
 ├── aimux-providers       # Provider implementations — registry-backed + typed (docs/api/providers.md)
-├── aimux-stream          # SSE / NDJSON stream parsing
+├── aimux-stream          # SSE parsing, streamed tool-call tracking
 ├── aimux-provider-utils  # One-exchange HTTP helpers, response handlers, API-key loading
 ├── aimux-ffi             # C ABI (opaque handles + JSON results + owned aimux_error_t *) for non-native bindings
 └── tools/                # aimux-cli (cache probe) · aimux-replay · aimux-web (console)
@@ -122,7 +122,7 @@ cargo add aimux-core aimux-providers
 |-------|-------------|-----------|
 | `aimux-core` | Core abstractions: `LanguageModel` / `Provider` / `Message` / `StreamPart` | [crates.io](https://crates.io/crates/aimux-core) |
 | `aimux-providers` | Provider implementations — [registry-backed + typed](docs/api/providers.md) | [crates.io](https://crates.io/crates/aimux-providers) |
-| `aimux-stream` | SSE / NDJSON stream parsing | [crates.io](https://crates.io/crates/aimux-stream) |
+| `aimux-stream` | SSE parsing, streamed tool-call tracking | [crates.io](https://crates.io/crates/aimux-stream) |
 | `aimux-provider-utils` | One-exchange HTTP helpers and typed response handlers | [crates.io](https://crates.io/crates/aimux-provider-utils) |
 | `aimux-ffi` | C ABI for non-native bindings | [crates.io](https://crates.io/crates/aimux-ffi) |
 

aimux-provider-utils/src/response_handler.rs

@@ -340,6 +340,9 @@ pub fn create_binary_response_handler() -> ResponseHandler<Bytes> {
 }
 
 /// Parse an SSE response and deserialize each `data:` field as `T`.
+///
+/// Invalid JSON in a complete event remains a recoverable frame error. SSE
+/// decoding failures terminate the byte stream and are non-retryable API errors.
 #[must_use]
 pub fn create_event_source_response_handler<T>()
 -> ResponseHandler<BoxStream<'static, Result<T, AiMuxError>>>
@@ -392,21 +395,35 @@ where
                 match item {
                     Ok(event) if event.data == "[DONE]" => continue,
                     Ok(event) => yield serde_json::from_str::<T>(&event.data).map_err(AiMuxError::from),
-                    Err(aimux_stream::SseError::Stream(message)) => {
+                    Err(aimux_stream::SseError::Stream(error)) => {
                         // Preserve response transport failures as ApiCallError
-                        // items. Framing/parser failures remain JsonParse below.
+                        // items, eligible for retry before output is emitted.
                         yield Err(AiMuxError::ApiCall(Box::new(ApiCallError {
                             response_headers: Some(stream_headers.clone()),
                             is_retryable: true,
                             ..ApiCallError::new(
-                                message,
+                                error.to_string(),
+                                stream_url.clone(),
+                                stream_request_body_values.clone(),
+                            )
+                        })));
+                        break;
+                    }
+                    Err(error) => {
+                        // The SSE decoder ends after this error. Reporting a
+                        // recoverable JsonParse would let providers treat that
+                        // EOF as success and flush incomplete output/tool calls.
+                        yield Err(AiMuxError::ApiCall(Box::new(ApiCallError {
+                            response_headers: Some(stream_headers.clone()),
+                            is_retryable: false,
+                            ..ApiCallError::new(
+                                error.to_string(),
                                 stream_url.clone(),
                                 stream_request_body_values.clone(),
                             )
                         })));
                         break;
                     }
-                    Err(error) => yield Err(AiMuxError::JsonParse(error.to_string())),
                 }
             }
         });

aimux-providers/tests/sse_terminal_error_test.rs

@@ -0,0 +1,98 @@
+//! Integration checks for aimux's strict UTF-8 decoding (unlike upstream's
+//! lossy decoder): a terminal byte error must stay terminal through the
+//! response handler and provider, without a successful finish or tool flush.
+
+use aimux_core::error::AiMuxError;
+use aimux_core::language_model::LanguageModel;
+use aimux_core::options::CallOptions;
+use aimux_core::stream_part::StreamPart;
+use aimux_providers::{OpenAIConfig, OpenAIProvider};
+use futures::StreamExt;
+use serde_json::{Value, json};
+use wiremock::matchers::{method, path};
+use wiremock::{Mock, MockServer, ResponseTemplate};
+
+async fn corrupted_stream(prefix: Value) -> Vec<Result<StreamPart, AiMuxError>> {
+    let server = MockServer::start().await;
+    let suffix = json!({
+        "choices": [{"delta": {"content": "world"}, "finish_reason": "stop"}]
+    });
+    let mut body = format!("data: {prefix}\n\n").into_bytes();
+    body.extend_from_slice(b"data: \xff\n\n");
+    body.extend_from_slice(format!("data: {suffix}\n\ndata: [DONE]\n\n").as_bytes());
+    Mock::given(method("POST"))
+        .and(path("/chat/completions"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_bytes(body),
+        )
+        .mount(&server)
+        .await;
+    let provider = OpenAIProvider::new(OpenAIConfig::new("test-key").with_base_url(server.uri()));
+    provider
+        .model("gpt-4o")
+        .do_stream(&CallOptions::new(vec![]))
+        .await
+        .expect("the first event is valid")
+        .stream
+        .collect()
+        .await
+}
+
+fn assert_terminal_decode_error(parts: &[Result<StreamPart, AiMuxError>]) {
+    let errors: Vec<_> = parts
+        .iter()
+        .filter_map(|part| part.as_ref().err())
+        .collect();
+    assert_eq!(errors.len(), 1);
+    assert!(matches!(errors[0], AiMuxError::ApiCall(_)));
+    assert!(!errors[0].is_recoverable_stream_error());
+    assert!(!errors[0].is_retryable());
+    assert!(parts.last().is_some_and(Result::is_err));
+    assert!(
+        !parts
+            .iter()
+            .any(|part| matches!(part, Ok(StreamPart::Finish { .. })))
+    );
+}
+
+#[tokio::test]
+async fn invalid_utf8_after_text_does_not_emit_a_successful_finish() {
+    let parts = corrupted_stream(json!({
+        "choices": [{"delta": {"content": "hello "}}]
+    }))
+    .await;
+    let text: String = parts
+        .iter()
+        .filter_map(|part| match part {
+            Ok(StreamPart::TextDelta { delta, .. }) => Some(delta.as_str()),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(text, "hello ");
+    assert_terminal_decode_error(&parts);
+}
+
+#[tokio::test]
+async fn invalid_utf8_does_not_finalize_an_incomplete_tool_call() {
+    let parts = corrupted_stream(json!({
+        "choices": [{"delta": {"tool_calls": [{
+            "index": 0,
+            "id": "call-1",
+            "type": "function",
+            "function": {"name": "lookup", "arguments": "{"}
+        }]}}]
+    }))
+    .await;
+    assert!(
+        parts
+            .iter()
+            .any(|part| matches!(part, Ok(StreamPart::ToolInputStart { .. })))
+    );
+    assert!(!parts.iter().any(|part| matches!(
+        part,
+        Ok(StreamPart::ToolCall { .. } | StreamPart::ToolInputEnd { .. })
+    )));
+    assert_terminal_decode_error(&parts);
+}

aimux-stream/Cargo.toml

@@ -4,18 +4,21 @@ version.workspace = true
 edition.workspace = true
 license.workspace = true
 description = "Streaming primitives for aimux"
-keywords = ["llm", "streaming", "sse", "ndjson"]
+keywords = ["llm", "streaming", "sse", "tool-calls"]
 categories = ["asynchronous", "parser-implementations"]
 documentation = "https://docs.rs/aimux-stream"
 
 [dependencies]
 futures = { workspace = true }
-tokio = { workspace = true }
 pin-project-lite = { workspace = true }
 bytes = { workspace = true }
-serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
+sse-stream = { version = "0.3.0", default-features = false, features = ["memchr"] }
+
+[dev-dependencies]
+sha2 = "0.10"
+tokio = { workspace = true }
 
 [lints]
 workspace = true

aimux-stream/src/lib.rs

@@ -1,16 +1,15 @@
 //! # aimux-stream
 //!
-//! Low-level streaming primitives for SSE (Server-Sent Events) and NDJSON parsing,
+//! Low-level streaming primitives for SSE (Server-Sent Events) parsing and
+//! streamed tool-call tracking,
 //! used by provider implementations to decode model API response streams.
 
 pub mod lines;
-pub mod ndjson;
 pub mod sse;
 pub mod streaming_tool_call_tracker;
 
 // Re-export the most commonly used items.
 pub use lines::extract_lines;
-pub use ndjson::{NdjsonError, NdjsonStream};
 pub use sse::{SseError, SseEvent, SseStream};
 pub use streaming_tool_call_tracker::{
     StreamingToolCallDelta, StreamingToolCallFunction, StreamingToolCallTracker,