Fix pre-start stream failures returning HTTP 200 (#1026)

## Problem

Provider streams could commit HTTP 200 before an upstream-backed first
SSE frame was available. When setup or retry failed before usable stream
output, Claude and Codex saw a successful but broken stream instead of a
retryable non-200 error.

## Changes

| Before | After |
| --- | --- |
| API egress returned `StreamingResponse` before probing the provider
iterator. | API egress waits for the first chunk before committing
success headers. |
| Pre-start provider failures became synthetic SSE success streams. |
Pre-start provider failures raise typed errors and return Anthropic or
OpenAI JSON with non-200 status. |
| Post-start unexpected stream failures could truncate protocol output.
| Post-start failures emit terminal Anthropic error or Responses
`response.failed` frames where possible. |
| Provider tests expected pre-start final failures as SSE tails. |
Provider tests assert typed pre-start errors and preserve midstream and
tool-salvage behavior. |

<!-- greptile_comment -->

<details open><summary><h3>Greptile Summary</h3></summary>

This PR changes streaming responses so HTTP success is not committed
before the first protocol chunk. The main changes are:

- First-chunk gated streaming helpers for Anthropic and OpenAI Responses
egress.
- Non-200 JSON error responses for provider failures before stream
output starts.
- Terminal Anthropic `error` and Responses `response.failed` frames for
post-start interruptions.
- Provider transport updates that raise typed pre-start errors while
preserving retry, recovery, and tool salvage paths.
- Targeted API and provider tests plus a patch version and lockfile
update.
</details>

<h3>Confidence Score: 5/5</h3>

Safe to merge with low risk.

The changed paths keep provider retry and recovery ownership in
transports, gate HTTP success before the first chunk, and preserve
cancellation behavior. Tests cover the main Anthropic and OpenAI
Responses pre-start and post-start failure paths.

No files require special attention.

<details><summary><h3><a href="https://www.greptile.com/trex"><img
alt="T-Rex"
src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg"
height="20" align="absmiddle"></a> T-Rex Logs</h3></summary>

**What T-Rex did**
- Ran the pre-change stream-gating API tests against the previous commit
98515be55a98ca2bd0be78434c4897c33e3296d8, and they passed 35 tests in
5.08s (EXIT\_CODE: 0).
- Ran the post-change stream-gating API tests against the updated HEAD
d65f4ca107a024790df03cb4d30d6346027597b8, and they passed 35 tests in
2.27s (EXIT\_CODE: 0).
- Ran the focused API test suite against the current code and it passed
35 tests in 3.85s (EXIT\_CODE: 0).

<a
href="https://app.greptile.com/trex/runs/13779430/artifacts"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifactsDark.svg?v=4"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"><img
alt="View all artifacts"
src="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"></picture></a>

_{<a href="https://www.greptile.com/trex"><img alt="T-Rex"
src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg"
height="14" align="absmiddle"></a> Ran code and verified through
T-Rex}
</details>

<details open><summary><h3>Important Files Changed</h3></summary>

| Filename | Overview |
|----------|----------|
| api/response_streams.py | Introduces protocol-agnostic first-chunk
gating and Anthropic post-start terminal error fallback. |
| api/handlers/messages.py | Awaits first-chunk gated Anthropic stream
responses and maps pre-start provider or unexpected failures to JSON
errors. |
| api/handlers/responses.py | Adds first-chunk gated Responses streaming
and OpenAI-shaped pre-start error serialization. |
| providers/error_mapping.py | Adds shared mapping from final pre-start
stream exceptions to HTTP-serializable provider errors. |
| providers/transports/openai_chat/stream.py | Raises mapped provider
errors for uncommitted OpenAI-chat failures while preserving recovery
and complete tool salvage paths. |
| providers/transports/anthropic_messages/stream.py | Raises mapped
provider errors for uncommitted native stream failures while preserving
committed and salvageable tails. |
| core/openai_responses/stream.py | Preserves pre-start exceptions while
converting post-start Anthropic stream failures into same-assembler
`response.failed` events. |
| tests/api/test_response_streams.py | Adds unit coverage for
first-chunk gating, pre-start error JSON, and post-start terminal
Anthropic frames. |
| tests/api/test_openai_responses.py | Covers Responses pre-start
provider errors and same-id post-start failure frames. |
| tests/providers/test_openai_compat_5xx_retry.py | Covers exhausted
OpenAI-compatible 5xx and connection retries raising provider errors
before stream commit. |
| pyproject.toml | Bumps the package patch version for production
behavior changes. |
| uv.lock | Refreshes the lockfile package version to match the patch
bump. |

</details>

<details open><summary><h3>Sequence Diagram</h3></summary>

<a href="#gh-light-mode-only">

```mermaid
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant Client
participant Handler as API Handler
participant Egress as api/response_streams.py
participant Provider as Provider Stream
participant Assembler as Responses Assembler

Client->>Handler: POST /v1/messages or /v1/responses
Handler->>Provider: create async SSE iterator
Handler->>Egress: await first-chunk gated response
Egress->>Provider: anext(body)
alt Provider fails before first chunk
Provider-->>Egress: ProviderError / exception
Egress-->>Handler: protocol JSON error response
Handler-->>Client: non-200 JSON
else First protocol chunk is available
Provider-->>Egress: first SSE chunk
Egress-->>Client: HTTP 200 StreamingResponse
Egress-->>Client: replay first chunk and tail
alt Anthropic post-start failure
Provider-->>Egress: exception
Egress-->>Client: terminal event: error
else Responses post-start failure
Provider-->>Assembler: exception after response.created
Assembler-->>Client: response.failed with same response.id
end
end
```

</a>
<a href="#gh-dark-mode-only">

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant Client
participant Handler as API Handler
participant Egress as api/response_streams.py
participant Provider as Provider Stream
participant Assembler as Responses Assembler

Client->>Handler: POST /v1/messages or /v1/responses
Handler->>Provider: create async SSE iterator
Handler->>Egress: await first-chunk gated response
Egress->>Provider: anext(body)
alt Provider fails before first chunk
Provider-->>Egress: ProviderError / exception
Egress-->>Handler: protocol JSON error response
Handler-->>Client: non-200 JSON
else First protocol chunk is available
Provider-->>Egress: first SSE chunk
Egress-->>Client: HTTP 200 StreamingResponse
Egress-->>Client: replay first chunk and tail
alt Anthropic post-start failure
Provider-->>Egress: exception
Egress-->>Client: terminal event: error
else Responses post-start failure
Provider-->>Assembler: exception after response.created
Assembler-->>Client: response.failed with same response.id
end
end
```

</a>
</details>

_{Reviews (1): Last reviewed commit: ["Fix pre-start stream failure
status"](https://github.com/alishahryar1/free-claude-code/commit/d65f4ca107a024790df03cb4d30d6346027597b8)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=42891757)}

<!-- /greptile_comment -->

ARCHITECTURE.md

@@ -247,6 +247,12 @@ owns Anthropic token counting. Shared provider execution lives in
 [api/provider_execution.py](api/provider_execution.py), which resolves a
 provider, preflights the upstream request, emits trace events, counts input
 tokens, and returns an Anthropic SSE iterator.
+[api/response_streams.py](api/response_streams.py) owns public streaming egress
+commit timing. It waits for the first protocol chunk before returning a
+successful `StreamingResponse`, so provider setup failures can still become real
+non-200 JSON errors that Claude Code and Codex can retry. After the first chunk
+has escaped, HTTP status is committed; any unexpected failure must be represented
+as a protocol terminal frame where feasible.
 
 ```mermaid
 sequenceDiagram
@@ -438,6 +444,11 @@ classification also lives in this shared layer so stream recovery, provider
 backoff, and provider error mapping agree on retryable overload/rate-limit
 signals.
 
+Provider transports raise typed provider errors for final stream failures before
+any downstream-visible SSE chunk has escaped the recovery holdback. Once output
+has committed, transports keep ownership of midstream recovery, continuation,
+tool salvage, and protocol-specific success/error tails.
+
 [core/openai_responses/](core/openai_responses/) owns OpenAI Responses support:
 
 - the `OpenAIResponsesAdapter` facade used by the API layer;
@@ -460,6 +471,10 @@ builders, and error mapping. API code should depend on the adapter, not on
 those internal module owners directly. Responses output payloads stay
 OpenAI-shaped; Anthropic terminal metadata is used internally only when it
 affects streamed behavior.
+Post-start Responses failures are assembler-owned: the active
+`ResponsesStreamAssembler` emits `response.failed` so the terminal event keeps
+the same `response.id`, output ledger, and usage state as the earlier
+`response.created`.
 
 Responses custom tools are also boundary-owned. The adapter accepts native
 Responses `custom` tool declarations, represents them internally as Anthropic

api/handlers/messages.py

@@ -11,8 +11,16 @@ from api.model_router import ModelRouter, RoutedMessagesRequest
 from api.models.anthropic import MessagesRequest
 from api.optimization_handlers import try_optimizations
 from api.provider_execution import ProviderExecutionService, TokenCounter
-from api.request_errors import require_non_empty_messages, unexpected_http_exception
-from api.response_streams import anthropic_sse_streaming_response
+from api.request_errors import (
+    http_status_for_unexpected_api_exception,
+    log_unexpected_api_exception,
+    require_non_empty_messages,
+    unexpected_http_exception,
+)
+from api.response_streams import (
+    EmptyStreamError,
+    anthropic_sse_streaming_response,
+)
 from api.web_tools.egress import WebFetchEgressPolicy, web_fetch_allowed_scheme_set
 from api.web_tools.request import (
     is_web_server_tool_request,
@@ -21,7 +29,11 @@ from api.web_tools.request import (
 from api.web_tools.streaming import stream_web_server_tool_response
 from config.provider_catalog import PROVIDER_CATALOG
 from config.settings import Settings
-from core.anthropic import aggregate_anthropic_sse_to_message, get_token_count
+from core.anthropic import (
+    aggregate_anthropic_sse_to_message,
+    get_token_count,
+    get_user_facing_error_message,
+)
 from core.trace import trace_event
 from providers.base import BaseProvider
 from providers.exceptions import InvalidRequestError, ProviderError
@@ -48,6 +60,12 @@ _MessagesResult = _MessagesStreamResult | _MessagesCompleteResult
 MessageIntercept = Callable[[RoutedMessagesRequest], _MessagesResult | None]
 
 
+def _unexpected_stream_error_message(exc: BaseException) -> str:
+    if isinstance(exc, Exception):
+        return get_user_facing_error_message(exc)
+    return str(exc).strip() or f"{type(exc).__name__} occurred."
+
+
 class MessagesHandler:
     """Handle Anthropic-compatible Messages requests."""
 
@@ -116,7 +134,36 @@ class MessagesHandler:
                     content={"type": "error", "error": error},
                 )
             return JSONResponse(content=message)
-        return anthropic_sse_streaming_response(result.body)
+        return await anthropic_sse_streaming_response(
+            result.body,
+            pre_start_error_response=self._pre_start_error_response,
+        )
+
+    def _pre_start_error_response(self, exc: BaseException) -> JSONResponse:
+        if isinstance(exc, ProviderError):
+            return JSONResponse(
+                status_code=exc.status_code,
+                content=exc.to_anthropic_format(),
+            )
+        log_unexpected_api_exception(
+            self._settings,
+            exc,
+            context=(
+                "CREATE_MESSAGE_EMPTY_STREAM"
+                if isinstance(exc, EmptyStreamError)
+                else "CREATE_MESSAGE_STREAM_START_ERROR"
+            ),
+        )
+        return JSONResponse(
+            status_code=http_status_for_unexpected_api_exception(exc),
+            content={
+                "type": "error",
+                "error": {
+                    "type": "api_error",
+                    "message": _unexpected_stream_error_message(exc),
+                },
+            },
+        )
 
     def _reject_unsupported_server_tools(self, routed: RoutedMessagesRequest) -> None:
         if routed.resolved.provider_id not in _OPENAI_CHAT_UPSTREAM_IDS:

api/handlers/responses.py

@@ -13,7 +13,10 @@ from api.request_errors import (
     log_unexpected_api_exception,
     require_non_empty_messages,
 )
-from api.response_streams import openai_responses_sse_streaming_response
+from api.response_streams import (
+    EGRESS_STREAM_INTERRUPTED_MESSAGE,
+    openai_responses_sse_streaming_response,
+)
 from config.settings import Settings
 from core.anthropic import get_user_facing_error_message
 from core.openai_responses import OpenAIResponsesAdapter
@@ -23,6 +26,12 @@ from providers.exceptions import InvalidRequestError, ProviderError
 ProviderGetter = Callable[[str], BaseProvider]
 
 
+def _unexpected_stream_error_message(exc: BaseException) -> str:
+    if isinstance(exc, Exception):
+        return get_user_facing_error_message(exc)
+    return str(exc).strip() or f"{type(exc).__name__} occurred."
+
+
 class ResponsesHandler:
     """Handle streaming OpenAI Responses-compatible requests."""
 
@@ -72,12 +81,14 @@ class ResponsesHandler:
                 raw_log_label="FULL_RESPONSES_PAYLOAD",
                 raw_log_payload=request_payload,
             )
-            return openai_responses_sse_streaming_response(
+            return await openai_responses_sse_streaming_response(
                 self._responses_adapter.iter_sse_from_anthropic(
                     streamed,
                     request_payload,
+                    stream_error_message=EGRESS_STREAM_INTERRUPTED_MESSAGE,
                 ),
                 headers=self._responses_adapter.sse_headers,
+                pre_start_error_response=self._pre_start_error_response,
             )
         except OpenAIResponsesAdapter.ConversionError as exc:
             invalid_request = InvalidRequestError(str(exc))
@@ -109,3 +120,25 @@ class ResponsesHandler:
                     error_type="api_error",
                 ),
             )
