diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 566b98b4..0026d9b2 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -32,6 +32,8 @@ from backend.config.paths import SESSIONS_DIR from backend.apps.agents.core.error_classify import ( _NON_TRANSIENT_PATTERNS, _TRANSIENT_CAPACITY_PATTERNS, + CAPACITY_BACKOFFS, + capacity_retry_wait, _is_auth_error, _is_free_trial_exhausted, _is_long_context_error, @@ -2008,7 +2010,6 @@ class AgentManager: # user just sees a pause, not a red error card. Hard errors # (auth, plan limit, invalid args) fall through to the existing # error handler unchanged. - _CAPACITY_BACKOFFS = [5, 15, 45, 90, 180] async def _emit_consolidated_thinking(force_provider_unavailable: bool = False) -> None: """Build the running aggregate Message and broadcast it. @@ -2854,16 +2855,13 @@ class AgentManager: pass _ticker_task = None stderr_snapshot = "\n".join(_stderr_buffer[-50:]) - if ( - _is_transient_capacity_error(e, extra_text=stderr_snapshot) - and capacity_retry_attempt < len(_CAPACITY_BACKOFFS) - ): - wait = _CAPACITY_BACKOFFS[capacity_retry_attempt] + wait = capacity_retry_wait(e, capacity_retry_attempt, extra_text=stderr_snapshot) + if wait is not None: capacity_retry_attempt += 1 mid_stream = _current_turn_emitted logger.warning( f"Transient upstream error on session {session_id} " - f"(attempt {capacity_retry_attempt}/{len(_CAPACITY_BACKOFFS)}, " + f"(attempt {capacity_retry_attempt}/{len(CAPACITY_BACKOFFS)}, " f"mid_stream={mid_stream}); sleeping {wait}s before retry. " f"exc={e!r} stderr_tail={stderr_snapshot[-400:]!r}" ) diff --git a/backend/apps/agents/core/error_classify.py b/backend/apps/agents/core/error_classify.py index ad1cb15f..6d7a7039 100644 --- a/backend/apps/agents/core/error_classify.py +++ b/backend/apps/agents/core/error_classify.py @@ -1,4 +1,7 @@ import re +from typing import Optional + +from typeguard import typechecked # Secret shapes that must never ride along when we ship a stderr tail or an # error string to telemetry. own_key mode means the subprocess stderr can echo @@ -211,3 +214,18 @@ def _is_transient_capacity_error(exc: BaseException, extra_text: str = "") -> bo if re.search(r"no\s+pool\s+capacity", combined, re.IGNORECASE): return True return False + + +# Exponential-ish backoff schedule (seconds) for silently retrying a transient upstream +# capacity error before giving up and surfacing the rate-limit pill. +CAPACITY_BACKOFFS = [5, 15, 45, 90, 180] + + +@typechecked +def capacity_retry_wait(exc: BaseException, attempt: int, extra_text: str = "") -> Optional[int]: + """Seconds to wait before retrying a transient upstream capacity error (429 / overload / + 5xx / network blip), or None when the error isn't transient or the backoff budget for + this turn is already spent. Keeps the retry DECISION testable; the loop owns the wait.""" + if _is_transient_capacity_error(exc, extra_text=extra_text) and 0 <= attempt < len(CAPACITY_BACKOFFS): + return CAPACITY_BACKOFFS[attempt] + return None diff --git a/backend/tests/test_capacity_retry.py b/backend/tests/test_capacity_retry.py new file mode 100644 index 00000000..5909f3a9 --- /dev/null +++ b/backend/tests/test_capacity_retry.py @@ -0,0 +1,34 @@ +"""Rigorous coverage for capacity_retry_wait, the transient-error backoff decision lifted +into error_classify.py next to the classifier it uses. It was previously inline + untestable +in the agent loop's retry while-loop.""" + +from backend.apps.agents.core.error_classify import CAPACITY_BACKOFFS, capacity_retry_wait + +# The classifier matches this proxy copy verbatim (a guaranteed-transient signal). +TRANSIENT = "No pool capacity available. Try again shortly." + + +def test_transient_returns_the_scheduled_backoff_for_each_attempt(): + waits = [capacity_retry_wait(Exception(TRANSIENT), i) for i in range(len(CAPACITY_BACKOFFS))] + assert waits == CAPACITY_BACKOFFS # escalates 5 -> 15 -> 45 -> 90 -> 180 + + +def test_budget_exhausted_returns_none(): + assert capacity_retry_wait(Exception(TRANSIENT), len(CAPACITY_BACKOFFS)) is None + assert capacity_retry_wait(Exception(TRANSIENT), len(CAPACITY_BACKOFFS) + 3) is None + + +def test_negative_attempt_returns_none(): + assert capacity_retry_wait(Exception(TRANSIENT), -1) is None + + +def test_non_transient_error_never_retries(): + assert capacity_retry_wait(Exception("invalid_request_error: bad params"), 0) is None + assert capacity_retry_wait(ValueError("a totally unrelated bug"), 0) is None + + +def test_transient_signal_can_arrive_only_via_the_stderr_tail(): + # the CLI's ProcessError stringifies to something generic; the real cause is in stderr + generic = Exception("upstream hiccup") + assert capacity_retry_wait(generic, 0) is None # nothing transient yet + assert capacity_retry_wait(generic, 0, extra_text=TRANSIENT) == 5 # stderr reveals it