diff --git a/src/marsys/agents/exceptions.py b/src/marsys/agents/exceptions.py index 47a1343b..4ac5b27a 100644 --- a/src/marsys/agents/exceptions.py +++ b/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") diff --git a/src/marsys/models/adapters/streaming.py b/src/marsys/models/adapters/streaming.py index 0defdde5..7a4d212c 100644 --- a/src/marsys/models/adapters/streaming.py +++ b/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 diff --git a/tests/models/test_adapter_harmonize.py b/tests/models/test_adapter_harmonize.py index f7432017..bf0b1790 100644 --- a/tests/models/test_adapter_harmonize.py +++ b/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(): diff --git a/tests/models/test_adapter_streaming.py b/tests/models/test_adapter_streaming.py index 1a6ea220..863628b5 100644 --- a/tests/models/test_adapter_streaming.py +++ b/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) # ---------------------------------------------------------------------------