+
+    def _pre_start_error_response(self, exc: BaseException) -> JSONResponse:
+        if isinstance(exc, ProviderError):
+            return JSONResponse(
+                status_code=exc.status_code,
+                content=self._responses_adapter.error_payload(
+                    message=exc.message,
+                    error_type=exc.error_type,
+                ),
+            )
+        log_unexpected_api_exception(
+            self._settings,
+            exc,
+            context="CREATE_RESPONSE_STREAM_START_ERROR",
+        )
+        return JSONResponse(
+            status_code=http_status_for_unexpected_api_exception(exc),
+            content=self._responses_adapter.error_payload(
+                message=_unexpected_stream_error_message(exc),
+                error_type="api_error",
+            ),
+        )

api/response_streams.py

@@ -1,29 +1,123 @@
 """FastAPI streaming response wrappers for public API wire formats."""
 
-from collections.abc import AsyncIterator, Mapping
+import asyncio
+from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping
 
-from fastapi.responses import StreamingResponse
+from fastapi.responses import Response, StreamingResponse
 
-from core.anthropic.streaming import ANTHROPIC_SSE_RESPONSE_HEADERS
+from core.anthropic.streaming import (
+    ANTHROPIC_SSE_RESPONSE_HEADERS,
+    anthropic_terminal_error_frame,
+)
+from core.trace import trace_event
 
+EGRESS_STREAM_INTERRUPTED_MESSAGE = (
+    "The upstream response stream ended unexpectedly; the request could not be "
+    "completed."
+)
+
+PreStartErrorResponse = Callable[[BaseException], Response]
+TerminalFrameEmitter = Callable[[BaseException], str]
+
+
+class EmptyStreamError(RuntimeError):
+    """Raised when a public stream ends before emitting any protocol chunk."""
+
+
+def _trace_egress_failure(exc: BaseException) -> None:
+    trace_event(
+        stage="egress",
+        event="api.response.egress_error_frame_emitted",
+        source="api",
+        exc_type=type(exc).__name__,
+    )
+
+
+async def _first_chunk_streaming_response(
+    body: AsyncIterator[str],
+    *,
+    headers: Mapping[str, str],
+    pre_start_error_response: PreStartErrorResponse,
+    terminal_frame: TerminalFrameEmitter | None,
+) -> Response:
+    try:
+        first_chunk = await anext(body)
+    except StopAsyncIteration:
+        return pre_start_error_response(
+            EmptyStreamError("Stream ended before emitting a response.")
+        )
+    except GeneratorExit:
+        raise
+    except asyncio.CancelledError:
+        raise
+    except BaseExceptionGroup as exc:
+        return pre_start_error_response(exc)
+    except Exception as exc:
+        return pre_start_error_response(exc)
 
-def anthropic_sse_streaming_response(body: AsyncIterator[str]) -> StreamingResponse:
-    """Return a streaming response for Anthropic-style SSE streams."""
     return StreamingResponse(
-        body,
+        _replay_first_chunk_then_stream(
+            first_chunk,
+            body,
+            terminal_frame=terminal_frame,
+        ),
         media_type="text/event-stream",
+        headers=dict(headers),
+    )
+
+
+async def _replay_first_chunk_then_stream(
+    first_chunk: str,
+    body: AsyncIterator[str],
+    *,
+    terminal_frame: TerminalFrameEmitter | None,
+) -> AsyncGenerator[str]:
+    yield first_chunk
+    try:
+        async for chunk in body:
+            yield chunk
+    except GeneratorExit:
+        raise
+    except asyncio.CancelledError:
+        raise
+    except BaseExceptionGroup as exc:
+        if terminal_frame is None:
+            raise
+        _trace_egress_failure(exc)
+        yield terminal_frame(exc)
+    except Exception as exc:
+        if terminal_frame is None:
+            raise
+        _trace_egress_failure(exc)
+        yield terminal_frame(exc)
+
+
+async def anthropic_sse_streaming_response(
+    body: AsyncIterator[str],
+    *,
+    pre_start_error_response: PreStartErrorResponse,
+) -> Response:
+    """Return a streaming response for Anthropic-style SSE streams."""
+    return await _first_chunk_streaming_response(
+        body,
         headers=ANTHROPIC_SSE_RESPONSE_HEADERS,
+        pre_start_error_response=pre_start_error_response,
+        terminal_frame=lambda _exc: anthropic_terminal_error_frame(
+            EGRESS_STREAM_INTERRUPTED_MESSAGE
+        ),
     )
 
 
-def openai_responses_sse_streaming_response(
+async def openai_responses_sse_streaming_response(
     body: AsyncIterator[str],
     *,
     headers: Mapping[str, str],
-) -> StreamingResponse:
+    pre_start_error_response: PreStartErrorResponse,
+) -> Response:
     """Return a streaming response for OpenAI Responses-style SSE."""
