fix(models): status-less stream faults classify for every provider family

src/marsys/agents/exceptions.py

@@ -1046,35 +1046,67 @@ def from_provider_response(
                 if "token" in message.lower() and ("expired" in message.lower() or "invalid" in message.lower()):
                     classification = APIErrorClassification.AUTHENTICATION_FAILED.value
 
-        elif provider in ("anthropic", "anthropic-oauth") and raw_response and raw_response.get("error"):
-            # NO status code: an in-stream SSE error event (Anthropic delivers stream
-            # failures as `{"type":"error","error":{...}}` under HTTP 200). Classify by
-            # the documented error type — the official streaming docs equate
-            # overloaded_error to HTTP 529, so it must be retryable exactly like a 5xx.
+        elif raw_response and raw_response.get("error"):
+            # NO status code: an in-stream SSE error event (providers deliver stream failures
+            # under HTTP 200, so there is no status to classify by). The message/type/code are
+            # read BEFORE the provider dispatch: whatever the provider, and however novel its
+            # error type, its real words survive to the caller — never a synthetic "API Error".
+            # A status-less fault that matches no provider arm below stays UNKNOWN and
+            # non-retryable, message intact.
             error_data = raw_response.get("error", {}) or {}
             message = error_data.get("message", message)
             api_error_type = error_data.get("type")
-            if api_error_type == "overloaded_error":
-                classification = APIErrorClassification.SERVICE_UNAVAILABLE.value
-                is_retryable = True
-                retry_after = 10
-            elif api_error_type == "rate_limit_error":
-                classification = APIErrorClassification.RATE_LIMIT.value
-                is_retryable = True
-                retry_after = 60
-            elif api_error_type == "authentication_error":
-                classification = APIErrorClassification.AUTHENTICATION_FAILED.value
-            elif api_error_type == "permission_error":
-                classification = APIErrorClassification.PERMISSION_DENIED.value
-            elif api_error_type == "invalid_request_error":
-                classification = APIErrorClassification.INVALID_REQUEST.value
-            elif api_error_type == "api_error":
-                # The provider's own "internal error" type (≙ HTTP 500).
-                classification = APIErrorClassification.SERVICE_UNAVAILABLE.value
-                is_retryable = True
-                retry_after = 10
-            # Unknown in-stream types keep UNKNOWN classification but the REAL
-            # provider message above — never a destroyed/synthetic one.
+            api_error_code = error_data.get("code")
+
+            if provider in ("anthropic", "anthropic-oauth", "bedrock"):
+                # Anthropic's documented stream error types — the official streaming docs equate
+                # overloaded_error to HTTP 529, so it must be retryable exactly like a 5xx.
+                # Bedrock serves the Messages API and emits the same error envelope, so it
+                # classifies identically to first-party Anthropic — the same share the
+                # status-code arm above already makes; without it a Bedrock in-stream fault
+                # falls through unclassified and a retryable overload dispositions as terminal.
+                if api_error_type == "overloaded_error":
+                    classification = APIErrorClassification.SERVICE_UNAVAILABLE.value
+                    is_retryable = True
+                    retry_after = 10
+                elif api_error_type == "rate_limit_error":
+                    classification = APIErrorClassification.RATE_LIMIT.value
+                    is_retryable = True
+                    retry_after = 60
+                elif api_error_type == "authentication_error":
+                    classification = APIErrorClassification.AUTHENTICATION_FAILED.value
+                elif api_error_type == "permission_error":
+                    classification = APIErrorClassification.PERMISSION_DENIED.value
+                elif api_error_type == "invalid_request_error":
+                    classification = APIErrorClassification.INVALID_REQUEST.value
+                elif api_error_type == "api_error":
+                    # The provider's own "internal error" type (≙ HTTP 500).
+                    classification = APIErrorClassification.SERVICE_UNAVAILABLE.value
+                    is_retryable = True
+                    retry_after = 10
+
+            elif provider in ("openai", "azure"):
+                # The Responses API names a stream fault by its `code` (a closed vocabulary
+                # upstream); `type` is the fallback key. Azure serves the same wire contract as
+                # first-party OpenAI — the same share the status-code arm above already makes.
+                key = api_error_code or api_error_type
+                if key == "insufficient_quota":
+                    classification = APIErrorClassification.INSUFFICIENT_CREDITS.value
+                elif key in ("rate_limit_exceeded", "rate_limit_error"):
+                    classification = APIErrorClassification.RATE_LIMIT.value
+                    is_retryable = True
+                    retry_after = 60
+                elif key in ("server_error", "response.failed"):
+                    # `server_error` is the provider's own 5xx; a bare `response.failed` (no
+                    # code) is the provider saying the response failed without saying why —
+                    # the same provider-side-fault shape, so the same retryable disposition.
+                    classification = APIErrorClassification.SERVICE_UNAVAILABLE.value
+                    is_retryable = True
+                    retry_after = 10
+                # Every other code — request-shaped faults (invalid_prompt, image_*…) and the
+                # adapters' synthetic max_retries / incomplete_stream markers — keeps UNKNOWN
+                # and non-retryable, real message intact: flipping any of them to retryable is
+                # a policy decision, not a classification repair.
 
         elif (
             provider in ("anthropic", "anthropic-oauth")

src/marsys/models/adapters/streaming.py

@@ -339,7 +339,14 @@ def feed(self, data: Dict[str, Any]) -> bool:
                 return False
 
         elif event_type == "error":
-            self.error = data.get("error", {}) or {"type": "unknown"}
+            # The Responses `error` event is FLAT — code/message/param sit on the event itself,
+            # not under a nested "error" key (that nesting is Anthropic's grammar, and these
+            # accumulators exist precisely because the grammars differ). Reading it nested returns
+            # {} and destroys the provider's verdict before classification, so a retryable fault
+            # dispositions as unknown/terminal. Normalize to the {code, message} shape the
+            # `response.failed` arm above already yields: one shape downstream.
+            flat = {k: data.get(k) for k in ("code", "message") if data.get(k) is not None}
+            self.error = flat or {"type": "unknown"}
             return False
 
         return True

tests/models/test_adapter_harmonize.py

@@ -229,27 +229,56 @@ def test_oauth_stream_error_event_raises_classified_retryable(monkeypatch):
 
 
 @pytest.mark.parametrize(
-    "error_type, expected_classification, expected_retryable",
+    "provider, error, expected_classification, expected_retryable",
     [
-        ("overloaded_error", APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
-        ("rate_limit_error", APIErrorClassification.RATE_LIMIT.value, True),
-        ("api_error", APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
-        ("authentication_error", APIErrorClassification.AUTHENTICATION_FAILED.value, False),
-        ("invalid_request_error", APIErrorClassification.INVALID_REQUEST.value, False),
-        ("never_seen_before", APIErrorClassification.UNKNOWN.value, False),
+        # Anthropic family — classified by the documented stream error `type`.
+        ("anthropic-oauth", {"type": "overloaded_error"}, APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
+        ("anthropic-oauth", {"type": "rate_limit_error"}, APIErrorClassification.RATE_LIMIT.value, True),
+        ("anthropic-oauth", {"type": "api_error"}, APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
+        ("anthropic-oauth", {"type": "authentication_error"}, APIErrorClassification.AUTHENTICATION_FAILED.value, False),
+        ("anthropic-oauth", {"type": "invalid_request_error"}, APIErrorClassification.INVALID_REQUEST.value, False),
+        ("anthropic-oauth", {"type": "never_seen_before"}, APIErrorClassification.UNKNOWN.value, False),
+        # Bedrock shares Anthropic's wire contract — a bedrock in-stream overload is as
+        # retryable as a first-party one, not an unclassified terminal fault.
+        ("bedrock", {"type": "overloaded_error"}, APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
+        ("bedrock", {"type": "rate_limit_error"}, APIErrorClassification.RATE_LIMIT.value, True),
+        # OpenAI family (azure serves the identical Responses contract) — classified by `code`
+        # first, `type` as the fallback key.
+        ("azure", {"code": "server_error"}, APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
+        ("azure", {"code": "rate_limit_exceeded"}, APIErrorClassification.RATE_LIMIT.value, True),
+        ("azure", {"code": "rate_limit_exceeded", "type": "rate_limit_error"}, APIErrorClassification.RATE_LIMIT.value, True),
+        # code and type mapping DIFFERENTLY: the code must win (the Responses vocabulary names
+        # the fault; `type` is only the fallback key).
+        ("azure", {"code": "insufficient_quota", "type": "rate_limit_error"}, APIErrorClassification.INSUFFICIENT_CREDITS.value, False),
+        ("openai", {"code": "insufficient_quota"}, APIErrorClassification.INSUFFICIENT_CREDITS.value, False),
+        # a bare response.failed — the provider said the response failed and not why.
+        ("azure", {"type": "response.failed"}, APIErrorClassification.SERVICE_UNAVAILABLE.value, True),
+        # Request-shaped codes stay UNKNOWN/non-retryable with the real words.
+        ("azure", {"code": "invalid_prompt"}, APIErrorClassification.UNKNOWN.value, False),
+        # The adapters' synthetic markers keep today's disposition: flipping either to
+        # retryable is a policy ruling, not a classification repair — these rows pin that a
+        # future flip is deliberate, never drift.
+        ("azure", {"type": "max_retries"}, APIErrorClassification.UNKNOWN.value, False),
+        ("azure", {"type": "incomplete_stream"}, APIErrorClassification.UNKNOWN.value, False),
+        # A provider outside both families keeps UNKNOWN — but its words still survive.
+        ("openrouter", {"code": "whatever"}, APIErrorClassification.UNKNOWN.value, False),
     ],
 )
-def test_status_less_stream_errors_classify_by_type(error_type, expected_classification, expected_retryable):
+def test_status_less_stream_errors_classify_by_type(provider, error, expected_classification, expected_retryable):
     """`from_provider_response` accepts a plain error dict (no Response, no status —
-    the in-stream case) and classifies by the documented error type. Unknown types
-    keep UNKNOWN but the REAL provider message survives."""
+    the in-stream case) and classifies it per provider family. Unmapped codes/types
+    keep UNKNOWN but the REAL provider message survives for every provider."""
     err = ModelAPIError.from_provider_response(
-        provider="anthropic-oauth",
-        response={"error": {"type": error_type, "message": "the real provider words"}},
+        provider=provider,
+        response={"error": {**error, "message": "the real provider words"}},
     )
     assert err.classification == expected_classification
     assert err.is_retryable is expected_retryable
     assert "the real provider words" in str(err)
+    if expected_retryable:
+        # A retryable verdict must carry a usable delay — a retry ladder reading
+        # retry_after=None/0 degenerates to a hot loop or a policy-side guess.
+        assert err.retry_after and err.retry_after > 0
 
 
 def test_oauth_truncation_empty_harmonizes_valid_with_placeholder():

tests/models/test_adapter_streaming.py

@@ -379,6 +379,39 @@ def test_responses_failed_event_is_terminal():
     assert acc.error == {"code": "server_error", "message": "x"}
 
 
+def test_responses_flat_error_event_keeps_the_providers_words():
+    # The Responses `error` event is FLAT — code/message sit on the event itself, unlike
+    # Anthropic's nested {"error": {...}}. Reading it nested yields {} and the provider's
+    # verdict is destroyed before classification, so a retryable fault dispositions as
+    # unknown/terminal downstream. Production turns died exactly that way.
+    acc = ResponsesStreamAccumulator()
+    ok = acc.feed({"type": "error", "code": "rate_limit_exceeded",
+                   "message": "You exceeded your current quota of requests.",
+                   "param": None, "sequence_number": 7})
+    assert not ok
+    assert acc.error == {"code": "rate_limit_exceeded",
+                         "message": "You exceeded your current quota of requests."}
+
+
+def test_responses_failed_without_error_object_yields_the_marker():
+    # A `response.failed` carrying no error object yields the bare marker the classifier's
+    # code-less arm keys on — this pins the wire→classifier join for the "provider said it
+    # failed and not why" shape.
+    acc = ResponsesStreamAccumulator()
+    ok = acc.feed({"type": "response.failed", "response": {"id": "resp_x", "status": "failed"}})
+    assert not ok
+    assert acc.error == {"type": "response.failed"}
+
+
+def test_responses_bare_error_event_still_terminates():
+    # An error event carrying neither code nor message still ends the stream with a
+    # non-empty error marker (the pre-existing fallback).
+    acc = ResponsesStreamAccumulator()
+    ok = acc.feed({"type": "error"})
+    assert not ok
+    assert acc.error == {"type": "unknown"}
+
+
 # ---------------------------------------------------------------------------
 # end-to-end arun_streaming over a fake aiohttp session
 # ---------------------------------------------------------------------------
@@ -582,6 +615,46 @@ async def test_openai_stream_without_completion_is_a_typed_failure():
         await adapter.arun_streaming([{"role": "user", "content": "q"}], max_tokens=4096)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "fault_event, expected_classification",
+    [
+        # the flat top-level `error` event — the grammar the accumulator once read nested
+        ({"type": "error", "code": "rate_limit_exceeded",
+          "message": "You exceeded your current quota of requests."}, "rate_limit"),
+        # the enveloped `response.failed` grammar
+        ({"type": "response.failed", "response": {"error": {
+            "code": "server_error",
+            "message": "The server had an error while processing your request."}}},
+         "service_unavailable"),
+    ],
+)
+async def test_openai_in_stream_fault_surfaces_classified_with_the_providers_words(
+    fault_event, expected_classification
+):
+    # End-to-end over the fake session: a mid-stream fault under HTTP 200 must surface as a
+    # CLASSIFIED, retryable-aware ModelAPIError carrying the provider's real words — that verdict
+    # is the input the caller's turn-level retry ladder runs on, and an UNKNOWN/non-retryable
+    # default silently turns a transient provider blip into permanently destroyed work.
+    from marsys.agents.exceptions import ModelAPIError
+
+    faulted = RESPONSES_STREAM[:4] + [fault_event]  # never reaches response.completed
+    session = _FakeSession([_FakeStreamResponse(200, _sse_lines(faulted))])
+    adapter = AsyncOpenAIAdapter(
+        model_name="gpt-test", api_key="k",
+        base_url="https://api.openai.com/v1", max_tokens=4096, streaming=True,
+    )
+    adapter._session = session
+
+    with pytest.raises(ModelAPIError) as exc_info:
+        await adapter.arun_streaming([{"role": "user", "content": "q"}], max_tokens=4096)
+    err = exc_info.value
+    assert err.classification == expected_classification
+    assert err.is_retryable is True
+    words = fault_event.get("message") or fault_event["response"]["error"]["message"]
+    assert words in str(err)  # the provider's words, not a synthetic "API Error"
+
+
 # ---------------------------------------------------------------------------
 # the tap kwarg never reaches payload builders (the unknown-param warn trap)
 # ---------------------------------------------------------------------------