diff --git a/failures.py b/failures.py new file mode 100644 index 0000000..51ad26b --- /dev/null +++ b/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 diff --git a/loop.py b/loop.py index 116195c..e28fc28 100644 --- a/loop.py +++ b/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) diff --git a/tests/test_failures.py b/tests/test_failures.py new file mode 100644 index 0000000..1e3ba24 --- /dev/null +++ b/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 diff --git a/tests/test_loop.py b/tests/test_loop.py index 6cc0e73..1101c7c 100644 --- a/tests/test_loop.py +++ b/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 ────────────────────────