-    return StreamingResponse(
+    return await _first_chunk_streaming_response(
         body,
-        media_type="text/event-stream",
-        headers=dict(headers),
+        headers=headers,
+        pre_start_error_response=pre_start_error_response,
+        terminal_frame=None,
     )

core/anthropic/streaming/__init__.py

@@ -3,6 +3,7 @@
 from .emitter import (
     ANTHROPIC_SSE_RESPONSE_HEADERS,
     AnthropicSseEmitter,
+    anthropic_terminal_error_frame,
     format_sse_event,
     map_stop_reason,
 )
@@ -43,6 +44,7 @@ __all__ = [
     "ToolSchema",
     "TruncatedProviderStreamError",
     "accept_tool_json_repair",
+    "anthropic_terminal_error_frame",
     "continuation_suffix",
     "format_sse_event",
     "is_retryable_stream_error",

core/anthropic/streaming/emitter.py

@@ -31,6 +31,14 @@ def format_sse_event(event_type: str, data: dict[str, Any]) -> str:
     return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
 
 
+def anthropic_terminal_error_frame(message: str) -> str:
+    """Serialize a terminal Anthropic SSE error event for egress failures."""
+    return format_sse_event(
+        "error",
+        {"type": "error", "error": {"type": "api_error", "message": message}},
+    )
+
+
 class AnthropicSseEmitter:
     """Serialize Anthropic SSE events and optionally log raw event bodies."""
 

core/openai_responses/adapter.py

@@ -22,8 +22,13 @@ class OpenAIResponsesAdapter:
         self,
         chunks: AsyncIterable[Any],
         request: Mapping[str, Any],
+        *,
+        stream_error_message: str | None = None,
     ) -> AsyncIterator[str]:
-        return iter_responses_sse_from_anthropic(chunks, request)
+        kwargs: dict[str, str] = {}
+        if stream_error_message is not None:
+            kwargs["stream_error_message"] = stream_error_message
+        return iter_responses_sse_from_anthropic(chunks, request, **kwargs)
 
     def error_payload(self, *, message: str, error_type: str) -> dict[str, Any]:
         return openai_error_payload(message=message, error_type=error_type)

core/openai_responses/stream.py

@@ -1,23 +1,69 @@
 """Translate Anthropic SSE streams into OpenAI Responses SSE streams."""
 
+import asyncio
 from collections.abc import AsyncIterable, AsyncIterator, Mapping
 from typing import Any
 
+from core.trace import trace_event
+
 from .anthropic_sse import iter_sse_events
 from .streaming import ResponsesStreamAssembler
 
+DEFAULT_STREAM_INTERRUPTED_MESSAGE = (
+    "The upstream response stream ended unexpectedly; the request could not be "
+    "completed."
+)
+
 
 async def iter_responses_sse_from_anthropic(
     chunks: AsyncIterable[Any],
     request: Mapping[str, Any],
+    *,
+    stream_error_message: str = DEFAULT_STREAM_INTERRUPTED_MESSAGE,
 ) -> AsyncIterator[str]:
     """Yield Responses SSE events translated from an Anthropic SSE stream."""
 
     assembler = ResponsesStreamAssembler(request)
-    async for event in iter_sse_events(chunks):
-        for chunk in assembler.process_anthropic_event(event):
+    emitted_any_chunk = False
+    try:
+        async for event in iter_sse_events(chunks):
+            for chunk in assembler.process_anthropic_event(event):
+                yield chunk
+                emitted_any_chunk = True
+            if assembler.terminal:
+                return
+        for chunk in assembler.finish_if_needed():
+            yield chunk
+            emitted_any_chunk = True
+    except GeneratorExit:
+        raise
+    except asyncio.CancelledError:
+        raise
+    except BaseExceptionGroup as exc:
+        if not emitted_any_chunk:
+            raise
+        trace_event(
+            stage="responses",
+            event="responses.stream.terminal_failure_frame",
+            source="openai_responses",
+            exc_type=type(exc).__name__,
+        )
+        for chunk in assembler.fail_response(
+            {"error": {"type": "api_error", "message": stream_error_message}}
+        ):
+            yield chunk
+        return
+    except Exception as exc:
+        if not emitted_any_chunk:
+            raise
+        trace_event(
+            stage="responses",
+            event="responses.stream.terminal_failure_frame",
+            source="openai_responses",
+            exc_type=type(exc).__name__,
+        )
+        for chunk in assembler.fail_response(
+            {"error": {"type": "api_error", "message": stream_error_message}}
+        ):
             yield chunk
-        if assembler.terminal:
-            return
-    for chunk in assembler.finish_if_needed():
-        yield chunk
+        return

plan.md

@@ -0,0 +1,324 @@
+# First-Frame Gated Streaming PR Plan
+
+## Summary
+
+Fix issue #1020 at the real boundary: FCC currently commits downstream
+`HTTP 200` as soon as FastAPI receives a `StreamingResponse`, before the
+provider-backed stream has proven it can emit a valid first SSE chunk. If the
+provider fails before that first chunk, Claude Code sees `200` with an empty or
+malformed stream instead of a non-200 response it can retry.
+
+The PR should introduce an API egress first-frame gate and align provider
+transports so pre-start failures raise typed HTTP-mappable errors instead of
+being converted into synthetic successful SSE streams. Once the first chunk has
+escaped, HTTP status can no longer change, so post-start failures still need a
+protocol-correct terminal stream frame.
+
+Because this changes production API/provider behavior on `main`, bump the
+current patch version from `3.4.12` to `3.4.13` unless `main` advances first,
+then refresh `uv.lock`.
+
+## Customer-Facing Contract
+
+- `fcc-server` should not return `HTTP 200` for `/v1/messages` or
+  `/v1/responses` until the stream can produce its first protocol chunk.
+- Claude Code should receive a non-200 Anthropic-shaped error when upstream
+  provider setup/retry fails before any stream output is viable.
+- Codex should receive a non-200 OpenAI-shaped error when the Responses stream
+  fails before `response.created`.
+- After streaming has started, clients should receive a parseable terminal
+  protocol frame instead of a truncated connection where feasible.
+- Provider retries, midstream recovery, tool salvage, thinking/reasoning, tool
+  calls, local web tools, non-streaming `/v1/messages`, and messaging behavior
+  should remain unchanged except for the pre-start HTTP status fix.
+
+## Grill-Me Decisions
+
+### Is this caused by NIM or FCC?
+
+Recommended answer: FCC owns the bug. NIM or any upstream can trigger the
+failure by returning 429/5xx/504 or closing early, but FCC currently commits
+downstream `HTTP 200` before upstream viability is known. Reproducing current
+`api.response_streams.anthropic_sse_streaming_response()` shows
+`http.response.start 200` is sent immediately, before the first delayed body
+chunk and even when the body raises before its first chunk.
+
+### Is a terminal SSE error frame enough?
+
+Recommended answer: no. A terminal frame fixes only post-start truncation. It
+does not restore Claude Code's HTTP retry behavior because the response is
+still `HTTP 200`. The PR needs first-frame gating for pre-start failures plus
+terminal framing for post-start failures.
+
+### Should API egress own provider retry?
+
+Recommended answer: no. Provider transports keep upstream retries, recovery,
+tool salvage, and provider-specific fallbacks. API egress owns only the HTTP
+commit boundary: do not commit success until there is a first chunk; after
+success is committed, close the protocol cleanly if possible.
+
+### Should providers keep emitting synthetic pre-start SSE errors?
+
+Recommended answer: no. A provider-side final error before downstream-visible
+output should raise a mapped `ProviderError`. Synthetic Anthropic SSE error
+tails are only appropriate when the stream has already started or provider
+state has produced output that must be closed in protocol shape.
+
+### Should `/v1/responses` use a fresh assembler for egress failures?
+
+Recommended answer: no. Responses streams are stateful. A post-start
+`response.failed` must be produced by the same `ResponsesStreamAssembler` that
+emitted `response.created`, preserving `response.id`, active output flushes,
+usage, and response metadata.
+
+## Architecture Target
+
+### API Egress
+
+`api/response_streams.py` should own public HTTP streaming commit timing.
+
+Add an async first-frame helper that:
+
+1. pulls the first chunk from an `AsyncIterator[str]` before constructing the
+   public `StreamingResponse`;
+2. returns a protocol JSON error response if the iterator raises before the
+   first chunk;
+3. returns a `StreamingResponse` that replays the first chunk and streams the
+   rest when the first chunk exists;
+4. wraps the post-first-chunk tail with a terminal-frame guard for unexpected
+   non-cancellation failures.
+
+The helper should be protocol-agnostic. Protocol-specific call sites provide:
+
+- streaming headers;
+- pre-start JSON error envelope builder;
+- post-start terminal frame behavior.
+
+Do not import `core/openai_responses` internals directly into API egress. API
+handlers may use their adapter/facade objects.
+
+### Provider Execution
+
+`api/provider_execution.py` should keep resolving providers, preflight,
+request tracing, raw-payload logging, token counting, and `traced_async_stream`.
+It should not construct `StreamingResponse` and should not own protocol error
+serialization.
+
+### Provider Error Mapping
+
+Provider transports need a single helper for pre-start final failures, for
+example under `providers/error_mapping.py`, that converts any final stream
+exception into a `ProviderError`:
+
+- preserve existing mapped provider statuses for authentication, bad request,
+  rate limit, overload, and upstream 5xx cases;
+- preserve existing user-facing error-message sanitization and request-id
+  appending;
+- wrap internal stream exceptions such as `TruncatedProviderStreamError` in an
+  upstream-style `APIError` rather than letting raw runtime exceptions escape;
+- keep verbose raw exception detail behind existing diagnostic flags.
+
+Do not make API egress inspect OpenAI/httpx exception classes directly. API
+egress should receive either a first chunk or a typed exception it can serialize
+for the product protocol.
+
+### OpenAI-Chat Transport
+
+`providers/transports/openai_chat/stream.py` should distinguish:
+
+- **uncommitted pre-start final error**: raise mapped `ProviderError` so the API
+  first-frame gate returns non-200;
+- **committed/buffered stream failure**: preserve existing recovery/error-tail
+  behavior;
+- **early retry/recovery success**: unchanged;
+- **complete tool salvage**: unchanged.
+
+The important classification is not merely whether `message_start` was created
+internally. It is whether anything has escaped the recovery holdback to the API
+iterator. If the holdback has not committed and no buffered events are being
+flushed as client-visible output, pre-start final errors should raise.
+
+### Native Anthropic Transport
+
+`providers/transports/anthropic_messages/stream.py` should apply the same
+boundary:
+
+- no committed/buffered downstream-visible event: raise mapped `ProviderError`;
+- committed or buffered native stream: use native ledger error-tail behavior.
+
+This keeps local native providers and future native providers consistent with
+OpenAI-chat behavior.
+
+### OpenAI Responses
+
+`core/openai_responses/stream.py` and
+`core/openai_responses/streaming/assembler.py` should own post-start Responses
+terminal failures.
+
+Do not add a stateless `OpenAIResponsesAdapter.egress_error_frame()` that mints
+a fresh response id. Instead, the iterator returned by
+`OpenAIResponsesAdapter.iter_sse_from_anthropic()` should:
+
+- let pre-`response.created` failures propagate to API egress;
+- after `response.created`, catch unexpected non-cancellation failures, call
+  `ResponsesStreamAssembler.fail_response(...)` on the active assembler, yield
+  the resulting `response.failed`, then re-raise or trace according to the
+  existing egress tracing policy.
+
+This preserves `response.failed.response.id == response.created.response.id`.
+
+### Anthropic Messages
+
+For `/v1/messages`, post-start unexpected failures may use a stateless terminal
+Anthropic `event: error` as the final API egress fallback. Provider-owned error
+tails remain preferred when provider code can close content blocks and emit
+`message_delta`/`message_stop`.
+
+## Implementation Plan
+
+1. Add first-frame response helpers in `api/response_streams.py`.
+   - Add a small private result type for either first chunk or pre-start
+     exception.
+   - Add `async def anthropic_sse_streaming_response(...)` or a new clearly
+     named async builder, because first-frame probing must await the iterator.
+   - Add the equivalent Responses builder with injected OpenAI error-envelope
+     handling.
+   - Keep wrappers protocol-agnostic and make call sites explicit.
+
+2. Update `MessagesHandler._to_public_response()`.
+   - Await the new Anthropic streaming response builder for `stream != false`.
+   - Convert pre-start `ProviderError` to Anthropic JSON with the provider
+     status code.
+   - Convert unexpected pre-start exceptions to a safe 500 Anthropic JSON error
+     using existing safe logging rules.
+   - Keep `stream: false` aggregation unchanged.
+
+3. Update `ResponsesHandler.create()`.
+   - Await the new Responses streaming response builder.
+   - Convert pre-start `ProviderError` to OpenAI-shaped JSON using
+     `OpenAIResponsesAdapter.error_payload()`.
+   - Convert unexpected pre-start exceptions to safe 500 OpenAI-shaped JSON.
+   - Keep request conversion errors and `stream: false` rejection unchanged.
+
+4. Add a provider-side pre-start failure exception path.
+   - Introduce a small neutral helper in provider/shared error mapping that
+     always returns a `ProviderError` for a final pre-start stream exception.
+   - In OpenAI-chat final-error handling, if the recovery holdback is not
+     committed and no event should be exposed, raise `map_error(...)` instead
+     of yielding `emit_error_tail(...)`.
+   - In native Anthropic final-error handling, raise mapped provider errors when
+     no native event has been committed/buffered to the API.
+   - Preserve existing synthetic SSE tails once events have escaped or must be
+     closed.
+
+5. Harden post-start terminal fallback.
+   - Add Anthropic terminal-frame serialization under
+     `core/anthropic/streaming/` only as a last-resort egress fallback.
+   - In Responses stream conversion, add same-assembler failure handling for
+     post-`response.created` exceptions.
+   - Do not mint fresh Responses IDs for a terminal failure.
+
+6. Update architecture docs.
+   - Document that API egress owns first-frame HTTP commit gating.
+   - Document that providers raise pre-start final failures but own retries and
+     midstream recovery.
+   - Document that Responses terminal failures are assembler-owned because
+     Responses streams are stateful.
+
+7. Bump version and lockfile.
+   - Update `[project].version` from `3.4.12` to `3.4.13` unless `main`
+     advances.
+   - Run `uv lock`.
+
+## Test Plan
+
+### API Egress Tests
+
+- ASGI-level test proving `http.response.start 200` is not sent until the first
+  chunk is available.
+- ASGI-level test proving a pre-first-chunk `ProviderError` returns non-200 JSON
+  and sends no SSE body.
+- ASGI-level test proving a pre-first-chunk unexpected exception returns safe
+  500 JSON and does not leak raw exception text by default.
+- ASGI-level test proving post-first-chunk exceptions yield a terminal frame
+  before the exception closes the ASGI body.
+- Cancellation and `GeneratorExit` tests proving no terminal frame is emitted
+  into a dead socket.
+
+### Messages API Tests
+
+- `/v1/messages` provider pre-start `RateLimitError` returns HTTP 429
+  Anthropic-shaped JSON.
+- `/v1/messages` provider pre-start `APIError(status_code=504)` returns HTTP
+  non-200 using the existing provider mapping, not HTTP 200.
+- `/v1/messages` delayed valid provider stream still returns `text/event-stream`
+  and valid Anthropic SSE.
+- `/v1/messages stream:false` aggregation remains unchanged.
+
+### Responses API Tests
+
+- `/v1/responses` provider pre-start `RateLimitError` returns HTTP 429
+  OpenAI-shaped JSON.
+- `/v1/responses` delayed valid provider stream still returns
+  `text/event-stream`.
+- Responses post-start exception emits `response.failed` with the same
+  `response.id` as `response.created`.
+- Existing provider-emitted Anthropic `event:error` still converts to
+  `response.failed` with the same active response id.
+
+### Provider Transport Tests
+
+- OpenAI-chat exhausted pre-stream 429/5xx/transport failure raises mapped
+  `ProviderError` before any downstream-visible event.
+- OpenAI-chat retry success path remains unchanged.
+- OpenAI-chat midstream text failure still uses recovery/terminal tail behavior.
+- OpenAI-chat complete tool salvage remains unchanged.
+- Native Anthropic pre-send/pre-event failure raises mapped `ProviderError`.
+- Native Anthropic midstream native event failure still closes through native
+  recovery/error-tail behavior.
+
+### Contract Tests
+
+- API response stream helper is the only owner of first-frame HTTP commit
+  gating.
+- Responses egress terminal failure is assembler-owned, not generated by a
+  stateless adapter helper.
+- `api/provider_execution.py` remains free of `StreamingResponse` ownership.
+- Architecture relative links still resolve.
+
+## Verification Commands
+
+Run targeted tests first:
+
+```powershell
+uv run pytest tests/api/test_response_streams.py tests/api/test_api_handlers.py tests/api/test_openai_responses.py
+uv run pytest tests/providers/test_streaming_errors.py tests/providers/test_anthropic_messages.py tests/providers/test_openai_compat_5xx_retry.py tests/providers/test_anthropic_messages_429_retry.py
+uv run pytest tests/core/openai_responses/test_sse.py tests/contracts/test_import_boundaries.py tests/contracts/test_architecture_contracts.py
+```
+
+Then run the final local gate:
+
+```powershell
+.\scripts\ci.ps1
+```
+
+## Risks And Guardrails
+
+- Pulling the first chunk delays HTTP headers until upstream viability is known.
+  This is intentional for Claude/Codex retry correctness, but tests should prove
+  normal streams still begin promptly after the first provider chunk.
+- A provider can emit a synthetic SSE error as its first chunk. The PR should
+  avoid relying only on API first-chunk gating; providers must raise pre-start
+  final failures instead of creating synthetic success streams.
+- Responses post-start fallback must preserve assembler identity. Any test that
+  shape-asserts only `response.failed` without comparing IDs is insufficient.
+- Do not broaden API egress into retry/recovery ownership. That would erode the
+  provider transport boundaries established in `ARCHITECTURE.md`.
+
+## Out Of Scope
+
+- Changing retry counts, backoff, or adding `Retry-After` support.
+- Redesigning OpenAI-chat tool-call buffering.
+- Changing model routing, thinking/reasoning policy, or provider request bodies.
+- Changing messaging queue/cancellation behavior.
+- Restoring non-streaming `/v1/responses`.

providers/error_mapping.py

@@ -19,6 +19,7 @@ from providers.exceptions import (
     AuthenticationError,
     InvalidRequestError,
     OverloadedError,
+    ProviderError,
     RateLimitError,
 )
 from providers.rate_limit import GlobalRateLimiter
@@ -313,6 +314,40 @@ def user_visible_message_for_mapped_provider_error(
     return get_user_facing_error_message(mapped, read_timeout_s=read_timeout_s)
 
 
+def map_stream_start_error(
+    error: Exception,
+    *,
+    provider_name: str,
+    read_timeout_s: float | None,
+    request_id: str | None,
+    rate_limiter: GlobalRateLimiter | None = None,
+) -> ProviderError:
+    """Map a final pre-start stream failure into an HTTP-serializable provider error.
+
+    Providers call this only when no downstream-visible SSE chunk has escaped.
+    At that boundary the API can still return a real non-200 response, which is
+    preferable to synthesizing a successful SSE stream that starts with an error
+    message.
+    """
+    mapped = map_error(error, rate_limiter=rate_limiter)
+    detail = extract_provider_error_detail(error)
+    message = user_visible_message_for_mapped_provider_error(
+        mapped,
+        provider_name=provider_name,
+        read_timeout_s=read_timeout_s,
+        detail=detail,
+        request_id=request_id,
+    )
+    if isinstance(mapped, ProviderError):
+        return ProviderError(
+            message,
+            status_code=mapped.status_code,
+            error_type=mapped.error_type,
+            raw_error=mapped.raw_error,
+        )
+    return APIError(message, status_code=502, raw_error=str(error))
+
+
 def map_error(
     e: Exception, *, rate_limiter: GlobalRateLimiter | None = None
 ) -> Exception:

providers/transports/anthropic_messages/stream.py

@@ -14,6 +14,7 @@ from core.anthropic.streaming import (
     tool_schemas_by_name,
 )
 from core.trace import provider_native_messages_body_snapshot, trace_event
+from providers.error_mapping import map_stream_start_error
 from providers.transports.http import maybe_await_aclose
 
 from .recovery import AnthropicMessagesRecovery
@@ -205,28 +206,26 @@ class AnthropicMessagesStreamAdapter:
                         request_id=self._request_id,
                         error_message=error_message,
                         exc_type=type(error).__name__,
-                        mid_stream=(
-                            sent_any_event
-                            or decision.committed
-                            or decision.has_buffered
-                        ),
+                        mid_stream=(sent_any_event or decision.committed),
                     )
-                    if decision.committed or decision.has_buffered:
-                        if not decision.committed:
-                            for event in recovery.flush():
-                                sent_any_event = True
-                                yield event
+                    if decision.committed:
+                        for event in ledger.midstream_error_tail(error_message):
+                            yield event
+                    elif decision.has_buffered and complete_tool_salvageable:
+                        for event in recovery.flush():
+                            sent_any_event = True
+                            yield event
                         for event in ledger.midstream_error_tail(error_message):
                             yield event
                     else:
                         recovery.discard()
-                        for event in self._transport._emit_error_events(
-                            request=self._request,
-                            input_tokens=self._input_tokens,
-                            error_message=error_message,
-                            sent_any_event=False,
-                        ):
-                            yield event
+                        raise map_stream_start_error(
+                            error,
+                            provider_name=tag,
+                            read_timeout_s=self._transport._config.http_read_timeout,
+                            request_id=self._request_id,
+                            rate_limiter=self._transport._global_rate_limiter,
+                        ) from error
                     return
                 finally:
                     if response is not None and not response.is_closed:

providers/transports/openai_chat/stream.py

@@ -20,7 +20,7 @@ from core.anthropic.streaming import (
     map_stop_reason,
 )
 from core.trace import provider_chat_body_snapshot, trace_event
-from providers.error_mapping import map_error
+from providers.error_mapping import map_error, map_stream_start_error
 from providers.transports.http import maybe_await_aclose
 
 from .recovery import OpenAIChatRecovery
@@ -288,12 +288,22 @@ class OpenAIChatStreamAdapter:
                             )
                         ).__name__,
                     )
-                    if not decision.committed and decision.has_buffered:
+                    if (
+                        not decision.committed
+                        and decision.has_buffered
+                        and complete_tool_salvageable
+                    ):
                         for event in recovery.flush():
                             yield event
                     elif not decision.committed:
                         recovery.discard()
-                        ledger = self._new_ledger()
+                        raise map_stream_start_error(
+                            error,
+                            provider_name=tag,
+                            read_timeout_s=self._transport._config.http_read_timeout,
+                            request_id=self._request_id,
+                            rate_limiter=self._transport._global_rate_limiter,
+                        ) from error
                     for event in self._recovery.emit_error_tail(ledger, error_message):
                         yield event
                     return

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "free-claude-code"
-version = "3.4.12"
+version = "3.4.13"
 description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM"
 readme = "README.md"
 requires-python = ">=3.14.0"

tests/api/test_api.py

@@ -4,6 +4,7 @@ import pytest
 from fastapi.testclient import TestClient
 
 from api.app import create_app
+from providers.exceptions import RateLimitError
 from providers.nvidia_nim import NvidiaNimProvider
 
 app = create_app()
@@ -22,6 +23,13 @@ async def _mock_stream_response(*args, **kwargs):
     yield "[DONE]\n\n"
 
 
+async def _mock_pre_start_rate_limit(*args, **kwargs):
+    """Provider stream that fails before any downstream-visible SSE chunk."""
+    _stream_response_calls.append((args, kwargs))
+    raise RateLimitError("upstream is busy")
+    yield "unreachable"
+
+
 mock_provider.stream_response = _mock_stream_response
 
 
@@ -95,6 +103,29 @@ def test_create_message_stream(client: TestClient):
     assert b"message_start" in content or b"event:" in content
 
 
+def test_create_message_pre_start_provider_error_returns_non_200_json(
+    client: TestClient,
+):
+    """Pre-first-chunk provider errors should not commit HTTP 200."""
+    mock_provider.stream_response = _mock_pre_start_rate_limit
+    payload = {
+        "model": "claude-3-sonnet",
+        "messages": [{"role": "user", "content": "Hi"}],
+        "max_tokens": 100,
+        "stream": True,
+    }
+
+    response = client.post("/v1/messages", json=payload)
+
+    assert response.status_code == 429
+    assert response.headers["content-type"].startswith("application/json")
+    assert response.json() == {
+        "type": "error",
+        "error": {"type": "rate_limit_error", "message": "upstream is busy"},
+    }
+    mock_provider.stream_response = _mock_stream_response
+
+
 def test_create_message_accepts_system_role_messages(client: TestClient):
     """Create message accepts latest-client system messages."""
     mock_provider.stream_response = _mock_stream_response

tests/api/test_openai_responses.py

@@ -7,6 +7,7 @@ from fastapi.testclient import TestClient
 from api.app import create_app
 from core.anthropic.stream_contracts import parse_sse_text
 from core.anthropic.streaming import format_sse_event
+from providers.exceptions import RateLimitError
 
 
 class FakeProvider:
@@ -23,6 +24,29 @@ class FakeProvider:
             yield chunk
 
 
+class PreStartFailingProvider(FakeProvider):
+    def __init__(self) -> None:
+        super().__init__([])
+
+    async def stream_response(self, request_data, **_kwargs):
+        self.requests.append(request_data)
+        self.stream_kwargs.append(_kwargs)
+        raise RateLimitError("upstream is busy")
+        yield "unreachable"
+
+
+class PostStartFailingProvider(FakeProvider):
+    def __init__(self) -> None:
+        super().__init__([format_sse_event("message_start", {"type": "message_start"})])
+
+    async def stream_response(self, request_data, **_kwargs):
+        self.requests.append(request_data)
+        self.stream_kwargs.append(_kwargs)
+        for chunk in self.chunks:
+            yield chunk
+        raise RuntimeError("socket closed")
+
+
 @pytest.fixture
 def responses_client():
     provider = FakeProvider(_anthropic_text_stream("Hello from provider"))
