diff --git a/README.md b/README.md index 1a75b292c3..db05092ad3 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index 08d710ddd6..8422f399b6 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -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). 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 401957b7e8..559c6406c9 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 @@ -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 diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index bc2e54b29c..c5b751fc06 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -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 diff --git a/backend/packages/harness/deerflow/models/claude_provider.py b/backend/packages/harness/deerflow/models/claude_provider.py index dffaa33b92..68239eb24f 100644 --- a/backend/packages/harness/deerflow/models/claude_provider.py +++ b/backend/packages/harness/deerflow/models/claude_provider.py @@ -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 diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index 0d67eda661..e2cc85c86f 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -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 diff --git a/backend/packages/harness/deerflow/utils/retry_after.py b/backend/packages/harness/deerflow/utils/retry_after.py new file mode 100644 index 0000000000..7a56796f77 --- /dev/null +++ b/backend/packages/harness/deerflow/utils/retry_after.py @@ -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)) diff --git a/backend/tests/test_claude_provider_retry_after.py b/backend/tests/test_claude_provider_retry_after.py new file mode 100644 index 0000000000..a497024470 --- /dev/null +++ b/backend/tests/test_claude_provider_retry_after.py @@ -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 diff --git a/backend/tests/test_llm_error_handling_middleware.py b/backend/tests/test_llm_error_handling_middleware.py index 5c0e3178f9..d54f27d442 100644 --- a/backend/tests/test_llm_error_handling_middleware.py +++ b/backend/tests/test_llm_error_handling_middleware.py @@ -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,