alishahryar1--free-claude-code
af12e7b2bb
## Problem Concurrent transient failures could start independent retry, replay, continuation, and repair loops while holding provider concurrency slots. This multiplied upstream attempts and could delay or strand terminal errors under fan-out. ## Changes | Before | After | | --- | --- | | Retry paths owned separate attempt budgets. | One logical-execution session caps all upstream work at five attempts. | | Concurrent failures backed off independently. | One provider-owned recovery episode elects a single half-open probe while followers coalesce. | | Backoff occupied stream concurrency. | Concurrency is held only while an upstream operation or stream is active. | | Provider catalog calls and stream creation used separate admission paths. | Every upstream operation uses one provider-generation admission controller. | | Cancellation could leave recovery ownership or follower state unresolved. | Cancellation releases permits, transfers probe ownership, and unregisters waiting followers. | | Late in-flight failures could cross an exhausted episode boundary. | Every coalesced execution retains that generation's terminal outcome. | | Replay tests allowed loose lifecycle assertions. | Exact SSE contracts prove retries and continuations emit one unduplicated response. | | Recovery wrappers could mask final diagnostics. | Final responses and traces retain the raw provider failure and request ID. | <!-- greptile_comment --> <details open><summary><h3>Greptile Summary</h3></summary> This PR coordinates provider recovery and retry work under concurrent load. The main changes are: - One five-attempt budget for each logical execution. - Provider-wide recovery episodes with one elected probe. - Shared admission for streams, catalog calls, rate limits, and concurrency. - Concurrency permits held only during active upstream work. - Cancellation-safe probe ownership and preserved final diagnostics. </details> <h3>Confidence Score: 5/5</h3> This looks safe to merge. No blocking issues found in the changed code. <details><summary><h3><a href="https://www.greptile.com/trex"><img alt="T-Rex" src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg" height="20" align="absmiddle"></a> T-Rex Logs</h3></summary> **What T-Rex did** - Reviewed the coordinated-recovery-01-before.log to understand how the exhausted generation outcome was not preserved in a late in-flight failure. - Reviewed the coordinated-recovery-02-after.log to confirm that the updated implementation preserves the exhausted generation outcome for the same focused contract set. - Validated that the provider-admission-full-current.log shows the complete requested test file passed under Python 3.14 with uv run pytest -n 0. <a href="https://app.greptile.com/trex/runs/15050270/artifacts"><picture><source media="(prefers-color-scheme: dark)" srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifactsDark.svg?v=4"><source media="(prefers-color-scheme: light)" srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"><img alt="View all artifacts" src="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"></picture></a> <sub><a href="https://www.greptile.com/trex"><img alt="T-Rex" src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg" height="14" align="absmiddle"></a> Ran code and verified through T-Rex</sub> </details> <details open><summary><h3>Important Files Changed</h3></summary> | Filename | Overview | |----------|----------| | src/free_claude_code/providers/admission.py | Adds shared admission, retry budgets, recovery episodes, probe election, and cancellation handling. | | src/free_claude_code/providers/openai_chat/provider.py | Moves stream creation, replay, continuation, and repair onto one admission-owned retry session. | | src/free_claude_code/providers/stream_recovery.py | Selects replay, continuation, repair, or final failure using the remaining shared attempt budget. | | src/free_claude_code/providers/failure_policy.py | Adds recovery exhaustion handling and preserves the underlying provider error for final classification. | | src/free_claude_code/providers/runtime/factory.py | Creates one admission controller per provider generation and passes it through provider factories. | </details> <details open><summary><h3>Sequence Diagram</h3></summary> <a href="#gh-light-mode-only"> ```mermaid %%{init: {'theme': 'neutral'}}%% sequenceDiagram participant E as Execution participant A as Admission controller participant P as Provider participant F as Concurrent follower E->>A: Open attempt A->>P: Send upstream request P-->>E: Retryable failure E->>A: Open recovery episode F->>A: Request admission A-->>F: Coalesce and wait E->>A: Claim probe A->>P: Send half-open probe alt Probe succeeds P-->>E: Valid response E->>A: Close recovery episode A-->>F: Release waiter else Probe fails P-->>E: Retryable failure E->>A: Schedule next probe or finalize error end ``` </a> <a href="#gh-dark-mode-only"> ```mermaid %%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%% sequenceDiagram participant E as Execution participant A as Admission controller participant P as Provider participant F as Concurrent follower E->>A: Open attempt A->>P: Send upstream request P-->>E: Retryable failure E->>A: Open recovery episode F->>A: Request admission A-->>F: Coalesce and wait E->>A: Claim probe A->>P: Send half-open probe alt Probe succeeds P-->>E: Valid response E->>A: Close recovery episode A-->>F: Release waiter else Probe fails P-->>E: Retryable failure E->>A: Schedule next probe or finalize error end ``` </a> </details> <sub>Reviews (2): Last reviewed commit: ["Harden coordinated retry lifecycle invar..."](https://github.com/alishahryar1/free-claude-code/commit/2e871c8649d148b5eb71d21f80bf870ae2d11708) | [Re-trigger Greptile](https://app.greptile.com/api/retrigger?id=45554917)</sub> <!-- /greptile_comment -->
383 行
11 KiB
Python
383 行
11 KiB
Python
"""Raw provider failure classification into the canonical neutral model."""
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
import openai
|
|
import pytest
|
|
|
|
from free_claude_code.core.diagnostics import (
|
|
ERROR_DETAIL_DISPLAY_CAP_BYTES,
|
|
attach_upstream_error_body,
|
|
)
|
|
from free_claude_code.core.failures import ExecutionFailure, FailureKind
|
|
from free_claude_code.providers.failure_policy import (
|
|
ProviderRecoveryExhausted,
|
|
classify_provider_failure,
|
|
is_retryable_provider_error,
|
|
retryable_upstream_status,
|
|
)
|
|
|
|
|
|
def _openai_status_error(
|
|
error_type: type[openai.APIStatusError],
|
|
*,
|
|
status_code: int,
|
|
message: str,
|
|
body: object | None = None,
|
|
) -> openai.APIStatusError:
|
|
request = httpx.Request("POST", "https://provider.test/v1/chat/completions")
|
|
response = httpx.Response(status_code, request=request)
|
|
return error_type(
|
|
message,
|
|
response=response,
|
|
body=body or {"error": {"message": message}},
|
|
)
|
|
|
|
|
|
def _statusless_openai_error(message: str, body: object | None) -> openai.APIError:
|
|
return openai.APIError(
|
|
message,
|
|
request=httpx.Request("POST", "https://provider.test/v1/chat/completions"),
|
|
body=body,
|
|
)
|
|
|
|
|
|
def _http_status_error(status_code: int, message: str) -> httpx.HTTPStatusError:
|
|
request = httpx.Request("POST", "https://provider.test/v1/messages")
|
|
response = httpx.Response(
|
|
status_code,
|
|
request=request,
|
|
json={"error": {"message": message, "api_key": "SECRET"}},
|
|
)
|
|
return httpx.HTTPStatusError(message, request=request, response=response)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ClassificationCase:
|
|
name: str
|
|
error: Callable[[], Exception]
|
|
kind: FailureKind
|
|
status_code: int
|
|
retryable: bool
|
|
|
|
|
|
_CASES = (
|
|
_ClassificationCase(
|
|
"openai_authentication",
|
|
lambda: _openai_status_error(
|
|
openai.AuthenticationError,
|
|
status_code=401,
|
|
message="Unauthorized",
|
|
),
|
|
FailureKind.AUTHENTICATION,
|
|
401,
|
|
False,
|
|
),
|
|
_ClassificationCase(
|
|
"openai_rate_limit",
|
|
lambda: _openai_status_error(
|
|
openai.RateLimitError,
|
|
status_code=429,
|
|
message="Too many requests",
|
|
),
|
|
FailureKind.RATE_LIMIT,
|
|
429,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"openai_bad_request",
|
|
lambda: _openai_status_error(
|
|
openai.BadRequestError,
|
|
status_code=400,
|
|
message="bad tool shape",
|
|
),
|
|
FailureKind.INVALID_REQUEST,
|
|
400,
|
|
False,
|
|
),
|
|
_ClassificationCase(
|
|
"openai_overload_marker",
|
|
lambda: _openai_status_error(
|
|
openai.InternalServerError,
|
|
status_code=500,
|
|
message="No capacity available",
|
|
),
|
|
FailureKind.OVERLOADED,
|
|
529,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"openai_generic_503_preserved",
|
|
lambda: _openai_status_error(
|
|
openai.InternalServerError,
|
|
status_code=503,
|
|
message="generic server failure",
|
|
),
|
|
FailureKind.UPSTREAM,
|
|
503,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"statusless_openai_rate_limit_body",
|
|
lambda: _statusless_openai_error(
|
|
"stream embedded error",
|
|
{"error": {"message": "too many requests", "code": 429}},
|
|
),
|
|
FailureKind.RATE_LIMIT,
|
|
429,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"statusless_openai_overload_body",
|
|
lambda: _statusless_openai_error(
|
|
"ResourceExhausted: limit reached",
|
|
{"error": {"message": "ResourceExhausted: limit reached"}},
|
|
),
|
|
FailureKind.OVERLOADED,
|
|
529,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"statusless_openai_unknown_is_not_retryable",
|
|
lambda: _statusless_openai_error(
|
|
"stream embedded error",
|
|
{"error": {"message": "unknown provider failure"}},
|
|
),
|
|
FailureKind.UPSTREAM,
|
|
500,
|
|
False,
|
|
),
|
|
_ClassificationCase(
|
|
"http_403_keeps_authentication_quirk",
|
|
lambda: _http_status_error(403, "Forbidden"),
|
|
FailureKind.AUTHENTICATION,
|
|
401,
|
|
False,
|
|
),
|
|
_ClassificationCase(
|
|
"http_502_keeps_overload_quirk",
|
|
lambda: _http_status_error(502, "Bad gateway"),
|
|
FailureKind.OVERLOADED,
|
|
529,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"http_599_preserves_status",
|
|
lambda: _http_status_error(599, "Upstream failure"),
|
|
FailureKind.UPSTREAM,
|
|
599,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"http_405_is_not_retryable",
|
|
lambda: _http_status_error(405, "Wrong endpoint"),
|
|
FailureKind.UPSTREAM,
|
|
405,
|
|
False,
|
|
),
|
|
_ClassificationCase(
|
|
"read_timeout_keeps_pre_start_status",
|
|
lambda: httpx.ReadTimeout(
|
|
"",
|
|
request=httpx.Request("POST", "https://provider.test/v1/messages"),
|
|
),
|
|
FailureKind.TIMEOUT,
|
|
502,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"openai_connection_error_keeps_status",
|
|
lambda: openai.APIConnectionError(
|
|
request=httpx.Request("POST", "https://provider.test/v1/chat/completions")
|
|
),
|
|
FailureKind.UNAVAILABLE,
|
|
500,
|
|
True,
|
|
),
|
|
_ClassificationCase(
|
|
"unknown_exception_keeps_gateway_status",
|
|
lambda: RuntimeError("unexpected provider failure"),
|
|
FailureKind.UPSTREAM,
|
|
502,
|
|
False,
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("case", _CASES, ids=lambda case: case.name)
|
|
def test_raw_provider_failure_maps_to_canonical_failure(
|
|
case: _ClassificationCase,
|
|
) -> None:
|
|
failure = classify_provider_failure(
|
|
case.error(),
|
|
provider_name="TEST_PROVIDER",
|
|
read_timeout_s=30.0,
|
|
request_id="req_classification",
|
|
)
|
|
|
|
assert isinstance(failure, ExecutionFailure)
|
|
assert failure.kind is case.kind
|
|
assert failure.status_code == case.status_code
|
|
assert failure.retryable is case.retryable
|
|
assert failure.message.strip()
|
|
assert "Request ID: req_classification" in failure.message
|
|
assert "SECRET" not in failure.message
|
|
|
|
|
|
def test_classification_preserves_useful_body_while_redacting_credentials() -> None:
|
|
error = _http_status_error(
|
|
400,
|
|
"unsupported model format authorization: Bearer AUTH_SECRET",
|
|
)
|
|
|
|
failure = classify_provider_failure(
|
|
error,
|
|
provider_name="LOCAL",
|
|
read_timeout_s=60.0,
|
|
request_id="req_body",
|
|
)
|
|
|
|
assert failure.kind is FailureKind.INVALID_REQUEST
|
|
assert failure.status_code == 400
|
|
assert "Upstream provider LOCAL returned HTTP 400." in failure.message
|
|
assert "unsupported model format" in failure.message
|
|
assert "Request ID: req_body" in failure.message
|
|
assert "AUTH_SECRET" not in failure.message
|
|
assert "SECRET" not in failure.message
|
|
|
|
|
|
def test_auth_failure_preserves_model_error_body_instead_of_masking_it() -> None:
|
|
error = _openai_status_error(
|
|
openai.AuthenticationError,
|
|
status_code=401,
|
|
message="Unauthorized",
|
|
body={
|
|
"type": "error",
|
|
"error": {
|
|
"type": "ModelError",
|
|
"message": ("Model qwen3.7-max is not supported for format oa-compat"),
|
|
},
|
|
},
|
|
)
|
|
|
|
failure = classify_provider_failure(
|
|
error,
|
|
provider_name="OPENCODE_GO",
|
|
read_timeout_s=60.0,
|
|
request_id="req_model",
|
|
)
|
|
|
|
assert failure.kind is FailureKind.AUTHENTICATION
|
|
assert failure.status_code == 401
|
|
assert "Category: ModelError" in failure.message
|
|
assert "Provider authentication failed. Check API key." in failure.message
|
|
assert "Model qwen3.7-max is not supported for format oa-compat" in failure.message
|
|
assert "Request ID: req_model" in failure.message
|
|
|
|
|
|
def test_empty_http_error_body_is_reported_explicitly() -> None:
|
|
request = httpx.Request("POST", "https://provider.test/v1/messages")
|
|
response = httpx.Response(500, request=request, content=b"")
|
|
error = httpx.HTTPStatusError(
|
|
"Server Error",
|
|
request=request,
|
|
response=response,
|
|
)
|
|
|
|
failure = classify_provider_failure(
|
|
error,
|
|
provider_name="EMPTY",
|
|
read_timeout_s=30.0,
|
|
request_id="req_empty",
|
|
)
|
|
|
|
assert failure.kind is FailureKind.UPSTREAM
|
|
assert failure.status_code == 500
|
|
assert "Upstream provider EMPTY returned HTTP 500." in failure.message
|
|
assert "(empty upstream error body)" in failure.message
|
|
|
|
|
|
def test_http_405_diagnostic_names_rejected_upstream_endpoint() -> None:
|
|
failure = classify_provider_failure(
|
|
_http_status_error(405, "Method Not Allowed"),
|
|
provider_name="LOCAL",
|
|
read_timeout_s=30.0,
|
|
request_id="req_405",
|
|
)
|
|
|
|
assert failure.kind is FailureKind.UPSTREAM
|
|
assert failure.status_code == 405
|
|
assert (
|
|
"Upstream provider LOCAL rejected the request method or endpoint (HTTP 405)."
|
|
in failure.message
|
|
)
|
|
assert "Request ID: req_405" in failure.message
|
|
|
|
|
|
def test_connection_cause_chain_is_redacted_and_capped() -> None:
|
|
request = httpx.Request("POST", "https://provider.test/v1/chat/completions")
|
|
error = openai.APIConnectionError(request=request)
|
|
error.__cause__ = httpx.ConnectError(
|
|
"connect failed authorization: Bearer CAUSE_SECRET "
|
|
+ "x" * (ERROR_DETAIL_DISPLAY_CAP_BYTES + 10),
|
|
request=request,
|
|
)
|
|
|
|
failure = classify_provider_failure(
|
|
error,
|
|
provider_name="NIM",
|
|
read_timeout_s=30.0,
|
|
request_id="req_cause",
|
|
)
|
|
|
|
assert "Caused by:" in failure.message
|
|
assert "ConnectError: connect failed authorization: <redacted>" in failure.message
|
|
assert "CAUSE_SECRET" not in failure.message
|
|
assert f"truncated after {ERROR_DETAIL_DISPLAY_CAP_BYTES} bytes" in failure.message
|
|
assert "Request ID: req_cause" in failure.message
|
|
|
|
|
|
def test_attached_streamed_error_body_remains_bounded() -> None:
|
|
request = httpx.Request("POST", "https://provider.test/v1/messages")
|
|
response = httpx.Response(500, request=request, content=b"")
|
|
error = httpx.HTTPStatusError(
|
|
"Server Error",
|
|
request=request,
|
|
response=response,
|
|
)
|
|
attach_upstream_error_body(
|
|
error,
|
|
"x" * (ERROR_DETAIL_DISPLAY_CAP_BYTES + 10),
|
|
)
|
|
|
|
failure = classify_provider_failure(
|
|
error,
|
|
provider_name="LONG",
|
|
read_timeout_s=30.0,
|
|
request_id="req_long",
|
|
)
|
|
|
|
assert f"truncated after {ERROR_DETAIL_DISPLAY_CAP_BYTES} bytes" in failure.message
|
|
assert "x" * 100 in failure.message
|
|
|
|
|
|
def test_shared_recovery_exhaustion_preserves_last_provider_failure() -> None:
|
|
raw_error = _http_status_error(503, "provider still unavailable")
|
|
exhausted = ProviderRecoveryExhausted(raw_error)
|
|
|
|
failure = classify_provider_failure(
|
|
exhausted,
|
|
provider_name="SHARED",
|
|
read_timeout_s=30.0,
|
|
request_id="req_exhausted",
|
|
)
|
|
|
|
assert failure.kind is FailureKind.OVERLOADED
|
|
assert failure.status_code == 529
|
|
assert "provider still unavailable" in failure.message
|
|
assert "Request ID: req_exhausted" in failure.message
|
|
assert retryable_upstream_status(exhausted) is None
|
|
assert not is_retryable_provider_error(exhausted)
|