@@ -73,6 +97,49 @@ def test_create_response_stream_routes_through_provider(
     assert routed.max_tokens == 32
 
 
+def test_create_response_pre_start_provider_error_returns_openai_error() -> None:
+    provider = PreStartFailingProvider()
+    app = create_app(lifespan_enabled=False)
+    with (
+        patch("api.dependencies.resolve_provider", return_value=provider),
+        TestClient(app) as client,
+    ):
+        response = client.post(
+            "/v1/responses",
+            json={
+                "model": "nvidia_nim/test-model",
+                "input": "Hello",
+            },
+        )
+
+    assert response.status_code == 429
+    payload = response.json()
+    assert payload["error"]["type"] == "rate_limit_error"
+    assert payload["error"]["message"] == "upstream is busy"
+
+
+def test_create_response_post_start_failure_preserves_response_id() -> None:
+    provider = PostStartFailingProvider()
+    app = create_app(lifespan_enabled=False)
+    with (
+        patch("api.dependencies.resolve_provider", return_value=provider),
+        TestClient(app) as client,
+    ):
+        response = client.post(
+            "/v1/responses",
+            json={
+                "model": "nvidia_nim/test-model",
+                "input": "Hello",
+            },
+        )
+
+    assert response.status_code == 200
+    events = parse_sse_text(response.text)
+    assert [event.event for event in events] == ["response.created", "response.failed"]
+    assert events[-1].data["response"]["id"] == events[0].data["response"]["id"]
+    assert events[-1].data["response"]["status"] == "failed"
+
+
 def test_create_response_stream_bypasses_local_message_optimizations() -> None:
     provider = FakeProvider(_anthropic_text_stream("Provider response"))
     app = create_app(lifespan_enabled=False)

tests/api/test_response_streams.py

@@ -0,0 +1,111 @@
+"""Tests for public SSE response start gating."""
+
+import asyncio
+import json
+from collections.abc import AsyncGenerator
+
+import pytest
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from api.response_streams import (
+    EGRESS_STREAM_INTERRUPTED_MESSAGE,
+    anthropic_sse_streaming_response,
+)
+from core.anthropic.stream_contracts import parse_sse_text
+from providers.exceptions import RateLimitError
+
+
+async def _body_chunks(chunks: list[str]) -> AsyncGenerator[str]:
+    for chunk in chunks:
+        yield chunk
+
+
+async def _body_raises(exc: BaseException) -> AsyncGenerator[str]:
+    raise exc
+    yield "unreachable"
+
+
+async def _body_then_raises(
+    chunks: list[str], exc: BaseException
+) -> AsyncGenerator[str]:
+    for chunk in chunks:
+        yield chunk
+    raise exc
+
+
+def _json_error(exc: BaseException) -> JSONResponse:
+    if isinstance(exc, RateLimitError):
+        return JSONResponse(
+            status_code=exc.status_code,
+            content=exc.to_anthropic_format(),
+        )
+    return JSONResponse(
+        status_code=500,
+        content={
+            "type": "error",
+            "error": {"type": "api_error", "message": "failed"},
+        },
+    )
+
+
+async def _drain(response: StreamingResponse) -> str:
+    parts = [
+        chunk.decode("utf-8") if isinstance(chunk, bytes) else str(chunk)
+        async for chunk in response.body_iterator
+    ]
+    return "".join(parts)
+
+
+@pytest.mark.asyncio
+async def test_anthropic_response_waits_for_first_chunk_before_returning() -> None:
+    ready = asyncio.Event()
+
+    async def body() -> AsyncGenerator[str]:
+        await ready.wait()
+        yield 'event: message_start\ndata: {"type":"message_start"}\n\n'
+
+    task = asyncio.create_task(
+        anthropic_sse_streaming_response(
+            body(),
+            pre_start_error_response=_json_error,
+        )
+    )
+
+    await asyncio.sleep(0)
+    assert not task.done()
+
+    ready.set()
+    response = await asyncio.wait_for(task, timeout=1)
+    assert isinstance(response, StreamingResponse)
+    assert "message_start" in await _drain(response)
+
+
+@pytest.mark.asyncio
+async def test_anthropic_pre_start_provider_error_returns_non_200_json() -> None:
+    response = await anthropic_sse_streaming_response(
+        _body_raises(RateLimitError("provider says slow down")),
+        pre_start_error_response=_json_error,
+    )
+
+    assert isinstance(response, JSONResponse)
+    assert response.status_code == 429
+    body = json.loads(bytes(response.body))
+    assert body["error"]["type"] == "rate_limit_error"
+    assert body["error"]["message"] == "provider says slow down"
+
+
+@pytest.mark.asyncio
+async def test_anthropic_post_start_exception_emits_terminal_error_frame() -> None:
+    response = await anthropic_sse_streaming_response(
+        _body_then_raises(
+            ['event: message_start\ndata: {"type":"message_start"}\n\n'],
+            RuntimeError("socket cut"),
+        ),
+        pre_start_error_response=_json_error,
+    )
+
+    assert isinstance(response, StreamingResponse)
+    text = await _drain(response)
+    events = parse_sse_text(text)
+    assert [event.event for event in events] == ["message_start", "error"]
+    assert events[-1].data["error"]["message"] == EGRESS_STREAM_INTERRUPTED_MESSAGE

tests/providers/test_anthropic_messages.py

@@ -7,7 +7,7 @@ import httpx
 import pytest
 
 from config.constants import ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS
-from core.anthropic.stream_contracts import event_index, parse_sse_text
+from core.anthropic.stream_contracts import parse_sse_text
 from core.anthropic.streaming import (
     MIDSTREAM_RECOVERY_ATTEMPTS,
     AnthropicStreamLedger,
@@ -15,9 +15,9 @@ from core.anthropic.streaming import (
     format_sse_event,
 )
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.transports.anthropic_messages import AnthropicMessagesTransport
 from providers.transports.anthropic_messages.recovery import AnthropicMessagesRecovery
-from tests.stream_contract import assert_canonical_stream_error_envelope
 
 
 class NativeProvider(AnthropicMessagesTransport):
@@ -251,7 +251,7 @@ async def test_stream_uses_retry_builds_request_and_closes_response(
 
 
 @pytest.mark.asyncio
-async def test_stream_maps_non_200_to_error_event_and_closes_response(
+async def test_stream_maps_pre_start_non_200_to_provider_error_and_closes_response(
     provider_config,
 ):
     provider = NativeProvider(provider_config)
@@ -266,25 +266,21 @@ async def test_stream_maps_non_200_to_error_event_and_closes_response(
             new_callable=AsyncMock,
             return_value=response,
         ),
+        pytest.raises(ProviderError) as exc_info,
     ):
-        events = [
-            event async for event in provider.stream_response(req, request_id="REQ_123")
-        ]
+        [event async for event in provider.stream_response(req, request_id="REQ_123")]
 
     assert response.is_closed
-    assert_canonical_stream_error_envelope(
-        events, user_message_substr="Upstream provider TEST_NATIVE returned HTTP 500."
-    )
-    blob = "".join(events)
-    assert "Internal Server Error" in blob
-    assert "REQ_123" in blob
+    assert "Upstream provider TEST_NATIVE returned HTTP 500." in exc_info.value.message
+    assert "Internal Server Error" in exc_info.value.message
+    assert "REQ_123" in exc_info.value.message
 
 
 @pytest.mark.asyncio
-async def test_midstream_error_closes_open_block_and_uses_fresh_content_index(
+async def test_precommit_native_error_raises_without_leaking_open_block(
     provider_config,
 ):
-    """After upstream message_start + content_block_start, synthetic errors must not reuse index 0."""
+    """A native error before holdback commit raises instead of sending HTTP 200 SSE."""
     provider = NativeProvider(provider_config)
     req = MockRequest()
     mid = "msg_midstream_err"
@@ -325,17 +321,11 @@ async def test_midstream_error_closes_open_block_and_uses_fresh_content_index(
             new_callable=AsyncMock,
             return_value=response,
         ),
+        pytest.raises(ProviderError) as exc_info,
     ):
-        events = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
-    assert_canonical_stream_error_envelope(
-        events, user_message_substr="mid-stream failure"
-    )
-    parsed = parse_sse_text("".join(events))
-    starts = [e for e in parsed if e.event == "content_block_start"]
-    assert event_index(starts[0]) == 0
-    assert event_index(starts[-1]) == 1
-    assert {event_index(e) for e in parsed if e.event == "content_block_stop"} == {0, 1}
+    assert "mid-stream failure" in exc_info.value.message
 
 
 @pytest.mark.asyncio

tests/providers/test_anthropic_messages_429_retry.py

@@ -8,13 +8,13 @@ import pytest
 
 from core.anthropic.stream_contracts import event_names, parse_sse_text
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.rate_limit import GlobalRateLimiter
 from tests.providers.test_anthropic_messages import (
     FakeResponse,
     MockRequest,
     NativeProvider,
 )
-from tests.stream_contract import assert_canonical_stream_error_envelope
 
 
 def _assert_minimal_success_stream(events: list[str]) -> None:
@@ -236,15 +236,13 @@ async def test_native_stream_5xx_retry_exhausted(provider_config, status_code, s
                     return_value=bad,
                 ) as mock_send,
                 patch("asyncio.sleep", new_callable=AsyncMock),
+                pytest.raises(ProviderError) as exc_info,
             ):
-                events = [e async for e in provider.stream_response(req)]
+                [e async for e in provider.stream_response(req)]
 
             assert mock_send.await_count == 5
             assert bad.is_closed
-            assert_canonical_stream_error_envelope(
-                events,
-                user_message_substr=substr,
-            )
+            assert substr in exc_info.value.message
     finally:
         GlobalRateLimiter.reset_instance()
 
@@ -293,8 +291,9 @@ async def test_native_stream_connection_error_retry_exhausted(provider_config):
                 patch(
                     "providers.transports.anthropic_messages.stream.trace_event"
                 ) as trace,
+                pytest.raises(ProviderError) as exc_info,
             ):
-                events = [
+                [
                     e
                     async for e in provider.stream_response(
                         req, request_id="req_native_conn"
@@ -309,10 +308,7 @@ async def test_native_stream_connection_error_retry_exhausted(provider_config):
             ]
             assert error_traces[-1]["request_id"] == "req_native_conn"
             assert error_traces[-1]["exc_type"] == "ConnectError"
-            assert_canonical_stream_error_envelope(
-                events,
-                user_message_substr="Provider exception:\nconnect failed",
-            )
+            assert "Provider exception:\nconnect failed" in exc_info.value.message
     finally:
         GlobalRateLimiter.reset_instance()
 
@@ -352,13 +348,12 @@ async def test_non_retryable_4xx_http_error_not_retried(provider_config):
                     new_callable=AsyncMock,
                     return_value=err,
                 ) as mock_send,
+                pytest.raises(ProviderError) as exc_info,
             ):
-                events = [e async for e in provider.stream_response(req)]
+                [e async for e in provider.stream_response(req)]
 
             mock_send.assert_awaited_once()
             assert err.is_closed
-            assert_canonical_stream_error_envelope(
-                events, user_message_substr="Invalid request sent to provider"
-            )
+            assert "Invalid request sent to provider" in exc_info.value.message
     finally:
         GlobalRateLimiter.reset_instance()

tests/providers/test_llamacpp.py

@@ -8,8 +8,8 @@ import pytest
 from config.constants import ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS
 from core.anthropic.stream_contracts import parse_sse_text
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.llamacpp import LlamaCppProvider
-from tests.stream_contract import assert_canonical_stream_error_envelope
 
 
 class MockMessage:
@@ -45,6 +45,15 @@ class MockRequest:
         }
 
 
+async def _minimal_native_lines():
+    yield "event: message_start"
+    yield 'data: {"type":"message_start"}'
+    yield ""
+    yield "event: message_stop"
+    yield 'data: {"type":"message_stop"}'
+    yield ""
+
+
 @pytest.fixture
 def llamacpp_config():
     return ProviderConfig(
@@ -125,11 +134,7 @@ async def test_stream_response_omits_thinking_when_globally_disabled(llamacpp_co
     mock_response = MagicMock()
     mock_response.status_code = 200
 
-    async def empty_aiter():
-        if False:
-            yield ""
-
-    mock_response.aiter_lines = empty_aiter
+    mock_response.aiter_lines = _minimal_native_lines
 
     with (
         patch.object(provider._client, "build_request") as mock_build,
@@ -208,11 +213,7 @@ async def test_stream_response_adds_max_tokens_if_missing(llamacpp_provider):
     mock_response = MagicMock()
     mock_response.status_code = 200
 
-    async def empty_aiter():
-        if False:
-            yield ""
-
-    mock_response.aiter_lines = empty_aiter
+    mock_response.aiter_lines = _minimal_native_lines
 
     with (
         patch.object(req, "model_dump", return_value={"model": "test"}),
@@ -233,7 +234,7 @@ async def test_stream_response_adds_max_tokens_if_missing(llamacpp_provider):
 
 @pytest.mark.asyncio
 async def test_stream_error_status_code(llamacpp_provider):
-    """Non-200 status code raises an error that gets caught and yielded as an SSE API error."""
+    """Pre-start non-200 status code raises for API-level non-200 handling."""
     req = MockRequest()
 
     mock_response = MagicMock()
@@ -256,20 +257,21 @@ async def test_stream_error_status_code(llamacpp_provider):
             return_value=mock_response,
         ),
     ):
-        events = [
-            e
-            async for e in llamacpp_provider.stream_response(req, request_id="TEST_ID")
-        ]
+        with pytest.raises(ProviderError) as exc_info:
+            [
+                e
+                async for e in llamacpp_provider.stream_response(
+                    req, request_id="TEST_ID"
+                )
+            ]
 
-        assert_canonical_stream_error_envelope(
-            events, user_message_substr="Provider API request failed"
-        )
-        assert "TEST_ID" in "".join(events)
+        assert "Provider API request failed" in exc_info.value.message
+        assert "TEST_ID" in exc_info.value.message
 
 
 @pytest.mark.asyncio
 async def test_stream_network_error(llamacpp_provider):
-    """Network errors are caught and yielded as SSE API error events."""
+    """Pre-start network errors raise for API-level non-200 handling."""
     req = MockRequest()
 
     with (
@@ -283,18 +285,18 @@ async def test_stream_network_error(llamacpp_provider):
             side_effect=httpx.ConnectError("Connection refused"),
         ),
     ):
-        events = [
-            e
-            async for e in llamacpp_provider.stream_response(req, request_id="TEST_ID2")
-        ]
+        with pytest.raises(ProviderError) as exc_info:
+            [
+                e
+                async for e in llamacpp_provider.stream_response(
+                    req, request_id="TEST_ID2"
+                )
+            ]
 
-        blob = "".join(events)
-        assert_canonical_stream_error_envelope(
-            events, user_message_substr="Could not connect to provider."
-        )
-        assert "Provider exception" in blob
-        assert "Connection refused" in blob
-        assert "TEST_ID2" in blob
+        assert "Could not connect to provider." in exc_info.value.message
+        assert "Provider exception" in exc_info.value.message
+        assert "Connection refused" in exc_info.value.message
+        assert "TEST_ID2" in exc_info.value.message
 
 
 @pytest.mark.asyncio
@@ -320,17 +322,15 @@ async def test_stream_error_405_mentions_upstream_provider(llamacpp_provider):
             new_callable=AsyncMock,
             return_value=mock_response,
         ),
