From ac5fd462819f042df6de7de8beb6bba63e527ead Mon Sep 17 00:00:00 2001 From: Willem Jiang Date: Tue, 21 Jul 2026 08:07:49 +0800 Subject: [PATCH] feat(backend): bound LLM call concurrency and shed burst-rate (429) retries (#4294) * Create a feature of Process-global LLM concurrency cap * Added configuration of llm_call of max_concurrent_calls * Classify limit_burst_rate and expose retry params via config.yaml * refactor(middleware): encapsulate LLM concurrency state in a dataclass Address PR #4294 review feedback (github-code-quality bot): the bare module-level globals _GLOBAL_CONCURRENCY_LOOP / _GLOBAL_CONCURRENCY_LIMIT were flagged as unused - a false positive, since both are read on the recreate condition, but the `global`-declaration pattern tripped the analyzer. Replace the three globals + `global` declaration with a single _ConcurrencyState dataclass singleton mutated in place. Behavior is unchanged (lazy recreate when the running loop or configured limit changes); the state is now co-located and no longer relies on bare globals. dataclasses is already an established harness convention. Co-Authored-By: Claude * fix(middleware): make LLM concurrency limiter process-wide + jitter burst retry Addresses PR #4294 review (fancyboi999, CHANGES_REQUESTED) - two P1 issues. P1 #1: the asyncio.Semaphore limiter was loop-bound, so it recreated per event loop and the cap was NOT process-wide: lead-agent calls (main loop) and subagent calls (the isolated persistent loop in subagents/executor.py) each got their own semaphore, and the sync graph path (wrap_model_call) bypassed the cap entirely. Recreating on loop/limit change also abandoned permits held by the prior instance. Replace it with a _ProcessWideLimiter built on threading primitives (not loop-bound): one limiter shared across every event loop and both sync/async wrappers. The cap is mutable via set_limit (never recreates, so in-flight permits are never abandoned); permits release in finally and async waiters unregister on cancellation, so cancellation never leaks capacity. Wire it into wrap_model_call (sync) too - previously a direct handler() call. P1 #2: the first (and only) burst-rate retry was deterministic at 5000ms. prev_delay_ms was seeded from the 1000ms normal base, so for burst the window collapsed to randint(5000, max(5000, 1000*3)) = randint(5000, 5000) - a fleet that failed together realigned on the same 5s tick. Seed the first retry from the reason-specific base (prev_delay_ms=None on loop init) so the burst window is [burst_base, cap] = [5000, 8000], non-degenerate. Retry-After is still honored verbatim. Tests: rename semaphore tests -> limiter; add an autouse fixture resetting the process singleton; add regressions the reviewer asked for - cross-loop (lead + isolated-loop subagent), two concurrent sync calls, limit-change while a permit is held (same instance, permit preserved), cancellation no-leak, and burst first-retry non-degeneracy with default config (real and seeded RNG) plus a concurrent de-synchronization case. Verified the burst guard goes red on the old logic ({5000}) and green on the new. Co-Authored-By: Claude * Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> * Potential fix for pull request finding 'Statement has no effect' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> * fix(middleware): lossless limiter handoff, generation-aware cap, burst-rate CB gate Address the open P1/P2 review findings on #4294: - P1 #1 (cancellation handoff): reserve the permit for a specific waiter at dequeue time (grant-at-dequeue, _AsyncWaiter.granted) so a waiter cancelled in the post-dequeue / pre-reacquire window hands its reservation to the next waiter (_handoff_granted_permit_locked) instead of stranding it. No cancellation window remains. - P1 #2 (hot-reload generation): move cap updates out of the per-attempt path; give the limiter one generation-aware owner (set_limit_if_newer with a monotonic instance seq proxy for config freshness) so a stale in-flight run cannot rewrite a freshly-lowered cap. max_concurrent_calls is now genuinely hot-reloadable, resolving the reload-boundary inconsistency by option (b) - no STARTUP_ONLY_FIELDS change (retry params truly hot-reload). - P2 (circuit breaker): gate _record_failure on reason != "burst_rate" so burst-rate (limit_burst_rate) exhaustion - a transient slope-throttle, not "provider down" - does not trip the CB and fast-fail ALL calls. - P3: clamp the jitter window to the cap before drawing (uniform spread instead of piling at the cap); document the per-process / GATEWAY_WORKERS cap semantics in config + the field description. Tests: add the reviewer-requested regressions (cancel-after-dequeue handoff; stale-instance-doesn't-overwrite-lowered-cap across sync + isolated-loop async; burst_rate-exhaustion-doesn't-trip-CB sync + async). Each is red on the prior buggy logic and green on the fix. _build_middleware now routes llm_call knobs through AppConfig so __init__ applies the cap. 71 middleware tests pass; 212 across the blast radius (1 pre-existing skip). Co-Authored-By: Claude * fix(middleware): startup-only LLM concurrency cap; report effective retry budget Addresses review feedback on #4294 (fancyboi999 CHANGES_REQUESTED on acfc7617): P1 - the generation guard measured construction order, not config freshness, so a stale AppConfig(cap=3) constructed after a fresher AppConfig(cap=1) could restore the higher cap; and on a downscale 3->1 release() handed excess permits to queued waiters, keeping in_flight pegged at the old cap. Replace the pseudo-generation path with a startup-only cap: the first middleware __init__ resolves and freezes the cap; later instances (newer or older config) are no-ops. No runtime cap mutation means no downscale race and no freshness/construction-order race. Per-call gate is now `limiter is None` only, so a reloaded instance with max_concurrent_calls=0 cannot silently drop the frozen cap. Removes _owner_seq / set_limit_if_newer / _grant_to_queued_locked / _next_instance_seq; file 946 -> 926 lines. P2 - burst-rate calls are capped at 2 attempts but the retry log line, the llm_retry stream event max_attempts, and the user-facing message still used self.retry_max_attempts (3), so the frontend showed 1/3 then stopped after attempt 2. Thread the effective max_attempts (_max_attempts_for) through the logger, _emit_retry_event, and _build_retry_message. Also: document max_concurrent_calls as startup-only in the config field description and config.example.yaml (prose only - the startup-only: prefix is top-level AppConfig-field granularity and would mislabel the otherwise hot-reloadable llm_call section / break the reload_boundary drift test). Rewrite the cap-mutation tests for startup-only semantics; add P2 retry-budget event tests (sync+async, teeth-verified red on the bug); fix bot nits (empty except blocks -> gather(return_exceptions=True); bare await statements -> assigned+asserted). Co-Authored-By: Claude --------- Co-authored-by: Claude Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> --- .../llm_error_handling_middleware.py | 496 ++++++- .../harness/deerflow/config/app_config.py | 58 + .../test_llm_error_handling_middleware.py | 1202 ++++++++++++++++- config.example.yaml | 43 + 4 files changed, 1765 insertions(+), 34 deletions(-) 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 71c00f4aa..e04ee4d5f 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 @@ -4,8 +4,10 @@ from __future__ import annotations import asyncio import logging +import random import threading import time +from collections import deque from collections.abc import Awaitable, Callable from email.utils import parsedate_to_datetime from typing import Any, override @@ -62,6 +64,18 @@ _AUTH_PATTERNS = ( "未授权", ) +# Provider burst-rate (``limit_burst_rate``) signals. This is a *rate-of-change* +# limit, not a quota limit: the provider throttles when request RPM ramps up too +# steeply (e.g. the 08:30 morning peak going 0 -> full throttle in seconds). +# Matched against both the error message and the error ``code``/``type``. +_BURST_PATTERNS = ( + "limit_burst_rate", + "rate increased too quickly", + "burst rate", + "请求速率增长过快", + "突发速率", +) + # Per-exception retry budget overrides. # # Some transient errors are retriable in principle but expensive to retry at @@ -82,6 +96,18 @@ _RETRY_BUDGET_OVERRIDES: dict[str, int] = { "StreamChunkTimeoutError": 2, } +# Per-reason retry budget overrides, applied in addition to the per-exception +# overrides above; the tightest bound wins (so neither loosens the other) and +# the user-configured ``retry_max_attempts`` still caps everything. +# +# A burst-rate (``limit_burst_rate``) 429 gets a tight budget on purpose: +# retrying into the burst adds demand to the very request-rate slope being +# throttled, so we keep at most one retry (with a longer backoff) and then shed +# load rather than hammering the provider. Keys are ``_classify_error`` reasons. +_REASON_RETRY_BUDGETS: dict[str, int] = { + "burst_rate": 2, +} + # Exception class names that indicate the upstream stream-chunk watchdog # fired because the model stalled mid-flight. These deserve a more specific # user-facing message than the generic "temporarily unavailable" copy, @@ -98,12 +124,261 @@ _STREAM_DROP_EXCEPTIONS: frozenset[str] = frozenset( ) +# Process-global LLM call concurrency cap. ONE limiter is shared across every +# ``LLMErrorHandlingMiddleware`` instance and every call path: the lead agent +# (main event loop), subagents (the isolated persistent loop in +# subagents/executor.py), ``asyncio.run`` tests, and the sync graph path. That +# matters because a provider burst-rate (``limit_burst_rate``) limit fires on +# the *slope* of the request rate, so the cap must bound aggregate in-flight +# calls process-wide - a per-loop cap (which is what asyncio.Semaphore would +# give) is defeated the moment subagent fan-out runs on a second loop. +# +# Correctness invariants the design below preserves: +# * Lossless waiter handoff: a permit handed to a waiter is *reserved* for +# that waiter at dequeue time (``granted=True``). If the waiter is +# cancelled before it wakes, the reserved permit is re-handed to the next +# waiter (or freed) - so a cancellation in the post-dequeue/pre-reacquire +# window never strands the next waiter with capacity idle. +# * Startup-only cap: the cap is resolved ONCE, at the first middleware +# construction (``_apply_configured_cap``), and frozen thereafter. Later +# ``__init__`` calls never touch the cap - whether they hold a newer or an +# older ``AppConfig`` snapshot. This removes the pseudo-generation path +# entirely: with no cap mutation at runtime there is no downscale that +# could hand excess permits to queued waiters (keeping ``in_flight`` pegged +# at the old cap), and no construction-order race where a stale config +# constructed after a fresher one could restore a higher cap. Per-attempt +# callers only acquire/release. Changing the cap requires a gateway +# restart (see ``LlmCallConfig.max_concurrent_calls``). + + +class _AsyncWaiter: + """A parked async caller awaiting a transferred permit. + + ``granted`` is flipped to ``True`` (under the limiter lock) at the exact + moment a permit is reserved for this waiter - by ``release`` handing off a + returning permit, or by another cancelling waiter handing off its reserved + permit. The reservation is atomic with the dequeue, so the invariant + ``granted is True <=> not in _async_waiters`` always holds: once granted, + the permit is already counted in ``_in_flight`` and the waiter need only + wake and return. A cancelled waiter therefore knows from ``granted`` + whether it owes a handoff (granted) or is merely unregistering (not yet + granted). + """ + + __slots__ = ("loop", "event", "granted") + + def __init__(self, loop: asyncio.AbstractEventLoop, event: asyncio.Event) -> None: + self.loop = loop + self.event = event + self.granted = False + + +class _ProcessWideLimiter: + """In-flight call limiter shared across event loops and sync/async wrappers. + + ``asyncio.Semaphore`` binds to the first event loop that uses it and raises + if acquired from another, so it cannot cap lead-agent and subagent calls + together (they run on different loops), nor the sync graph path. This + limiter is built on ``threading`` primitives (not loop-bound): every call + path shares one in-flight counter and one cap. + + The cap is **immutable**: it is set once at construction (by + ``_apply_configured_cap`` on the first middleware ``__init__``) and never + mutated afterwards. Because the cap never changes at runtime there is no + downscale race (a lowered cap could otherwise keep admitting queued + waiters until ``in_flight`` drains) and no config-freshness race (a stale + snapshot constructed later could otherwise restore a higher cap). Per- + attempt callers (``acquire_sync``/``acquire_async``/``release``) never + touch the cap. Permits are released in a ``finally`` and an async waiter + that is cancelled after its permit was reserved hands the reservation to + the next waiter, so capacity never leaks and a cancellation never strands + a later waiter. + """ + + def __init__(self, limit: int) -> None: + self._lock = threading.Lock() + self._cond = threading.Condition(self._lock) + self._in_flight = 0 + self._limit = max(0, limit) + # FIFO of async callers waiting on capacity. Each waiter lives on its + # caller's loop; release/handoff wakes one across loops via + # call_soon_threadsafe so the wakeup runs on the right loop. + self._async_waiters: deque[_AsyncWaiter] = deque() + + @property + def limit(self) -> int: + return self._limit + + @property + def in_flight(self) -> int: + return self._in_flight + + def acquire_sync(self) -> None: + """Block the calling thread until a permit is available, then take one.""" + with self._cond: + while not self._try_acquire_locked(): + self._cond.wait() + + def release(self) -> None: + """Return one permit, handing it to a waiter if one is queued. + + If an async waiter is queued, the returning permit *transfers* to it + (ownership moves; ``_in_flight`` is unchanged) and its event is set so + it wakes already owning a permit. Otherwise the permit returns to the + free pool (``_in_flight -= 1``) and one sync waiter is notified to grab + it on its next ``_try_acquire_locked`` re-check. + """ + with self._cond: + if self._async_waiters: + waiter = self._async_waiters.popleft() + waiter.granted = True + if not self._wake_locked(waiter): + # Owner loop closed: the transferred permit is stranded; + # hand it to the next waiter or free it. + self._handoff_granted_permit_locked() + return + if self._in_flight > 0: + self._in_flight -= 1 + self._cond.notify() + + async def acquire_async(self) -> None: + """Acquire a permit without blocking the event loop. + + Free capacity -> take one immediately. Otherwise park on an + ``asyncio.Event``; ``release`` / a cap-raise transfers a permit to us + (``granted=True``) and sets the event. On cancellation, if a permit was + already reserved for us, hand it to the next waiter (or free it) so the + reservation is never lost; if we were still queued (not yet granted), + just unregister - no permit was reserved for us, so there is nothing to + release. + """ + loop = asyncio.get_running_loop() + while True: + waiter = _AsyncWaiter(loop=loop, event=asyncio.Event()) + with self._cond: + if self._try_acquire_locked(): + return + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "LLM call parking on process-wide limiter (in_flight=%d, limit=%d, queued=%d)", + self._in_flight, + self._limit, + len(self._async_waiters) + 1, + ) + self._async_waiters.append(waiter) + try: + await waiter.event.wait() + except asyncio.CancelledError: + with self._cond: + if waiter.granted: + # A permit was reserved for us but we're cancelling + # before waking. Pass the reservation to the next + # waiter (or free it) so it is not stranded. + self._handoff_granted_permit_locked() + else: + # Still queued, never granted (granted is set only when + # dequeued, under the lock): just unregister. + self._async_waiters.remove(waiter) + raise + return # woken => granted => we own a permit (already in _in_flight) + + def _try_acquire_locked(self) -> bool: + if self._in_flight < self._limit: + self._in_flight += 1 + return True + return False + + def _handoff_granted_permit_locked(self) -> None: + """Transfer an already-reserved permit to the next queued waiter, or free it. + + Used when a waiter that had a permit reserved cancels before waking, or + when a reservation target's loop is dead. The permit is already counted + in ``_in_flight``; transferring keeps it counted (ownership moves to the + next waiter), freeing returns it to the pool. Either way ``_in_flight`` + stays correct and the reservation is never lost. + """ + while self._async_waiters: + waiter = self._async_waiters.popleft() + waiter.granted = True + if self._wake_locked(waiter): + return # ownership transferred; _in_flight unchanged + # dead loop; try the next waiter + # No async waiter to take it: free the permit and wake a sync waiter. + if self._in_flight > 0: + self._in_flight -= 1 + self._cond.notify() + + def _wake_locked(self, waiter: _AsyncWaiter) -> bool: + """Schedule ``event.set`` on the waiter's loop. False if the loop is dead.""" + try: + waiter.loop.call_soon_threadsafe(waiter.event.set) + return True + except RuntimeError: + return False # owner loop closed: the wakeup cannot land + + +_LIMITER_LOCK = threading.Lock() +_PROCESS_LIMITER: _ProcessWideLimiter | None = None + +# Whether the process-wide cap has been resolved yet. The cap is startup-only: +# the first ``LLMErrorHandlingMiddleware`` ``__init__`` resolves it (creating a +# limiter for a positive cap, or leaving it ``None`` for a disabled cap) and +# every subsequent ``__init__`` is a no-op - regardless of whether its +# ``AppConfig`` snapshot is newer or older than the first. This is the single +# owner of the cap; per-attempt callers only acquire/release. +_CAP_RESOLVED: bool = False + + +def _get_process_limiter() -> _ProcessWideLimiter | None: + """Return the process-wide LLM-call limiter, or ``None`` when the cap is + disabled (or before the first middleware construction resolves it). + + Per-attempt callers use this to acquire/release only - it never changes the + cap. ``limiter is None`` is the sole gate for "cap disabled": a per-call + short-circuit on the instance's configured value would let a later + (reloaded) instance with ``max_concurrent_calls=0`` silently drop the cap + mid-process, which is exactly the hot-reload churn the startup-only design + removes. + """ + return _PROCESS_LIMITER + + +def _apply_configured_cap(limit: int) -> None: + """Resolve the process-wide cap from the first middleware ``__init__``. + + Startup-only: the very first call wins and freezes the cap. A positive + ``limit`` creates the limiter at that cap; ``limit <= 0`` resolves the cap + as disabled (limiter stays ``None``, callers short-circuit on + ``limiter is None``). Every later call - whether it carries a newer or an + older ``AppConfig`` snapshot, and whether it would raise or lower the cap - + is ignored, so the cap can never be mutated at runtime. Changing it requires + a gateway restart. + """ + global _PROCESS_LIMITER, _CAP_RESOLVED + if _CAP_RESOLVED: + return # cap already frozen at first construction; this instance is a no-op + with _LIMITER_LOCK: + if _CAP_RESOLVED: + return + _CAP_RESOLVED = True + if limit > 0: + _PROCESS_LIMITER = _ProcessWideLimiter(limit) + + class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): """Retry transient LLM errors and surface graceful assistant messages.""" retry_max_attempts: int = 3 retry_base_delay_ms: int = 1000 retry_cap_delay_ms: int = 8000 + # Longer backoff base used only for burst-rate (limit_burst_rate) 429s, so + # the single burst retry lands after the throttle window subsides. + burst_retry_base_delay_ms: int = 5000 + # Process-wide cap on concurrently in-flight LLM calls. 0 disables the cap + # (default) so existing deployments see no behavior change; set to a + # positive int to bound aggregate concurrency and smooth provider + # burst-rate (limit_burst_rate) spikes. See _get_process_limiter. + max_concurrent_llm_calls: int = 0 def __init__(self, *, app_config: AppConfig, **kwargs: Any) -> None: super().__init__(**kwargs) @@ -111,6 +386,23 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): self.circuit_failure_threshold = app_config.circuit_breaker.failure_threshold self.circuit_recovery_timeout_sec = app_config.circuit_breaker.recovery_timeout_sec + # Retry / backoff / concurrency knobs are all configured via the + # ``llm_call`` section of config.yaml; they override the class defaults + # above so operators can tune them without code changes. + llm_call = app_config.llm_call + self.retry_max_attempts = llm_call.retry_max_attempts + self.retry_base_delay_ms = llm_call.retry_base_delay_ms + self.retry_cap_delay_ms = llm_call.retry_cap_delay_ms + self.burst_retry_base_delay_ms = llm_call.burst_retry_base_delay_ms + self.max_concurrent_llm_calls = llm_call.max_concurrent_calls + + # Resolve the process-wide cap (startup-only: the first ``__init__`` in + # the process wins and freezes it; later instances - newer or older + # config - are no-ops). Per-attempt callers only acquire/release, so the + # cap can never be mutated at runtime and there is no downscale or + # config-freshness race to admit waiters above the live cap. + _apply_configured_cap(self.max_concurrent_llm_calls) + # Circuit Breaker state self._circuit_lock = threading.Lock() self._circuit_failure_count = 0 @@ -118,17 +410,24 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): self._circuit_state = "closed" self._circuit_probe_in_flight = False - def _max_attempts_for(self, exc: BaseException) -> int: + def _max_attempts_for(self, exc: BaseException, reason: str = "transient") -> int: """Return the effective max attempt count for this exception. - Falls back to `self.retry_max_attempts` unless the exception class name - appears in the per-exception override table. + The user-configured ``retry_max_attempts`` is the ceiling; per-exception + (``_RETRY_BUDGET_OVERRIDES``, keyed by class name) and per-reason + (``_REASON_RETRY_BUDGETS``, keyed by ``_classify_error`` reason) + overrides can only *tighten* it. The tightest bound wins, so a burst-rate + 429 never gets more attempts than its dedicated budget even if the + operator raised the global cap. """ - override = _RETRY_BUDGET_OVERRIDES.get(type(exc).__name__) - if override is None: - return self.retry_max_attempts - - return min(override, self.retry_max_attempts) + candidates = [self.retry_max_attempts] + class_override = _RETRY_BUDGET_OVERRIDES.get(type(exc).__name__) + if class_override is not None: + candidates.append(class_override) + reason_override = _REASON_RETRY_BUDGETS.get(reason) + if reason_override is not None: + candidates.append(reason_override) + return min(candidates) def _check_circuit(self) -> bool: """Returns True if circuit is OPEN (fast fail), False otherwise.""" @@ -203,6 +502,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): return False, "quota" if _matches_any(lowered, _AUTH_PATTERNS): return False, "auth" + # Burst-rate (limit_burst_rate) 429 is retriable but needs its own + # policy: a tight retry budget and a longer backoff base (see + # _REASON_RETRY_BUDGETS / _build_retry_delay_ms). Detected before the + # generic 429->transient mapping so it isn't lumped in with ordinary + # transient errors. + if _matches_any(lowered, _BURST_PATTERNS) or _matches_any(str(error_code).lower(), _BURST_PATTERNS): + return True, "burst_rate" exc_name = exc.__class__.__name__ if exc_name in { @@ -232,17 +538,121 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): return False, "generic" - def _build_retry_delay_ms(self, attempt: int, exc: BaseException) -> int: + def _bounded_model_call_sync( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], ModelResponse], + ) -> ModelResponse: + """Run one sync model attempt under the process-global concurrency cap. + + The limiter wraps a *single* attempt only (not the retry loop), so + backoff sleeps release the slot for other callers. ``limiter is None`` + (cap disabled at startup) is a direct passthrough; a non-``None`` + limiter is always consulted - the cap is frozen at the first + ``__init__``, so a later instance whose ``max_concurrent_llm_calls`` is + 0 cannot silently drop it. Permits release on any exit (return or + raise) via ``finally`` so a raised handler never leaks a slot. + """ + limiter = _get_process_limiter() + if limiter is None: + return handler(request) + limiter.acquire_sync() + try: + return handler(request) + finally: + limiter.release() + + async def _bounded_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], Awaitable[ModelResponse]], + ) -> ModelResponse: + """Run one async model attempt under the process-global concurrency cap. + + The limiter wraps a *single* attempt only (not the retry loop), so + backoff sleeps release the slot for other callers - we bound in-flight + requests, not waiting ones. ``limiter is None`` (cap disabled at + startup) is a direct passthrough; a non-``None`` limiter is always + consulted (cap frozen at first ``__init__``). Permits release on any + exit (return, raise, or cancellation) via ``finally``; + ``acquire_async`` separately cleans up if cancelled while waiting, so + capacity never leaks. + """ + limiter = _get_process_limiter() + if limiter is None: + return await handler(request) + await limiter.acquire_async() + try: + return await handler(request) + finally: + limiter.release() + + 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 + 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: + ``delay = random(base, min(cap, max(base, seed * 3)))`` where ``seed`` + is the previous delay, or the reason-specific base on the first retry + (``prev_delay_ms is None``). The window is clamped to the cap *before* + drawing (not after) so the distribution stays uniform up to the cap + rather than piling up at it. ``reason="burst_rate"`` swaps in + ``burst_retry_base_delay_ms`` (longer than the normal base) so the + single burst retry lands after the throttle window subsides. + + Seeding the first retry from the *reason-specific* base (not always the + normal base) is what keeps the first-and-only burst retry + non-degenerate: with the normal base (1000ms) the burst window would + collapse to ``randint(5000, max(5000, 1000*3)) = randint(5000, 5000)`` + and every concurrent burst failure would realign on the same 5s tick. + Seeding from 5000ms gives ``randint(5000, min(8000, 15000)) = + randint(5000, 8000)`` with defaults, so a fleet that failed together + spreads out across the whole window. + + Deterministic exponential backoff (``base * 2^(attempt-1)``) makes + every concurrent retryer realign on the same backoff ticks; when a + whole fleet fails at once (e.g. a provider burst-rate limit at the + morning peak) that synchronized retry storm re-triggers the very limit + we are backing off from. Decorrelated jitter spreads those retries + across a random window so they don't re-peak in lockstep. + """ retry_after = _extract_retry_after_ms(exc) if retry_after is not None: return retry_after - backoff = self.retry_base_delay_ms * (2 ** max(0, attempt - 1)) - return min(backoff, self.retry_cap_delay_ms) + base = self.burst_retry_base_delay_ms if reason == "burst_rate" else self.retry_base_delay_ms + cap = self.retry_cap_delay_ms + seed = base if prev_delay_ms is None else prev_delay_ms + # Clamp the window to the cap *before* drawing so the jitter spreads + # uniformly across [base, min(cap, seed*3)] instead of concentrating at + # the cap: with defaults seed*3 (=15000) >> cap (=8000), drawing + # randint(base, seed*3) then min(delay, cap) would put ~70% of draws at + # exactly cap, re-clustering a fleet that the jitter is meant to spread. + high = min(cap, max(base, seed * 3)) + if high < base: + return cap # base exceeds cap (misconfiguration): the cap wins + return random.randint(base, high) - def _build_retry_message(self, attempt: int, wait_ms: int, reason: str) -> str: + def _build_retry_message( + self, + attempt: int, + wait_ms: int, + reason: str, + *, + max_attempts: int, + ) -> str: seconds = max(1, round(wait_ms / 1000)) - reason_text = "provider is busy" if reason == "busy" else "provider request failed temporarily" - return f"LLM request retry {attempt}/{self.retry_max_attempts}: {reason_text}. Retrying in {seconds}s." + reason_text = { + "busy": "provider is busy", + "burst_rate": "provider is throttling request burst rate", + }.get(reason, "provider request failed temporarily") + # ``max_attempts`` is the *effective* budget for this call (from + # ``_max_attempts_for``), not the configured ceiling: a burst-rate call + # is capped at 2 attempts, so its message must read ``1/2`` not ``1/3`` + # even when ``retry_max_attempts`` is the default 3 - otherwise the UI + # promises a retry that will never happen. + return f"LLM request retry {attempt}/{max_attempts}: {reason_text}. Retrying in {seconds}s." def _build_circuit_breaker_message(self) -> str: return "The configured LLM provider is currently unavailable due to continuous failures. Circuit breaker is engaged to protect the system. Please wait a moment before trying again." @@ -271,6 +681,8 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): return "The configured LLM provider rejected the request because the account is out of quota, billing is unavailable, or usage is restricted. Please fix the provider account and try again." if reason == "auth": return "The configured LLM provider rejected the request because authentication or access is invalid. Please check the provider credentials and try again." + if reason == "burst_rate": + return "The configured LLM provider is temporarily throttling requests because the request rate increased too quickly (burst-rate limit). Please wait a moment and try again." if reason in {"busy", "transient"}: # Stream-drop failures (chunk-gap timeout, peer-closed connection, # raw read error) almost always point at a single oversized @@ -297,7 +709,14 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): detail=_extract_error_detail(exc), ) - def _emit_retry_event(self, attempt: int, wait_ms: int, reason: str) -> None: + def _emit_retry_event( + self, + attempt: int, + wait_ms: int, + reason: str, + *, + max_attempts: int, + ) -> None: try: from langgraph.config import get_stream_writer @@ -306,10 +725,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): { "type": "llm_retry", "attempt": attempt, - "max_attempts": self.retry_max_attempts, + # Effective budget for this call (burst-rate == 2), not the + # configured ceiling - the frontend renders this and the + # ``message`` below, so both must describe the loop that runs. + "max_attempts": max_attempts, "wait_ms": wait_ms, "reason": reason, - "message": self._build_retry_message(attempt, wait_ms, reason), + "message": self._build_retry_message(attempt, wait_ms, reason, max_attempts=max_attempts), } ) except Exception: @@ -330,9 +752,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): ) attempt = 1 + prev_delay_ms: int | None = None while True: try: - response = handler(request) + response = self._bounded_model_call_sync(request, handler) self._record_success() return response except GraphBubbleUp: @@ -341,17 +764,18 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): raise except Exception as exc: retriable, reason = self._classify_error(exc) - max_attempts = self._max_attempts_for(exc) + max_attempts = self._max_attempts_for(exc, reason) if retriable and attempt < max_attempts: - wait_ms = self._build_retry_delay_ms(attempt, exc) + wait_ms = self._build_retry_delay_ms(prev_delay_ms, exc, reason) + prev_delay_ms = wait_ms logger.warning( "Transient LLM error on attempt %d/%d; retrying in %dms: %s", attempt, - self.retry_max_attempts, + max_attempts, wait_ms, _extract_error_detail(exc), ) - self._emit_retry_event(attempt, wait_ms, reason) + self._emit_retry_event(attempt, wait_ms, reason, max_attempts=max_attempts) time.sleep(wait_ms / 1000) attempt += 1 continue @@ -361,10 +785,14 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): _extract_error_detail(exc), exc_info=exc, ) - if retriable: + if retriable and reason != "burst_rate": self._record_failure() else: - # Non-retriable: release the probe without recording a failure. + # Non-retriable, OR burst_rate (a transient provider + # slope-throttle, not "provider down"): release the half-open + # probe without recording a failure so the circuit doesn't + # trip and fast-fail ALL calls for the recovery window - the + # exact self-inflicted outage #4290 is trying to prevent. self._release_half_open_probe() return self._build_user_fallback_message(exc, reason) @@ -383,9 +811,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): ) attempt = 1 + prev_delay_ms: int | None = None while True: try: - response = await handler(request) + response = await self._bounded_model_call(request, handler) self._record_success() return response except GraphBubbleUp: @@ -394,17 +823,18 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): raise except Exception as exc: retriable, reason = self._classify_error(exc) - max_attempts = self._max_attempts_for(exc) + max_attempts = self._max_attempts_for(exc, reason) if retriable and attempt < max_attempts: - wait_ms = self._build_retry_delay_ms(attempt, exc) + wait_ms = self._build_retry_delay_ms(prev_delay_ms, exc, reason) + prev_delay_ms = wait_ms logger.warning( "Transient LLM error on attempt %d/%d; retrying in %dms: %s", attempt, - self.retry_max_attempts, + max_attempts, wait_ms, _extract_error_detail(exc), ) - self._emit_retry_event(attempt, wait_ms, reason) + self._emit_retry_event(attempt, wait_ms, reason, max_attempts=max_attempts) await asyncio.sleep(wait_ms / 1000) attempt += 1 continue @@ -414,10 +844,14 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): _extract_error_detail(exc), exc_info=exc, ) - if retriable: + if retriable and reason != "burst_rate": self._record_failure() else: - # Non-retriable: release the probe without recording a failure. + # Non-retriable, OR burst_rate (a transient provider + # slope-throttle, not "provider down"): release the half-open + # probe without recording a failure so the circuit doesn't + # trip and fast-fail ALL calls for the recovery window - the + # exact self-inflicted outage #4290 is trying to prevent. self._release_half_open_probe() return self._build_user_fallback_message(exc, reason) diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index ae9c06f7c..ef1758fe9 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -65,6 +65,63 @@ class CircuitBreakerConfig(BaseModel): recovery_timeout_sec: int = Field(default=60, description="Time in seconds before attempting to recover the circuit") +class LlmCallConfig(BaseModel): + """Configuration for LLM call execution (concurrency / rate shaping). + + Distinct from :class:`CircuitBreakerConfig` (which handles a *failing* + provider) and from :class:`ModelConfig` (which describes model endpoints): + these knobs shape how many LLM calls run at once and how the retry/backoff + loop behaves. Capping concurrency caps the *slope* of the request rate, + which is what a provider burst-rate (``limit_burst_rate``) limit fires on. + """ + + max_concurrent_calls: int = Field( + default=0, + ge=0, + description=( + "Process-wide cap on concurrently in-flight LLM calls. 0 disables " + "the cap (default, preserving existing behavior). Set to a positive " + "int to smooth provider burst-rate (limit_burst_rate) spikes by " + "bounding the request-rate slope at the morning peak. Per-process, " + "not per-cluster: with GATEWAY_WORKERS > 1 the aggregate cap is " + "effectively max_concurrent_calls * GATEWAY_WORKERS (and a " + "multi-node rollout multiplies it further), so size the per-process " + "value accordingly and pair it with an nginx limit_req at the ingress " + "for a true cluster-wide slope cap. Startup-only: the cap is captured " + "at the first LLM run and frozen for the process lifetime, so editing " + "it in config.yaml takes effect only after a gateway restart (the " + "other llm_call.* knobs remain hot-reloadable). Freezing avoids the " + "downscale/config-freshness races a runtime-mutable cap would " + "introduce on a process-wide, cross-loop limiter." + ), + ) + retry_max_attempts: int = Field( + default=3, + ge=1, + description="Max LLM call attempts (1 = no retry) for retriable transient errors.", + ) + retry_base_delay_ms: int = Field( + default=1000, + ge=0, + description="Base (ms) for the decorrelated-jitter retry backoff; seeds the first retry delay.", + ) + retry_cap_delay_ms: int = Field( + default=8000, + ge=0, + description="Hard cap (ms) on any single retry backoff delay.", + ) + burst_retry_base_delay_ms: int = Field( + default=5000, + ge=0, + description=( + "Base (ms) for the backoff when the provider returns a burst-rate " + "(limit_burst_rate) 429. Higher than retry_base_delay_ms so the " + "single burst retry lands after the throttle window subsides. " + "Ignored when the provider sends Retry-After (honored verbatim)." + ), + ) + + class LoggingEnhanceConfig(BaseModel): """Request trace logging enhancement settings.""" @@ -175,6 +232,7 @@ class AppConfig(BaseModel): input_polish: InputPolishConfig = Field(default_factory=InputPolishConfig, description="Pre-send input polishing configuration.") suggestions: SuggestionsConfig = Field(default_factory=SuggestionsConfig, description="Follow-up suggestions configuration.") circuit_breaker: CircuitBreakerConfig = Field(default_factory=CircuitBreakerConfig, description="LLM circuit breaker configuration") + llm_call: LlmCallConfig = Field(default_factory=LlmCallConfig, description="LLM call execution configuration (concurrency / rate shaping)") channel_connections: ChannelConnectionsConfig = Field( default_factory=ChannelConnectionsConfig, description=format_field_description( diff --git a/backend/tests/test_llm_error_handling_middleware.py b/backend/tests/test_llm_error_handling_middleware.py index 016c53a96..08c1acf51 100644 --- a/backend/tests/test_llm_error_handling_middleware.py +++ b/backend/tests/test_llm_error_handling_middleware.py @@ -1,6 +1,10 @@ from __future__ import annotations import asyncio +import concurrent.futures +import threading +import time +from collections.abc import Iterator from types import SimpleNamespace from typing import Any @@ -11,10 +15,30 @@ from langgraph.errors import GraphBubbleUp from deerflow.agents.middlewares.llm_error_handling_middleware import ( LLMErrorHandlingMiddleware, ) -from deerflow.config.app_config import AppConfig +from deerflow.config.app_config import AppConfig, LlmCallConfig from deerflow.config.sandbox_config import SandboxConfig +@pytest.fixture(autouse=True) +def _reset_process_limiter() -> Iterator[None]: + """Reset the module-global limiter + cap-resolved flag between tests. + + The limiter is a process singleton shared across all middleware instances, + so cap / in-flight state from one test would otherwise bleed into the next. + The cap is startup-only (frozen at first ``__init__``), so ``_CAP_RESOLVED`` + must be reset too - otherwise the first test to construct a middleware + would freeze the cap for every later test regardless of what cap they ask + for. + """ + from deerflow.agents.middlewares import llm_error_handling_middleware as mod + + mod._PROCESS_LIMITER = None + mod._CAP_RESOLVED = False + yield + mod._PROCESS_LIMITER = None + mod._CAP_RESOLVED = False + + def _make_app_config() -> AppConfig: """Minimal AppConfig for middleware tests; circuit_breaker uses defaults.""" return AppConfig(sandbox=SandboxConfig(use="test")) @@ -37,10 +61,30 @@ class FakeError(Exception): self.response = SimpleNamespace(status_code=status_code, headers=headers or {}) if status_code is not None or headers else None +# Middleware-level attribute -> ``LlmCallConfig`` field. ``llm_call`` knobs are +# routed through ``AppConfig`` so ``__init__`` resolves the cap on the process +# limiter (startup-only: the first construction freezes it). Circuit-breaker +# knobs are read per-call from ``self`` (not via the limiter), so setattr after +# ``__init__`` still works for them. +_LLM_CALL_ATTR_MAP: dict[str, str] = { + "max_concurrent_llm_calls": "max_concurrent_calls", + "retry_max_attempts": "retry_max_attempts", + "retry_base_delay_ms": "retry_base_delay_ms", + "retry_cap_delay_ms": "retry_cap_delay_ms", + "burst_retry_base_delay_ms": "burst_retry_base_delay_ms", +} + + def _build_middleware(**attrs: int) -> LLMErrorHandlingMiddleware: - middleware = LLMErrorHandlingMiddleware(app_config=_make_app_config()) + llm_call_fields = {_LLM_CALL_ATTR_MAP[key]: value for key, value in attrs.items() if key in _LLM_CALL_ATTR_MAP} + app_config = AppConfig( + sandbox=SandboxConfig(use="test"), + llm_call=LlmCallConfig(**llm_call_fields), + ) + middleware = LLMErrorHandlingMiddleware(app_config=app_config) for key, value in attrs.items(): - setattr(middleware, key, value) + if key not in _LLM_CALL_ATTR_MAP: + setattr(middleware, key, value) return middleware @@ -819,3 +863,1155 @@ def test_async_index_error_exhausted_returns_user_fallback( assert result.additional_kwargs["error_reason"] == "transient" assert result.additional_kwargs["error_type"] == "IndexError" assert "temporarily unavailable" in str(result.content) + + +# ---------- Process-wide concurrency limiter ---------- + + +async def _run_concurrent( + middleware: LLMErrorHandlingMiddleware, + count: int, + event: asyncio.Event, +) -> tuple[int, int]: + """Fire ``count`` concurrent awrap_model_call tasks whose handlers park on + ``event`` until the test releases them. + + Returns ``(max_in_flight, in_flight_at_steady_state)`` so callers can assert + the concurrency cap. Handlers increment a shared counter on entry and + decrement on exit, so the counter reflects only calls that got past the + semaphore - parked-on-semaphore tasks do not count as in-flight. + """ + in_flight = 0 + max_in_flight = 0 + + async def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + in_flight += 1 + if in_flight > max_in_flight: + max_in_flight = in_flight + try: + await event.wait() + finally: + in_flight -= 1 + return AIMessage(content="ok") + + tasks = [asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler)) for _ in range(count)] + + # Yield until the steady state is reached: the cap is hit (capped case), or + # every task has been admitted (uncapped case). + limit = middleware.max_concurrent_llm_calls + target = count if limit <= 0 else min(count, limit) + for _ in range(100): + if max_in_flight >= target: + break + await asyncio.sleep(0) + steady_in_flight = in_flight + + event.set() + await asyncio.gather(*tasks) + return max_in_flight, steady_in_flight + + +@pytest.mark.anyio +async def test_limiter_caps_concurrent_llm_calls() -> None: + """With max_concurrent_llm_calls=2, five concurrent calls must never exceed + two in-flight at once - the rest park on the process-global semaphore. + """ + middleware = _build_middleware(max_concurrent_llm_calls=2) + event = asyncio.Event() + + max_in_flight, steady_in_flight = await _run_concurrent(middleware, 5, event) + + assert max_in_flight == 2 + # At steady state exactly two handlers are parked on the event; the other + # three are blocked on the semaphore and have not entered the handler. + assert steady_in_flight == 2 + + +@pytest.mark.anyio +async def test_limiter_disabled_by_default() -> None: + """max_concurrent_llm_calls defaults to 0 (disabled): no cap, so all five + concurrent calls run at once. Guards against the semaphore accidentally + engaging for existing deployments that never opted in. + """ + middleware = _build_middleware() # default max_concurrent_llm_calls=0 + event = asyncio.Event() + + max_in_flight, _ = await _run_concurrent(middleware, 5, event) + + assert max_in_flight == 5 + + +@pytest.mark.anyio +async def test_limiter_is_shared_across_instances() -> None: + """The semaphore is process-global, not per-middleware-instance: two + middlewares with the same limit share one cap, so four calls spread across + them still never exceed two in-flight. + """ + mw_a = _build_middleware(max_concurrent_llm_calls=2) + mw_b = _build_middleware(max_concurrent_llm_calls=2) + event = asyncio.Event() + + in_flight = 0 + max_in_flight = 0 + + async def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + in_flight += 1 + if in_flight > max_in_flight: + max_in_flight = in_flight + try: + await event.wait() + finally: + in_flight -= 1 + return AIMessage(content="ok") + + tasks = [ + asyncio.create_task(mw_a.awrap_model_call(SimpleNamespace(), handler)), + asyncio.create_task(mw_b.awrap_model_call(SimpleNamespace(), handler)), + asyncio.create_task(mw_a.awrap_model_call(SimpleNamespace(), handler)), + asyncio.create_task(mw_b.awrap_model_call(SimpleNamespace(), handler)), + ] + for _ in range(100): + if max_in_flight >= 2: + break + await asyncio.sleep(0) + + assert max_in_flight == 2 # shared global cap, not 2+2 per instance + event.set() + await asyncio.gather(*tasks) + + +@pytest.mark.anyio +async def test_limiter_releases_slot_during_backoff_sleep( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The semaphore wraps a single attempt only, not the retry loop. A call in + its backoff sleep must release its slot so another caller can proceed - + otherwise backoff would waste concurrency slots and worsen the burst we are + trying to smooth. + + ``fake_sleep`` parks call A in backoff on ``backoff_gate`` so the test has a + deterministic window in which to observe call B being admitted to the freed + slot. ``asyncio.Event.wait`` / ``asyncio.wait_for`` do not route through + ``asyncio.sleep``, so the monkeypatch does not disturb test orchestration. + """ + middleware = _build_middleware( + max_concurrent_llm_calls=1, + retry_max_attempts=2, + retry_base_delay_ms=10000, + retry_cap_delay_ms=10000, + ) + a_entered_backoff = asyncio.Event() + backoff_gate = asyncio.Event() + b_admitted = asyncio.Event() + attempts_a = 0 + + async def fake_sleep(_delay: float) -> None: + a_entered_backoff.set() + # Park call A in backoff until the test releases it. + await backoff_gate.wait() + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + async def handler_a(_request) -> AIMessage: + nonlocal attempts_a + attempts_a += 1 + # Always fails -> call A enters backoff after attempt 1, exhausts after 2. + raise FakeError("server busy", status_code=503) + + async def handler_b(_request) -> AIMessage: + b_admitted.set() + return AIMessage(content="b-ok") + + task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_a)) + # Wait until call A has failed attempt 1 and parked in its backoff sleep - + # at which point its semaphore slot has been released. + await asyncio.wait_for(a_entered_backoff.wait(), timeout=2.0) + + task_b = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_b)) + # Call B must be admitted to the single slot while A is still parked in + # backoff (A has not retried yet). + await asyncio.wait_for(b_admitted.wait(), timeout=2.0) + assert attempts_a == 1 + + # Release A: it retries once more, fails again, and exhausts its budget. + backoff_gate.set() + result_a = await task_a + result_b = await task_b + assert result_b.content == "b-ok" + assert result_a.additional_kwargs.get("deerflow_error_fallback") is True + + +# ---------- Decorrelated jitter ---------- + + +def test_retry_delay_decorrelated_jitter_within_bounds() -> None: + """First retry seeds from base: high = max(base, base*3) = base*3, so the + delay lands in [base, base*3]. + """ + middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000) + delay = middleware._build_retry_delay_ms(100, FakeError("server busy", status_code=503)) + assert 100 <= delay <= 300 + + +def test_retry_delay_grows_from_previous_delay() -> None: + """Decorrelated jitter grows off the previous delay, not a fixed schedule: + prev=1000 -> high = max(100, 3000) = 3000 -> delay in [100, 3000]. + """ + middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000) + delay = middleware._build_retry_delay_ms(1000, FakeError("server busy", status_code=503)) + assert 100 <= delay <= 3000 + + +def test_retry_delay_respects_cap() -> None: + """The cap always bounds the jittered delay, even when prev*3 would exceed it.""" + middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=200) + delay = middleware._build_retry_delay_ms(1000, FakeError("server busy", status_code=503)) + assert delay <= 200 + + +def test_retry_delay_base_equals_cap_is_deterministic() -> None: + """When base == cap the jittered delay collapses to exactly cap regardless + of the RNG draw - this is what keeps the fast/retry-budget tests stable. + """ + middleware = _build_middleware(retry_base_delay_ms=25, retry_cap_delay_ms=25) + for _ in range(20): + delay = middleware._build_retry_delay_ms(25, FakeError("server busy", status_code=503)) + assert delay == 25 + + +def test_retry_delay_honors_retry_after_without_jitter() -> None: + """An explicit Retry-After is honored verbatim - the server said exactly + when to come back, so jitter must not perturb it. + """ + middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000) + exc = FakeError("rate limited", status_code=429, headers={"retry-after-ms": "5000"}) + delay = middleware._build_retry_delay_ms(100, exc) + assert delay == 5000 + + +@pytest.mark.anyio +async def test_async_retry_loop_emits_jittered_delays_within_bounds( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """End-to-end: the async retry loop emits decorrelated-jitter delays, each + within [base, cap], and prev_delay threads through so the second delay can + grow off the first. + """ + middleware = _build_middleware( + retry_max_attempts=3, + retry_base_delay_ms=100, + retry_cap_delay_ms=10000, + ) + waits: list[float] = [] + + async def fake_sleep(delay: float) -> None: + waits.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + attempts = 0 + + async def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + if attempts < 3: + raise FakeError("server busy", status_code=503) + return AIMessage(content="ok") + + result = await middleware.awrap_model_call(SimpleNamespace(), handler) + + assert result.content == "ok" + assert attempts == 3 + assert len(waits) == 2 + for w in waits: + assert 0.1 <= w <= 10.0 + + +# ---------- Config wiring ---------- + + +def test_max_concurrent_llm_calls_defaults_to_disabled() -> None: + """With no llm_call config, the cap defaults to 0 (disabled) - existing + deployments see no behavior change. + """ + middleware = LLMErrorHandlingMiddleware(app_config=_make_app_config()) + assert middleware.max_concurrent_llm_calls == 0 + + +def test_max_concurrent_llm_calls_wired_from_config() -> None: + """llm_call.max_concurrent_calls flows through AppConfig into the middleware + attribute that ``__init__`` applies to the process limiter via + ``_apply_configured_cap`` (the single generation-aware cap owner). + """ + app_config = AppConfig( + sandbox=SandboxConfig(use="test"), + llm_call=LlmCallConfig(max_concurrent_calls=8), + ) + middleware = LLMErrorHandlingMiddleware(app_config=app_config) + assert middleware.max_concurrent_llm_calls == 8 + + +@pytest.mark.anyio +async def test_configured_cap_bounds_concurrency_end_to_end() -> None: + """A cap set via config.yaml (not via setattr) actually bounds in-flight + calls end-to-end: AppConfig -> middleware -> process-wide limiter. + """ + app_config = AppConfig( + sandbox=SandboxConfig(use="test"), + llm_call=LlmCallConfig(max_concurrent_calls=2), + ) + middleware = LLMErrorHandlingMiddleware(app_config=app_config) + event = asyncio.Event() + + max_in_flight, steady_in_flight = await _run_concurrent(middleware, 5, event) + + assert max_in_flight == 2 + assert steady_in_flight == 2 + + +# ---------- Burst-rate (limit_burst_rate) classification ---------- + + +def test_classify_error_limit_burst_rate_by_code() -> None: + """A 429 with error code ``limit_burst_rate`` classifies as burst_rate, + not generic transient - so it gets the tight budget + longer backoff. + """ + middleware = _build_middleware() + exc = FakeError("Request rate increased too quickly", status_code=429, code="limit_burst_rate") + assert middleware._classify_error(exc) == (True, "burst_rate") + + +def test_classify_error_limit_burst_rate_by_message() -> None: + """Burst-rate is also detectable from the message alone (no code field), + matching how Volcano Engine phrases the error. + """ + middleware = _build_middleware() + exc = FakeError("Request rate increased too quickly. To ensure system stability, please adjust your client logic.", status_code=429) + assert middleware._classify_error(exc) == (True, "burst_rate") + + +def test_classify_error_normal_429_stays_transient() -> None: + """A non-burst 429 (generic 'too many requests') must NOT be classified as + burst_rate - it keeps the full retry budget and normal backoff. + """ + middleware = _build_middleware() + exc = FakeError("Too many requests", status_code=429) + assert middleware._classify_error(exc) == (True, "transient") + + +def test_classify_error_burst_takes_precedence_over_transient_status() -> None: + """Burst-rate detection runs before the generic 429->transient mapping.""" + middleware = _build_middleware() + exc = FakeError("limit_burst_rate triggered", status_code=429, code="limit_burst_rate") + assert middleware._classify_error(exc) == (True, "burst_rate") + + +def test_max_attempts_for_burst_rate_is_tight() -> None: + """burst_rate gets a 2-attempt budget (1 + 1 retry) even when the global + cap is higher - retrying into the burst adds demand to the throttled slope. + """ + middleware = _build_middleware(retry_max_attempts=3) + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + assert middleware._max_attempts_for(exc, "burst_rate") == 2 + + +def test_max_attempts_for_burst_rate_respects_user_cap() -> None: + """If the operator lowered retry_max_attempts below the burst budget, the + user cap wins - overrides only ever tighten, never loosen.""" + middleware = _build_middleware(retry_max_attempts=1) + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + assert middleware._max_attempts_for(exc, "burst_rate") == 1 + + +def test_burst_rate_delay_uses_longer_base() -> None: + """burst_rate backoff uses burst_retry_base_delay_ms (not the normal base), + so the single retry lands after the throttle window subsides.""" + middleware = _build_middleware( + retry_base_delay_ms=10, + retry_cap_delay_ms=200, + burst_retry_base_delay_ms=100, + ) + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + # prev = burst base (100) -> high = max(100, 300) = 300 -> delay in [100, 200] (capped) + delay = middleware._build_retry_delay_ms(100, exc, reason="burst_rate") + assert 100 <= delay <= 200 + + +def test_normal_transient_delay_uses_normal_base_not_burst() -> None: + """Non-burst transient errors keep using the normal base - the burst base + only applies to reason='burst_rate'.""" + middleware = _build_middleware( + retry_base_delay_ms=10, + retry_cap_delay_ms=200, + burst_retry_base_delay_ms=100, + ) + exc = FakeError("server busy", status_code=503) + # prev = 10 -> high = max(10, 30) = 30 -> delay in [10, 30] + delay = middleware._build_retry_delay_ms(10, exc, reason="transient") + assert 10 <= delay <= 30 + + +def test_burst_rate_delay_prefers_retry_after() -> None: + """An explicit Retry-After is honored verbatim for burst-rate errors - the + server said exactly when to come back, so no jitter / longer base applies.""" + middleware = _build_middleware( + retry_base_delay_ms=10, + retry_cap_delay_ms=200, + burst_retry_base_delay_ms=100, + ) + exc = FakeError( + "rate increased too quickly", + status_code=429, + code="limit_burst_rate", + headers={"retry-after-ms": "5000"}, + ) + assert middleware._build_retry_delay_ms(100, exc, reason="burst_rate") == 5000 + + +def test_burst_rate_exhausted_returns_distinct_message() -> None: + """When the burst retry budget is exhausted, the user-facing message names + the burst-rate throttle rather than the generic 'temporarily unavailable'.""" + middleware = _build_middleware() + exc = FakeError("Request rate increased too quickly", status_code=429, code="limit_burst_rate") + message = middleware._build_user_message(exc, reason="burst_rate") + assert "burst-rate" in message + assert "temporarily unavailable" not in message + + +@pytest.mark.anyio +async def test_async_burst_rate_uses_tight_budget_and_longer_base( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """End-to-end: a handler raising limit_burst_rate retries at most once + (budget 2) with a delay drawn from the burst base, then surfaces a fallback. + """ + middleware = _build_middleware( + retry_max_attempts=3, + retry_base_delay_ms=10, + retry_cap_delay_ms=200, + burst_retry_base_delay_ms=100, + ) + waits: list[float] = [] + + async def fake_sleep(delay: float) -> None: + waits.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + attempts = 0 + + async def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + raise FakeError("Request rate increased too quickly", status_code=429, code="limit_burst_rate") + + result = await middleware.awrap_model_call(SimpleNamespace(), handler) + + assert attempts == 2 # tight budget: 1 first attempt + 1 retry + assert len(waits) == 1 + # Longer burst base: delay in [burst_base, cap] = [0.1s, 0.2s] + assert 0.1 <= waits[0] <= 0.2 + assert result.additional_kwargs.get("deerflow_error_fallback") is True + assert result.additional_kwargs.get("error_reason") == "burst_rate" + + +def test_burst_rate_exhaustion_does_not_trip_circuit_breaker(monkeypatch: pytest.MonkeyPatch) -> None: + """P2: burst-rate (limit_burst_rate) is a transient provider slope-throttle, + not "provider down". Exhausting its retry budget must NOT count toward the + circuit breaker - otherwise N consecutive burst failures flip the CB open + and fast-fail ALL calls for the recovery window, the exact self-inflicted + outage #4290 is trying to prevent. The burst reason is already distinctively + classified by this PR, so it is the natural place to exclude it. + """ + monkeypatch.setattr("time.sleep", lambda _d: None) + middleware = _build_middleware(circuit_failure_threshold=3, retry_max_attempts=3) + + def handler(_request) -> AIMessage: + raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + + # Exceed the failure threshold with burst-rate failures: the CB must stay + # closed (burst-rate is transient-by-design, not a provider outage). + for _ in range(5): + result = middleware.wrap_model_call(SimpleNamespace(), handler) + assert result.additional_kwargs.get("error_reason") == "burst_rate" + + assert middleware._circuit_failure_count == 0 + assert middleware._circuit_state == "closed" + assert middleware._check_circuit() is False # still admitting calls + + # A subsequent successful call still goes through (CB never opened). + def ok_handler(_request) -> AIMessage: + return AIMessage(content="ok") + + result = middleware.wrap_model_call(SimpleNamespace(), ok_handler) + assert result.content == "ok" + + +@pytest.mark.anyio +async def test_async_burst_rate_exhaustion_does_not_trip_circuit_breaker( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Async mirror: burst-rate exhaustion on the async path also stays out of + the circuit breaker (the gate lives in both ``wrap_model_call`` and + ``awrap_model_call``). + """ + + async def _noop_sleep(_d: float) -> None: + return None + + monkeypatch.setattr(asyncio, "sleep", _noop_sleep) + middleware = _build_middleware(circuit_failure_threshold=3, retry_max_attempts=3) + + async def handler(_request) -> AIMessage: + raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + + for _ in range(5): + result = await middleware.awrap_model_call(SimpleNamespace(), handler) + assert result.additional_kwargs.get("error_reason") == "burst_rate" + + assert middleware._circuit_failure_count == 0 + assert middleware._circuit_state == "closed" + assert middleware._check_circuit() is False + + +# ---------- Effective retry budget in retry events (review P2) ---------- + + +def _capture_retry_events(monkeypatch: pytest.MonkeyPatch) -> list[dict]: + """Capture the ``llm_retry`` stream events emitted by ``_emit_retry_event``. + + ``_emit_retry_event`` lazily imports ``get_stream_writer`` from + ``langgraph.config`` on each call, so monkeypatching the module attribute is + enough to intercept the payloads without standing up a LangGraph run context + (the real writer raises ``RuntimeError`` outside one - which the middleware + swallows - so the patched writer must simply return a callable). + """ + captured: list[dict] = [] + + def _writer(payload: dict) -> None: + captured.append(payload) + + monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: _writer) + return captured + + +def test_burst_rate_retry_event_reports_effective_budget_sync(monkeypatch: pytest.MonkeyPatch) -> None: + """P2: a burst-rate (limit_burst_rate) call is capped at 2 attempts, so the + emitted ``llm_retry`` event and its user-facing message must report the + *effective* budget (``max_attempts=2``, message ``1/2``), not the configured + ceiling (``retry_max_attempts=3``) - otherwise the frontend promises a retry + that never happens. Exactly two handler attempts occur (1 first + 1 retry). + """ + monkeypatch.setattr("time.sleep", lambda _d: None) + events = _capture_retry_events(monkeypatch) + middleware = _build_middleware(retry_max_attempts=3) # default global budget + + attempts = 0 + + def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert attempts == 2 # 1 first attempt + 1 retry; budget is 2, not 3 + assert result.additional_kwargs.get("error_reason") == "burst_rate" + assert len(events) == 1 # exactly one retry event (attempt 1 -> 2) + assert events[0]["max_attempts"] == 2 # effective budget, not the ceiling 3 + assert events[0]["attempt"] == 1 + assert "1/2" in events[0]["message"] + + +@pytest.mark.anyio +async def test_burst_rate_retry_event_reports_effective_budget_async( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Async mirror of the sync P2 test: the async burst-rate path also reports + the effective budget (``max_attempts=2``, message ``1/2``) and runs exactly + two handler attempts. + """ + + async def _noop_sleep(_d: float) -> None: + return None + + monkeypatch.setattr(asyncio, "sleep", _noop_sleep) + events = _capture_retry_events(monkeypatch) + middleware = _build_middleware(retry_max_attempts=3) + + attempts = 0 + + async def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + + result = await middleware.awrap_model_call(SimpleNamespace(), handler) + + assert attempts == 2 + assert result.additional_kwargs.get("error_reason") == "burst_rate" + assert len(events) == 1 + assert events[0]["max_attempts"] == 2 + assert events[0]["attempt"] == 1 + assert "1/2" in events[0]["message"] + + +# ---------- Retry params config wiring ---------- + + +def test_retry_params_default_values() -> None: + """With no llm_call config, retry params fall back to their documented + defaults (matching the previous hard-coded class attributes).""" + middleware = LLMErrorHandlingMiddleware(app_config=_make_app_config()) + assert middleware.retry_max_attempts == 3 + assert middleware.retry_base_delay_ms == 1000 + assert middleware.retry_cap_delay_ms == 8000 + assert middleware.burst_retry_base_delay_ms == 5000 + assert middleware.max_concurrent_llm_calls == 0 + + +def test_retry_params_wired_from_config() -> None: + """All retry/backoff knobs flow from config.yaml -> AppConfig -> middleware.""" + app_config = AppConfig( + sandbox=SandboxConfig(use="test"), + llm_call=LlmCallConfig( + retry_max_attempts=7, + retry_base_delay_ms=123, + retry_cap_delay_ms=999, + burst_retry_base_delay_ms=777, + max_concurrent_calls=4, + ), + ) + middleware = LLMErrorHandlingMiddleware(app_config=app_config) + assert middleware.retry_max_attempts == 7 + assert middleware.retry_base_delay_ms == 123 + assert middleware.retry_cap_delay_ms == 999 + assert middleware.burst_retry_base_delay_ms == 777 + assert middleware.max_concurrent_llm_calls == 4 + + +def test_burst_retry_base_delay_is_configurable_end_to_end() -> None: + """A burst base set via config actually drives the burst retry delay.""" + app_config = AppConfig( + sandbox=SandboxConfig(use="test"), + llm_call=LlmCallConfig(burst_retry_base_delay_ms=100, retry_cap_delay_ms=200), + ) + middleware = LLMErrorHandlingMiddleware(app_config=app_config) + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + delay = middleware._build_retry_delay_ms(100, exc, reason="burst_rate") + assert 100 <= delay <= 200 + + +# ---------- Process-wide limiter across call paths (review P1) ---------- + + +def _run_on_isolated_loop(coro_factory: Any) -> concurrent.futures.Future: + """Run ``coro_factory()`` on a freshly-created event loop in a worker thread. + + Mirrors how ``subagents/executor.py`` runs subagent calls on an isolated + persistent loop separate from the lead agent's loop. Returns a + ``concurrent.futures.Future`` holding the result/exception. + """ + done: concurrent.futures.Future = concurrent.futures.Future() + + def runner() -> None: + loop = asyncio.new_event_loop() + try: + asyncio.set_event_loop(loop) + done.set_result(loop.run_until_complete(coro_factory())) + except Exception as exc: # propagate to the awaiting test + done.set_exception(exc) + finally: + loop.close() + + threading.Thread(target=runner, daemon=True).start() + return done + + +@pytest.mark.anyio +async def test_limiter_is_process_wide_across_event_loops() -> None: + """A lead-agent call (main loop) and a subagent call (isolated loop) share + one process-wide cap - the limiter is NOT loop-bound (unlike + asyncio.Semaphore). With cap=1, two concurrent calls on different loops must + never both be in-flight. Regression for the per-loop cap the reviewer found. + """ + middleware = _build_middleware(max_concurrent_llm_calls=1) + gate = threading.Event() # cross-loop-safe block + in_flight = 0 + max_in_flight = 0 + lock = threading.Lock() + + async def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + with lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + await asyncio.to_thread(gate.wait) + finally: + with lock: + in_flight -= 1 + return AIMessage(content="ok") + + # Call A on the test (main) loop; call B on a separate event loop / thread. + task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler)) + fut_b = _run_on_isolated_loop(lambda: middleware.awrap_model_call(SimpleNamespace(), handler)) + + # Let both loops run; with cap=1 only one handler may be in-flight at a time + # across BOTH loops. + for _ in range(100): + await asyncio.sleep(0) + with lock: + if max_in_flight >= 1: + break + with lock: + assert max_in_flight == 1 + + gate.set() + result_a = await task_a + assert result_a.content == "ok" + result_b = await asyncio.wait_for(asyncio.wrap_future(fut_b), timeout=5) + assert result_b.content == "ok" + with lock: + assert in_flight == 0 # both calls completed; no handler left in-flight + + +def test_limiter_caps_concurrent_sync_calls() -> None: + """The sync graph path now acquires the limiter too (previously bypassed it + entirely). With cap=1, two concurrent sync calls must never both be + in-flight. Regression for the sync-wrapper bypass the reviewer found. + """ + middleware = _build_middleware(max_concurrent_llm_calls=1) + gate = threading.Event() + in_flight = 0 + max_in_flight = 0 + lock = threading.Lock() + + def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + with lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + gate.wait() + with lock: + in_flight -= 1 + return AIMessage(content="ok") + + results: list[AIMessage] = [] + + def run_one() -> None: + results.append(middleware.wrap_model_call(SimpleNamespace(), handler)) + + t1 = threading.Thread(target=run_one) + t2 = threading.Thread(target=run_one) + t1.start() + t2.start() + # Wait until one handler is in-flight; the other blocks on the limiter. + for _ in range(400): + with lock: + if max_in_flight >= 1: + break + time.sleep(0.005) + with lock: + assert max_in_flight == 1 + + gate.set() + t1.join(timeout=5) + t2.join(timeout=5) + assert not t1.is_alive() and not t2.is_alive() + assert len(results) == 2 + + +def test_cap_is_frozen_at_first_construction_and_unchanged_by_later_instances() -> None: + """Startup-only cap: the first ``__init__`` resolves the cap and freezes it; + later instances - whether they would raise it, lower it, or disable it - are + no-ops (same limiter instance, cap unchanged, in-flight permits preserved). + Replaces the prior generation-aware in-place update with a frozen-at-startup + controller. There is no cap mutation at runtime, so the downscale race + (review P1 Part A: a lowered cap handing excess permits to queued waiters, + keeping ``in_flight`` pegged at the old cap) is structurally unreachable, + and the construction-order / config-freshness race (Part B) has nothing to + race on. + """ + from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter + + first_mw = _build_middleware(max_concurrent_llm_calls=1) + limiter = _get_process_limiter() + assert limiter is not None + assert limiter.limit == 1 + + # A holder takes the single permit so we can observe whether later instances + # disturb in-flight state (they must not). + gate = threading.Event() + in_flight = 0 + max_in_flight = 0 + lock = threading.Lock() + + def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + with lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + gate.wait() + with lock: + in_flight -= 1 + return AIMessage(content="ok") + + holder = threading.Thread(target=lambda: first_mw.wrap_model_call(SimpleNamespace(), handler)) + holder.start() + try: + for _ in range(400): + with lock: + if max_in_flight >= 1: + break + time.sleep(0.005) + with lock: + assert max_in_flight == 1 + assert limiter.in_flight == 1 # holder holds the single permit + + # Later instances try to raise (3) and to disable (0): both ignored. + _build_middleware(max_concurrent_llm_calls=3) + assert _get_process_limiter() is limiter # same instance, not recreated + assert limiter.limit == 1 # frozen; raise attempt had no effect + _build_middleware(max_concurrent_llm_calls=0) + assert limiter.limit == 1 # frozen; disable attempt had no effect + assert limiter.in_flight == 1 # holder's permit untouched + finally: + # Always release the holder so a failing assertion doesn't leak the thread. + gate.set() + holder.join(timeout=5) + assert not holder.is_alive() + assert limiter.in_flight == 0 # permit returned, not leaked + + +def test_first_constructed_cap_wins_over_later_config_snapshot() -> None: + """Startup-only cap (review P1 Part B): the cap is frozen at the FIRST + middleware construction, so a later instance holding an OLDER config + snapshot (higher cap) cannot restore it - regardless of construction timing + or config freshness. This is the startup-only replacement for the prior + generation-aware guard: there is no "newer config wins" race because no + instance can mutate the cap at all. + + Mirrors the reviewer's reverse-construction probe: the lower (newer) cap is + constructed first, the higher (older/stale) cap second, and the older + snapshot must NOT raise the live cap. Three overlapping calls through the + older-cap instance are still bounded by the frozen cap of 1 - which also + proves the Part A invariant (a sustained queue never admits callers above + the frozen cap). + """ + from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter + + # "Newer" config (cap 1) constructed FIRST -> freezes the cap at 1. + _build_middleware(max_concurrent_llm_calls=1) + # "Older/stale" config (cap 3) constructed AFTER -> must not raise the cap. + older_mw = _build_middleware(max_concurrent_llm_calls=3) + limiter = _get_process_limiter() + assert limiter is not None + assert limiter.limit == 1 # frozen at first construction; older snapshot did not raise + + gate = threading.Event() + in_flight = 0 + max_in_flight = 0 + lock = threading.Lock() + + def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + with lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + gate.wait() + with lock: + in_flight -= 1 + return AIMessage(content="ok") + + # 3 overlapping calls through the OLDER-cap (cap 3) instance: the live cap is + # frozen at 1, so only one may be in-flight - the older snapshot cannot + # restore 3 AND callers cannot be admitted above the frozen cap. + threads = [threading.Thread(target=lambda: older_mw.wrap_model_call(SimpleNamespace(), handler)) for _ in range(3)] + for t in threads: + t.start() + try: + for _ in range(400): + with lock: + if max_in_flight >= 1: + break + time.sleep(0.005) + # Give the queued calls a chance to (incorrectly) exceed the frozen cap. + time.sleep(0.05) + with lock: + assert max_in_flight == 1 # never exceeded the frozen cap of 1 + assert limiter.limit == 1 # still 1; older snapshot did not restore 3 + finally: + # Always release the holder threads so a failing assertion doesn't hang. + gate.set() + for t in threads: + t.join(timeout=5) + assert all(not t.is_alive() for t in threads) + assert limiter.in_flight == 0 + + +@pytest.mark.anyio +async def test_frozen_cap_binds_calls_across_isolated_loop() -> None: + """Startup-only cap (cross-loop): the cap frozen at first construction binds + async calls across the lead loop AND an isolated subagent loop - a later + instance (higher cap) cannot raise it, so cross-loop in-flight calls never + exceed the frozen cap. Replaces the prior generation-aware cross-loop test. + """ + from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter + + # First construction (cap 1) freezes the cap; a later cap=3 instance can't raise it. + _build_middleware(max_concurrent_llm_calls=1) + older_mw = _build_middleware(max_concurrent_llm_calls=3) + limiter = _get_process_limiter() + assert limiter is not None + assert limiter.limit == 1 # frozen at 1; later cap=3 instance did not raise + + gate = threading.Event() + counter_lock = threading.Lock() + in_flight = 0 + max_in_flight = 0 + + async def handler(_request) -> AIMessage: + nonlocal in_flight, max_in_flight + with counter_lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + await asyncio.to_thread(gate.wait) + finally: + with counter_lock: + in_flight -= 1 + return AIMessage(content="ok") + + # Lead-loop call + isolated-loop call, both on the cap-3 (older) instance: + # the frozen cap is 1, so only one may be in-flight across BOTH loops. + task_lead = asyncio.create_task(older_mw.awrap_model_call(SimpleNamespace(), handler)) + fut_iso = _run_on_isolated_loop(lambda: older_mw.awrap_model_call(SimpleNamespace(), handler)) + try: + for _ in range(100): + await asyncio.sleep(0) + with counter_lock: + if max_in_flight >= 1: + break + await asyncio.sleep(0.05) # let the second call queue on the limiter + with counter_lock: + assert max_in_flight == 1 # never exceeded the frozen cap across loops + assert limiter.limit == 1 # still 1; later cap=3 instance did not raise + finally: + # Release the gate so parked handlers unblock and complete, then drain + # both tasks. return_exceptions surfaces any late failure as a value + # rather than masking the assertion above with a raise; gate.set() + # guarantees both complete, so no timeout/empty-except is needed. + gate.set() + await asyncio.gather(task_lead, asyncio.wrap_future(fut_iso), return_exceptions=True) + with counter_lock: + assert in_flight == 0 + + +@pytest.mark.anyio +async def test_limiter_cancellation_does_not_leak_capacity() -> None: + """A caller cancelled while waiting on the limiter must not leak a permit: + after it's cancelled and the in-flight call releases, a fresh call is still + admitted to the same cap. Regression for the cancellation-capacity-leak + concern the reviewer raised. + """ + middleware = _build_middleware(max_concurrent_llm_calls=1) + gate = threading.Event() + a_started = asyncio.Event() + + async def handler_a(_request) -> AIMessage: + a_started.set() + await asyncio.to_thread(gate.wait) + return AIMessage(content="a-ok") + + async def handler_b(_request) -> AIMessage: + return AIMessage(content="b-ok") + + # A acquires the single permit and blocks. + task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_a)) + await asyncio.wait_for(a_started.wait(), timeout=2) + + # B queues on the limiter (no permit available). + task_b = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_b)) + await asyncio.sleep(0.05) + assert not task_b.done() # B is waiting, not admitted + + # Cancel B while it waits on the limiter. + task_b.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.gather(task_b) + + # Release A; its permit returns. B's cancellation must not have consumed it. + gate.set() + result_a = await task_a + assert result_a.content == "a-ok" + + # A fresh call C must be admitted immediately - proving B's cancellation + # didn't leak capacity (in_flight back to 0, not stuck at 1). + task_c = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_b)) + result_c = await asyncio.wait_for(task_c, timeout=2) + assert result_c.content == "b-ok" + + +@pytest.mark.anyio +async def test_limiter_cancellation_after_dequeue_hands_off_to_next_waiter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """P1 #1 regression: cap=1, A holds the permit; B and C queue. A releases + (reserving/granting B); B is cancelled in the post-dequeue / pre-reacquire + handoff window; C must still complete WITHOUT another release - the reserved + permit is handed off to C rather than stranded with capacity idle. + + ``test_limiter_cancellation_does_not_leak_capacity`` cancels B while it is + still purely queued (before A releases); this test cancels B *after* it has + been dequeued+granted but *before* it wakes, which is the window the prior + limiter stranded the next waiter in. + """ + from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter + + middleware = _build_middleware(max_concurrent_llm_calls=1) + a_started = asyncio.Event() + gate = threading.Event() + + async def handler_a(_request) -> AIMessage: + a_started.set() + await asyncio.to_thread(gate.wait) + return AIMessage(content="a-ok") + + async def handler_ok(_request) -> AIMessage: + return AIMessage(content="ok") + + task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_a)) + await asyncio.wait_for(a_started.wait(), timeout=2) + + # B and C queue on the limiter (no permit available). + task_b = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_ok)) + task_c = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_ok)) + await asyncio.sleep(0.05) # let B and C register as waiters + assert not task_b.done() and not task_c.done() + + limiter = _get_process_limiter() + assert limiter is not None + + # Patch the wake so the FIRST granted waiter (B) does NOT wake yet - this + # is the post-dequeue / pre-reacquire handoff window. Later wakes (C, handed + # off from B's cancellation) proceed via the real path. ``_wake_locked`` is + # called under the limiter lock and must not block, so we only toggle which + # branch runs; B's event is simply never set, leaving B parked until cancel. + real_wake = type(limiter)._wake_locked + state = {"first_done": False} + + def patched_wake(waiter: Any) -> bool: + if not state["first_done"]: + state["first_done"] = True + return True # pretend the loop is alive; do NOT set B's event + return real_wake(limiter, waiter) + + monkeypatch.setattr(limiter, "_wake_locked", patched_wake) + + # Release A: reserves the permit for B (dequeues B, granted=True) but B does + # not wake (patched). B sits in the handoff window. + gate.set() + result_a = await task_a + assert result_a.content == "a-ok" + await asyncio.sleep(0.02) # let A's release grant B + assert state["first_done"] # B was dequeued+granted, never woken + + # Cancel B while it is granted-but-not-yet-awake. Its cancellation must hand + # the reserved permit to C (the next waiter), waking C via the real path. + task_b.cancel() + # gather(return_exceptions=True) surfaces B's cancellation as a value so we + # can assert on it directly, instead of a bare ``await task_b`` inside + # pytest.raises (whose discarded result trips the "statement has no effect" + # analyzer without adding any assertion strength here). + b_outcomes = await asyncio.gather(task_b, return_exceptions=True) + assert isinstance(b_outcomes[0], asyncio.CancelledError) + + # C completes WITHOUT another release - no stranded permit. + result_c = await asyncio.wait_for(task_c, timeout=2) + assert result_c.content == "ok" + + # No permit leaked: in_flight is back to 0 after C completes and releases. + assert limiter.in_flight == 0 + + +# ---------- Burst-rate first-retry jitter (review P1) ---------- + + +def test_burst_first_retry_is_non_degenerate_with_default_config() -> None: + """With shipped defaults (burst_base=5000, cap=8000, normal_base=1000), the + first burst retry must be drawn from a NON-degenerate window [5000, 8000], + not fixed at 5000ms. Regression for the deterministic-herd bug (prev was + seeded from the 1000ms normal base, collapsing the window to a point). + """ + middleware = _build_middleware() # defaults + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + # prev_delay_ms=None simulates the first retry (loops now init it to None). + delays = {middleware._build_retry_delay_ms(None, exc, reason="burst_rate") for _ in range(20)} + assert all(5000 <= d <= 8000 for d in delays) + assert len(delays) > 1 # non-degenerate: more than one distinct value observed + + +def test_burst_first_retry_uses_jitter_not_fixed_value(monkeypatch: pytest.MonkeyPatch) -> None: + """Controlled RNG: forcing randint to an upper-window value yields that + value (capped), proving the first burst retry is jittered rather than a + constant 5000ms.""" + middleware = _build_middleware() # defaults + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + monkeypatch.setattr( + "deerflow.agents.middlewares.llm_error_handling_middleware.random.randint", + lambda lo, hi: 7000, + ) + assert middleware._build_retry_delay_ms(None, exc, reason="burst_rate") == 7000 + + +@pytest.mark.anyio +async def test_async_burst_first_retry_non_degenerate_default_config( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """End-to-end on the real async wrapper with default burst config: the + first (and only) burst retry lands at a jittered value, not fixed at 5s.""" + middleware = _build_middleware() # defaults: burst_base=5000, cap=8000 + waits: list[float] = [] + + async def fake_sleep(delay: float) -> None: + waits.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + monkeypatch.setattr( + "deerflow.agents.middlewares.llm_error_handling_middleware.random.randint", + lambda lo, hi: 7000, + ) + + async def handler(_request) -> AIMessage: + raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + + result = await middleware.awrap_model_call(SimpleNamespace(), handler) + assert len(waits) == 1 + assert waits[0] == 7.0 # 7000ms, jittered - not the fixed 5.0s + assert result.additional_kwargs.get("error_reason") == "burst_rate" + + +def test_concurrent_burst_failures_get_distinct_jittered_delays() -> None: + """De-synchronization invariant: concurrent burst-rate failures get distinct + retry delays, so a fleet that failed together does not realign on one tick. + + Uses a seeded real RNG (not a monkeypatch) so the window computation still + runs - on the old code the window collapsed to ``randint(5000, 5000)`` and + every delay was 5000ms (``len(set) == 1``); the jittered window yields many + distinct values. + """ + import random as _random + + middleware = _build_middleware() # defaults + exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate") + saved_state = _random.getstate() + _random.seed(42) + try: + delays = [middleware._build_retry_delay_ms(None, exc, reason="burst_rate") for _ in range(50)] + finally: + _random.setstate(saved_state) + assert all(5000 <= d <= 8000 for d in delays) + assert len(set(delays)) > 1 # de-synchronized, not a single 5000ms tick diff --git a/config.example.yaml b/config.example.yaml index f63930449..aa706921c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1985,6 +1985,49 @@ authorization: # # Time in seconds before attempting to recover (default: 60) # recovery_timeout_sec: 60 +# ============================================================================ +# LLM Call Concurrency Configuration +# ============================================================================ +# Cap the number of concurrently in-flight LLM calls process-wide. A provider +# burst-rate (limit_burst_rate) error fires on the *slope* of the request rate, +# not on a static quota - so the morning peak (e.g. 08:30) ramping from ~0 to +# full throttle in seconds gets rejected even when total RPM is within budget. +# Capping concurrency caps that slope. Retries alone make it worse (they add +# demand inside the very burst being throttled); pair this cap with the +# decorrelated-jitter backoff already built into the LLM error-handling +# middleware, and ideally an nginx `limit_req` at the ingress. +# +# 0 disables the cap (default) - existing deployments see no behavior change. +# +# Per-process, not per-cluster: the cap bounds in-flight LLM calls within ONE +# gateway process. With GATEWAY_WORKERS > 1 the aggregate cap across the +# deployment is effectively `max_concurrent_calls * GATEWAY_WORKERS`, and a +# multi-node rollout multiplies it further - size the per-process value with +# that in mind (and pair it with an nginx `limit_req` at the ingress for a +# true cluster-wide slope cap). +# +# Startup-only: the cap is captured at the first LLM run and frozen for the +# process lifetime. Editing `max_concurrent_calls` here takes effect only after +# a gateway restart; the other `llm_call.*` knobs below remain hot-reloadable. +# Freezing the cap avoids the downscale and config-freshness races that a +# runtime-mutable, process-wide/cross-loop limiter would otherwise hit (a +# lowered cap could keep admitting queued waiters, or a stale config snapshot +# constructed after a fresher one could restore a higher cap). + +# llm_call: +# # Max concurrently in-flight LLM calls across the whole process (default: 0 = disabled) +# max_concurrent_calls: 0 +# # Max LLM call attempts for retriable transient errors, 1 = no retry (default: 3) +# retry_max_attempts: 3 +# # Base delay (ms) for the decorrelated-jitter retry backoff (default: 1000) +# retry_base_delay_ms: 1000 +# # Hard cap (ms) on any single retry backoff delay (default: 8000) +# retry_cap_delay_ms: 8000 +# # Backoff base (ms) used ONLY for burst-rate (limit_burst_rate) 429s - higher +# # than retry_base_delay_ms so the single burst retry lands after the throttle +# # window subsides. Ignored when the provider sends Retry-After (default: 5000) +# burst_retry_base_delay_ms: 5000 + # ============================================================================ # SSO / OIDC Authentication (optional) # ============================================================================