From 215758818edeeabbbdf8d848637bfaf82c2500bb Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Fri, 9 Oct 2026 06:55:42 -0700 Subject: [PATCH] fix(models): keep Claude retries behind request admission (#6545) * fix(models): keep Claude retries behind request admission * fix(models): keep Codex retries behind admission and log overrides * fix(models): retry paced Codex HTTP 529 errors through middleware * test(jina): allow independent HTTPX options in compatibility checks --------- Co-authored-by: Willem Jiang --- README.md | 4 + backend/docs/CONFIGURATION.md | 7 +- .../deerflow/agents/middlewares/AGENTS.md | 2 +- .../llm_error_handling_middleware.py | 2 +- .../harness/deerflow/models/AGENTS.md | 10 + .../deerflow/models/claude_provider.py | 7 + .../deerflow/models/openai_codex_provider.py | 8 + backend/tests/AGENTS.md | 6 + backend/tests/test_jina_response_limit.py | 4 +- backend/tests/test_jina_retry_after.py | 4 +- backend/tests/test_model_request_admission.py | 255 ++++++++++++++++++ 11 files changed, 303 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 12a96774a7..fdcdff4278 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,10 @@ literal prepend/append configuration without editing source templates. See Optional per-model [`request_admission`](backend/docs/CONFIGURATION.md#model-request-admission) paces requests to help stay within provider request-per-minute limits. It is disabled by default; see the linked guide to enable it. +Enabling it disables exposed SDK retries and the Claude and Codex adapters' +internal retry loops so middleware retries pass through request admission again. +Middleware retries also cover HTTP 529 overload responses. +A warning identifies `retry_max_attempts` values overridden by admission. For Google's official Gemini OpenAI-compatible endpoint, use the [Gemini reasoning profile](backend/docs/CONFIGURATION.md#gemini-via-googles-openai-compatible-endpoint). diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 341bb30ae7..fa5ea01ac4 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -83,8 +83,11 @@ disabling or regrouping active policies; conflicting settings fail model construction instead of resetting a live budget. When enabled, exposed SDK `max_retries` settings are set to zero: SDK retries -would bypass the admission hook. Agent middleware retries still work and each -new attempt is paced. Calls outside that middleware no longer get SDK retries. +would bypass the admission hook. Claude and Codex also set their provider-internal +`retry_max_attempts` to one, regardless of config or caller overrides, and log a +warning when reducing a larger value. Agent middleware retries still work, +including HTTP 529 overloads, and each new attempt is paced. Calls outside that +middleware no longer get SDK or Claude/Codex wrapper retries. Custom providers that bypass BaseChatModel admission hooks or perform hidden retries need their own integration. A caller-supplied `rate_limiter` cannot be combined with `request_admission`. diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index be2ade1c24..4f32b00e3a 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -78,7 +78,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation. their narrower discovery allowlists never rebuild the shared thread view or force eager sandbox acquisition. 8. **DanglingToolCallMiddleware** - Injects placeholder ToolMessages for AIMessage tool_calls that lack responses (e.g., user interruption), preserving raw provider tool-call payloads in `additional_kwargs["tool_calls"]`; malformed tool-call names and arguments are sanitized in the model-bound request so strict OpenAI-compatible providers do not reject the next request -9. **LLMErrorHandlingMiddleware** - Converts provider/model failures to recoverable assistant errors. Numeric/date retry hints share the 24-hour ceiling in `utils.retry_after`; unusable hints use jitter. Sync and async calls carry circuit-generation ownership; only the current owner may settle or release a half-open probe, so stale completions cannot affect a newer recovery attempt. Cancellation propagates unchanged without retry or failure accounting. +9. **LLMErrorHandlingMiddleware** - Assistant error recovery. HTTP 529 is transient by status; retries re-enter admission. Honor numeric/date hints up to 24h via `utils.retry_after`, otherwise jitter. Only current circuit-generation owners settle/release sync/async half-open probes. Cancellation propagates without retry/failure accounting. 10. **ArtifactResolutionMiddleware** - See [tool artifacts](TOOL_ARTIFACTS.md). 11. **Authorization / GuardrailMiddleware** - Up to two independent pre-tool-call gates run here. When `authorization.enabled`, the `AuthorizationProvider` instance already used for Layer 1 capability filtering is wrapped by `GuardrailAuthorizationAdapter` and reused for Layer 2 execution checks. A generated `tool_search` bypasses the adapter's second provider call only when the current build has a concrete deferred setup; its catalog was already filtered by Layer 1, and an ordinary same-named tool without that deferred setup receives no exemption. When `guardrails.enabled`, the explicitly configured `GuardrailProvider` is appended after authorization and still evaluates every call, including `tool_search`. Authorization therefore runs outermost and can deny before an external guardrail call; both use the existing middleware's fail-closed, audit, sync/async, and error-`ToolMessage` behavior. See the authorization RFC and [docs/GUARDRAILS.md](../../../../../docs/GUARDRAILS.md). diff --git a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py index 559c6406c9..05ced8800c 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py @@ -81,7 +81,7 @@ def _consume_empty_response_retry(request: ModelRequest) -> bool: return True -_RETRIABLE_STATUS_CODES = {408, 409, 425, 429, 500, 502, 503, 504} +_RETRIABLE_STATUS_CODES = {408, 409, 425, 429, 500, 502, 503, 504, 529} _BUSY_PATTERNS = ( "server busy", "temporarily unavailable", diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index 3fffbd8b75..2e80a14ed4 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -136,6 +136,16 @@ provider hints. Invalid or over-24-hour values retain local exponential backoff; finite negative values clamp to zero. Sync/async 429/500 regressions live in `tests/test_claude_provider_retry_after.py` and use offline provider doubles. +### Provider Retry Admission (`claude_provider.py`, `openai_codex_provider.py`) + +With DeerFlow `RequestAdmission`, `ClaudeChatModel` and `CodexChatModel` disable their own retry loops +(`retry_max_attempts=1`) even when config/caller kwargs request more attempts. +Log a warning with the previous attempt count when reducing a value above one; +already-single-attempt models should not warn. +The factory also disables SDK retries; middleware retries re-enter admission. +Unpaced models and unrelated custom rate limiters keep their wrapper retry policy. +`tests/test_model_request_admission.py` covers public sync/async calls through offline HTTP transports. + ### Claude Prompt Caching (`packages/harness/deerflow/models/claude_provider.py`) - The request payload shares objects with the caller: langchain-anthropic forwards Claude-native blocks (an image or document with a `source`, search results) and list-form system blocks by reference, and a reused tool binding passes its own tool dicts (the lead agent re-binds per call, so its tool dicts are fresh). Writing `cache_control` in place checkpointed the markers with the thread's messages, and the stale ones pushed every later request past the 4-breakpoint limit diff --git a/backend/packages/harness/deerflow/models/claude_provider.py b/backend/packages/harness/deerflow/models/claude_provider.py index 68239eb24f..2ffbf52112 100644 --- a/backend/packages/harness/deerflow/models/claude_provider.py +++ b/backend/packages/harness/deerflow/models/claude_provider.py @@ -84,8 +84,15 @@ class ClaudeChatModel(ChatAnthropic): is_oauth_token, load_claude_code_credential, ) + from deerflow.models.request_admission import RequestAdmission self._validate_retry_config() + if isinstance(self.rate_limiter, RequestAdmission): + # These wrapper retries, like SDK retries, bypass BaseChatModel's + # admission hook. Leave paced retries to the LLM middleware. + if self.retry_max_attempts != 1: + logger.warning("Request admission enabled; ignoring configured retry_max_attempts=%d; provider retries are handled by middleware", self.retry_max_attempts) + self.retry_max_attempts = 1 # Extract actual key value (SecretStr.str() returns '**********') current_key = "" diff --git a/backend/packages/harness/deerflow/models/openai_codex_provider.py b/backend/packages/harness/deerflow/models/openai_codex_provider.py index d05cc61278..c3471b160f 100644 --- a/backend/packages/harness/deerflow/models/openai_codex_provider.py +++ b/backend/packages/harness/deerflow/models/openai_codex_provider.py @@ -95,7 +95,15 @@ class CodexChatModel(BaseChatModel): def model_post_init(self, __context: Any) -> None: """Auto-load Codex CLI credentials.""" + from deerflow.models.request_admission import RequestAdmission + self._validate_retry_config() + if isinstance(self.rate_limiter, RequestAdmission): + # Wrapper retries bypass BaseChatModel's admission hook. Leave + # paced retries to the LLM middleware. + if self.retry_max_attempts != 1: + logger.warning("Request admission enabled; ignoring configured retry_max_attempts=%d; provider retries are handled by middleware", self.retry_max_attempts) + self.retry_max_attempts = 1 cred = self._load_codex_auth() if cred: diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index 3734254f03..56d285c595 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -68,6 +68,12 @@ an actionable reason; cover version detection offline in `test_support_compose.p Real rendering and production entry-point tests use only read-only Compose calls. Never start or stop a stack from these tests. +## Jina client options + +Retry/response-budget compatibility tests pin client count and caller-supplied +`proxy`/`trust_env` settings. Additional provider-owned HTTPX options must not +break those assertions. + ## Claude provider tests `test_claude_provider_prompt_caching.py` exercises real Anthropic SDK serialization diff --git a/backend/tests/test_jina_response_limit.py b/backend/tests/test_jina_response_limit.py index 0fbd668f69..8f208fe2bb 100644 --- a/backend/tests/test_jina_response_limit.py +++ b/backend/tests/test_jina_response_limit.py @@ -79,7 +79,9 @@ async def test_default_and_request_compatibility(transport, options): assert request.headers["x-timeout"] == "7" assert request.headers["content-type"] == "application/json" assert request.extensions["timeout"]["read"] == 7 - assert transport[2] == [{"proxy": "http://proxy.invalid", "trust_env": False}] + assert len(transport[2]) == 1 + assert transport[2][0]["proxy"] == "http://proxy.invalid" + assert transport[2][0]["trust_env"] is False assert stream.closed diff --git a/backend/tests/test_jina_retry_after.py b/backend/tests/test_jina_retry_after.py index 69edf2a0df..ea1282dda2 100644 --- a/backend/tests/test_jina_retry_after.py +++ b/backend/tests/test_jina_retry_after.py @@ -147,7 +147,9 @@ async def test_web_fetch_integration(rig, monkeypatch): rig.post.side_effect = [response(429, "7"), response(402, "1")] assert await tools.web_fetch_tool.ainvoke({"url": "https://example.com"}) == "Error: Jina API returned status 402: provider result" assert rig.waits == [7] - assert rig.options == [{"proxy": "http://proxy:8080", "trust_env": False}] + assert len(rig.options) == 1 + assert rig.options[0]["proxy"] == "http://proxy:8080" + assert rig.options[0]["trust_env"] is False assert [call.kwargs["timeout"] for call in rig.post.await_args_list] == [6, 1] assert set(tools.web_fetch_tool.args) == {"url"} diff --git a/backend/tests/test_model_request_admission.py b/backend/tests/test_model_request_admission.py index 80175ada41..43a7975fce 100644 --- a/backend/tests/test_model_request_admission.py +++ b/backend/tests/test_model_request_admission.py @@ -1,6 +1,8 @@ """Offline request pacing, queue lifecycle, and model-factory integration.""" import asyncio +import json +import logging import threading from concurrent.futures import ThreadPoolExecutor @@ -275,6 +277,259 @@ def test_disabled_factory_preserves_default(monkeypatch, registry): assert not admission._registry +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("status", [429, 500]) +async def test_claude_admission_does_not_retry_inside_provider(registry, mode, status): + import anthropic + import httpx + + from deerflow.config.app_config import AppConfig + from deerflow.models.factory import create_chat_model + + config = AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "models": [ + { + "name": "claude-paced", + "use": "deerflow.models.claude_provider:ClaudeChatModel", + "model": "claude-sonnet-4-6", + "api_key": "offline-test-key", + "request_admission": {"requests_per_minute": 1, "max_wait_seconds": 0.01}, + "retry_max_attempts": 7, + "max_retries": 7, + } + ], + } + ) + model = create_chat_model("claude-paced", app_config=config, attach_tracing=False, retry_max_attempts=5, max_retries=5) + calls = [] + + def respond(request): + calls.append(request) + if len(calls) == 1: + return httpx.Response(status, headers={"retry-after": "0"}, json={"type": "error", "error": {"type": "rate_limit_error" if status == 429 else "api_error", "message": "synthetic error"}}) + return httpx.Response( + 200, + json={ + "id": "msg-synthetic", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-6", + "content": [{"type": "text", "text": "unexpected retry"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 1}, + }, + ) + + error = anthropic.RateLimitError if status == 429 else anthropic.InternalServerError + if mode == "sync": + with httpx.Client(transport=httpx.MockTransport(respond)) as client: + model._client._client = client + with pytest.raises(error): + model.invoke("hello") + else: + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + model._async_client._client = client + with pytest.raises(error): + await model.ainvoke("hello") + assert len(calls) == 1 + assert model.max_retries == 0 + assert model.retry_max_attempts == 1 + assert config.get_model_config("claude-paced").retry_max_attempts == 7 + + +@pytest.mark.parametrize("kwargs,attempts", [({}, 3), ({"retry_max_attempts": 7}, 7)]) +def test_claude_without_admission_preserves_wrapper_retries(kwargs, attempts): + from deerflow.models.claude_provider import ClaudeChatModel + + model = ClaudeChatModel(model="claude-sonnet-4-6", api_key="offline-test-key", **kwargs) + assert model.retry_max_attempts == attempts + + +def test_claude_custom_rate_limiter_preserves_wrapper_retries(): + from langchain_core.rate_limiters import InMemoryRateLimiter + + from deerflow.models.claude_provider import ClaudeChatModel + + model = ClaudeChatModel(model="claude-sonnet-4-6", api_key="offline-test-key", rate_limiter=InMemoryRateLimiter(), retry_max_attempts=7) + assert model.retry_max_attempts == 7 + + +@pytest.fixture +def codex_credentials(monkeypatch): + from deerflow.models import openai_codex_provider as provider + from deerflow.models.credential_loader import CodexCliCredential + + monkeypatch.setattr(provider, "load_codex_cli_credential", lambda: CodexCliCredential("offline-test-token", "offline-test-account")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("status", [429, 500, 529]) +@pytest.mark.parametrize("enabled", [False, True]) +async def test_codex_factory_retry_policy_respects_admission(monkeypatch, registry, codex_credentials, mode, status, enabled): + import httpx + + from deerflow.config.app_config import AppConfig + from deerflow.models import openai_codex_provider as provider + from deerflow.models.factory import create_chat_model + + config = AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "models": [ + { + "name": "codex-paced", + "use": "deerflow.models.openai_codex_provider:CodexChatModel", + "model": "gpt-5.4", + "request_admission": {"requests_per_minute": 1, "max_wait_seconds": 0.01} if enabled else None, + "retry_max_attempts": 7, + } + ], + } + ) + model = create_chat_model("codex-paced", app_config=config, attach_tracing=False, retry_max_attempts=5) + calls = [] + sleeps = [] + + def respond(request): + calls.append(request) + if len(calls) == 1: + return httpx.Response(status, json={"error": "synthetic error"}) + event = {"type": "response.completed", "response": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "unexpected retry"}]}], "usage": {}}} + return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=f"data: {json.dumps(event)}\n\n") + + client_class = httpx.Client + monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=httpx.MockTransport(respond), **kwargs)) + monkeypatch.setattr(provider.time, "sleep", sleeps.append) + + if enabled: + with pytest.raises(httpx.HTTPStatusError) as caught: + if mode == "sync": + model.invoke("hello") + else: + await model.ainvoke("hello") + assert caught.value.response.status_code == status + assert len(calls) == 1 + assert sleeps == [] + assert isinstance(model.rate_limiter, admission.RequestAdmission) + assert model.retry_max_attempts == 1 + else: + result = model.invoke("hello") if mode == "sync" else await model.ainvoke("hello") + assert result.content == "unexpected retry" + assert len(calls) == 2 + assert len(sleeps) == 1 + assert model.rate_limiter is None + assert model.retry_max_attempts == 5 + assert config.get_model_config("codex-paced").retry_max_attempts == 7 + + +def test_codex_without_admission_preserves_default_retries(codex_credentials): + from deerflow.models.openai_codex_provider import CodexChatModel + + assert CodexChatModel().retry_max_attempts == 3 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("reason_phrase", [b"", b"Upstream error"]) +async def test_codex_529_middleware_retry_reenters_admission(monkeypatch, registry, clock, codex_credentials, mode, reason_phrase): + import httpx + from langchain.agents import create_agent + + from deerflow.agents.middlewares import llm_error_handling_middleware as errors + from deerflow.config.app_config import AppConfig + from deerflow.models import openai_codex_provider as provider + from deerflow.models.factory import create_chat_model + + monkeypatch.setattr(errors, "_PROCESS_LIMITER", None) + monkeypatch.setattr(errors, "_CAP_RESOLVED", False) + config = AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "llm_call": {"retry_max_attempts": 2, "retry_base_delay_ms": 25, "retry_cap_delay_ms": 25}, + "models": [ + { + "name": "codex-paced", + "use": "deerflow.models.openai_codex_provider:CodexChatModel", + "model": "gpt-5.4", + "request_admission": {"requests_per_minute": 60, "max_wait_seconds": 2}, + } + ], + } + ) + model = create_chat_model("codex-paced", app_config=config, attach_tracing=False) + calls = [] + sleeps = [] + + def respond(request): + calls.append(clock[0]) + if len(calls) == 1: + # HTTPStatusError omits this body; retry must recognize the status + # even when the reason phrase carries no overload/busy wording. + return httpx.Response(529, extensions={"reason_phrase": reason_phrase}, json={"error": "overloaded"}) + event = {"type": "response.completed", "response": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "recovered"}]}], "usage": {}}} + return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=f"data: {json.dumps(event)}\n\n") + + def sleep(delay): + sleeps.append(delay) + clock[0] += delay + + real_async_sleep = asyncio.sleep + + async def async_sleep(delay): + sleep(delay) + await real_async_sleep(0) + + client_class = httpx.Client + monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=httpx.MockTransport(respond), **kwargs)) + monkeypatch.setattr(provider.time, "sleep", sleep) + monkeypatch.setattr(asyncio, "sleep", async_sleep) + agent = create_agent(model=model, tools=[], middleware=[errors.LLMErrorHandlingMiddleware(app_config=config)]) + inputs = {"messages": [{"role": "user", "content": "hello"}]} + result = agent.invoke(inputs) if mode == "sync" else await agent.ainvoke(inputs) + + assert result["messages"][-1].content == "recovered" + assert model.retry_max_attempts == 1 + # Middleware's 25ms backoff alone cannot satisfy the one-second pacing + # interval: the second HTTP attempt must also wait for admission. + assert calls == pytest.approx([0, 1]) + assert sleeps[0] == pytest.approx(0.025) + assert len(sleeps) > 1 + assert all(0 <= delay <= 0.05 for delay in sleeps) + + +@pytest.mark.parametrize("provider", ["claude", "codex"]) +@pytest.mark.parametrize("limiter_kind", ["admission", "custom", "none"]) +@pytest.mark.parametrize("attempts", [1, 7]) +def test_provider_warns_only_when_admission_overrides_retries(codex_credentials, caplog, provider, limiter_kind, attempts): + from langchain_core.rate_limiters import InMemoryRateLimiter + + from deerflow.models.claude_provider import ClaudeChatModel + from deerflow.models.openai_codex_provider import CodexChatModel + + limiter = None + if limiter_kind == "admission": + limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60)) + elif limiter_kind == "custom": + limiter = InMemoryRateLimiter() + + model_class = ClaudeChatModel if provider == "claude" else CodexChatModel + kwargs = {"model": "claude-sonnet-4-6", "api_key": "offline-test-key"} if provider == "claude" else {} + logger_name = model_class.__module__ + with caplog.at_level(logging.WARNING, logger=logger_name): + model = model_class(rate_limiter=limiter, retry_max_attempts=attempts, **kwargs) + + assert model.retry_max_attempts == (1 if limiter_kind == "admission" else attempts) + if limiter_kind == "admission" and attempts != 1: + assert caplog.record_tuples == [(logger_name, logging.WARNING, f"Request admission enabled; ignoring configured retry_max_attempts={attempts}; provider retries are handled by middleware")] + else: + assert caplog.record_tuples == [] + + def test_admission_failures_are_not_retried_as_provider_errors(monkeypatch): from deerflow.agents.middlewares import llm_error_handling_middleware as errors from deerflow.config.app_config import AppConfig