+        pytest.raises(ProviderError) as exc_info,
     ):
-        events = [
-            e async for e in llamacpp_provider.stream_response(req, request_id="REQ405")
-        ]
+        [e async for e in llamacpp_provider.stream_response(req, request_id="REQ405")]
 
-    blob = "".join(events)
     assert (
         "Upstream provider LLAMACPP rejected the request method or endpoint (HTTP 405)."
-        in blob
+        in exc_info.value.message
     )
-    assert "REQ405" in blob
+    assert "REQ405" in exc_info.value.message
 
 
 def test_build_request_body_disabled_thinking_strips_native_thinking_history(

tests/providers/test_mistral.py

@@ -9,6 +9,7 @@ import pytest
 from httpx import Request, Response
 
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.mistral import MISTRAL_DEFAULT_BASE, MistralProvider
 
 
@@ -740,10 +741,11 @@ async def test_stream_response_unrelated_bad_request_does_not_retry(mistral_prov
     ) as mock_create:
         mock_create.side_effect = error
 
-        events = [e async for e in mistral_provider.stream_response(req)]
+        with pytest.raises(ProviderError) as exc_info:
+            [e async for e in mistral_provider.stream_response(req)]
 
     assert mock_create.await_count == 1
-    assert any("message_stop" in event for event in events)
+    assert "Invalid request sent to provider" in exc_info.value.message
 
 
 @pytest.mark.asyncio
@@ -758,10 +760,11 @@ async def test_stream_response_generic_thinking_error_does_not_retry(
     ) as mock_create:
         mock_create.side_effect = error
 
-        events = [e async for e in mistral_provider.stream_response(req)]
+        with pytest.raises(ProviderError) as exc_info:
+            [e async for e in mistral_provider.stream_response(req)]
 
     assert mock_create.await_count == 1
-    assert any("message_stop" in event for event in events)
+    assert "Invalid request sent to provider" in exc_info.value.message
 
 
 def test_retry_body_without_reasoning_returns_none(mistral_provider):

tests/providers/test_nvidia_nim.py

@@ -7,6 +7,7 @@ from httpx import Request, Response
 
 from config.nim import NimSettings
 from providers.defaults import NVIDIA_NIM_DEFAULT_BASE
+from providers.exceptions import ProviderError
 from providers.nvidia_nim import NvidiaNimProvider
 from providers.nvidia_nim.tool_schema import NIM_TOOL_ARGUMENT_ALIASES_KEY
 
@@ -499,12 +500,11 @@ async def test_stream_response_does_not_retry_unrelated_bad_request(provider_con
     ) as mock_create:
         mock_create.side_effect = _make_bad_request_error("unrelated bad request")
 
-        events = [e async for e in provider.stream_response(req)]
+        with pytest.raises(ProviderError) as exc_info:
+            [e async for e in provider.stream_response(req)]
 
     assert mock_create.await_count == 1
-    event_text = "".join(events)
-    assert "Invalid request sent to provider" in event_text
-    assert "event: message_stop" in event_text
+    assert "Invalid request sent to provider" in exc_info.value.message
 
 
 @pytest.mark.asyncio
@@ -891,11 +891,11 @@ async def test_stream_response_bad_request_without_reasoning_budget_does_not_ret
     ) as mock_create:
         mock_create.side_effect = error
 
-        events = [e async for e in nim_provider.stream_response(req)]
+        with pytest.raises(ProviderError) as exc_info:
+            [e async for e in nim_provider.stream_response(req)]
 
     assert mock_create.await_count == 1
-    assert any("Invalid request sent to provider" in event for event in events)
-    assert any("message_stop" in event for event in events)
+    assert "Invalid request sent to provider" in exc_info.value.message
 
 
 @pytest.mark.asyncio
@@ -910,11 +910,11 @@ async def test_stream_response_unrelated_internal_error_does_not_downgrade(
     ) as mock_create:
         mock_create.side_effect = error
 
-        events = [e async for e in nim_provider.stream_response(req)]
+        with pytest.raises(ProviderError) as exc_info:
+            [e async for e in nim_provider.stream_response(req)]
 
     assert mock_create.await_count == 1
-    assert any("Provider API request failed" in event for event in events)
-    assert any("message_stop" in event for event in events)
+    assert "Provider API request failed" in exc_info.value.message
 
 
 @pytest.mark.asyncio
@@ -931,8 +931,8 @@ async def test_stream_response_internal_reasoning_content_error_does_not_downgra
     ) as mock_create:
         mock_create.side_effect = error
 
-        events = [e async for e in nim_provider.stream_response(req)]
+        with pytest.raises(ProviderError) as exc_info:
+            [e async for e in nim_provider.stream_response(req)]
 
     assert mock_create.await_count == 1
-    assert any("Provider API request failed" in event for event in events)
-    assert any("message_stop" in event for event in events)
+    assert "Provider API request failed" in exc_info.value.message

tests/providers/test_ollama.py

@@ -7,8 +7,8 @@ import pytest
 
 from core.anthropic.stream_contracts import parse_sse_text
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.ollama import OLLAMA_DEFAULT_BASE, OllamaProvider
-from tests.stream_contract import assert_canonical_stream_error_envelope
 
 
 class MockMessage:
@@ -233,7 +233,7 @@ def test_build_request_body_disabled_thinking_strips_assistant_thinking_blocks(
 
 @pytest.mark.asyncio
 async def test_stream_error_status_code(ollama_provider):
-    """Non-200 status code is yielded as an SSE API error."""
+    """Pre-start non-200 status code raises for API-level non-200 handling."""
     req = MockRequest()
     mock_response = MagicMock()
     mock_response.status_code = 500
@@ -254,16 +254,15 @@ async def test_stream_error_status_code(ollama_provider):
             new_callable=AsyncMock,
             return_value=mock_response,
         ),
+        pytest.raises(ProviderError) as exc_info,
     ):
-        events = [
+        [
             event
             async for event in ollama_provider.stream_response(req, request_id="REQ")
         ]
 
-    assert_canonical_stream_error_envelope(
-        events, user_message_substr="Provider API request failed"
-    )
-    assert "REQ" in "".join(events)
+    assert "Provider API request failed" in exc_info.value.message
+    assert "REQ" in exc_info.value.message
 
 
 @pytest.mark.asyncio

tests/providers/test_openai_compat_5xx_retry.py

@@ -9,10 +9,10 @@ from httpx import Request, Response
 
 from config.nim import NimSettings
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.nvidia_nim import NvidiaNimProvider
 from providers.rate_limit import GlobalRateLimiter
 from tests.providers.test_nvidia_nim import MockRequest
-from tests.stream_contract import assert_canonical_stream_error_envelope
 
 
 def _internal_5xx(code: int) -> openai.InternalServerError:
@@ -148,10 +148,9 @@ async def test_nim_stream_connection_error_exhausted_emits_cause_chain():
             ) as mock_create,
             patch("asyncio.sleep", new_callable=AsyncMock),
             patch("providers.transports.openai_chat.stream.trace_event") as trace,
+            pytest.raises(ProviderError) as exc_info,
         ):
-            events = [
-                e async for e in provider.stream_response(req, request_id="req_conn")
-            ]
+            [e async for e in provider.stream_response(req, request_id="req_conn")]
 
         assert mock_create.await_count == 5
         error_traces = [
@@ -161,9 +160,8 @@ async def test_nim_stream_connection_error_exhausted_emits_cause_chain():
         ]
         assert error_traces[-1]["request_id"] == "req_conn"
         assert error_traces[-1]["exc_type"] == "APIConnectionError"
