feat(loop): classify failures + retry transient ones with backoff

failures.py

@@ -0,0 +1,63 @@
+"""Failure classification — map a coder/infra error to a retry policy.
+
+The loop used to treat every failure the same: Blocked, or (with a ladder) climb a
+model tier. But a rate limit or a transient git/network error is not the feature's
+fault, and a stronger model won't fix it — it should be retried with backoff, not
+permanently blocked. This is a small, ordered regex table (a lean distillation of
+protoMaker's failure-classifier, ~14 categories → the handful that matter for a
+single-board loop) returning whether an error is retryable, how long to back off,
+and the attempt cap. Pure + deterministic — no I/O, trivially testable.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Policy:
+    """How the loop should respond to a failure."""
+
+    category: str
+    retryable: bool
+    base_delay_s: float
+    max_attempts: int  # total dispatch attempts (1 = no retry)
+
+
+# Ordered: first match wins. Patterns are matched case-insensitively against the
+# error message. Only genuinely transient/infra classes are retryable — a stronger
+# model can't clear a rate limit, and a re-dispatch off the latest base can clear a
+# merge conflict, but an auth error or an unknown failure needs a human.
+_RULES: tuple[tuple[str, Policy], ...] = (
+    (
+        r"rate.?limit|\b429\b|quota|overloaded|too many requests|capacity",
+        Policy("rate_limit", True, 60.0, 5),
+    ),
+    (
+        r"timed out|timeout|connection|network|temporarily|econnreset|reset by peer"
+        r"|could not resolve|unavailable|\b50[234]\b",
+        Policy("transient", True, 15.0, 3),
+    ),
+    (
+        r"conflict|cannot be merged|merge failed|non-fast-forward|\brebase\b",
+        Policy("merge_conflict", True, 5.0, 2),
+    ),
+    (
+        r"\bauth\b|permission|forbidden|\b401\b|\b403\b|credential|not authorized",
+        Policy("auth", False, 0.0, 1),
+    ),
+)
+
+# Anything unmatched (incl. "no commits"/no-diff, which the escalation ladder owns)
+# → terminal: block, don't retry.
+TERMINAL = Policy("terminal", False, 0.0, 1)
+
+
+def classify(error: str) -> Policy:
+    """Classify an error message → a retry :class:`Policy`. Unknown → ``TERMINAL``."""
+    text = (error or "").lower()
+    for pattern, policy in _RULES:
+        if re.search(pattern, text):
+            return policy
+    return TERMINAL

loop.py

@@ -28,6 +28,7 @@
 import time
 
 from . import worktree
+from .failures import classify
 from .store import BoardError, escalation_enabled, get_store
 
 log = logging.getLogger("protoagent.plugins.project_board")
@@ -196,6 +197,7 @@ async def _drive(self, feature: dict):
         title = f"feat: {feature['title']}"
         prompt = self._build_prompt(feature)
         tier = store.current_tier(fid) if self.escalation_on else ""
+        retries = 0  # transient-failure retries at the current tier (reset on a climb)
         wt = branch = None
         try:
             while True:
@@ -211,8 +213,25 @@ async def _drive(self, feature: dict):
                     result = await worktree.dispatch_coder(coder, wt, prompt)  # reaps subprocess
                     pr_url = await worktree.open_pr(wt, branch, base=base, title=title, body=(result or "")[:4000])
                 except (worktree.NoChangesError, worktree.WorktreeError) as exc:
-                    # Capability failure (coder errored / no diff) → escalate when a
-                    # ladder exists; an infra failure (push/gh) is not escalable → block.
+                    policy = classify(str(exc))
+                    # 1. Transient infra (rate-limit / network / merge-conflict) → back
+                    #    off and retry the SAME tier. A stronger model can't clear a rate
+                    #    limit, and a re-dispatch off the latest base clears a conflict.
+                    if policy.retryable and retries < policy.max_attempts - 1:
+                        retries += 1
+                        log.info(
+                            "[project_board] %s %s — retry %d/%d in %ss: %s",
+                            fid,
+                            policy.category,
+                            retries + 1,
+                            policy.max_attempts,
+                            policy.base_delay_s,
+                            exc,
+                        )
+                        await asyncio.sleep(policy.base_delay_s)
+                        continue
+                    # 2. Capability failure (coder errored / no diff) + a ladder → climb
+                    #    a model tier (fresh retry budget at the new tier).
                     capability = isinstance(exc, worktree.NoChangesError) or str(exc).startswith(
                         "coder dispatch failed"
                     )
@@ -221,9 +240,11 @@ async def _drive(self, feature: dict):
                         if nxt:
                             log.info("[project_board] %s escalating %s→%s: %s", fid, tier, nxt, exc)
                             tier = nxt
+                            retries = 0
                             continue
-                    log.warning("[project_board] %s blocked: %s", fid, exc)
-                    store.flag_blocked(fid, str(exc))
+                    # 3. Terminal, or retries/ladder exhausted → Blocked.
+                    log.warning("[project_board] %s blocked (%s): %s", fid, policy.category, exc)
+                    store.flag_blocked(fid, f"{policy.category}: {exc}")
                     if wt:
                         await worktree.remove_worktree(repo, wt, branch or "")
                     self._inflight.pop(fid, None)

tests/test_failures.py

@@ -0,0 +1,61 @@
+"""Failure-classification tests — the retry-policy table.
+
+Pure function, so these are exhaustive and cheap: each category's patterns, the
+ordering (first match wins), and the terminal fallback for anything unknown.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from project_board.failures import TERMINAL, classify
+
+
+@pytest.mark.parametrize(
+    "msg,category,retryable",
+    [
+        # rate-limit / capacity
+        ("coder dispatch failed: 429 rate limit exceeded", "rate_limit", True),
+        ("Error: you are being rate-limited, slow down", "rate_limit", True),
+        ("model overloaded, try later", "rate_limit", True),
+        # transient infra
+        ("git push failed: connection reset by peer", "transient", True),
+        ("git fetch origin main timed out after 60s", "transient", True),
+        ("gh pr create failed: 503 service unavailable", "transient", True),
+        ("could not resolve host github.com", "transient", True),
+        # merge / rebase
+        ("git push failed: ! [rejected] (non-fast-forward)", "merge_conflict", True),
+        ("merge failed: CONFLICT in foo.py", "merge_conflict", True),
+        # auth — NOT retryable
+        ("gh pr create failed: 403 forbidden — bad credential", "auth", False),
+        ("not authorized: permission denied", "auth", False),
+        # capability / unknown → terminal
+        ("coder produced no commits vs base — nothing to PR", "terminal", False),
+        ("some weird unexpected explosion", "terminal", False),
+    ],
+)
+def test_classify_categories(msg, category, retryable):
+    p = classify(msg)
+    assert p.category == category
+    assert p.retryable is retryable
+
+
+def test_unknown_falls_back_to_terminal():
+    p = classify("???")
+    assert p is TERMINAL
+    assert p.retryable is False and p.max_attempts == 1
+
+
+def test_empty_message_is_terminal():
+    assert classify("").category == "terminal"
+
+
+def test_first_match_wins():
+    # both "429" (rate_limit) and "connection" (transient) present → rate_limit first.
+    assert classify("429 too many requests on connection").category == "rate_limit"
+
+
+def test_retryable_policies_carry_a_real_budget():
+    for msg in ("429 rate limit", "connection reset", "merge conflict"):
+        p = classify(msg)
+        assert p.retryable and p.max_attempts >= 2 and p.base_delay_s >= 0

