项目文件夹

文件
Ali Khokhar af12e7b2bb Coordinate provider recovery under concurrent load (#1205)
## 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 -->
2026-07-19 22:43:29 -07:00

238 行
7.0 KiB
Python

"""Tests for Hugging Face Inference Providers."""
from dataclasses import replace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from free_claude_code.config.provider_catalog import HUGGINGFACE_DEFAULT_BASE
from free_claude_code.core.anthropic import ReasoningReplayMode
from free_claude_code.core.reasoning import ReasoningEffort, ReasoningPolicy
from free_claude_code.providers.base import ProviderConfig
from tests.providers.request_factory import make_messages_request
from tests.providers.support import immediate_admission, profiled_provider
def make_request(**overrides):
return make_messages_request("openai/gpt-oss-120b:fastest", **overrides)
@pytest.fixture
def huggingface_config():
return ProviderConfig(
api_key="test_hf_key",
base_url=HUGGINGFACE_DEFAULT_BASE,
rate_limit=10,
rate_window=60,
)
@pytest.fixture
def huggingface_provider(huggingface_config):
return profiled_provider(
"huggingface", huggingface_config, admission=immediate_admission()
)
def test_default_base_url_constant():
assert HUGGINGFACE_DEFAULT_BASE == "https://router.huggingface.co/v1"
def test_init_uses_default_base_url_and_api_key(huggingface_config):
with patch(
"free_claude_code.providers.openai_chat.provider.AsyncOpenAI"
) as mock_openai:
provider = profiled_provider(
"huggingface", huggingface_config, admission=immediate_admission()
)
assert provider._api_key == "test_hf_key"
assert provider._base_url == HUGGINGFACE_DEFAULT_BASE
mock_openai.assert_called_once()
def test_init_strips_trailing_slash(huggingface_config):
config = replace(huggingface_config, base_url=f"{HUGGINGFACE_DEFAULT_BASE}/")
with patch("free_claude_code.providers.openai_chat.provider.AsyncOpenAI"):
provider = profiled_provider(
"huggingface", config, admission=immediate_admission()
)
assert provider._base_url == HUGGINGFACE_DEFAULT_BASE
def test_build_request_body_keeps_max_tokens(huggingface_provider):
with patch(
"free_claude_code.providers.openai_chat.request_policy.build_base_request_body"
) as mock_convert:
mock_convert.return_value = {
"model": "openai/gpt-oss-120b:fastest",
"messages": [{"role": "user", "name": "alice", "content": "hi"}],
"max_tokens": 42,
}
body = huggingface_provider._build_request_body(make_request())
mock_convert.assert_called_once()
assert (
mock_convert.call_args.kwargs["reasoning_replay"]
is ReasoningReplayMode.DISABLED
)
assert body["messages"][0].get("name") == "alice"
assert body["max_tokens"] == 42
assert "max_completion_tokens" not in body
def test_build_request_body_preserves_caller_extra_body(huggingface_provider):
extra_body = {"provider": "auto", "routing": {"bill_to": "my-org"}}
req = make_request(extra_body=extra_body)
body = huggingface_provider._build_request_body(req)
assert body["extra_body"] == extra_body
assert body["extra_body"] is not extra_body
assert body["extra_body"]["routing"] is not extra_body["routing"]
@pytest.mark.parametrize(
"reasoning",
(
ReasoningPolicy.off(),
ReasoningPolicy.on(effort=ReasoningEffort.MAX),
ReasoningPolicy.on(budget_tokens=4096),
),
)
def test_build_request_body_leaves_reasoning_control_to_selected_upstream(
huggingface_provider, reasoning
):
body = huggingface_provider._build_request_body(
make_request(),
reasoning=reasoning,
)
assert "reasoning_effort" not in body
assert "reasoning" not in body
assert "thinking" not in body
assert "extra_body" not in body
def test_build_request_body_does_not_replay_prior_thinking_blocks(
huggingface_provider,
):
req = make_request(
system=None,
messages=[
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": "hidden prior thought"},
{"type": "text", "text": "visible answer"},
],
}
],
)
body = huggingface_provider._build_request_body(req)
assert body["messages"] == [{"role": "assistant", "content": "visible answer"}]
assert "reasoning_content" not in body["messages"][0]
assert "hidden prior thought" not in str(body)
def test_build_request_body_does_not_replay_top_level_reasoning_content(
huggingface_provider,
):
req = make_request(
system=None,
messages=[
{
"role": "assistant",
"content": "visible answer",
"reasoning_content": "hidden prior reasoning",
}
],
)
body = huggingface_provider._build_request_body(req)
assert body["messages"] == [{"role": "assistant", "content": "visible answer"}]
assert "hidden prior reasoning" not in str(body)
@pytest.mark.asyncio
async def test_stream_response_text(huggingface_provider):
mock_chunk = MagicMock()
mock_chunk.choices = [
MagicMock(
delta=MagicMock(
content="Hello from Hugging Face",
reasoning_content=None,
tool_calls=None,
),
finish_reason="stop",
)
]
mock_chunk.usage = MagicMock(completion_tokens=5, prompt_tokens=10)
async def mock_stream():
yield mock_chunk
with patch.object(
huggingface_provider._client.chat.completions, "create", new_callable=AsyncMock
) as mock_create:
mock_create.return_value = mock_stream()
events = [
event
async for event in huggingface_provider.stream_response(make_request())
]
assert any(
'"text_delta"' in event and "Hello from Hugging Face" in event
for event in events
)
@pytest.mark.asyncio
async def test_stream_response_reasoning_content(huggingface_provider):
mock_chunk = MagicMock()
mock_chunk.choices = [
MagicMock(
delta=MagicMock(
content=None,
reasoning_content="Thinking via router",
tool_calls=None,
),
finish_reason="stop",
)
]
mock_chunk.usage = MagicMock(completion_tokens=2, prompt_tokens=10)
async def mock_stream():
yield mock_chunk
with patch.object(
huggingface_provider._client.chat.completions, "create", new_callable=AsyncMock
) as mock_create:
mock_create.return_value = mock_stream()
events = [
event
async for event in huggingface_provider.stream_response(make_request())
]
assert any(
'"thinking_delta"' in event and "Thinking via router" in event
for event in events
)
@pytest.mark.asyncio
async def test_cleanup(huggingface_provider):
huggingface_provider._client = AsyncMock()
await huggingface_provider.cleanup()
huggingface_provider._client.close.assert_called_once()