mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(codex): honor Retry-After in the provider retry loop (#7151)
* fix(codex): honor Retry-After in the provider retry loop * style(codex): format retry-after regression helper * fix(codex): honor HTTP-date Retry-After hints --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
36ee986d55
commit
5e5913f4cf
@ -639,7 +639,7 @@ 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.
|
||||
> 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 and HTTP-date hints in LLM error handling and Codex's retry path, and integer-second hints in Claude's retry path. Codex prioritizes numeric `retry-after-ms` hints; past HTTP-date hints retry immediately.
|
||||
>
|
||||
> 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.
|
||||
>
|
||||
|
||||
@ -136,6 +136,15 @@ reported as text instead of raising `AttributeError`. SSE failures do not enter
|
||||
the HTTP-status retry loop.
|
||||
Offline HTTP-stream coverage: `tests/test_codex_stream_terminal_events.py`.
|
||||
|
||||
### Codex Retry Delays (`openai_codex_provider.py`)
|
||||
|
||||
Prefer numeric `retry-after-ms` over numeric or HTTP-date `Retry-After` hints.
|
||||
Interpret HTTP-date values in UTC, including the obsolete timezone-less asctime
|
||||
form. Apply `utils.retry_after.bounded_retry_after_ms` to every parsed delay:
|
||||
past dates retry immediately; malformed, non-finite or over-24-hour hints retain
|
||||
local exponential backoff. Keep millisecond hints numeric-only. Offline coverage
|
||||
in `tests/test_codex_retry_after.py` includes public sync/async HTTP retries.
|
||||
|
||||
### Claude Code Credentials (`packages/harness/deerflow/models/credential_loader.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)
|
||||
|
||||
@ -14,6 +14,8 @@ Supports:
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from datetime import UTC
|
||||
from email.utils import parsedate_to_datetime
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
@ -23,6 +25,7 @@ from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, System
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
|
||||
from deerflow.models.credential_loader import CodexCliCredential, load_codex_cli_credential
|
||||
from deerflow.utils.retry_after import bounded_retry_after_ms
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -34,6 +37,47 @@ def _is_valid_call_id(call_id: Any) -> bool:
|
||||
return isinstance(call_id, str) and bool(call_id.strip())
|
||||
|
||||
|
||||
def _retry_after_ms(error: Exception) -> int | None:
|
||||
"""Read a bounded retry hint from the error's response headers, if there is one."""
|
||||
response = getattr(error, "response", None)
|
||||
if response is None:
|
||||
return None
|
||||
headers = getattr(response, "headers", None)
|
||||
if not headers:
|
||||
return None
|
||||
|
||||
for header in ("retry-after-ms", "Retry-After-Ms", "retry-after", "Retry-After"):
|
||||
raw = headers.get(header)
|
||||
if not raw:
|
||||
continue
|
||||
try:
|
||||
milliseconds = float(raw) * 1000 if "ms" not in (header.lower()) else float(raw)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
if "ms" in header.lower():
|
||||
continue
|
||||
try:
|
||||
target = parsedate_to_datetime(str(raw))
|
||||
# The obsolete asctime HTTP-date form has no timezone; it is UTC.
|
||||
if target.tzinfo is None:
|
||||
target = target.replace(tzinfo=UTC)
|
||||
milliseconds = (target.timestamp() - time.time()) * 1000
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
continue
|
||||
bounded = bounded_retry_after_ms(milliseconds)
|
||||
if bounded is not None:
|
||||
return bounded
|
||||
return None
|
||||
|
||||
|
||||
def _calc_backoff_ms(attempt: int, error: Exception | None = None) -> int:
|
||||
"""Prefer the server's retry hint, falling back to the local exponential schedule."""
|
||||
if error is not None:
|
||||
hint_ms = _retry_after_ms(error)
|
||||
if hint_ms is not None:
|
||||
return hint_ms
|
||||
return 2000 * (1 << (attempt - 1))
|
||||
|
||||
|
||||
def _convert_tool_choice(tool_choice: str | dict | bool | None) -> str | dict | None:
|
||||
"""Normalize LangChain function selection to the Responses API format."""
|
||||
if tool_choice is None or tool_choice is False:
|
||||
@ -297,7 +341,7 @@ class CodexChatModel(BaseChatModel):
|
||||
if e.response.status_code in (429, 500, 529):
|
||||
if attempt >= self.retry_max_attempts:
|
||||
raise
|
||||
wait_ms = 2000 * (1 << (attempt - 1))
|
||||
wait_ms = _calc_backoff_ms(attempt, e)
|
||||
logger.warning(f"Codex API error {e.response.status_code}, retrying {attempt}/{self.retry_max_attempts} after {wait_ms}ms")
|
||||
time.sleep(wait_ms / 1000)
|
||||
else:
|
||||
|
||||
250
backend/tests/test_codex_retry_after.py
Normal file
250
backend/tests/test_codex_retry_after.py
Normal file
@ -0,0 +1,250 @@
|
||||
"""Retry pacing for deerflow.models.openai_codex_provider.CodexChatModel.
|
||||
|
||||
The provider's retry loop used to pace every 429/500/529 on a hard-coded exponential
|
||||
schedule, ignoring the server's hint. These tests pin the helper that now decides the
|
||||
wait, and the loop that consumes it, without any network access.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from email.utils import format_datetime
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from langchain_core.messages import HumanMessage
|
||||
|
||||
from deerflow.models.openai_codex_provider import _calc_backoff_ms
|
||||
from deerflow.utils.retry_after import MAX_RETRY_AFTER_MS
|
||||
|
||||
|
||||
def _status_error(status: int, headers: dict[str, str] | None = None) -> httpx.HTTPStatusError:
|
||||
request = httpx.Request("POST", "https://example.invalid/responses")
|
||||
return httpx.HTTPStatusError(
|
||||
"boom",
|
||||
request=request,
|
||||
response=httpx.Response(status, headers=headers or {}, request=request),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_without_a_response_it_keeps_the_local_schedule():
|
||||
error = RuntimeError("no response here")
|
||||
assert _calc_backoff_ms(1, error) == 2000
|
||||
assert _calc_backoff_ms(3, error) == 8000
|
||||
|
||||
|
||||
def test_without_an_error_it_keeps_the_local_schedule():
|
||||
assert _calc_backoff_ms(1) == 2000
|
||||
assert _calc_backoff_ms(2) == 4000
|
||||
|
||||
|
||||
def test_retry_after_seconds_is_honoured():
|
||||
error = _status_error(429, {"Retry-After": "45"})
|
||||
assert _calc_backoff_ms(1, error) == 45_000
|
||||
|
||||
|
||||
def test_retry_after_ms_is_preferred_over_seconds():
|
||||
error = _status_error(429, {"retry-after-ms": "1500", "Retry-After": "45"})
|
||||
assert _calc_backoff_ms(1, error) == 1500
|
||||
|
||||
|
||||
def test_header_lookup_is_case_insensitive():
|
||||
error = _status_error(429, {"RETRY-AFTER": "7"})
|
||||
assert _calc_backoff_ms(1, error) == 7000
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["not-a-number", "", "nan", "inf"])
|
||||
def test_unusable_hints_fall_back_to_the_local_schedule(value: str):
|
||||
error = _status_error(429, {"Retry-After": value})
|
||||
assert _calc_backoff_ms(2, error) == 4000
|
||||
|
||||
|
||||
def test_a_negative_hint_clamps_to_zero_like_the_claude_provider():
|
||||
# utils.retry_after treats a finite negative value as 0 rather than dropping the hint;
|
||||
# the Codex provider now shares that contract instead of inventing its own.
|
||||
error = _status_error(429, {"Retry-After": "-3"})
|
||||
assert _calc_backoff_ms(2, error) == 0
|
||||
|
||||
|
||||
def test_a_hint_beyond_the_bound_falls_back_to_the_local_schedule():
|
||||
# 25 hours, one past the 24h bound.
|
||||
error = _status_error(429, {"Retry-After": str(25 * 60 * 60)})
|
||||
assert _calc_backoff_ms(1, error) == 2000
|
||||
|
||||
|
||||
def test_the_bound_itself_is_accepted():
|
||||
seconds = MAX_RETRY_AFTER_MS // 1000
|
||||
error = _status_error(429, {"Retry-After": str(seconds)})
|
||||
assert _calc_backoff_ms(1, error) == MAX_RETRY_AFTER_MS
|
||||
|
||||
|
||||
def test_a_zero_hint_means_retry_immediately():
|
||||
error = _status_error(429, {"Retry-After": "0"})
|
||||
assert _calc_backoff_ms(1, error) == 0
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fixed_clock(monkeypatch):
|
||||
now = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
monkeypatch.setattr("deerflow.models.openai_codex_provider.time.time", lambda: now.timestamp())
|
||||
return now
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("delta_seconds", "expected_ms"),
|
||||
[(-60, 0), (0, 0), (600, 600_000), (86400, MAX_RETRY_AFTER_MS), (86401, 2000)],
|
||||
)
|
||||
def test_http_date_retry_after_obeys_delay_boundaries(fixed_clock, delta_seconds, expected_ms):
|
||||
raw = format_datetime(fixed_clock + timedelta(seconds=delta_seconds), usegmt=True)
|
||||
error = _status_error(429, {"Retry-After": raw})
|
||||
assert _calc_backoff_ms(1, error) == expected_ms
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw",
|
||||
[
|
||||
"Thu, 01 Jan 2026 00:10:00 GMT",
|
||||
"Thursday, 01-Jan-26 00:10:00 GMT",
|
||||
"Thu Jan 1 00:10:00 2026",
|
||||
],
|
||||
)
|
||||
def test_http_date_retry_after_accepts_http_date_formats_in_utc(fixed_clock, raw):
|
||||
error = _status_error(429, {"RETRY-AFTER": raw})
|
||||
assert _calc_backoff_ms(1, error) == 600_000
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw",
|
||||
["Thu, 99 Jan 2026 00:10:00 GMT", "Thu, 01 Jan 2026 99:10:00 GMT", "Fri, 31 Dec 9999 23:59:59 GMT"],
|
||||
)
|
||||
def test_invalid_or_unbounded_http_dates_keep_local_backoff(fixed_clock, raw):
|
||||
error = _status_error(429, {"Retry-After": raw})
|
||||
assert _calc_backoff_ms(2, error) == 4000
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("milliseconds", "expected_ms"), [("1500", 1500), ("invalid", 600_000)])
|
||||
def test_http_date_retry_after_preserves_millisecond_hint_precedence(fixed_clock, milliseconds, expected_ms):
|
||||
raw = format_datetime(fixed_clock + timedelta(minutes=10), usegmt=True)
|
||||
error = _status_error(429, {"retry-after-ms": milliseconds, "Retry-After": raw})
|
||||
assert _calc_backoff_ms(1, error) == expected_ms
|
||||
|
||||
|
||||
def test_http_dates_are_not_accepted_as_millisecond_hints(fixed_clock):
|
||||
raw = format_datetime(fixed_clock + timedelta(minutes=10), usegmt=True)
|
||||
error = _status_error(429, {"retry-after-ms": raw})
|
||||
assert _calc_backoff_ms(1, error) == 2000
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The loop, through the public model surface
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_model(**kwargs):
|
||||
from unittest.mock import patch
|
||||
|
||||
from deerflow.models.credential_loader import CodexCliCredential
|
||||
from deerflow.models.openai_codex_provider import CodexChatModel
|
||||
|
||||
cred = CodexCliCredential(access_token="tok-test", account_id="acc-test")
|
||||
with patch(
|
||||
"deerflow.models.openai_codex_provider.load_codex_cli_credential",
|
||||
return_value=cred,
|
||||
):
|
||||
return CodexChatModel(model="gpt-5.4", reasoning_effort="medium", **kwargs)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("mode", ["sync", "async"])
|
||||
@pytest.mark.parametrize("status", [429, 500, 529])
|
||||
async def test_public_invocation_waits_for_http_date_retry_after(monkeypatch, fixed_clock, mode, status):
|
||||
from deerflow.models import openai_codex_provider as provider
|
||||
|
||||
model = _make_model(retry_max_attempts=2)
|
||||
calls = []
|
||||
sleeps = []
|
||||
raw = format_datetime(fixed_clock + timedelta(minutes=10), usegmt=True)
|
||||
|
||||
def respond(request):
|
||||
calls.append(request)
|
||||
if len(calls) == 1:
|
||||
return httpx.Response(status, headers={"Retry-After": raw}, json={"error": "synthetic throttle"})
|
||||
event = {
|
||||
"type": "response.completed",
|
||||
"response": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "recovered"}]}]},
|
||||
}
|
||||
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), trust_env=False, **kwargs))
|
||||
monkeypatch.setattr(provider.time, "sleep", sleeps.append)
|
||||
|
||||
result = model.invoke("hi") if mode == "sync" else await model.ainvoke("hi")
|
||||
|
||||
assert result.content == "recovered"
|
||||
assert len(calls) == 2
|
||||
assert sleeps == [600.0]
|
||||
|
||||
|
||||
def test_loop_sleeps_for_the_server_hint(monkeypatch):
|
||||
model = _make_model(retry_max_attempts=2)
|
||||
sleeps: list[float] = []
|
||||
errors = [_status_error(429, {"Retry-After": "45"})]
|
||||
|
||||
def fake_stream(headers, payload):
|
||||
if errors:
|
||||
raise errors.pop()
|
||||
return {"ok": True}
|
||||
|
||||
monkeypatch.setattr(model, "_stream_response", fake_stream)
|
||||
monkeypatch.setattr(
|
||||
"deerflow.models.openai_codex_provider.time.sleep",
|
||||
lambda seconds: sleeps.append(seconds),
|
||||
)
|
||||
|
||||
assert model._call_codex_api([HumanMessage(content="hi")]) == {"ok": True}
|
||||
assert sleeps == [45.0]
|
||||
|
||||
|
||||
def test_loop_uses_local_backoff_without_a_hint(monkeypatch):
|
||||
model = _make_model(retry_max_attempts=2)
|
||||
sleeps: list[float] = []
|
||||
errors = [_status_error(429)]
|
||||
|
||||
def fake_stream(headers, payload):
|
||||
if errors:
|
||||
raise errors.pop()
|
||||
return {"ok": True}
|
||||
|
||||
monkeypatch.setattr(model, "_stream_response", fake_stream)
|
||||
monkeypatch.setattr(
|
||||
"deerflow.models.openai_codex_provider.time.sleep",
|
||||
lambda seconds: sleeps.append(seconds),
|
||||
)
|
||||
|
||||
assert model._call_codex_api([HumanMessage(content="hi")]) == {"ok": True}
|
||||
assert sleeps == [2.0]
|
||||
|
||||
|
||||
def test_loop_does_not_retry_a_400(monkeypatch):
|
||||
model = _make_model(retry_max_attempts=3)
|
||||
sleeps: list[float] = []
|
||||
|
||||
def fake_stream(headers, payload):
|
||||
raise _status_error(400)
|
||||
|
||||
monkeypatch.setattr(model, "_stream_response", fake_stream)
|
||||
monkeypatch.setattr(
|
||||
"deerflow.models.openai_codex_provider.time.sleep",
|
||||
lambda seconds: sleeps.append(seconds),
|
||||
)
|
||||
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
model._call_codex_api([HumanMessage(content="hi")])
|
||||
assert sleeps == []
|
||||
Loading…
x
Reference in New Issue
Block a user