tests/test_loop.py

@@ -139,6 +139,57 @@ async def _open_pr(wt, branch, *, base, title, body):
     assert store.names() == ["flag_blocked"]  # blocked before any worktree work
 
 
+# ── _drive: failure classification + backoff (no real sleeps) ───────────────────
+
+
+async def _no_sleep(_delay):
+    return None
+
+
+async def test_drive_retries_a_transient_failure_then_succeeds(monkeypatch):
+    calls = {"n": 0}
+
+    async def _open_pr(wt, branch, *, base, title, body):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise worktree.WorktreeError("git push failed: connection reset by peer")
+        return "https://example/pr/1"
+
+    monkeypatch.setattr("project_board.loop.asyncio.sleep", _no_sleep)
+    loop, store = await _drive_with(monkeypatch, open_pr=_open_pr)
+    assert ("open_review", "bd-1", "https://example/pr/1") in store.calls
+    assert calls["n"] == 2  # one transient retry, then success
+    assert "flag_blocked" not in store.names()
+    assert loop._inflight == {}
+
+
+async def test_drive_blocks_after_exhausting_transient_retries(monkeypatch):
+    calls = {"n": 0}
+
+    async def _open_pr(wt, branch, *, base, title, body):
+        calls["n"] += 1
+        raise worktree.WorktreeError("gh pr create failed: 503 service unavailable")
+
+    monkeypatch.setattr("project_board.loop.asyncio.sleep", _no_sleep)
+    loop, store = await _drive_with(monkeypatch, open_pr=_open_pr)
+    assert "flag_blocked" in store.names()
+    assert calls["n"] == 3  # transient policy = 3 attempts, then Blocked
+    assert loop._inflight == {}
+
+
+async def test_drive_blocks_immediately_on_a_terminal_failure(monkeypatch):
+    calls = {"n": 0}
+
+    async def _open_pr(wt, branch, *, base, title, body):
+        calls["n"] += 1
+        raise worktree.WorktreeError("gh pr create failed: 403 forbidden — bad credential")
+
+    monkeypatch.setattr("project_board.loop.asyncio.sleep", _no_sleep)
+    loop, store = await _drive_with(monkeypatch, open_pr=_open_pr)
+    assert "flag_blocked" in store.names()
+    assert calls["n"] == 1  # auth is terminal → no retry
+
+
 # ── concurrency: _spawn_ready claims up to max_concurrent ────────────────────────