[eric] agents: extract capacity-retry backoff decision into error_classify.capacity_retry_wait + tests

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}"
                         )

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

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