mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(middleware): guard OverflowError when parsing non-finite Retry-After hints (#6455)
* fix(middleware): guard OverflowError when parsing non-finite Retry-After hints * fix(middleware): guard OverflowError when parsing non-finite Retry-After hints * fix(retry): bound provider retry hints across model paths Share a 24-hour ceiling for middleware numeric/date hints and Claude's integer-second hints. Non-finite or oversized values use local backoff; valid hints above the jitter cap remain unchanged and negatives clamp to zero. Parameterize unusable hints at both middleware test levels, including nan. Cover numeric/date boundaries and sync/async recovery through both retry paths. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
7797409355
commit
9472579ef0
@ -506,6 +506,8 @@ DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser
|
||||
|
||||
> Cancelling a model recovery probe, including while it is queued or waiting to retry, lets the next call check whether the provider has recovered. Cancellation does not count as a provider failure or release another call's active recovery probe.
|
||||
>
|
||||
> Model retries honor valid provider `Retry-After` hints up to 24 hours, including waits longer than the local backoff cap. Malformed, non-finite, or over-limit hints use normal backoff instead. This applies to numeric/date hints in LLM error handling and integer-second hints in Claude's retry path.
|
||||
>
|
||||
> With lease heartbeat enabled, a transient RunStore renewal error is retried only until the last confirmed lease expires; the stale worker then cancels local execution and suppresses checkpoint, completion-hook, delivery-receipt, and thread-status finalization. A remote tool side effect already in flight may still be outside local cancellation.
|
||||
>
|
||||
> Reconciliation uses an atomic takeover claim that re-checks the lease after candidate selection, so a successful owner renewal wins over orphan recovery and only one reconciler can report a run as recovered. When multiple Gateway workers share the Docker/AIO or E2B sandbox backend, also configure `sandbox.ownership.type: redis`; E2B uses the leases during background startup and periodic reconciliation so duplicate/orphan cleanup cannot terminate a live peer's sandbox.
|
||||
|
||||
@ -74,7 +74,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. 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** - 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.
|
||||
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).
|
||||
|
||||
|
||||
@ -27,6 +27,7 @@ from deerflow.agents.middlewares.model_response import append_visible_text, fini
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.models.request_admission import AdmissionError
|
||||
from deerflow.utils.custom_events import aemit_custom_event, emit_custom_event
|
||||
from deerflow.utils.retry_after import bounded_retry_after_ms
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -695,7 +696,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
|
||||
def _build_retry_delay_ms(self, prev_delay_ms: int | None, exc: BaseException, reason: str = "transient") -> int:
|
||||
"""Compute the next retry delay (ms) using decorrelated jitter.
|
||||
|
||||
An explicit ``Retry-After`` from the provider is honored as-is (no
|
||||
A finite ``Retry-After`` up to 24 hours is honored as-is (no
|
||||
jitter) - the server told us exactly when to come back, and for a
|
||||
burst-rate 429 this is strongly preferred over any computed delay.
|
||||
Otherwise AWS-style "decorrelated jitter" is applied:
|
||||
@ -1074,12 +1075,12 @@ def _extract_retry_after_ms(exc: BaseException) -> int | None:
|
||||
|
||||
try:
|
||||
multiplier = 1 if "ms" in header_name.lower() else 1000
|
||||
return max(0, int(float(raw) * multiplier))
|
||||
except (TypeError, ValueError):
|
||||
return bounded_retry_after_ms(float(raw) * multiplier)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
try:
|
||||
target = parsedate_to_datetime(str(raw))
|
||||
delta = target.timestamp() - time.time()
|
||||
return max(0, int(delta * 1000))
|
||||
return bounded_retry_after_ms(delta * 1000)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
return None
|
||||
|
||||
|
||||
@ -129,6 +129,13 @@ Offline HTTP-stream coverage: `tests/test_codex_stream_terminal_events.py`.
|
||||
- `ClaudeChatModel.model_post_init` calls `load_claude_code_credential()` for every instance, and `create_chat_model` builds fresh instances per run (lead agent, title, summarization, subagents)
|
||||
- `$CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR` is a one-shot handoff: a pipe returns EOF and a file keeps its advanced offset. `_read_secret_from_file_descriptor` therefore caches a non-empty secret per `(env_var, fd)` under a lock held across the read. Do not drop the cache or the lock — later instances would get no credential, and the Anthropic SDK raises `TypeError: Could not resolve authentication method` before sending. Empty reads, `OSError`, and UTF-8 decode failures are not cached; unreadable handoffs return `None` so the loader can try credential files. Warnings must not include token contents. The key is the descriptor number on purpose — a closed handoff keeps serving its token, and a secret placed on a recycled number in-process is not re-read unless the cache is cleared. The cache is per process, so a new process (e.g. a uvicorn `--reload` worker) cannot recover a drained descriptor. Pinned by `tests/test_credential_loader.py` and `tests/test_claude_fd_encoding.py`, including two-instance `ClaudeChatModel` tests
|
||||
|
||||
### Claude Retry Delays (`claude_provider.py`)
|
||||
|
||||
`_calc_backoff_ms` uses `utils.retry_after.bounded_retry_after_ms` for integer-second
|
||||
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.
|
||||
|
||||
### 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
|
||||
|
||||
@ -30,6 +30,8 @@ from langchain_anthropic import ChatAnthropic
|
||||
from langchain_core.messages import BaseMessage
|
||||
from pydantic import PrivateAttr
|
||||
|
||||
from deerflow.utils.retry_after import bounded_retry_after_ms
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MAX_RETRIES = 3
|
||||
@ -427,8 +429,10 @@ class ClaudeChatModel(ChatAnthropic):
|
||||
retry_after = error.response.headers.get("Retry-After")
|
||||
if retry_after:
|
||||
try:
|
||||
total_ms = int(retry_after) * 1000
|
||||
except (ValueError, TypeError):
|
||||
hint_ms = bounded_retry_after_ms(int(retry_after) * 1000)
|
||||
if hint_ms is not None:
|
||||
total_ms = hint_ms
|
||||
except (ValueError, TypeError, OverflowError):
|
||||
pass
|
||||
|
||||
return total_ms
|
||||
|
||||
@ -1,3 +1,10 @@
|
||||
### Model Retry Hints
|
||||
|
||||
`retry_after.py::bounded_retry_after_ms` shares the 24-hour provider-hint ceiling
|
||||
between LLM middleware (numeric/date hints) and Claude backoff. Non-finite or
|
||||
over-limit delays return `None` for local backoff; finite negatives clamp to zero.
|
||||
Validate before integer truncation and preserve valid hints above the jitter cap.
|
||||
|
||||
### Port Allocation Bounds
|
||||
|
||||
`network.py::PortAllocator.allocate` caps its exclusive search endpoint at
|
||||
|
||||
14
backend/packages/harness/deerflow/utils/retry_after.py
Normal file
14
backend/packages/harness/deerflow/utils/retry_after.py
Normal file
@ -0,0 +1,14 @@
|
||||
"""Bounds for provider-supplied model retry delays."""
|
||||
|
||||
import math
|
||||
|
||||
MAX_RETRY_AFTER_MS = 24 * 60 * 60 * 1000
|
||||
|
||||
|
||||
def bounded_retry_after_ms(delay_ms: int | float) -> int | None:
|
||||
"""Return a finite delay up to 24 hours, or None to use local backoff."""
|
||||
if isinstance(delay_ms, float) and not math.isfinite(delay_ms):
|
||||
return None
|
||||
if delay_ms > MAX_RETRY_AFTER_MS:
|
||||
return None
|
||||
return max(0, int(delay_ms))
|
||||
74
backend/tests/test_claude_provider_retry_after.py
Normal file
74
backend/tests/test_claude_provider_retry_after.py
Normal file
@ -0,0 +1,74 @@
|
||||
"""Offline regressions for provider-owned Retry-After backoff."""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
import anthropic
|
||||
import httpx
|
||||
import pytest
|
||||
from langchain_anthropic import ChatAnthropic
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
|
||||
from deerflow.models.claude_provider import ClaudeChatModel
|
||||
|
||||
|
||||
@pytest.mark.parametrize("attempt", [1, 2])
|
||||
@pytest.mark.parametrize("raw", ["1e999", "inf", "-inf", "nan", "1e15", "999999999999999999999999999999", pytest.param("9" * 1000, id="huge-integer"), "86401", "bad", None])
|
||||
def test_unusable_claude_retry_after_uses_default_backoff(attempt: int, raw: str | None) -> None:
|
||||
error = Exception("rate limited")
|
||||
error.response = SimpleNamespace(headers={"Retry-After": raw})
|
||||
assert ClaudeChatModel._calc_backoff_ms(attempt, error) == 2400 * 2 ** (attempt - 1)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("raw", "expected_ms"), [("-2", 0), ("0", 0), ("60", 60000), ("86400", 86400000)])
|
||||
def test_usable_claude_retry_after_is_preserved(raw: str, expected_ms: int) -> None:
|
||||
error = Exception("rate limited")
|
||||
error.response = SimpleNamespace(headers={"Retry-After": raw})
|
||||
assert ClaudeChatModel._calc_backoff_ms(1, error) == expected_ms
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("mode", ["sync", "async"])
|
||||
@pytest.mark.parametrize("error_type", [anthropic.RateLimitError, anthropic.InternalServerError])
|
||||
@pytest.mark.parametrize("raw", ["999999999999999999999999999999", "86401", "1e999", "60"])
|
||||
async def test_claude_retry_loops_use_safe_delay(monkeypatch: pytest.MonkeyPatch, mode: str, error_type: type[anthropic.APIStatusError], raw: str) -> None:
|
||||
# Avoid credential discovery/client construction; only the provider call is fake.
|
||||
monkeypatch.setattr(ClaudeChatModel, "model_post_init", lambda self, context: None)
|
||||
model = ClaudeChatModel(model="claude-sonnet-4-6", anthropic_api_key="sk-ant-fake", retry_max_attempts=2)
|
||||
model._is_oauth = False
|
||||
response = httpx.Response(429 if error_type is anthropic.RateLimitError else 500, headers={"Retry-After": raw}, request=httpx.Request("POST", "https://example.invalid/messages"))
|
||||
error = error_type("provider failure", response=response, body={})
|
||||
expected = ChatResult(generations=[ChatGeneration(message=AIMessage(content="recovered"))])
|
||||
attempts = 0
|
||||
waits: list[float] = []
|
||||
|
||||
def provider_call(self, messages, **kwargs) -> ChatResult:
|
||||
nonlocal attempts
|
||||
attempts += 1
|
||||
if attempts == 1:
|
||||
raise error
|
||||
return expected
|
||||
|
||||
async def async_provider_call(self, messages, **kwargs) -> ChatResult:
|
||||
return provider_call(self, messages, **kwargs)
|
||||
|
||||
def fake_sleep(delay: float) -> None:
|
||||
assert delay == (60 if raw == "60" else 2.4)
|
||||
waits.append(delay)
|
||||
|
||||
async def fake_async_sleep(delay: float) -> None:
|
||||
fake_sleep(delay)
|
||||
|
||||
monkeypatch.setattr(ChatAnthropic, "_generate", provider_call)
|
||||
monkeypatch.setattr(ChatAnthropic, "_agenerate", async_provider_call)
|
||||
monkeypatch.setattr(time, "sleep", fake_sleep)
|
||||
monkeypatch.setattr(asyncio, "sleep", fake_async_sleep)
|
||||
if mode == "sync":
|
||||
result = model._generate([HumanMessage(content="hi")])
|
||||
else:
|
||||
result = await model._agenerate([HumanMessage(content="hi")])
|
||||
assert result is expected
|
||||
assert attempts == 2
|
||||
assert len(waits) == 1
|
||||
@ -5,6 +5,8 @@ import concurrent.futures
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from email.utils import format_datetime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
@ -20,6 +22,7 @@ from langgraph.errors import GraphBubbleUp
|
||||
from deerflow.agents.middlewares.llm_error_handling_middleware import (
|
||||
EmptyModelResponseError,
|
||||
LLMErrorHandlingMiddleware,
|
||||
_extract_retry_after_ms,
|
||||
)
|
||||
from deerflow.config.app_config import AppConfig, LlmCallConfig
|
||||
from deerflow.config.sandbox_config import SandboxConfig
|
||||
@ -1871,6 +1874,97 @@ def test_retry_delay_honors_retry_after_without_jitter() -> None:
|
||||
assert delay == 5000
|
||||
|
||||
|
||||
@pytest.mark.parametrize("header_name", ["Retry-After", "retry-after", "Retry-After-Ms", "retry-after-ms"])
|
||||
@pytest.mark.parametrize("header_value", ["1e999", "inf", "-inf", "nan", "1e15", "1e308", "99999999999999999999", "not-a-number"])
|
||||
def test_unusable_retry_after_falls_back_to_jitter(header_name: str, header_value: str) -> None:
|
||||
"""Unusable provider delays must fall back to bounded jitter."""
|
||||
middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
|
||||
exc = FakeError("rate limited", status_code=429, headers={header_name: header_value})
|
||||
assert 100 <= middleware._build_retry_delay_ms(100, exc) <= 300
|
||||
|
||||
|
||||
@pytest.mark.parametrize("header_name", ["Retry-After", "retry-after", "Retry-After-Ms", "retry-after-ms"])
|
||||
@pytest.mark.parametrize("header_value", ["1e999", "inf", "-inf", "nan", "1e15", "1e308", "99999999999999999999", "not-a-number"])
|
||||
def test_extract_unusable_retry_after_ms_returns_none(header_name: str, header_value: str) -> None:
|
||||
"""Mirror the delay-level cases, including nan's existing safe fallback."""
|
||||
exc = FakeError("rate limited", status_code=429, headers={header_name: header_value})
|
||||
assert _extract_retry_after_ms(exc) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("header_name", "header_value", "expected_ms"),
|
||||
[
|
||||
("Retry-After", "60", 60000),
|
||||
("retry-after-ms", "60000", 60000),
|
||||
("Retry-After", "2.5", 2500),
|
||||
("retry-after-ms", "2.5", 2),
|
||||
("Retry-After", "-2", 0),
|
||||
("retry-after-ms", "-2", 0),
|
||||
("Retry-After", "0", 0),
|
||||
("retry-after-ms", "0", 0),
|
||||
("Retry-After", "86400", 86400000),
|
||||
("retry-after-ms", "86400000", 86400000),
|
||||
("Retry-After", "86400.001", None),
|
||||
("retry-after-ms", "86400001", None),
|
||||
],
|
||||
)
|
||||
def test_retry_after_numeric_boundaries(header_name: str, header_value: str, expected_ms: int | None) -> None:
|
||||
exc = FakeError("rate limited", status_code=429, headers={header_name: header_value})
|
||||
assert _extract_retry_after_ms(exc) == expected_ms
|
||||
middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
|
||||
delay = middleware._build_retry_delay_ms(100, exc)
|
||||
if expected_ms is None:
|
||||
assert 100 <= delay <= 300
|
||||
else:
|
||||
assert delay == expected_ms
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("delta_seconds", "expected_ms"), [(-60, 0), (60, 60000), (86400, 86400000), (86401, None)])
|
||||
def test_retry_after_date_boundaries(monkeypatch: pytest.MonkeyPatch, delta_seconds: int, expected_ms: int | None) -> None:
|
||||
now = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
monkeypatch.setattr(time, "time", lambda: now.timestamp())
|
||||
raw = format_datetime(now + timedelta(seconds=delta_seconds), usegmt=True)
|
||||
exc = FakeError("rate limited", status_code=429, headers={"Retry-After": raw})
|
||||
assert _extract_retry_after_ms(exc) == expected_ms
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("mode", ["sync", "async"])
|
||||
@pytest.mark.parametrize("header_name", ["Retry-After", "retry-after-ms"])
|
||||
@pytest.mark.parametrize("header_value", ["1e999", "nan", "1e15", "99999999999999999999", "Fri, 31 Dec 9999 23:59:59 GMT"])
|
||||
async def test_retry_loops_recover_from_unusable_hints(monkeypatch: pytest.MonkeyPatch, mode: str, header_name: str, header_value: str) -> None:
|
||||
middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=100, retry_cap_delay_ms=300)
|
||||
attempts = 0
|
||||
waits: list[float] = []
|
||||
|
||||
def fake_sleep(delay: float) -> None:
|
||||
assert 0.1 <= delay <= 0.3
|
||||
waits.append(delay)
|
||||
|
||||
async def fake_async_sleep(delay: float) -> None:
|
||||
fake_sleep(delay)
|
||||
|
||||
def handler(_request) -> AIMessage:
|
||||
nonlocal attempts
|
||||
attempts += 1
|
||||
if attempts == 1:
|
||||
raise FakeError("rate limited", status_code=429, headers={header_name: header_value})
|
||||
return AIMessage(content="recovered")
|
||||
|
||||
async def async_handler(request) -> AIMessage:
|
||||
return handler(request)
|
||||
|
||||
monkeypatch.setattr(time, "sleep", fake_sleep)
|
||||
monkeypatch.setattr(asyncio, "sleep", fake_async_sleep)
|
||||
if mode == "sync":
|
||||
result = middleware.wrap_model_call(SimpleNamespace(), handler)
|
||||
else:
|
||||
result = await middleware.awrap_model_call(SimpleNamespace(), async_handler)
|
||||
assert result.content == "recovered"
|
||||
assert attempts == 2
|
||||
assert len(waits) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_async_retry_loop_emits_jittered_delays_within_bounds(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user