SSE stream error events are silently discarded — add error detection to streaming pipeline

[Kamilbenkirane]

Bug

When a provider sends an error event mid-stream (e.g., Anthropic's {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}), the error is silently discarded. The consumer receives partial content with no indication that an error occurred.

Reproduction

The silent discard path:

1. stream_post() in http.py yields the error event (valid JSON ✅)
2. Provider's _parse_chunk_content() → None (not a recognized content event type)
3. Provider's _parse_chunk_usage() → None
4. Provider's _parse_chunk_finish_reason() → None
5. _parse_chunk() → None (all three returned None)
6. Stream.__anext__() silently skips it (if chunk is not None: ...)
7. Error vanishes — never logged, never raised, never stored in metadata

This affects all SSE-based text providers: Anthropic, OpenAI, Google, Cohere, DeepSeek, Groq, Mistral, Moonshot, xAI, OpenResponses.

Additional Notes

• SSE_EVENT_ERROR = "error" is defined in src/celeste/providers/anthropic/messages/config.py (line 51) but never imported or referenced anywhere — dead code.
• http.py's stream_post() does not call raise_for_status() before iterating SSE events (unlike stream_post_ndjson which does).
• The only provider with any error detection is BytePlus Images (stuffs error into chunk metadata, doesn't raise) and Gradium TTS (raises, but WebSocket-based, not SSE).

Proposed Fix

Add error detection to the base streaming pipeline. Two options:

Option A — Provider streaming mixin (recommended):
Each provider's streaming mixin adds a _handle_stream_error(event_data) method that checks for provider-specific error event types and raises. Called at the start of _parse_chunk().

Option B — Base Stream.__anext__:
Add a generic check before _parse_chunk() that looks for common error patterns (event.get("error"), event.get("type") == "error").

Option A is preferred because error event formats differ across providers (Anthropic uses {"type": "error"}, OpenAI uses different patterns, etc.).

Affected Files

• src/celeste/streaming.py — optionally add base error hook
• src/celeste/providers/*/streaming.py — add _handle_stream_error() to each provider's streaming mixin
• src/celeste/http.py — consider adding raise_for_status() to stream_post() for HTTP-level errors

🤖 Generated with Claude Code

[Kamilbenkirane]

Proposed approach

Bug confirmed. Mid-stream error events from providers are silently discarded — _parse_chunk() returns None for unrecognized events, so the consumer gets partial content with no indication of failure.

Approach: ClassVar-driven base implementation

Following the patterns established in PRs #160, #161, and #172, we can handle this with minimal duplication.

Two generic SSE error patterns cover all providers:

• Type-based: {"type": "error", "error": {"type": "...", "message": "..."}}
• Field-based: {"error": {"message": "...", "type": "..."}}

The base Stream class handles both via a _error_type_fields ClassVar that controls how the error type is extracted from the error dict. Only 2 outlier providers (OpenResponses, Google Interactions) override the method entirely.

Changes summary

1. StreamEventError exception — new exception in celeste.exceptions
2. Core Stream — _error_type_fields: ClassVar + base _parse_stream_error() + call in _parse_chunk()
3. Google GenerateContent — ClassVar override: _error_type_fields = ("status", "code")
4. Google Interactions — method override (dual event_type/type field — unique to this provider)
5. OpenResponses — method override (reads code/message from root level)
6. Delete 4 redundant implementations — ChatCompletions, Anthropic, Cohere, OpenAI Images (all covered by base)
7. HTTP raise_for_status() — added to stream_post() for HTTP-level errors
8. Tests — unit tests for error detection

This keeps provider-specific knowledge out of the core (only generic SSE patterns), follows the ClassVar consolidation pattern, and provides a built-in safety net for new providers.