From 01bb4b3132b71b18ace5e0b04d0f3f531e9b1569 Mon Sep 17 00:00:00 2001 From: Genadij Blinov Date: Sun, 6 Sep 2026 07:32:34 +0300 Subject: [PATCH 1/3] fix(providers): retry transient OpenAI stream errors --- src/conductor/providers/_pydantic_ai/retry.py | 6 +++ .../test_providers/test_pydantic_ai_retry.py | 43 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/src/conductor/providers/_pydantic_ai/retry.py b/src/conductor/providers/_pydantic_ai/retry.py index 5b176098..1a24269c 100644 --- a/src/conductor/providers/_pydantic_ai/retry.py +++ b/src/conductor/providers/_pydantic_ai/retry.py @@ -169,6 +169,12 @@ def _is_retryable_error(exception: Exception) -> bool: if type(exception) is ModelAPIError: return True + # The OpenAI SDK raises a bare APIError for error objects delivered after + # an SSE response has started. Unlike HTTP APIStatusError subclasses, that + # exception has no status code; retry only server/rate-limit payload types. + if openai is not None and type(exception) is openai.APIError: + return exception.type in {"server_error", "rate_limit_error"} + error_type_name = type(exception).__name__ retryable_names = { diff --git a/tests/test_providers/test_pydantic_ai_retry.py b/tests/test_providers/test_pydantic_ai_retry.py index 8e4d6d65..2b4fb7d3 100644 --- a/tests/test_providers/test_pydantic_ai_retry.py +++ b/tests/test_providers/test_pydantic_ai_retry.py @@ -228,6 +228,17 @@ def test_raw_openai_api_status_5xx_and_429_are_retryable(self) -> None: assert _is_retryable_error(_make_openai_status_error(503)) is True assert _is_retryable_error(_make_openai_status_error(429)) is True + def test_raw_openai_stream_validation_error_is_fatal(self) -> None: + # Requirement: a structured client-side error inside an SSE stream must not + # become retryable merely because the SDK reports it as a base APIError. + error = openai.APIError( + message="invalid request", + request=_make_http_request(), + body={"type": "invalid_request_error", "code": "invalid_value"}, + ) + + assert _is_retryable_error(error) is False + def test_raw_openai_api_status_4xx_are_fatal(self) -> None: """Raw openai APIStatusError 400/401/403/404 must be fatal.""" assert _is_retryable_error(_make_openai_status_error(400)) is False @@ -344,6 +355,38 @@ def callback(event_type: str, payload: dict[str, Any]) -> None: } mock_sleep.assert_called_once_with(2.0) + @pytest.mark.asyncio + async def test_real_openai_stream_server_error_retries_once_then_succeeds(self) -> None: + # Requirement: a transient error delivered inside an OpenAI-compatible SSE stream + # must consume the configured retry attempt instead of failing after the first call. + attempts = 0 + request = _make_http_request() + stream_error = openai.APIError( + message="upstream overloaded", + request=request, + body={"type": "server_error", "code": "internal_server_error"}, + ) + + async def factory() -> str: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise stream_error + return "success" + + config = RetryConfig(max_attempts=2, base_delay=0.0, jitter=0.0) + with patch("conductor.providers._pydantic_ai.retry.asyncio.sleep") as mock_sleep: + result = await execute_with_retry( + factory, + retry_config=config, + event_callback=None, + agent_name="openai-stream-retryer", + ) + + assert result == "success" + assert attempts == 2 + mock_sleep.assert_called_once_with(0.0) + @pytest.mark.asyncio async def test_real_openai_bad_request_error_is_not_retried(self) -> None: """Requirement: openai.BadRequestError is fatal and must not be retried; From 42eeb730ffc9ce81465c9dd15ab67640de5a13b8 Mon Sep 17 00:00:00 2001 From: Genadij Blinov Date: Tue, 8 Sep 2026 22:11:13 +0300 Subject: [PATCH 2/3] fix(providers): match stream-error retry classification to actual OpenAI payloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The retryable set for a bare openai.APIError used Anthropic's "rate_limit_error" string, which a real OpenAI 429 never sends — its mid-stream rate limits arrive as type "requests"/"tokens" with code "rate_limit_exceeded", so they still failed on the first attempt. Widen the marker set to the vocabulary OpenAI and Anthropic-shaped gateways actually emit, with sources cited, and treat a payload with no parseable type (a non-object error value from an Ollama/vLLM gateway, or an Azure-style {"code": ...} shape) like a broken stream: by the time a stream has started, auth and request validation have passed. The module docstring also claimed only the translated ModelHTTPError/ ModelAPIError types are relied on at runtime — now false, since the bare-APIError branch depends on pydantic-ai continuing not to translate it. Rewrite it to say so, drop the stale 1.44.0 floor note, and pin the upstream assumptions with canary tests: pydantic-ai's _map_api_errors leaves a bare APIError untranslated, and a real openai.Stream puts the SSE error object itself into APIError.body. Every marker in the set is now pinned in both payload positions, and the exact-type guard has a test that fails under an isinstance mutation. --- CHANGELOG.md | 15 ++ src/conductor/providers/_pydantic_ai/retry.py | 85 ++++++-- .../test_providers/test_pydantic_ai_retry.py | 184 +++++++++++++++++- 3 files changed, 262 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f3a369a1..6be69ade 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased](https://github.com/microsoft/conductor/compare/v0.1.36...HEAD) +### Fixed + +- **`openai`: retry transient errors delivered inside an SSE stream** (#506) — + the OpenAI SDK raises a bare `openai.APIError` (no HTTP status) for an + `error` object embedded in a stream, which pydantic-ai does not translate, + so a configured `retry:` policy was skipped and the run failed after the + first attempt. Now retried: OpenAI mid-stream 5xx (`server_error` / + `internal_server_error`), OpenAI rate limits (`type` `requests` / `tokens` + with code `rate_limit_exceeded`), Anthropic-shaped gateway errors proxied + unchanged (`rate_limit_error` / `overloaded_error` / `api_error`), and + stream errors with no parseable payload `type` (a non-object `error` value + from an Ollama/vLLM gateway, or an Azure-style `{"code": ...}` shape), + which are treated like broken streams. Still fatal: recognized client-side + payload types (e.g. `invalid_request_error`) and every HTTP 4xx. + ### Added - **Opt-in `runtime.provider.setting_sources` on `claude-agent-sdk`** (#501) — diff --git a/src/conductor/providers/_pydantic_ai/retry.py b/src/conductor/providers/_pydantic_ai/retry.py index 1a24269c..8f008a58 100644 --- a/src/conductor/providers/_pydantic_ai/retry.py +++ b/src/conductor/providers/_pydantic_ai/retry.py @@ -9,20 +9,31 @@ Conductor retries the whole call. pydantic-ai translates the Anthropic SDK's own exceptions before Conductor -ever sees them — a private helper (``_map_api_errors`` in pydantic-ai 2.x; -written inline in ``AnthropicModel`` at the 1.44.0 floor pinned in -pyproject.toml, with identical resulting behavior) wraps the SDK call: an -HTTP error response (``APIStatusError``, including ``RateLimitError``) -becomes ``ModelHTTPError``, and a transport failure (``APIConnectionError``, -including ``APITimeoutError``) becomes a bare ``ModelAPIError``. Those are -the exception types actually observed on this path, not the SDK's own +ever sees them — a private helper (``_map_api_errors`` in +``pydantic_ai.models.anthropic``) wraps the SDK call: an HTTP error response +(``APIStatusError``, including ``RateLimitError``) becomes ``ModelHTTPError``, +and a transport failure (``APIConnectionError``, including +``APITimeoutError``) becomes a bare ``ModelAPIError``. Those are the +exception types actually observed on the Anthropic path, not the SDK's own classes, so ``_is_retryable_error`` and ``_get_retry_after`` classify those -translated types directly (issue #454). Only the public ``ModelHTTPError``/ -``ModelAPIError`` types are relied on at runtime, so a change to the private -translator degrades this comment, not the code. The translation also drops -the original response headers, so a server's ``retry-after`` value is only +translated types directly (issue #454). The translation also drops the +original response headers, so a server's ``retry-after`` value is only recoverable through ``__cause__``, which the translator sets to the untranslated SDK exception via ``from e``. + +The OpenAI path is translated by the same-named ``_map_api_errors`` in +``pydantic_ai.models.openai``, which catches only ``APIStatusError`` +(-> ``ModelHTTPError``) and ``APIConnectionError`` (-> ``ModelAPIError``). +It deliberately does not catch the bare ``openai.APIError`` that +``openai/_streaming.py`` raises for an ``error`` object embedded in an SSE +body, so that one SDK class does reach Conductor untranslated and is +classified directly in ``_is_retryable_error``. If pydantic-ai widens that +``except`` clause, the branch goes dead and mid-stream errors arrive as +``ModelAPIError`` — which is unconditionally retryable, reversing the fatal +classification for client-side payload types. A canary test +(``test_pydantic_ai_does_not_translate_bare_openai_api_error``) pins the +upstream behavior so that change fails loudly instead of silently +re-classifying errors. """ from __future__ import annotations @@ -142,9 +153,12 @@ def _is_retryable_error(exception: Exception) -> bool: Extended to classify the ``ModelHTTPError``/``ModelAPIError`` types pydantic-ai actually raises on this path (see the module docstring; - issue #454). The SDK-class-name and ``anthropic.APIStatusError`` - fallback below is unreachable in production but kept intentionally — - see the comment at its definition. + issue #454), plus the bare ``openai.APIError`` the OpenAI SDK raises for + an error object embedded in an SSE body, which pydantic-ai does **not** + translate and which is therefore reachable in production. The + SDK-class-name and ``anthropic.APIStatusError`` fallback below remains + unreachable in production but is kept intentionally — see the comment + at its definition. """ if isinstance(exception, ProviderError): return exception.is_retryable @@ -169,11 +183,46 @@ def _is_retryable_error(exception: Exception) -> bool: if type(exception) is ModelAPIError: return True - # The OpenAI SDK raises a bare APIError for error objects delivered after - # an SSE response has started. Unlike HTTP APIStatusError subclasses, that - # exception has no status code; retry only server/rate-limit payload types. + # The OpenAI SDK raises a bare APIError for an error object embedded in + # an SSE body after the response has started (openai/_streaming.py). + # Unlike the HTTP APIStatusError subclasses, that exception carries no + # status code — only the payload's free-form `type`/`code` passthroughs + # (no SDK enum backs them), so the retryable vocabulary is spelled out + # here with its sources: + # - "server_error" / "internal_server_error": OpenAI mid-stream 5xx + # - "requests" / "tokens" + code "rate_limit_exceeded": an OpenAI 429 + # (see the code literals in openai/types/beta/threads/run.py) + # - "rate_limit_error" / "overloaded_error" / "api_error": + # Anthropic-shaped gateways (e.g. LiteLLM) proxying that vocabulary + # unchanged onto OpenAI-compatible endpoints + # An exact-type check, not isinstance: APIStatusError subclasses (e.g. + # BadRequestError) must keep their status-based classification below. if openai is not None and type(exception) is openai.APIError: - return exception.type in {"server_error", "rate_limit_error"} + payload_type = getattr(exception, "type", None) + payload_code = getattr(exception, "code", None) + if payload_type is None: + # A mid-stream error whose payload carries no `type` — a + # non-object `error` value ("upstream overloaded" from an + # Ollama/vLLM gateway) or a shape like Azure's {"code": "429"} — + # is indistinguishable from a broken stream, and by the time a + # stream has started, auth and request validation have already + # passed. Treat it like the transport failures above rather + # than burning the run. + return True + retryable_markers = { + "server_error", + "internal_server_error", + "requests", + "tokens", + "rate_limit_exceeded", + "rate_limit_error", + "overloaded_error", + "api_error", + } + # `in` is type-agnostic: the SDK does not coerce, so a payload like + # {"type": 500} arrives as an int, simply misses the set, and stays + # fatal (unknown marker) rather than crashing the comparison. + return payload_type in retryable_markers or payload_code in retryable_markers error_type_name = type(exception).__name__ diff --git a/tests/test_providers/test_pydantic_ai_retry.py b/tests/test_providers/test_pydantic_ai_retry.py index 2b4fb7d3..3da39a7d 100644 --- a/tests/test_providers/test_pydantic_ai_retry.py +++ b/tests/test_providers/test_pydantic_ai_retry.py @@ -229,8 +229,9 @@ def test_raw_openai_api_status_5xx_and_429_are_retryable(self) -> None: assert _is_retryable_error(_make_openai_status_error(429)) is True def test_raw_openai_stream_validation_error_is_fatal(self) -> None: - # Requirement: a structured client-side error inside an SSE stream must not - # become retryable merely because the SDK reports it as a base APIError. + # Requirement: a structured client-side error shaped like the one the + # SDK raises from an SSE body must not become retryable merely because + # the SDK reports it as a base APIError. error = openai.APIError( message="invalid request", request=_make_http_request(), @@ -239,6 +240,179 @@ def test_raw_openai_stream_validation_error_is_fatal(self) -> None: assert _is_retryable_error(error) is False + def test_raw_openai_stream_rate_limit_error_is_retryable(self) -> None: + # Requirement: a rate limit delivered mid-stream must be retryable in + # every shape OpenAI and Anthropic-shaped gateways send it. OpenAI's + # real 429 payload is type "requests"/"tokens" with code + # "rate_limit_exceeded" (literals in openai/types/beta/threads/run.py); + # "rate_limit_error" is the Anthropic vocabulary a gateway like + # LiteLLM proxies unchanged. The third row matches on the code marker + # alone — it is what pins "rate_limit_exceeded" in the retryable set. + request = _make_http_request() + for body in ( + {"type": "requests", "code": "rate_limit_exceeded"}, + {"type": "tokens", "code": "rate_limit_exceeded"}, + {"type": "rate_limit_error"}, + {"type": "throttled", "code": "rate_limit_exceeded"}, + ): + error = openai.APIError("slow down", request, body=body) + assert _is_retryable_error(error) is True, body + + def test_raw_openai_stream_gateway_overload_errors_are_retryable(self) -> None: + # Requirement: Anthropic-shaped gateways proxy their transient 5xx + # vocabulary ("overloaded_error", "api_error") onto OpenAI-compatible + # endpoints unchanged; both must retry. + request = _make_http_request() + for body in ({"type": "overloaded_error"}, {"type": "api_error"}): + error = openai.APIError("upstream overloaded", request, body=body) + assert _is_retryable_error(error) is True, body + + @pytest.mark.parametrize( + "marker", + [ + "server_error", + "internal_server_error", + "requests", + "tokens", + "rate_limit_exceeded", + "rate_limit_error", + "overloaded_error", + "api_error", + ], + ) + def test_every_documented_stream_marker_is_retryable_as_type_and_code( + self, marker: str + ) -> None: + # Requirement: every marker in the retryable set is load-bearing in + # both payload positions the classifier checks, so deleting one from + # the set is a loud failure rather than a silent behavior change. + request = _make_http_request() + assert ( + _is_retryable_error(openai.APIError("boom", request, body={"type": marker})) is True + ), marker + assert ( + _is_retryable_error( + openai.APIError("boom", request, body={"type": "unrelated", "code": marker}) + ) + is True + ), marker + + def test_openai_status_subclasses_are_not_captured_by_the_bare_arm(self) -> None: + # Requirement: the bare-APIError arm is an exact-type check — an + # APIStatusError subclass must keep its status-based classification + # even when its payload carries a marker from the retryable set (or + # a fatal one). This is the test that fails if the check is loosened + # to isinstance. + request = _make_http_request() + assert ( + _is_retryable_error( + openai.BadRequestError( + "bad", + response=httpx.Response(400, request=request), + body={"type": "server_error"}, + ) + ) + is False + ) + assert ( + _is_retryable_error( + openai.InternalServerError( + "boom", + response=httpx.Response(500, request=request), + body={"type": "invalid_request_error"}, + ) + ) + is True + ) + + @pytest.mark.parametrize("body", [None, "upstream overloaded", [1, 2], {"code": "x"}]) + def test_raw_openai_api_error_without_payload_type_is_retryable(self, body: object) -> None: + # Requirement: a mid-stream error with no parseable `type` — a + # non-object `error` value from an Ollama/vLLM gateway, or an + # Azure-style {"code": ...} shape — is treated like a broken stream + # (retryable), since auth and request validation have already passed + # once streaming starts. + assert _is_retryable_error(openai.APIError("boom", _make_http_request(), body=body)) is True + + def test_raw_openai_api_error_with_non_string_type_is_fatal(self) -> None: + # Requirement: the SDK does not coerce `type` — a payload of + # {"type": 500} arrives as an int. It must not crash classification, + # and an unknown marker stays fatal. + assert ( + _is_retryable_error(openai.APIError("boom", _make_http_request(), body={"type": 500})) + is False + ) + + def test_pydantic_ai_does_not_translate_bare_openai_api_error(self) -> None: + # Canary: the bare-APIError classification arm is only reachable while + # pydantic-ai's ``_map_api_errors`` lets a bare ``openai.APIError`` + # through untranslated. If upstream widens its ``except`` clause, this + # test fails — and the fatal classification for client-side payload + # types must be revisited, because those errors would then arrive as + # ``ModelAPIError``, which is unconditionally retryable. + from pydantic_ai.models.openai import _map_api_errors + + error = openai.APIError( + "boom", + _make_http_request(), + body={"type": "invalid_request_error"}, + ) + with pytest.raises(openai.APIError) as exc_info, _map_api_errors("gpt-4o"): + raise error + assert exc_info.value is error + + def test_sdk_stream_error_payload_shape_is_what_we_classify(self) -> None: + # Canary: drive a real ``openai.Stream`` over an SSE body carrying an + # error event and assert the SDK puts the error object itself into + # ``APIError.body`` (so `.type` sits at the top level, which is what + # the classifier reads). If a future SDK release nests the payload one + # level deeper, `.type` becomes None and the marker-based + # classification silently stops matching — this test fails instead. + with openai.OpenAI(api_key="test-key") as client: + request = httpx.Request("POST", "http://example.com/v1/chat/completions") + response = httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=( + b'data: {"error": {"type": "server_error", "message": "boom"}}\n\n' + b"data: [DONE]\n\n" + ), + request=request, + ) + stream = openai.Stream(cast_to=object, response=response, client=client) + + with pytest.raises(openai.APIError) as exc_info: + for _ in stream: + pass + + assert exc_info.value.type == "server_error" + assert exc_info.value.body == {"type": "server_error", "message": "boom"} + assert _is_retryable_error(exc_info.value) is True + + def test_sdk_stream_error_with_non_object_error_value(self) -> None: + # Canary: a gateway sending {"error": "upstream overloaded"} yields a + # bare APIError with .type None, the raw string body, and the SDK's + # fallback message — the exact shape the type-less retry branch + # exists for. + with openai.OpenAI(api_key="test-key") as client: + request = httpx.Request("POST", "http://example.com/v1/chat/completions") + response = httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=b'data: {"error": "upstream overloaded"}\n\ndata: [DONE]\n\n', + request=request, + ) + stream = openai.Stream(cast_to=object, response=response, client=client) + + with pytest.raises(openai.APIError) as exc_info: + for _ in stream: + pass + + assert exc_info.value.type is None + assert exc_info.value.body == "upstream overloaded" + assert str(exc_info.value) == "An error occurred during streaming" + assert _is_retryable_error(exc_info.value) is True + def test_raw_openai_api_status_4xx_are_fatal(self) -> None: """Raw openai APIStatusError 400/401/403/404 must be fatal.""" assert _is_retryable_error(_make_openai_status_error(400)) is False @@ -357,8 +531,10 @@ def callback(event_type: str, payload: dict[str, Any]) -> None: @pytest.mark.asyncio async def test_real_openai_stream_server_error_retries_once_then_succeeds(self) -> None: - # Requirement: a transient error delivered inside an OpenAI-compatible SSE stream - # must consume the configured retry attempt instead of failing after the first call. + # Requirement: a transient error shaped like the one the SDK raises + # from an SSE body must consume the configured retry attempt instead + # of failing after the first call. (The SDK's real SSE parsing is + # pinned separately by test_sdk_stream_error_payload_shape_is_what_we_classify.) attempts = 0 request = _make_http_request() stream_error = openai.APIError( From 4ab0868274d3d69bc78692eba7ddc90413f1e7b1 Mon Sep 17 00:00:00 2001 From: Genadij Blinov Date: Tue, 8 Sep 2026 22:13:55 +0300 Subject: [PATCH 3/3] fix(providers): surface declined and retry_on-filtered retries honestly Three gaps in how execute_with_retry reports a failure the retry policy did not take: - A retryable error declined by a narrowed retry_on filter escaped through a bare raise, so a raw SDK exception (e.g. a mid-stream openai.APIError) flew past callers that catch ProviderError. The gate now wraps it in a non-retryable ProviderError naming the declined category, chained to the original. - A retry taken logs a warning and emits agent_retry; a retry declined logged only at debug, which Conductor's default no-handler logging never emits. The decline is now logged at warning level too. - A bare openai.APIError's str() carries neither the payload type/code nor a non-object body, and the SDK substitutes a generic "An error occurred during streaming" message, so a fatal stream error told the user to check their API key with no evidence attached. The ProviderError message now includes the payload details via _describe_stream_error. --- CHANGELOG.md | 7 +- src/conductor/providers/_pydantic_ai/retry.py | 59 +++++++++++- .../test_providers/test_pydantic_ai_retry.py | 90 ++++++++++++++++++- 3 files changed, 148 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6be69ade..8278caac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 stream errors with no parseable payload `type` (a non-object `error` value from an Ollama/vLLM gateway, or an Azure-style `{"code": ...}` shape), which are treated like broken streams. Still fatal: recognized client-side - payload types (e.g. `invalid_request_error`) and every HTTP 4xx. + payload types (e.g. `invalid_request_error`) and every HTTP 4xx. Errors a + narrowed `retry_on:` declines are now wrapped in `ProviderError` naming the + declined category instead of escaping as raw SDK exceptions, a declined + retry is logged at warning level (a taken one already was), and a fatal + bare `APIError`'s message now carries the payload `type`/`code` the SDK + leaves out of `str(e)`. ### Added diff --git a/src/conductor/providers/_pydantic_ai/retry.py b/src/conductor/providers/_pydantic_ai/retry.py index 8f008a58..57730e02 100644 --- a/src/conductor/providers/_pydantic_ai/retry.py +++ b/src/conductor/providers/_pydantic_ai/retry.py @@ -368,6 +368,30 @@ def _get_retry_after(exception: Exception) -> float | None: return None +def _describe_stream_error(exception: BaseException | None) -> str: + """Render an exception for an error message, adding payload details for a + bare ``openai.APIError``. + + ``str()`` of a bare ``APIError`` carries only the message — not the + payload ``type``/``code``, and not the body when a gateway sent a + non-object ``error`` value — and the SDK substitutes the generic "An + error occurred during streaming" when the payload has no usable message, + so an unenriched failure is undiagnosable. + """ + if exception is None or openai is None or type(exception) is not openai.APIError: + return str(exception) + details: list[str] = [] + if exception.type is not None: + details.append(f"type={exception.type!r}") + if exception.code is not None: + details.append(f"code={exception.code!r}") + if exception.body is not None and not isinstance(exception.body, dict): + details.append(f"body={repr(exception.body)[:200]}") + if not details: + return str(exception) + return f"{exception} ({', '.join(details)})" + + def _extract_status_code(exception: Exception) -> int | None: """Extract HTTP status code from exception if available. @@ -497,16 +521,29 @@ async def execute_with_retry[T]( ) if not is_retryable: + # Asymmetric with a retry *taken*, which logs a warning and + # emits agent_retry: a declined retry must also leave a + # trace. Conductor installs no logging handlers, so the + # debug line above never reaches an operator, and nothing + # would otherwise record that a retry policy existed and + # was declined. + logger.warning( + "[No retry] %s classified non-retryable on attempt %s/%s: %s", + type(e).__name__, + attempt, + retry_config.max_attempts, + e, + ) status_code = _extract_status_code(e) if status_code is not None: raise ProviderError( - f"Pydantic AI provider error: {e}", + f"Pydantic AI provider error: {_describe_stream_error(e)}", suggestion="Check API key, model name, and request parameters", status_code=status_code, is_retryable=False, ) from e raise ProviderError( - f"Pydantic AI call failed: {e}", + f"Pydantic AI call failed: {_describe_stream_error(e)}", suggestion="Check API key, model name, and request parameters", is_retryable=False, ) from e @@ -514,7 +551,20 @@ async def execute_with_retry[T]( if retry_config.retry_on is not None: error_category = _classify_error(e) if error_category not in retry_config.retry_on: - raise + # A retry policy was configured and declined this error. + # Keep the ProviderError contract: a bare `raise` here + # would let a raw SDK exception (e.g. a mid-stream + # openai.APIError) escape past callers that catch + # ProviderError. + raise ProviderError( + f"Pydantic AI call failed: {_describe_stream_error(e)}", + suggestion=( + f"Error category {error_category!r} is not in retry_on=" + f"{retry_config.retry_on}; widen retry_on to retry it." + ), + status_code=_extract_status_code(e), + is_retryable=False, + ) from e if attempt >= retry_config.max_attempts: break @@ -596,7 +646,8 @@ async def execute_with_retry[T]( suggestion = f"Check API connectivity and rate limits. Last error: {last_error}" raise ProviderError( - f"Pydantic AI call failed after {retry_config.max_attempts} attempts: {last_error}", + f"Pydantic AI call failed after {retry_config.max_attempts} attempts: " + f"{_describe_stream_error(last_error)}", suggestion=suggestion, is_retryable=False, ) from last_error diff --git a/tests/test_providers/test_pydantic_ai_retry.py b/tests/test_providers/test_pydantic_ai_retry.py index 3da39a7d..af4833fb 100644 --- a/tests/test_providers/test_pydantic_ai_retry.py +++ b/tests/test_providers/test_pydantic_ai_retry.py @@ -8,6 +8,7 @@ from __future__ import annotations import asyncio +import logging from collections.abc import Callable, Coroutine from typing import Any from unittest.mock import Mock, patch @@ -940,8 +941,12 @@ async def test_validation_error_not_retried(self) -> None: @pytest.mark.asyncio async def test_retry_on_filter_honored(self) -> None: - """Per-agent retry_on must filter which error categories are retried.""" + """Per-agent retry_on must filter which error categories are retried, + and a declined error must surface as a non-retryable ProviderError + chained to the original — never as a raw SDK exception escaping past + callers that catch ProviderError.""" callback = Mock() + original = MockAPIStatusError("500", 500) config = RetryConfig( max_attempts=3, @@ -949,11 +954,11 @@ async def test_retry_on_filter_honored(self) -> None: jitter=0.0, retry_on=["timeout"], ) - factory = _make_factory([MockAPIStatusError("500", 500)]) + factory = _make_factory([original]) with ( patch("conductor.providers._pydantic_ai.retry.asyncio.sleep") as mock_sleep, - pytest.raises(MockAPIStatusError), + pytest.raises(ProviderError) as exc_info, ): await execute_with_retry( factory, @@ -962,9 +967,88 @@ async def test_retry_on_filter_honored(self) -> None: agent_name="retryer", ) + assert exc_info.value.is_retryable is False + assert exc_info.value.__cause__ is original + assert exc_info.value.suggestion is not None + assert "provider_error" in exc_info.value.suggestion + assert "retry_on" in exc_info.value.suggestion callback.assert_not_called() mock_sleep.assert_not_called() + @pytest.mark.asyncio + async def test_retry_on_filtered_stream_error_surfaces_as_provider_error(self) -> None: + # Requirement: a stream error this PR makes retryable still reaches the + # retry_on gate — when the filter declines it, the error must escape as + # a ProviderError chained to the original, not a raw openai.APIError. + stream_error = openai.APIError( + "upstream overloaded", + _make_http_request(), + body={"type": "server_error", "code": "internal_server_error"}, + ) + config = RetryConfig(max_attempts=3, base_delay=0.0, jitter=0.0, retry_on=["timeout"]) + factory = _make_factory([stream_error]) + + with pytest.raises(ProviderError) as exc_info: + await execute_with_retry( + factory, + retry_config=config, + event_callback=None, + agent_name="openai-stream-filtered", + ) + + assert exc_info.value.is_retryable is False + assert exc_info.value.__cause__ is stream_error + assert exc_info.value.suggestion is not None + assert "retry_on" in exc_info.value.suggestion + + @pytest.mark.asyncio + async def test_fatal_stream_error_message_carries_payload_details(self) -> None: + # Requirement: a fatal bare APIError must surface its payload type/code + # in the ProviderError message — str(e) alone does not carry them, and + # the SDK falls back to a generic "An error occurred during streaming" + # message when the payload has none. + error = openai.APIError( + "invalid request", + _make_http_request(), + body={"type": "invalid_request_error", "code": "invalid_value"}, + ) + config = RetryConfig(max_attempts=1, base_delay=0.0, jitter=0.0) + factory = _make_factory([error]) + + with pytest.raises(ProviderError) as exc_info: + await execute_with_retry( + factory, + retry_config=config, + event_callback=None, + agent_name="openai-stream-fatal", + ) + + message = str(exc_info.value) + assert "invalid_request_error" in message + assert "invalid_value" in message + + @pytest.mark.asyncio + async def test_declined_retry_logs_a_warning(self, caplog: pytest.LogCaptureFixture) -> None: + # Requirement: a declined retry must leave a trace. Conductor installs + # no logging handlers, so the debug-level classification line never + # reaches an operator — the decision is logged at warning level, + # mirroring the retry-taken path. + factory = _make_factory([ValueError("fatal")]) + config = RetryConfig(max_attempts=3, base_delay=0.0, jitter=0.0) + + with ( + caplog.at_level(logging.WARNING, logger="conductor.providers._pydantic_ai.retry"), + pytest.raises(ProviderError), + ): + await execute_with_retry( + factory, + retry_config=config, + event_callback=None, + agent_name="retryer", + ) + + assert any("classified non-retryable" in record.getMessage() for record in caplog.records) + @pytest.mark.asyncio async def test_retry_after_header_respected(self) -> None: """Retry-after header must override calculated backoff."""