mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-07-09 16:00:45 +00:00
Fix managed Claude diagnostics and transient retries (#965)
This commit is contained in:
Ali Khokhar
GitHub
parent
6a48811a9a
commit
6a56b18882
15 changed files with 522 additions and 29 deletions
|
|
@ -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
|
||||
|
|
|
|||
49
cli/managed/diagnostics.py
Normal file
49
cli/managed/diagnostics.py
Normal file
|
|
@ -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())
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
(
|
||||
|
|
|
|||
156
core/anthropic/streaming/transient_errors.py
Normal file
156
core/anthropic/streaming/transient_errors.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
2
uv.lock
generated
2
uv.lock
generated
|
|
@ -561,7 +561,7 @@ wheels = [
|
|||
|
||||
[[package]]
|
||||
name = "free-claude-code"
|
||||
version = "2.4.1"
|
||||
version = "2.4.2"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "aiohttp" },
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue