From ccf46b88cfee7bed408c1f49b38dc09e456eb39b Mon Sep 17 00:00:00 2001 From: Ali Khokhar Date: Tue, 7 Jul 2026 00:13:47 -0700 Subject: [PATCH] Retry pre-stream provider transport failures (#1003) --- core/anthropic/errors.py | 4 + providers/base.py | 8 +- providers/error_mapping.py | 55 ++++++++ providers/rate_limit.py | 48 +++++-- .../transports/anthropic_messages/stream.py | 1 + providers/transports/openai_chat/stream.py | 2 + pyproject.toml | 2 +- .../test_anthropic_messages_429_retry.py | 115 ++++++++++++++++ tests/providers/test_error_mapping.py | 45 +++++- tests/providers/test_llamacpp.py | 4 +- .../providers/test_openai_compat_5xx_retry.py | 102 ++++++++++++++ tests/providers/test_provider_rate_limit.py | 128 ++++++++++++++++++ .../test_provider_transport_logging.py | 42 ++++++ uv.lock | 2 +- 14 files changed, 544 insertions(+), 14 deletions(-) diff --git a/core/anthropic/errors.py b/core/anthropic/errors.py index afe093d6..81615d76 100644 --- a/core/anthropic/errors.py +++ b/core/anthropic/errors.py @@ -21,6 +21,10 @@ def get_user_facing_error_message( return "Provider request timed out." if isinstance(e, httpx.ConnectTimeout): return "Could not connect to provider." + if isinstance(e, httpx.ConnectError): + return "Could not connect to provider." + if isinstance(e, httpx.RemoteProtocolError): + return "Provider connection was interrupted before a response was received." if isinstance(e, TimeoutError): if read_timeout_s is not None: return f"Provider request timed out after {read_timeout_s:g}s." diff --git a/providers/base.py b/providers/base.py index caacce21..72f1a384 100644 --- a/providers/base.py +++ b/providers/base.py @@ -96,11 +96,13 @@ class BaseProvider(ABC): from loguru import logger from core.trace import trace_event + from providers.error_mapping import exception_cause_types response = getattr(error, "response", None) http_status = ( getattr(response, "status_code", None) if response is not None else None ) + cause_types = exception_cause_types(error) trace_event( stage="provider", event="provider.response.transport_error", @@ -109,19 +111,21 @@ class BaseProvider(ABC): request_id=request_id, exc_type=type(error).__name__, http_status=http_status, + cause_types=cause_types, ) if self._config.log_api_error_tracebacks: - logger.error( + logger.opt(exception=error).error( "{}_ERROR:{} {}: {}", tag, req_tag, type(error).__name__, error ) return logger.error( - "{}_ERROR:{} exc_type={} http_status={}", + "{}_ERROR:{} exc_type={} http_status={} cause_types={}", tag, req_tag, type(error).__name__, http_status, + ",".join(cause_types) if cause_types else None, ) @abstractmethod diff --git a/providers/error_mapping.py b/providers/error_mapping.py index 35c8d7fc..a5fa6b4a 100644 --- a/providers/error_mapping.py +++ b/providers/error_mapping.py @@ -1,6 +1,7 @@ """Provider-specific exception mapping and user-visible diagnostics.""" import json +import re from dataclasses import dataclass from typing import Any @@ -24,6 +25,12 @@ from providers.rate_limit import GlobalRateLimiter _BODY_ATTR = "_fcc_provider_error_body" _BODY_TRUNCATED_ATTR = "_fcc_provider_error_body_truncated" +_MAX_CAUSE_CHAIN_DEPTH = 4 +_SECRET_TEXT_PATTERNS = ( + re.compile(r"(?i)(authorization\s*[:=]\s*)(bearer\s+)?[^\s,;]+"), + re.compile(r"(?i)((?:api[_-]?key|token|secret)\s*[:=]\s*)[^\s,;]+"), + re.compile(r"(?i)(bearer\s+)[^\s,;]+"), +) @dataclass(frozen=True) @@ -33,6 +40,7 @@ class ProviderErrorDetail: status_code: int | None = None body_text: str | None = None exception_text: str | None = None + cause_chain_text: str | None = None error_type_hint: str | None = None body_truncated: bool = False @@ -100,6 +108,49 @@ def _cap_text_bytes(text: str, max_bytes: int) -> tuple[str, bool]: return f"{capped}\n... [truncated after {max_bytes} bytes]", True +def _sanitize_exception_text(text: str) -> str: + sanitized = text + for pattern in _SECRET_TEXT_PATTERNS: + sanitized = pattern.sub(r"\1", sanitized) + return sanitized + + +def _exception_causes(exc: BaseException) -> tuple[BaseException, ...]: + causes: list[BaseException] = [] + seen: set[int] = {id(exc)} + current: BaseException | None = exc + while current is not None and len(causes) < _MAX_CAUSE_CHAIN_DEPTH: + next_exc = current.__cause__ or current.__context__ + if next_exc is None or id(next_exc) in seen: + break + seen.add(id(next_exc)) + causes.append(next_exc) + current = next_exc + return tuple(causes) + + +def exception_cause_types(exc: BaseException) -> tuple[str, ...]: + """Return safe exception cause type names for default metadata logging.""" + return tuple(type(cause).__name__ for cause in _exception_causes(exc)) + + +def _exception_cause_chain_text(exc: BaseException) -> str | None: + lines: list[str] = [] + for cause in _exception_causes(exc): + raw_text = str(cause).strip() + if raw_text: + lines.append( + f"{type(cause).__name__}: {_sanitize_exception_text(raw_text)}" + ) + else: + lines.append(type(cause).__name__) + if not lines: + return None + text = "\n".join(lines) + capped, _ = _cap_text_bytes(text, PROVIDER_ERROR_BODY_DISPLAY_CAP_BYTES) + return capped + + def _error_type_hint_from_body(body: Any, body_text: str | None) -> str | None: parsed = body if isinstance(parsed, bytes): @@ -152,6 +203,7 @@ def extract_provider_error_detail(exc: Exception) -> ProviderErrorDetail: exception_text = str(exc).strip() or None if exception_text is not None: + exception_text = _sanitize_exception_text(exception_text) exception_text, _ = _cap_text_bytes( exception_text, PROVIDER_ERROR_BODY_DISPLAY_CAP_BYTES ) @@ -160,6 +212,7 @@ def extract_provider_error_detail(exc: Exception) -> ProviderErrorDetail: status_code=_status_code_from_exception(exc), body_text=body_text, exception_text=exception_text, + cause_chain_text=_exception_cause_chain_text(exc), error_type_hint=_error_type_hint_from_body(raw_body, body_text), body_truncated=display_truncated, ) @@ -194,6 +247,8 @@ def format_provider_error_message( lines = [stable_message] if detail.exception_text and detail.exception_text != stable_message: lines.extend(("", "Provider exception:", detail.exception_text)) + if detail.cause_chain_text: + lines.extend(("", "Caused by:", detail.cause_chain_text)) _append_request_id_lines(lines, request_id) return "\n".join(lines) diff --git a/providers/rate_limit.py b/providers/rate_limit.py index 034ffd36..3c2a0504 100644 --- a/providers/rate_limit.py +++ b/providers/rate_limit.py @@ -7,6 +7,8 @@ from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager from typing import Any, ClassVar, TypeVar +import httpx +import openai from loguru import logger from core.anthropic.streaming import retryable_transient_status @@ -36,6 +38,26 @@ def retryable_upstream_status(exc: BaseException) -> int | None: return None +def retryable_upstream_transport_error(exc: BaseException) -> bool: + """Return whether a pre-response transport failure can be retried.""" + if isinstance(exc, openai.AuthenticationError | openai.BadRequestError): + return False + return isinstance( + exc, + ( + TimeoutError, + httpx.TimeoutException, + httpx.ConnectError, + httpx.ReadError, + httpx.WriteError, + httpx.RemoteProtocolError, + httpx.NetworkError, + openai.APITimeoutError, + openai.APIConnectionError, + ), + ) + + class GlobalRateLimiter: """ Global singleton rate limiter that blocks all requests @@ -227,7 +249,9 @@ class GlobalRateLimiter: Waits for the proactive limiter before each attempt. On ``429`` (rate limit) or upstream ``5xx`` server errors, applies exponential backoff with jitter - and sets the reactive block before retrying. + and sets the reactive block before retrying. Pre-response transport errors + use the same attempt budget and backoff schedule without setting the + reactive provider block. Args: fn: Async callable to execute. @@ -252,14 +276,20 @@ class GlobalRateLimiter: return await fn(*args, **kwargs) except Exception as e: status = retryable_upstream_status(e) - if status is None: + transport_error = status is None and retryable_upstream_transport_error( + e + ) + if status is None and not transport_error: raise - label = ( - "Rate limited (429)" - if status == 429 - else f"Upstream server error ({status})" - ) + if status is None: + label = f"Provider transport error ({type(e).__name__})" + else: + label = ( + "Rate limited (429)" + if status == 429 + else f"Upstream server error ({status})" + ) last_exc = e if attempt >= max_retries: logger.warning( @@ -285,11 +315,13 @@ class GlobalRateLimiter: event="provider.retry.scheduled", source="provider", status_code=status, + exc_type=type(e).__name__, attempt=attempt_no, max_attempts=total_attempts, delay_s=round(delay, 3), ) - self.set_blocked(delay) + if status is not None: + self.set_blocked(delay) await asyncio.sleep(delay) assert last_exc is not None diff --git a/providers/transports/anthropic_messages/stream.py b/providers/transports/anthropic_messages/stream.py index 9f460870..2a3f9cd1 100644 --- a/providers/transports/anthropic_messages/stream.py +++ b/providers/transports/anthropic_messages/stream.py @@ -202,6 +202,7 @@ class AnthropicMessagesStreamAdapter: event="provider.response.error", source="provider", provider=tag, + request_id=self._request_id, error_message=error_message, exc_type=type(error).__name__, mid_stream=( diff --git a/providers/transports/openai_chat/stream.py b/providers/transports/openai_chat/stream.py index e3c4d5c8..be719e63 100644 --- a/providers/transports/openai_chat/stream.py +++ b/providers/transports/openai_chat/stream.py @@ -275,6 +275,8 @@ class OpenAIChatStreamAdapter: event="provider.response.error", source="provider", provider=tag, + request_id=self._request_id, + exc_type=type(error).__name__, error_message=error_message, mapped_error_type=type( map_error( diff --git a/pyproject.toml b/pyproject.toml index a27fc8ff..c36c375b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "free-claude-code" -version = "3.4.7" +version = "3.4.8" description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM" readme = "README.md" requires-python = ">=3.14.0" diff --git a/tests/providers/test_anthropic_messages_429_retry.py b/tests/providers/test_anthropic_messages_429_retry.py index d1753dd9..0b68ec9b 100644 --- a/tests/providers/test_anthropic_messages_429_retry.py +++ b/tests/providers/test_anthropic_messages_429_retry.py @@ -140,6 +140,53 @@ async def test_native_stream_retries_on_http_5xx_then_streams( GlobalRateLimiter.reset_instance() +@pytest.mark.asyncio +async def test_native_stream_retries_on_pre_send_connection_error_then_streams( + provider_config, +): + """Pre-response HTTPX transport errors retry through execute_with_retry.""" + GlobalRateLimiter.reset_instance() + try: + provider = NativeProvider(provider_config) + req = MockRequest() + request_obj = httpx.Request("POST", "https://custom.test/v1/messages") + ok_lines = [ + "event: message_start", + 'data: {"type":"message_start"}', + "", + "event: message_stop", + 'data: {"type":"message_stop"}', + "", + ] + ok_response = FakeResponse(lines=ok_lines) + + send_calls = {"n": 0} + + async def send_side_effect(*_a, **_kw): + send_calls["n"] += 1 + if send_calls["n"] == 1: + raise httpx.ConnectError("connect failed", request=request_obj) + return ok_response + + with ( + patch.object(provider._client, "build_request", return_value=request_obj), + patch.object( + provider._client, + "send", + new_callable=AsyncMock, + side_effect=send_side_effect, + ), + patch("asyncio.sleep", new_callable=AsyncMock), + ): + events = [e async for e in provider.stream_response(req)] + + assert send_calls["n"] == 2 + assert ok_response.is_closed + _assert_minimal_success_stream(events) + finally: + GlobalRateLimiter.reset_instance() + + @pytest.mark.parametrize( ("status_code", "substr"), [ @@ -202,6 +249,74 @@ async def test_native_stream_5xx_retry_exhausted(provider_config, status_code, s GlobalRateLimiter.reset_instance() +@pytest.mark.asyncio +async def test_native_stream_connection_error_retry_exhausted(provider_config): + """Repeated pre-response connection failures exhaust at 5 attempts.""" + GlobalRateLimiter.reset_instance() + try: + + @asynccontextmanager + async def _slot(): + yield + + with patch( + "providers.transports.anthropic_messages.transport.GlobalRateLimiter" + ) as mock_gl: + instance = mock_gl.get_scoped_instance.return_value + real = GlobalRateLimiter( + rate_limit=100, + rate_window=60, + max_concurrency=5, + ) + instance.wait_if_blocked = real.wait_if_blocked + instance.execute_with_retry = real.execute_with_retry + instance.set_blocked = real.set_blocked + instance.concurrency_slot.side_effect = _slot + + provider = NativeProvider(provider_config) + req = MockRequest() + request_obj = httpx.Request("POST", "https://custom.test/v1/messages") + + with ( + patch.object( + provider._client, "build_request", return_value=request_obj + ), + patch.object( + provider._client, + "send", + new_callable=AsyncMock, + side_effect=httpx.ConnectError( + "connect failed", request=request_obj + ), + ) as mock_send, + patch("asyncio.sleep", new_callable=AsyncMock), + patch( + "providers.transports.anthropic_messages.stream.trace_event" + ) as trace, + ): + events = [ + e + async for e in provider.stream_response( + req, request_id="req_native_conn" + ) + ] + + assert mock_send.await_count == 5 + error_traces = [ + call.kwargs + for call in trace.call_args_list + if call.kwargs.get("event") == "provider.response.error" + ] + 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", + ) + finally: + GlobalRateLimiter.reset_instance() + + @pytest.mark.asyncio async def test_non_retryable_4xx_http_error_not_retried(provider_config): """HTTP 400 from upstream is not retried; single send (passthrough limiter).""" diff --git a/tests/providers/test_error_mapping.py b/tests/providers/test_error_mapping.py index e740225a..a953b16a 100644 --- a/tests/providers/test_error_mapping.py +++ b/tests/providers/test_error_mapping.py @@ -5,7 +5,7 @@ from unittest.mock import MagicMock, patch import openai import pytest -from httpx import HTTPStatusError, ReadTimeout, Request, Response +from httpx import ConnectError, HTTPStatusError, ReadTimeout, Request, Response from config.constants import PROVIDER_ERROR_BODY_DISPLAY_CAP_BYTES from core.anthropic import ( @@ -15,6 +15,7 @@ from core.anthropic import ( ) from providers.error_mapping import ( attach_provider_error_body, + exception_cause_types, extract_provider_error_detail, format_provider_error_message, map_error, @@ -313,6 +314,48 @@ def test_empty_http_error_body_is_explicitly_reported(): assert "(empty upstream error body)" in msg +def test_connection_error_without_response_includes_sanitized_cause_chain(): + request = Request("POST", "http://test") + exc = openai.APIConnectionError(request=request) + exc.__cause__ = ConnectError( + "connect failed authorization: Bearer SECRET token=ALSO_SECRET", + request=request, + ) + mapped = map_error(exc) + detail = extract_provider_error_detail(exc) + msg = format_provider_error_message( + mapped, + detail, + provider_name="NIM", + read_timeout_s=30.0, + request_id="req_conn", + ) + + assert "Provider API request failed." in msg + assert "Provider exception:" in msg + assert "Connection error." in msg + assert "Caused by:" in msg + assert "ConnectError: connect failed authorization: " in msg + assert "SECRET" not in msg + assert "Request ID: req_conn" in msg + assert exception_cause_types(exc) == ("ConnectError",) + + +def test_connection_error_cause_chain_is_capped_for_display(): + request = Request("POST", "http://test") + exc = openai.APIConnectionError(request=request) + exc.__cause__ = ConnectError( + "x" * (PROVIDER_ERROR_BODY_DISPLAY_CAP_BYTES + 10), + request=request, + ) + detail = extract_provider_error_detail(exc) + + assert detail.cause_chain_text is not None + assert f"truncated after {PROVIDER_ERROR_BODY_DISPLAY_CAP_BYTES} bytes" in ( + detail.cause_chain_text + ) + + def test_attached_provider_error_body_is_capped_for_display(): response = Response( status_code=500, diff --git a/tests/providers/test_llamacpp.py b/tests/providers/test_llamacpp.py index a2d8e7b2..bfc4333b 100644 --- a/tests/providers/test_llamacpp.py +++ b/tests/providers/test_llamacpp.py @@ -290,8 +290,10 @@ async def test_stream_network_error(llamacpp_provider): blob = "".join(events) assert_canonical_stream_error_envelope( - events, user_message_substr="Connection refused" + events, user_message_substr="Could not connect to provider." ) + assert "Provider exception" in blob + assert "Connection refused" in blob assert "TEST_ID2" in blob diff --git a/tests/providers/test_openai_compat_5xx_retry.py b/tests/providers/test_openai_compat_5xx_retry.py index a17c2a3f..76fbb749 100644 --- a/tests/providers/test_openai_compat_5xx_retry.py +++ b/tests/providers/test_openai_compat_5xx_retry.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, MagicMock, patch +import httpx import openai import pytest from httpx import Request, Response @@ -11,6 +12,7 @@ from providers.base import ProviderConfig 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: @@ -21,6 +23,14 @@ def _internal_5xx(code: int) -> openai.InternalServerError: ) +def _connection_error(message: str = "connect failed") -> openai.APIConnectionError: + error = openai.APIConnectionError( + request=Request("POST", "https://test.api.nvidia.com/v1/chat/completions") + ) + error.__cause__ = httpx.ConnectError(message) + return error + + @pytest.mark.parametrize("status_code", [500, 502, 503, 504]) @pytest.mark.asyncio async def test_nim_stream_retries_on_openai_5xx_then_streams(status_code): @@ -67,6 +77,98 @@ async def test_nim_stream_retries_on_openai_5xx_then_streams(status_code): GlobalRateLimiter.reset_instance() +@pytest.mark.asyncio +async def test_nim_stream_retries_on_pre_stream_connection_error_then_streams(): + GlobalRateLimiter.reset_instance() + try: + config = ProviderConfig( + api_key="test_key", + base_url="https://test.api.nvidia.com/v1", + rate_limit=100, + rate_window=60, + http_read_timeout=600.0, + http_write_timeout=15.0, + http_connect_timeout=5.0, + ) + provider = NvidiaNimProvider(config, nim_settings=NimSettings()) + req = MockRequest() + + mock_chunk = MagicMock() + mock_chunk.choices = [ + MagicMock( + delta=MagicMock(content="Recovered", reasoning_content=""), + finish_reason="stop", + ) + ] + mock_chunk.usage = None + + async def mock_stream(): + yield mock_chunk + + with ( + patch.object( + provider._client.chat.completions, + "create", + new_callable=AsyncMock, + ) as mock_create, + patch("asyncio.sleep", new_callable=AsyncMock), + ): + mock_create.side_effect = [_connection_error(), mock_stream()] + events = [e async for e in provider.stream_response(req)] + + assert mock_create.await_count == 2 + assert any("Recovered" in e for e in events) + finally: + GlobalRateLimiter.reset_instance() + + +@pytest.mark.asyncio +async def test_nim_stream_connection_error_exhausted_emits_cause_chain(): + GlobalRateLimiter.reset_instance() + try: + config = ProviderConfig( + api_key="test_key", + base_url="https://test.api.nvidia.com/v1", + rate_limit=100, + rate_window=60, + http_read_timeout=600.0, + http_write_timeout=15.0, + http_connect_timeout=5.0, + ) + provider = NvidiaNimProvider(config, nim_settings=NimSettings()) + req = MockRequest() + error = _connection_error("upstream disconnected") + + with ( + patch.object( + provider._client.chat.completions, + "create", + new_callable=AsyncMock, + side_effect=error, + ) as mock_create, + patch("asyncio.sleep", new_callable=AsyncMock), + patch("providers.transports.openai_chat.stream.trace_event") as trace, + ): + events = [ + e async for e in provider.stream_response(req, request_id="req_conn") + ] + + assert mock_create.await_count == 5 + error_traces = [ + call.kwargs + for call in trace.call_args_list + if call.kwargs.get("event") == "provider.response.error" + ] + 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", + ) + finally: + GlobalRateLimiter.reset_instance() + + @pytest.mark.parametrize( ("status_code", "expect_substr"), [ diff --git a/tests/providers/test_provider_rate_limit.py b/tests/providers/test_provider_rate_limit.py index f46b8bf1..e8658058 100644 --- a/tests/providers/test_provider_rate_limit.py +++ b/tests/providers/test_provider_rate_limit.py @@ -1,6 +1,8 @@ import asyncio import time +from unittest.mock import AsyncMock, patch +import httpx import openai import pytest import pytest_asyncio @@ -11,6 +13,7 @@ from providers.rate_limit import ( UPSTREAM_TRANSIENT_TOTAL_ATTEMPTS, GlobalRateLimiter, retryable_upstream_status, + retryable_upstream_transport_error, ) @@ -41,6 +44,40 @@ def test_retryable_upstream_status_reads_statusless_resource_exhausted_text() -> assert retryable_upstream_status(exc) == 503 +def test_retryable_upstream_transport_error_classifies_connection_failures() -> None: + request = Request("POST", "http://x") + assert retryable_upstream_transport_error( + openai.APIConnectionError(request=request) + ) + assert retryable_upstream_transport_error(httpx.ConnectError("connect failed")) + assert retryable_upstream_transport_error(httpx.ReadError("read failed")) + assert retryable_upstream_transport_error(httpx.WriteError("write failed")) + assert retryable_upstream_transport_error( + httpx.RemoteProtocolError("server disconnected") + ) + assert retryable_upstream_transport_error(TimeoutError("timed out")) + + +def test_retryable_upstream_transport_error_rejects_request_errors() -> None: + from httpx import Response + + request = Request("POST", "http://x") + assert not retryable_upstream_transport_error( + openai.BadRequestError( + "bad request", + response=Response(400, request=request), + body={}, + ) + ) + assert not retryable_upstream_transport_error( + httpx.HTTPStatusError( + "bad request", + request=request, + response=Response(400, request=request), + ) + ) + + class TestProviderRateLimiter: """Tests for providers.rate_limit.GlobalRateLimiter.""" @@ -399,6 +436,97 @@ class TestProviderRateLimiter: assert result == "ok" assert call_count == 2 + @pytest.mark.asyncio + async def test_execute_with_retry_succeeds_on_openai_connection_retry(self): + """Pre-response OpenAI SDK connection errors retry then succeed.""" + limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60) + + call_count = 0 + request = Request("POST", "http://x") + + async def fail_then_ok(): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise openai.APIConnectionError(request=request) + return "ok" + + with ( + patch.object(limiter, "set_blocked") as set_blocked, + patch("asyncio.sleep", new_callable=AsyncMock) as sleep, + ): + result = await limiter.execute_with_retry( + fail_then_ok, + max_retries=2, + base_delay=0.01, + max_delay=0.1, + jitter=0, + ) + + assert result == "ok" + assert call_count == 2 + set_blocked.assert_not_called() + sleep.assert_awaited_once() + + @pytest.mark.asyncio + async def test_execute_with_retry_succeeds_on_httpx_transport_retry(self): + """Pre-response HTTPX transport errors retry then succeed.""" + limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60) + + call_count = 0 + + async def fail_then_ok(): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise httpx.RemoteProtocolError("server disconnected") + return "ok" + + with ( + patch.object(limiter, "set_blocked") as set_blocked, + patch("asyncio.sleep", new_callable=AsyncMock) as sleep, + ): + result = await limiter.execute_with_retry( + fail_then_ok, + max_retries=2, + base_delay=0.01, + max_delay=0.1, + jitter=0, + ) + + assert result == "ok" + assert call_count == 2 + set_blocked.assert_not_called() + sleep.assert_awaited_once() + + @pytest.mark.asyncio + async def test_execute_with_retry_exhaust_transport_error_attempts(self): + """Transport retries exhaust after the shared 5 total attempts.""" + limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60) + + call_count = 0 + + async def always_fail(): + nonlocal call_count + call_count += 1 + raise httpx.ConnectError("connect failed") + + with ( + patch.object(limiter, "set_blocked") as set_blocked, + patch("asyncio.sleep", new_callable=AsyncMock) as sleep, + pytest.raises(httpx.ConnectError), + ): + await limiter.execute_with_retry( + always_fail, + base_delay=0.01, + max_delay=0.1, + jitter=0, + ) + + assert call_count == UPSTREAM_TRANSIENT_TOTAL_ATTEMPTS + set_blocked.assert_not_called() + assert sleep.await_count == DEFAULT_UPSTREAM_MAX_RETRIES + @pytest.mark.parametrize("status_code", [500, 502, 503, 504]) @pytest.mark.asyncio async def test_execute_with_retry_exhaust_openai_5xx_raises(self, status_code): diff --git a/tests/providers/test_provider_transport_logging.py b/tests/providers/test_provider_transport_logging.py index ecc526bf..6141dd41 100644 --- a/tests/providers/test_provider_transport_logging.py +++ b/tests/providers/test_provider_transport_logging.py @@ -4,6 +4,8 @@ import logging from contextlib import asynccontextmanager from unittest.mock import AsyncMock, MagicMock, patch +import httpx +import openai import pytest from config.constants import NATIVE_MESSAGES_ERROR_BODY_LOG_CAP_BYTES @@ -231,6 +233,46 @@ async def test_openai_compat_stream_failure_default_logs_exclude_exception_str(c assert "exc_type=RuntimeError" in messages +@pytest.mark.asyncio +async def test_openai_compat_stream_failure_default_logs_cause_types_only(caplog): + config = ProviderConfig( + api_key="k", + base_url="http://localhost:1/v1", + log_api_error_tracebacks=False, + ) + provider = NvidiaNimProvider(config, nim_settings=NimSettings()) + req = make_openai_compat_stream_request() + error = openai.APIConnectionError( + request=httpx.Request("POST", "http://localhost:1/v1/chat/completions") + ) + error.__cause__ = httpx.ConnectError("SECRET_CAUSE_DETAIL") + + @asynccontextmanager + async def _noop_slot(): + yield + + with ( + patch.object( + provider, + "_create_stream", + new_callable=AsyncMock, + side_effect=error, + ), + patch.object( + provider._global_rate_limiter, + "concurrency_slot", + _noop_slot, + ), + caplog.at_level(logging.ERROR), + ): + _ = [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 + assert "exc_type=APIConnectionError" in messages + assert "cause_types=ConnectError" in messages + + @pytest.mark.asyncio async def test_openai_compat_stream_failure_respects_verbose_flag(caplog): config = ProviderConfig( diff --git a/uv.lock b/uv.lock index 792e7cf8..192d4b36 100644 --- a/uv.lock +++ b/uv.lock @@ -561,7 +561,7 @@ wheels = [ [[package]] name = "free-claude-code" -version = "3.4.7" +version = "3.4.8" source = { editable = "." } dependencies = [ { name = "aiohttp" },