-        assert_canonical_stream_error_envelope(
-            events,
-            user_message_substr="Caused by:\nConnectError: upstream disconnected",
+        assert (
+            "Caused by:\nConnectError: upstream disconnected" in exc_info.value.message
         )
     finally:
         GlobalRateLimiter.reset_instance()
@@ -206,10 +204,10 @@ async def test_nim_stream_openai_5xx_exhausted_emits_user_message(
             patch("asyncio.sleep", new_callable=AsyncMock),
         ):
             mock_create.side_effect = _internal_5xx(status_code)
-            events = [e async for e in provider.stream_response(req)]
+            with pytest.raises(ProviderError) as exc_info:
+                [e async for e in provider.stream_response(req)]
 
         assert mock_create.await_count == 5
-        blob = "".join(events)
-        assert expect_substr in blob.lower()
+        assert expect_substr in exc_info.value.message.lower()
     finally:
         GlobalRateLimiter.reset_instance()

tests/providers/test_provider_transport_logging.py

@@ -11,6 +11,7 @@ import pytest
 from config.constants import NATIVE_MESSAGES_ERROR_BODY_LOG_CAP_BYTES
 from config.nim import NimSettings
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.nvidia_nim import NvidiaNimProvider
 from providers.transports.anthropic_messages import stream as native_stream
 from tests.provider_request_mocks import make_openai_compat_stream_request
@@ -71,8 +72,9 @@ async def test_native_non_200_logs_exclude_body_text_by_default(
             return_value=response,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_UPSTREAM_BODY" not in messages
@@ -96,8 +98,9 @@ async def test_native_non_200_logs_body_when_verbose(caplog, provider_config):
             return_value=response,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_UPSTREAM_BODY" in messages
@@ -124,8 +127,9 @@ async def test_native_non_200_verbose_logs_only_capped_error_body(
             return_value=response,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_TAIL_NOT_LOGGED" not in messages
@@ -151,8 +155,9 @@ async def test_native_non_200_default_does_not_read_oversized_body(
             return_value=response,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "LEAK_MARKER" not in messages
@@ -189,8 +194,9 @@ async def test_native_stream_failure_logs_exclude_exception_str_by_default(
         ),
         patch.object(native_stream, "iter_sse_events", boom),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_DETAIL" not in messages
@@ -225,8 +231,9 @@ async def test_openai_compat_stream_failure_default_logs_exclude_exception_str(c
             _noop_slot,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_OPENAI_COMPAT" not in messages
@@ -264,8 +271,9 @@ async def test_openai_compat_stream_failure_default_logs_cause_types_only(caplog
             _noop_slot,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_CAUSE_DETAIL" not in messages
@@ -300,8 +308,9 @@ async def test_openai_compat_stream_failure_respects_verbose_flag(caplog):
             _noop_slot,
         ),
         caplog.at_level(logging.ERROR),
+        pytest.raises(ProviderError),
     ):
-        _ = [e async for e in provider.stream_response(req)]
+        [e async for e in provider.stream_response(req)]
 
     messages = " | ".join(r.getMessage() for r in caplog.records)
     assert "SECRET_OPENAI_COMPAT" in messages

tests/providers/test_streaming_errors.py

@@ -10,7 +10,6 @@ import pytest
 
 from config.nim import NimSettings
 from core.anthropic.stream_contracts import (
-    assert_anthropic_stream_contract,
     parse_sse_text,
 )
 from core.anthropic.streaming import (
@@ -20,6 +19,7 @@ from core.anthropic.streaming import (
     make_text_recovery_body,
 )
 from providers.base import ProviderConfig
+from providers.exceptions import ProviderError
 from providers.nvidia_nim import NvidiaNimProvider
 from providers.transports.openai_chat.recovery import OpenAIChatRecovery
 from providers.transports.openai_chat.tool_calls import (
@@ -128,6 +128,12 @@ async def _collect_stream(provider, request):
     return [e async for e in provider.stream_response(request)]
 
 
+async def _collect_stream_error(provider, request, **kwargs) -> ProviderError:
+    with pytest.raises(ProviderError) as exc_info:
+        [e async for e in provider.stream_response(request, **kwargs)]
+    return exc_info.value
+
+
 def _assert_no_content_deltas_after_error_text(
     events: list[str], error_substr: str
 ) -> None:
@@ -172,14 +178,11 @@ class TestStreamingExceptionHandling:
     """Tests for error paths during stream_response."""
 
     @pytest.mark.asyncio
-    async def test_api_error_emits_sse_error_event(self):
-        """When API raises during streaming, SSE error event is emitted."""
+    async def test_pre_start_api_error_raises_provider_error(self):
+        """Before holdback commit, provider failures raise for API-level non-200."""
         provider = _make_provider()
         request = _make_request()
 
-        mock_stream = AsyncMock()
-        mock_stream.__aiter__ = MagicMock(side_effect=RuntimeError("API failed"))
-
         with (
             patch.object(
                 provider._client.chat.completions,
@@ -194,21 +197,13 @@ class TestStreamingExceptionHandling:
                 return_value=False,
             ),
         ):
-            events = await _collect_stream(provider, request)
+            error = await _collect_stream_error(provider, request)
 
-        # Should have message_start, error text block, close blocks, message_delta, message_stop
-        event_text = "".join(events)
-        assert "message_start" in event_text
-        assert "API failed" in event_text
-        assert "message_stop" in event_text
-        parsed = parse_sse_text(event_text)
-        assert parsed[0].event == "message_start"
-        assert sum(event.event == "message_start" for event in parsed) == 1
-        _assert_no_content_deltas_after_error_text(events, "API failed")
+        assert "API failed" in error.message
 
     @pytest.mark.asyncio
-    async def test_read_timeout_with_empty_message_emits_fallback(self):
-        """ReadTimeout(TimeoutError()) should emit a visible, non-empty timeout message."""
+    async def test_read_timeout_with_empty_message_raises_fallback(self):
+        """ReadTimeout(TimeoutError()) should raise a non-empty timeout message."""
         provider = _make_provider()
         request = _make_request()
 
@@ -225,24 +220,20 @@ class TestStreamingExceptionHandling:
                 new_callable=AsyncMock,
                 return_value=False,
             ),
+            patch("asyncio.sleep", new_callable=AsyncMock),
         ):
-            events = [
-                e
-                async for e in provider.stream_response(
-                    request,
-                    request_id="req_timeout123",
-                )
-            ]
+            error = await _collect_stream_error(
+                provider,
+                request,
+                request_id="req_timeout123",
+            )
 
-        event_text = "".join(events)
-        assert "timed out after" in event_text
-        assert "Request ID: req_timeout123" in event_text
-        assert "message_stop" in event_text
-        _assert_no_content_deltas_after_error_text(events, "timed out after")
+        assert "timed out after" in error.message
+        assert "Request ID: req_timeout123" in error.message
 
     @pytest.mark.asyncio
-    async def test_error_after_partial_content(self):
-        """Error after partial content: blocks closed, error emitted."""
+    async def test_error_after_precommit_partial_content_raises(self):
+        """Precommit partial text is discarded so the API can return non-200."""
         provider = _make_provider()
         request = _make_request()
 
@@ -263,13 +254,9 @@ class TestStreamingExceptionHandling:
                 return_value=False,
             ),
         ):
-            events = await _collect_stream(provider, request)
+            error = await _collect_stream_error(provider, request)
 
-        event_text = "".join(events)
-        assert "Hello" in event_text
-        assert "Connection lost" in event_text
-        assert "message_stop" in event_text
-        _assert_no_content_deltas_after_error_text(events, "Connection lost")
+        assert "Connection lost" in error.message
 
     @pytest.mark.asyncio
     async def test_error_after_native_tool_call_uses_top_level_error_event(self):
@@ -568,28 +555,21 @@ class TestStreamingExceptionHandling:
             new_callable=AsyncMock,
             side_effect=error,
         ):
-            events = [
-                e
-                async for e in provider.stream_response(
-                    request,
-                    request_id="REQ405",
-                )
-            ]
+            stream_error = await _collect_stream_error(
+                provider,
+                request,
+                request_id="REQ405",
+            )
 
-        event_text = "".join(events)
         assert (
             "Upstream provider NIM rejected the request method or endpoint (HTTP 405)."
-            in event_text
-        )
-        assert "Request ID: REQ405" in event_text
-        _assert_no_content_deltas_after_error_text(
-            events,
-            "Upstream provider NIM rejected the request method or endpoint (HTTP 405).",
+            in stream_error.message
         )
+        assert "Request ID: REQ405" in stream_error.message
 
     @pytest.mark.asyncio
     async def test_stream_with_openai_bad_request_surfaces_upstream_body(self):
-        """OpenAI SDK bodies should be emitted so users can copy exact provider errors."""
+        """OpenAI SDK bodies should be raised so users can copy exact provider errors."""
         provider = _make_provider()
         request = _make_request()
         response = httpx.Response(
@@ -610,33 +590,20 @@ class TestStreamingExceptionHandling:
             new_callable=AsyncMock,
             side_effect=error,
         ):
-            events = [
-                e
-                async for e in provider.stream_response(
-                    request,
-                    request_id="REQ_BODY",
-                )
-            ]
+            stream_error = await _collect_stream_error(
+                provider,
+                request,
+                request_id="REQ_BODY",
+            )
 
-        event_text = "".join(events)
-        message_text = "".join(
-            str(ev.data.get("delta", {}).get("text", ""))
-            for ev in parse_sse_text(event_text)
-            if ev.event == "content_block_delta"
-            and ev.data.get("delta", {}).get("type") == "text_delta"
-        )
-        assert "Upstream provider NIM returned HTTP 400." in event_text
-        assert "Category: BadRequest" in event_text
-        assert "Thinking mode does not support this tool_choice" in event_text
+        assert "Upstream provider NIM returned HTTP 400." in stream_error.message
+        assert "Category: BadRequest" in stream_error.message
+        assert "Thinking mode does not support this tool_choice" in stream_error.message
         assert (
             '{"error":{"type":"BadRequest","message":"Thinking mode does not support this tool_choice"}}'
-            in message_text
-        )
-        assert "Request ID: REQ_BODY" in event_text
-        _assert_no_content_deltas_after_error_text(
-            events,
-            "Upstream provider NIM returned HTTP 400.",
+            in stream_error.message
         )
+        assert "Request ID: REQ_BODY" in stream_error.message
 
     @pytest.mark.asyncio
     async def test_error_after_native_tool_call_top_level_error_includes_body(self):
@@ -956,10 +923,10 @@ class TestStreamingExceptionHandling:
             new_callable=AsyncMock,
             return_value=stream,
         ):
-            events = await _collect_stream(provider, request)
+            error = await _collect_stream_error(provider, request)
 
         assert stream.closed is True
-        assert "provider stream failed" in "".join(events).lower()
+        assert "provider stream failed" in error.message.lower()
 
     @pytest.mark.asyncio
     async def test_truncated_recovery_stream_falls_back_to_error_tail(self):
@@ -1395,13 +1362,9 @@ class TestStreamChunkEdgeCases:
                 return_value=False,
             ),
         ):
-            events = await _collect_stream(provider, request)
+            error = await _collect_stream_error(provider, request)
 
-        event_text = "".join(events)
-        assert "Partial" in event_text
-        assert "Connection reset" in event_text
-        assert "message_stop" in event_text
-        _assert_no_content_deltas_after_error_text(events, "Connection reset")
+        assert "Connection reset" in error.message
 
     def test_stream_malformed_tool_args_chunked(self):
         """Chunked tool args that never form valid JSON are flushed with {}."""
@@ -1456,7 +1419,6 @@ async def test_openai_compat_stream_ends_with_contract_when_tool_name_never_arri
             return_value=False,
         ),
     ):
-        events = await _collect_stream(provider, request)
-    text = "".join(events)
-    assert_anthropic_stream_contract(parse_sse_text(text))
-    assert "text_delta" in text
+        error = await _collect_stream_error(provider, request)
+
+    assert "Provider stream ended without finish_reason." in error.message

uv.lock

@@ -561,7 +561,7 @@ wheels = [
 
 [[package]]
 name = "free-claude-code"
-version = "3.4.12"
+version = "3.4.13"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },