diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index c3f821ac..aedff70b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -406,7 +406,10 @@ transport packages are upstream adapters: OpenAI-chat providers convert chat chunks into ledger operations, and native Anthropic providers parse upstream SSE, apply native block policy, and re-emit normalized Anthropic SSE from the shared ledger. Transport bases stay focused on provider hooks, client setup, request -construction, rate limiting, and model listing. +construction, rate limiting, and model listing. Status-less upstream transient +classification also lives in this shared layer so stream recovery, provider +backoff, and provider error mapping agree on retryable overload/rate-limit +signals. [core/openai_responses/](core/openai_responses/) owns OpenAI Responses support: @@ -521,7 +524,9 @@ Discord and Telegram messaging. Managed task invocations set non-interactive terminal settings, optional `--resume`, optional `--fork-session`, and `--output-format stream-json`. The managed session parser extracts persistent Claude session IDs and yields Claude stream-json events to -the messaging event parser. +the messaging event parser. Managed Claude also owns subprocess stderr +diagnostic classification so known benign Claude Code notices do not become +messaging task errors, while unknown stderr remains fatal. Codex is supported through `fcc-codex` and Codex extensions. FCC does not keep an internal managed-Codex session runner because no user-facing messaging setting diff --git a/cli/managed/diagnostics.py b/cli/managed/diagnostics.py new file mode 100644 index 00000000..8bfe587d --- /dev/null +++ b/cli/managed/diagnostics.py @@ -0,0 +1,49 @@ +"""Managed Claude Code diagnostic classification.""" + +from __future__ import annotations + +from dataclasses import dataclass + +_BENIGN_STDERR_MARKERS = ("claude.ai connectors are disabled",) + + +@dataclass(frozen=True, slots=True) +class ManagedClaudeStderrDiagnostics: + """Classified stderr lines from one managed Claude Code invocation.""" + + benign_lines: tuple[str, ...] + fatal_lines: tuple[str, ...] + + @property + def fatal_text(self) -> str | None: + text = "\n".join(self.fatal_lines).strip() + return text or None + + @property + def has_benign(self) -> bool: + return bool(self.benign_lines) + + +def classify_managed_claude_stderr( + stderr_text: str, +) -> ManagedClaudeStderrDiagnostics: + """Classify known benign Claude Code diagnostics separately from failures.""" + benign_lines: list[str] = [] + fatal_lines: list[str] = [] + for line in _stderr_lines(stderr_text): + lowered = line.lower() + if any(marker in lowered for marker in _BENIGN_STDERR_MARKERS): + benign_lines.append(line) + else: + fatal_lines.append(line) + return ManagedClaudeStderrDiagnostics( + benign_lines=tuple(benign_lines), + fatal_lines=tuple(fatal_lines), + ) + + +def _stderr_lines(stderr_text: str) -> tuple[str, ...]: + stripped = stderr_text.strip() + if not stripped: + return () + return tuple(line.strip() for line in stripped.splitlines() if line.strip()) diff --git a/cli/managed/session.py b/cli/managed/session.py index d7c335bf..efb5a5ec 100644 --- a/cli/managed/session.py +++ b/cli/managed/session.py @@ -16,6 +16,7 @@ from .claude import ( build_managed_claude_invocation, parse_managed_claude_stdout_line, ) +from .diagnostics import classify_managed_claude_stderr # Cap stderr capture so a runaway child cannot exhaust memory; pipe is still drained. _MAX_STDERR_CAPTURE_BYTES = 256 * 1024 @@ -188,7 +189,17 @@ class ManagedClaudeSession: stderr_text = None if stderr_bytes: - stderr_text = stderr_bytes.decode("utf-8", errors="replace").strip() + raw_stderr_text = stderr_bytes.decode( + "utf-8", errors="replace" + ).strip() + if raw_stderr_text: + diagnostics = classify_managed_claude_stderr(raw_stderr_text) + if diagnostics.has_benign: + logger.info( + "Claude CLI benign stderr diagnostics: lines={}", + len(diagnostics.benign_lines), + ) + stderr_text = diagnostics.fatal_text if stderr_text: if self._log_raw_cli_diagnostics: logger.error("Claude CLI stderr: {}", stderr_text) diff --git a/core/anthropic/streaming/__init__.py b/core/anthropic/streaming/__init__.py index 8584e710..fa095d75 100644 --- a/core/anthropic/streaming/__init__.py +++ b/core/anthropic/streaming/__init__.py @@ -25,6 +25,7 @@ from .recovery import ( parse_complete_tool_input, tool_schemas_by_name, ) +from .transient_errors import is_transient_overload_error, retryable_transient_status __all__ = [ "ANTHROPIC_SSE_RESPONSE_HEADERS", @@ -45,9 +46,11 @@ __all__ = [ "continuation_suffix", "format_sse_event", "is_retryable_stream_error", + "is_transient_overload_error", "make_text_recovery_body", "make_tool_repair_body", "map_stop_reason", "parse_complete_tool_input", + "retryable_transient_status", "tool_schemas_by_name", ] diff --git a/core/anthropic/streaming/recovery.py b/core/anthropic/streaming/recovery.py index debb7ce0..088b33b1 100644 --- a/core/anthropic/streaming/recovery.py +++ b/core/anthropic/streaming/recovery.py @@ -17,6 +17,8 @@ from loguru import logger from core.trace import trace_event +from .transient_errors import retryable_transient_status + EARLY_TRANSPARENT_TOTAL_ATTEMPTS = 5 EARLY_TRANSPARENT_MAX_RETRIES = EARLY_TRANSPARENT_TOTAL_ATTEMPTS - 1 MIDSTREAM_RECOVERY_ATTEMPTS = 5 @@ -231,14 +233,8 @@ def is_retryable_stream_error(exc: BaseException) -> bool: return True if isinstance(exc, openai.AuthenticationError | openai.BadRequestError): return False - if isinstance(exc, httpx.HTTPStatusError): - status = exc.response.status_code - return status == 429 or 500 <= status <= 599 - if isinstance(exc, openai.RateLimitError): + if retryable_transient_status(exc) is not None: return True - if isinstance(exc, openai.APIStatusError): - status = getattr(exc, "status_code", None) - return isinstance(status, int) and (status == 429 or 500 <= status <= 599) return isinstance( exc, ( diff --git a/core/anthropic/streaming/transient_errors.py b/core/anthropic/streaming/transient_errors.py new file mode 100644 index 00000000..371d37ae --- /dev/null +++ b/core/anthropic/streaming/transient_errors.py @@ -0,0 +1,156 @@ +"""Provider-neutral transient upstream error classification.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from contextlib import suppress +from typing import Any + +import httpx +import openai + +_RATE_LIMIT_MARKERS = frozenset( + { + "rate_limit", + "rate limit", + "too many requests", + } +) +_OVERLOAD_MARKERS = frozenset( + { + "resourceexhausted", + "resource exhausted", + "limit reached", + "overloaded", + "capacity", + } +) +_INTERNAL_ERROR_MARKERS = frozenset( + { + "internal_server_error", + "internal server error", + } +) + + +def retryable_transient_status(exc: BaseException) -> int | None: + """Return an HTTP-like retryable status inferred from an upstream exception.""" + if isinstance(exc, openai.RateLimitError): + return 429 + if isinstance(exc, httpx.HTTPStatusError): + status = exc.response.status_code + return status if _is_retryable_status(status) else None + + status = _status_from_exception(exc) + if _is_retryable_status(status): + return status + + body = getattr(exc, "body", None) + body_status = _status_from_body(body) + if _is_retryable_status(body_status): + return body_status + + text = transient_error_text(exc) + if _has_marker(text, _RATE_LIMIT_MARKERS): + return 429 + if _has_marker(text, _OVERLOAD_MARKERS): + return 503 + if _has_marker(text, _INTERNAL_ERROR_MARKERS): + return 500 + return None + + +def is_transient_overload_error(exc: BaseException) -> bool: + """Return whether an upstream exception indicates overload/capacity pressure.""" + return _has_marker(transient_error_text(exc), _OVERLOAD_MARKERS) + + +def transient_error_text(exc: BaseException) -> str: + """Return normalized exception/body/response text for transient classifiers.""" + parts = [str(exc)] + body = getattr(exc, "body", None) + if body is not None: + parts.append(_body_to_text(body)) + response = getattr(exc, "response", None) + if response is not None: + with suppress(Exception): + parts.append(response.text) + return " ".join(part for part in parts if part).lower() + + +def _status_from_exception(exc: BaseException) -> int | None: + status = getattr(exc, "status_code", None) + return status if isinstance(status, int) else None + + +def _status_from_body(body: Any) -> int | None: + for item in _body_candidates(body): + if not isinstance(item, Mapping): + continue + for key in ("status", "status_code", "code"): + status = _coerce_status(item.get(key)) + if status is not None: + return status + type_status = _status_from_type_fields(item) + if type_status is not None: + return type_status + return None + + +def _body_candidates(body: Any) -> tuple[Any, ...]: + if isinstance(body, str): + try: + parsed = json.loads(body) + except ValueError: + return (body,) + return _body_candidates(parsed) + if isinstance(body, bytes): + return _body_candidates(body.decode("utf-8", errors="replace")) + if isinstance(body, Mapping): + nested = body.get("error") + return (body, nested) if isinstance(nested, Mapping) else (body,) + return (body,) + + +def _coerce_status(value: Any) -> int | None: + if isinstance(value, int): + return value + if isinstance(value, str) and value.isdigit(): + return int(value) + return None + + +def _status_from_type_fields(item: Mapping[str, Any]) -> int | None: + values = [] + for key in ("type", "code"): + value = item.get(key) + if isinstance(value, str): + values.append(value.lower()) + text = " ".join(values) + if _has_marker(text, _RATE_LIMIT_MARKERS): + return 429 + if _has_marker(text, _OVERLOAD_MARKERS): + return 503 + if _has_marker(text, _INTERNAL_ERROR_MARKERS): + return 500 + return None + + +def _body_to_text(body: Any) -> str: + if isinstance(body, bytes): + return body.decode("utf-8", errors="replace") + if isinstance(body, str): + return body + try: + return json.dumps(body, ensure_ascii=False, separators=(",", ":")) + except TypeError: + return str(body) + + +def _has_marker(text: str, markers: frozenset[str]) -> bool: + return any(marker in text for marker in markers) + + +def _is_retryable_status(status: int | None) -> bool: + return isinstance(status, int) and (status == 429 or 500 <= status <= 599) diff --git a/providers/error_mapping.py b/providers/error_mapping.py index cc4b8883..35c8d7fc 100644 --- a/providers/error_mapping.py +++ b/providers/error_mapping.py @@ -9,6 +9,10 @@ import openai from config.constants import PROVIDER_ERROR_BODY_DISPLAY_CAP_BYTES from core.anthropic import get_user_facing_error_message +from core.anthropic.streaming import ( + is_transient_overload_error, + retryable_transient_status, +) from providers.exceptions import ( APIError, AuthenticationError, @@ -275,8 +279,8 @@ def map_error( return InvalidRequestError(message, raw_error=str(e)) if isinstance(e, openai.InternalServerError): raw_message = str(e) - sdk_status = getattr(e, "status_code", None) - if "overloaded" in raw_message.lower() or "capacity" in raw_message.lower(): + sdk_status = retryable_transient_status(e) or getattr(e, "status_code", None) + if is_transient_overload_error(e): return OverloadedError(message, raw_error=raw_message) if isinstance(sdk_status, int) and 500 <= sdk_status <= 599: stable = APIError("_", status_code=sdk_status) @@ -287,8 +291,24 @@ def map_error( ) return APIError(message, status_code=500, raw_error=str(e)) if isinstance(e, openai.APIError): + status = retryable_transient_status(e) + if status == 429: + limiter.set_blocked(60) + return RateLimitError( + "Provider rate limit reached. Please retry shortly.", raw_error=str(e) + ) + if is_transient_overload_error(e): + return OverloadedError( + "Provider is currently overloaded. Please retry.", raw_error=str(e) + ) + effective_status = status or getattr(e, "status_code", None) + if not isinstance(effective_status, int): + effective_status = 500 + stable = APIError("_", status_code=effective_status) return APIError( - message, status_code=getattr(e, "status_code", 500), raw_error=str(e) + get_user_facing_error_message(stable), + status_code=effective_status, + raw_error=str(e), ) if isinstance(e, httpx.HTTPStatusError): diff --git a/providers/rate_limit.py b/providers/rate_limit.py index ce85b674..034ffd36 100644 --- a/providers/rate_limit.py +++ b/providers/rate_limit.py @@ -7,10 +7,9 @@ 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 from core.rate_limit import StrictSlidingWindowLimiter from core.trace import trace_event @@ -31,18 +30,9 @@ def retryable_upstream_status(exc: BaseException) -> int | None: ``429`` plus any upstream ``5xx`` use the same exponential backoff and scoped limiter blocking semantics as today's rate-limit path. """ - if isinstance(exc, openai.RateLimitError): - return 429 - if isinstance(exc, httpx.HTTPStatusError): - status = exc.response.status_code - if _upstream_http_retryable(status): - return status - return None - if isinstance(exc, openai.APIError): - status = getattr(exc, "status_code", None) - if isinstance(status, int) and 500 <= status <= 599: - return status - return None + status = retryable_transient_status(exc) + if status is not None and _upstream_http_retryable(status): + return status return None diff --git a/pyproject.toml b/pyproject.toml index 97bc814b..44ffd87b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "free-claude-code" -version = "2.4.1" +version = "2.4.2" description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM" readme = "README.md" requires-python = ">=3.14.0" diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index 591b8bf9..08755066 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -347,6 +347,86 @@ class TestManagedClaudeSession: assert events[-1]["type"] == "exit" assert events[-1]["code"] == 0 + @pytest.mark.asyncio + async def test_start_task_ignores_benign_claude_connectors_stderr(self): + """Known Claude diagnostics on stderr are not surfaced as task failures.""" + from cli.managed.session import ManagedClaudeSession + + session = ManagedClaudeSession("/tmp", "http://localhost:8082/v1") + + mock_process = AsyncMock() + mock_process.stdout.read.side_effect = [ + b'{"type": "message", "content": "Hi"}\n', + b"", + ] + mock_process.stderr.read.side_effect = [ + b"claude.ai connectors are disabled in this environment\n", + b"", + ] + mock_process.wait.return_value = 0 + + with patch( + "asyncio.create_subprocess_exec", new_callable=AsyncMock + ) as mock_exec: + mock_exec.return_value = mock_process + + events = [e async for e in session.start_task("Hello")] + + assert [e for e in events if e.get("type") == "error"] == [] + assert events[-1] == {"type": "exit", "code": 0, "stderr": None} + + @pytest.mark.asyncio + async def test_start_task_mixed_stderr_reports_only_fatal_lines(self): + """Benign stderr diagnostics are filtered without hiding real failures.""" + from cli.managed.session import ManagedClaudeSession + + session = ManagedClaudeSession("/tmp", "http://localhost:8082/v1") + + mock_process = AsyncMock() + mock_process.stdout.read.side_effect = [b""] + mock_process.stderr.read.side_effect = [ + (b"claude.ai connectors are disabled in this environment\nFatal error\n"), + b"", + ] + mock_process.wait.return_value = 1 + + with patch( + "asyncio.create_subprocess_exec", new_callable=AsyncMock + ) as mock_exec: + mock_exec.return_value = mock_process + + events = [e async for e in session.start_task("Hello")] + + assert len(events) == 2 + assert events[0] == {"type": "error", "error": {"message": "Fatal error"}} + assert events[1] == {"type": "exit", "code": 1, "stderr": "Fatal error"} + + @pytest.mark.asyncio + async def test_start_task_nonzero_with_only_benign_stderr_has_no_stderr_error( + self, + ): + """A benign stderr line is not duplicated as the process failure reason.""" + from cli.managed.session import ManagedClaudeSession + + session = ManagedClaudeSession("/tmp", "http://localhost:8082/v1") + + mock_process = AsyncMock() + mock_process.stdout.read.side_effect = [b""] + mock_process.stderr.read.side_effect = [ + b"claude.ai connectors are disabled in this environment\n", + b"", + ] + mock_process.wait.return_value = 1 + + with patch( + "asyncio.create_subprocess_exec", new_callable=AsyncMock + ) as mock_exec: + mock_exec.return_value = mock_process + + events = [e async for e in session.start_task("Hello")] + + assert events == [{"type": "exit", "code": 1, "stderr": None}] + @pytest.mark.asyncio async def test_drain_stderr_bounded_retains_cap_but_drains_to_eof(self): """Oversized stderr is fully drained so the pipe cannot deadlock; capture is bounded.""" diff --git a/tests/cli/test_managed_claude.py b/tests/cli/test_managed_claude.py index f638d2c0..c43d2b60 100644 --- a/tests/cli/test_managed_claude.py +++ b/tests/cli/test_managed_claude.py @@ -11,6 +11,7 @@ from cli.managed.claude import ( extract_managed_claude_session_id, parse_managed_claude_stdout_line, ) +from cli.managed.diagnostics import classify_managed_claude_stderr def _config(**overrides: object) -> ManagedClaudeConfig: @@ -98,6 +99,27 @@ def test_managed_claude_env_uses_sentinel_when_proxy_auth_blank() -> None: assert env["ANTHROPIC_AUTH_TOKEN"] == "fcc-no-auth" +def test_managed_claude_stderr_classifier_filters_known_benign_notice() -> None: + diagnostics = classify_managed_claude_stderr( + "claude.ai connectors are disabled in this environment" + ) + + assert diagnostics.has_benign + assert diagnostics.benign_lines == ( + "claude.ai connectors are disabled in this environment", + ) + assert diagnostics.fatal_text is None + + +def test_managed_claude_stderr_classifier_preserves_unknown_lines() -> None: + diagnostics = classify_managed_claude_stderr( + "claude.ai connectors are disabled in this environment\nFatal error" + ) + + assert diagnostics.has_benign + assert diagnostics.fatal_text == "Fatal error" + + def test_managed_claude_extracts_session_ids() -> None: assert extract_managed_claude_session_id({"session_id": "direct"}) == "direct" assert extract_managed_claude_session_id({"sessionId": "camel"}) == "camel" diff --git a/tests/core/anthropic/test_stream_recovery.py b/tests/core/anthropic/test_stream_recovery.py index 5e16988b..b4755c52 100644 --- a/tests/core/anthropic/test_stream_recovery.py +++ b/tests/core/anthropic/test_stream_recovery.py @@ -1,6 +1,7 @@ """Unit tests for resilient stream recovery helpers.""" import httpx +import openai from core.anthropic.streaming import ( EARLY_TRANSPARENT_MAX_RETRIES, @@ -16,6 +17,16 @@ from core.anthropic.streaming import ( ) +def _statusless_openai_api_error( + message: str, body: object | None = None +) -> openai.APIError: + return openai.APIError( + message, + request=httpx.Request("POST", "https://provider.test/messages"), + body=body, + ) + + def test_early_transparent_retry_total_attempts_is_five() -> None: assert EARLY_TRANSPARENT_TOTAL_ATTEMPTS == 5 assert EARLY_TRANSPARENT_MAX_RETRIES == 4 @@ -41,6 +52,44 @@ def test_retryable_stream_error_classifies_transport_and_http_status() -> None: ) +def test_retryable_stream_error_classifies_statusless_api_error_body_status() -> None: + assert is_retryable_stream_error( + _statusless_openai_api_error( + "stream embedded error", + {"error": {"message": "internal failure", "code": 500}}, + ) + ) + + +def test_retryable_stream_error_classifies_statusless_internal_error_type() -> None: + assert is_retryable_stream_error( + _statusless_openai_api_error( + "stream embedded error", + {"error": {"message": "internal failure", "type": "internal_server_error"}}, + ) + ) + + +def test_retryable_stream_error_classifies_resource_exhausted_text() -> None: + assert is_retryable_stream_error( + _statusless_openai_api_error( + "ResourceExhausted: limit reached while generating response", + {"error": {"message": "ResourceExhausted: limit reached"}}, + ) + ) + + +def test_retryable_stream_error_does_not_retry_bad_request_status() -> None: + request = httpx.Request("POST", "https://provider.test/messages") + assert not is_retryable_stream_error( + openai.BadRequestError( + "bad request", + response=httpx.Response(400, request=request), + body={"error": {"message": "bad request"}}, + ) + ) + + def test_stream_recovery_session_advances_early_retry_and_discards_holdback() -> None: session = RecoveryController(provider_name="TEST", request_id="REQ") @@ -60,6 +109,24 @@ def test_stream_recovery_session_advances_early_retry_and_discards_holdback() -> assert session.flush() == [] +def test_stream_recovery_session_retries_statusless_transient_api_error() -> None: + session = RecoveryController(provider_name="TEST", request_id="REQ") + + decision = session.advance_failure( + _statusless_openai_api_error( + "ResourceExhausted: limit reached while generating response", + {"error": {"message": "ResourceExhausted: limit reached"}}, + ), + stream_opened=True, + generated_output=False, + complete_tool_salvageable=False, + ) + + assert decision.action == RecoveryFailureAction.EARLY_RETRY + assert decision.retryable + assert decision.early_retry_attempt == 1 + + def test_stream_recovery_session_respects_early_retry_limit() -> None: session = RecoveryController(provider_name="TEST", request_id=None) diff --git a/tests/providers/test_error_mapping.py b/tests/providers/test_error_mapping.py index e9a5f29c..e740225a 100644 --- a/tests/providers/test_error_mapping.py +++ b/tests/providers/test_error_mapping.py @@ -41,6 +41,12 @@ def _make_openai_error(cls, message="test error", status_code=None): return cls(message, response=response, body=body) +def _make_statusless_openai_api_error( + message: str, body: object | None = None +) -> openai.APIError: + return openai.APIError(message, request=Request("POST", "http://test"), body=body) + + class TestMapError: """Tests for map_error function.""" @@ -124,6 +130,45 @@ class TestMapError: result = map_error(exc) assert isinstance(result, APIError) + def test_statusless_api_error_resource_exhausted_maps_to_overloaded(self): + """Status-less SDK APIError overload text maps to user-visible overload.""" + exc = _make_statusless_openai_api_error( + "ResourceExhausted: limit reached while generating response", + {"error": {"message": "ResourceExhausted: limit reached", "code": 500}}, + ) + + result = map_error(exc) + + assert isinstance(result, OverloadedError) + assert result.status_code == 529 + + def test_statusless_api_error_rate_limit_body_blocks_limiter(self): + """Status-less SDK APIError with 429 body uses scoped reactive limiting.""" + exc = _make_statusless_openai_api_error( + "stream embedded error", + {"error": {"message": "too many requests", "code": 429}}, + ) + limiter = MagicMock() + + result = map_error(exc, rate_limiter=limiter) + + assert isinstance(result, RateLimitError) + assert result.status_code == 429 + limiter.set_blocked.assert_called_once_with(60) + + def test_statusless_api_error_without_transient_marker_maps_to_generic_500(self): + """Unknown status-less SDK APIError stays generic instead of pretending success.""" + exc = _make_statusless_openai_api_error( + "stream embedded error", + {"error": {"message": "unknown provider failure"}}, + ) + + result = map_error(exc) + + assert isinstance(result, APIError) + assert result.status_code == 500 + assert result.message == "Provider API request failed." + def test_unmapped_exception_passthrough(self): """Non-openai exceptions are returned as-is.""" exc = RuntimeError("unexpected") diff --git a/tests/providers/test_provider_rate_limit.py b/tests/providers/test_provider_rate_limit.py index fd40230c..f46b8bf1 100644 --- a/tests/providers/test_provider_rate_limit.py +++ b/tests/providers/test_provider_rate_limit.py @@ -1,13 +1,16 @@ import asyncio import time +import openai import pytest import pytest_asyncio +from httpx import Request from providers.rate_limit import ( DEFAULT_UPSTREAM_MAX_RETRIES, UPSTREAM_TRANSIENT_TOTAL_ATTEMPTS, GlobalRateLimiter, + retryable_upstream_status, ) @@ -16,6 +19,28 @@ def test_upstream_transient_retry_total_attempts_is_five() -> None: assert DEFAULT_UPSTREAM_MAX_RETRIES == 4 +def _statusless_api_error(message: str, body: object | None) -> openai.APIError: + return openai.APIError(message, request=Request("POST", "http://x"), body=body) + + +def test_retryable_upstream_status_reads_statusless_api_error_body_status() -> None: + exc = _statusless_api_error( + "stream embedded error", + {"error": {"message": "internal failure", "code": 500}}, + ) + + assert retryable_upstream_status(exc) == 500 + + +def test_retryable_upstream_status_reads_statusless_resource_exhausted_text() -> None: + exc = _statusless_api_error( + "ResourceExhausted: limit reached while generating response", + {"error": {"message": "ResourceExhausted: limit reached"}}, + ) + + assert retryable_upstream_status(exc) == 503 + + class TestProviderRateLimiter: """Tests for providers.rate_limit.GlobalRateLimiter.""" @@ -350,6 +375,30 @@ class TestProviderRateLimiter: assert result == "ok" assert call_count == 2 + @pytest.mark.asyncio + async def test_execute_with_retry_succeeds_on_statusless_transient_api_error(self): + """Status-less SDK APIError transient markers participate in backoff retry.""" + 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 _statusless_api_error( + "ResourceExhausted: limit reached while generating response", + {"error": {"message": "ResourceExhausted: limit reached"}}, + ) + return "ok" + + 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 + @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/uv.lock b/uv.lock index c4e8f3e7..5df3ea24 100644 --- a/uv.lock +++ b/uv.lock @@ -561,7 +561,7 @@ wheels = [ [[package]] name = "free-claude-code" -version = "2.4.1" +version = "2.4.2" source = { editable = "." } dependencies = [ { name = "aiohttp" },