feat(loop): fail-closed review gate + delta re-reviews (ADR 0078 Phase A2, v0.32.0)

loop.py

@@ -36,6 +36,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 import re
 import time
@@ -138,6 +139,10 @@ def __init__(self, cfg: dict):
         self.review_gate = bool(self.cfg.get("review_gate", False))
         self.review_workflow = str(self.cfg.get("review_workflow", "code-review")).strip() or "code-review"
         self.review_fix_max = max(0, int(self.cfg.get("review_fix_max", 2)))
+        # Cap on consecutive UNRUNNABLE gate attempts (runner missing / workflow dying /
+        # panel steps failing) before escalating to the operator via flag_blocked —
+        # fail closed without re-burning the workflow every poll forever (ADR 0078 D3).
+        self.review_run_max = max(1, int(self.cfg.get("review_run_max", 3)))
         # Goal-verification gate (OPT-IN, default off). When on, a DETERMINISTIC pre-PR
         # check (no LLM, no diff dump): a code change must ship a test — CI runs tests but
         # can't require their presence, so the gate does. A miss → re-dispatch/escalate
@@ -320,6 +325,15 @@ def __init__(self, cfg: dict):
         # Review-gate bounce re-dispatches so far (fid → count), same-tier — the
         # review sibling of _ci_fix_attempts (plan M5).
         self._review_fix_attempts: dict[str, int] = {}
+        # Consecutive review runs that could not complete (panel step failed / no
+        # runner) — after review_run_max the feature is Blocked for the operator
+        # instead of re-burning the workflow every poll (ADR 0078 D3: fail closed,
+        # escalate; never judge from a partial panel).
+        self._review_run_failures: dict[str, int] = {}
+        # Last parsed findings JSON per fid — fed back as the recipe's
+        # prior_findings input so a bounce re-review is a DELTA review
+        # (GitHub-native review memory, ADR 0078 D5).
+        self._review_prior: dict[str, str] = {}
 
     def _store(self):
         return get_store(**self._store_kw)
@@ -533,6 +547,8 @@ async def _reconcile_prs(self):
                         self._ci_fix_attempts.pop(fid, None)
                         self._rebase_attempts.pop(fid, None)
                         self._review_fix_attempts.pop(fid, None)
+                        self._review_run_failures.pop(fid, None)
+                        self._review_prior.pop(fid, None)
                         # A merge with the gate still unhappy is a human override —
                         # reality wins, but it must be visible, not silent.
                         if self.review_gate and LABEL_CHANGES_REQUESTED in (f.get("labels") or []):
@@ -550,6 +566,8 @@ async def _reconcile_prs(self):
                     self._ci_fix_attempts.pop(fid, None)
                     self._rebase_attempts.pop(fid, None)
                     self._review_fix_attempts.pop(fid, None)
+                    self._review_run_failures.pop(fid, None)
+                    self._review_prior.pop(fid, None)
                     log.info("[project_board] reconcile → blocked (PR closed): %s (%s)", fid, pr_url)
                 elif state == "OPEN":
                     # Keep a stale/conflicting PR mergeable BEFORE the CI reconcile: a
@@ -974,18 +992,46 @@ async def _review_gate(self, store, fid: str, pr_url: str, repo: str) -> None:
         store.set_review_substate(fid, LABEL_REVIEW_PENDING)
         output = await self._run_review_workflow(fid, pr_url)
         if output is None:
-            # Could not review (no runner + no reviewer, or the run itself died).
-            # Leave review-pending set — the PR reconcile retries the gate next poll,
-            # so a transient host hiccup doesn't quietly demote the gate to advisory.
-            log.warning("[project_board] %s review gate could not run — will retry on the next poll", fid)
+            # Could not review (no runner + no reviewer, a dead run, or a PARTIAL
+            # panel — a failed finder step is not a review; judging from it is how
+            # an unreviewed PR gets promoted, ADR 0078 D3). Leave review-pending so
+            # the PR reconcile retries next poll — but bounded: a persistently
+            # unrunnable gate escalates to the operator instead of re-burning the
+            # workflow every poll forever.
+            n = self._review_run_failures.get(fid, 0) + 1
+            self._review_run_failures[fid] = n
+            if n >= self.review_run_max:
+                store.set_review_substate(fid, None)
+                store.flag_blocked(
+                    fid,
+                    f"review gate could not complete after {n} attempt(s) (runner missing, "
+                    f"workflow dying, or panel steps failing) — needs operator attention: {pr_url}",
+                )
+                self._review_run_failures.pop(fid, None)
+                log.warning("[project_board] %s blocked (review gate unrunnable %d times)", fid, n)
+                return
+            log.warning(
+                "[project_board] %s review gate could not run (%d/%d) — will retry on the next poll",
+                fid,
+                n,
+                self.review_run_max,
+            )
             return
+        self._review_run_failures.pop(fid, None)
         findings = self._parse_findings(output)
         if findings is None:
             # Host predates the findings convention (ADR 0077) — the gate can't
             # judge, so it must not pretend to. Record and leave in review.
             store.set_review_substate(fid, None, note="review gate: host lacks graph.review.findings — gate inert")
             log.warning("[project_board] %s review gate inert (no findings parser on this host)", fid)
             return
+        # Remember this round's findings — the next run (a bounce re-review) passes
+        # them back as the recipe's prior_findings input, making it a DELTA review
+        # (drop fixed, carry still-open) instead of a from-scratch re-litigation.
+        try:
+            self._review_prior[fid] = json.dumps([f.to_dict() for f in findings]) if findings else ""
+        except Exception:  # noqa: BLE001 — memory is an optimization, never a gate failure
+            self._review_prior.pop(fid, None)
         blocking = [f for f in findings if f.verdict != "refuted" and f.severity in ("blocker", "major")]
         if not blocking:
             store.set_review_substate(
@@ -1044,7 +1090,23 @@ async def _run_review_workflow(self, fid: str, pr_url: str) -> str | None:
             runner = None
         if runner is not None and number:
             try:
-                result = await runner(self.review_workflow, {"pr": number, "repo": repo_slug})
+                inputs: dict = {"pr": number, "repo": repo_slug}
+                prior = self._review_prior.get(fid)
+                if prior:
+                    inputs["prior_findings"] = prior
+                result = await runner(self.review_workflow, inputs)
+                failed = list((result or {}).get("failed") or [])
+                if failed:
+                    # A partial panel is NOT a review (ADR 0078 D3): a starved/errored
+                    # finder means unreviewed angles, and a verdict synthesized from
+                    # the survivors reads as clean coverage it never had.
+                    log.warning(
+                        "[project_board] %s review workflow %r had failed step(s) %s — fail closed, not a review",
+                        fid,
+                        self.review_workflow,
+                        failed,
+                    )
+                    return None
                 return str((result or {}).get("output") or "") or None
             except Exception as exc:  # noqa: BLE001 — a dead workflow ≠ a dead loop
                 log.warning("[project_board] %s review workflow %r failed: %s", fid, self.review_workflow, exc)

protoagent.plugin.yaml

@@ -1,6 +1,6 @@
 id: project_board
 name: Project Board (coding orchestration)
-version: 0.31.0
+version: 0.32.0
 description: >-
   A board-driven coding-orchestration plugin: a lean 6-state board (backlog → ready
   → in_progress → in_review → done, + a blocked flag) backed by **beads** (`br`), an
@@ -48,6 +48,9 @@ config:
   review_fix_max: 2         # review-bounce re-dispatches before the feature is Blocked
                             # (mirrors ci_fix_max; same-tier — findings are fixable, not a
                             # model-capability ceiling)
+  review_run_max: 3         # consecutive UNRUNNABLE gate attempts (no runner / dead
+                            # workflow / failed panel steps — a partial panel is NOT a
+                            # review, ADR 0078 D3) before Blocking for the operator
   goal_verify: false        # OPT-IN. On = an independent model call checks the coder's diff
                             # against the feature's acceptance_criteria (test coverage is
                             # mandatory even when unlisted) BEFORE opening the PR.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "project-board"
-version = "0.31.0"
+version = "0.32.0"
 description = "Board-driven coding-orchestration plugin for protoAgent (beads board + ACP spawn loop + planning layer + console view)."
 requires-python = ">=3.11"
 

tests/test_loop.py

@@ -1801,6 +1801,11 @@ class _Finding:
         verdict: str = ""
         note: str = field(default="")
 
+        def to_dict(self):
+            from dataclasses import asdict
+
+            return asdict(self)
+
     def parse_findings(text):
         start, end = text.find("["), text.rfind("]")
         if start == -1 or end <= start:
@@ -1924,3 +1929,86 @@ def test_parse_pr_url():
         "protoLabsAI/protoContent",
     )
     assert _parse_pr_url("https://example.com/not-a-pr") == ("", "")
+
+
+# ── fail-closed gate + delta re-review carry (ADR 0078 Phase A2) ─────────────────
+
+
+def test_review_gate_config_run_max():
+    assert BoardLoop({}).review_run_max == 3
+    assert BoardLoop({"review_run_max": 0}).review_run_max == 1  # floor: at least one try
+
+
+async def test_review_gate_partial_panel_is_not_a_review(monkeypatch):
+    """A workflow result with failed steps must NOT be judged — the gate treats it
+    as unreviewed (fail closed): review-pending stays, no requeue, no block."""
+    _inject_fake_findings(monkeypatch)
+    import sys as _sys
+    import types as _types
+
+    calls = []
+
+    async def _runner(name, inputs):
+        calls.append(inputs)
+        return {"output": "clean.\n```json\n[]\n```", "steps": {}, "failed": ["find_crossfile"]}
+
+    rt = _types.ModuleType("runtime")
+    rt_state = _types.ModuleType("runtime.state")
+    rt_state.STATE = _types.SimpleNamespace(workflow_run=_runner)
+    rt.state = rt_state
+    monkeypatch.setitem(_sys.modules, "runtime", rt)
+    monkeypatch.setitem(_sys.modules, "runtime.state", rt_state)
+
+    store = _GateStore()
+    loop = BoardLoop({"review_gate": True})
+    monkeypatch.setattr(loop, "_resolve_delegate", lambda n, t: None)  # no reviewer fallback
+    await loop._review_gate(store, "bd-1", "https://github.com/o/r/pull/9", "/repo")
+    assert calls, "the runner must have been invoked"
+    # Fail closed: the clean-looking partial output was NOT judged.
+    assert store.review_states == [("review-pending", "")]
+    assert ("requeue", "bd-1") not in store.calls
+    assert not any(c[0] == "flag_blocked" for c in store.calls)
+    assert loop._review_run_failures["bd-1"] == 1
+
+
+async def test_review_gate_unrunnable_escalates_after_run_max(monkeypatch):
+    _inject_fake_findings(monkeypatch)
+    store = _GateStore()
+    loop = _gate_loop(monkeypatch, None, cfg={"review_run_max": 2})
+    loop._review_run_failures["bd-1"] = 1  # one prior unrunnable attempt
+    await loop._review_gate(store, "bd-1", "https://github.com/o/r/pull/9", "/repo")
+    blocked = [c for c in store.calls if c[0] == "flag_blocked"]
+    assert blocked and "operator attention" in blocked[0][2]
+    assert "bd-1" not in loop._review_run_failures
+
+
+async def test_review_gate_passes_prior_findings_on_the_next_run(monkeypatch):
+    """Round 1 findings ride into round 2's workflow inputs (delta re-review)."""
+    _inject_fake_findings(monkeypatch)
+    import sys as _sys
+    import types as _types
+
+    seen_inputs = []
+
+    async def _runner(name, inputs):
+        seen_inputs.append(dict(inputs))
+        return {"output": f"```json\n{_BLOCKER}\n```", "steps": {}, "failed": []}
+
+    rt = _types.ModuleType("runtime")
+    rt_state = _types.ModuleType("runtime.state")
+    rt_state.STATE = _types.SimpleNamespace(workflow_run=_runner)
+    rt.state = rt_state
+    monkeypatch.setitem(_sys.modules, "runtime", rt)
+    monkeypatch.setitem(_sys.modules, "runtime.state", rt_state)
+
+    async def _diff(pr_url, cwd="."):
+        return "diff --git a/x b/x"
+
+    monkeypatch.setattr(worktree, "pr_diff", _diff)
+    store = _GateStore()
+    loop = BoardLoop({"review_gate": True, "review_fix_max": 5})
+    await loop._review_gate(store, "bd-1", "https://github.com/o/r/pull/9", "/repo")
+    assert "prior_findings" not in seen_inputs[0]  # first pass — nothing to carry
+    await loop._review_gate(store, "bd-1", "https://github.com/o/r/pull/9", "/repo")
+    assert "prior_findings" in seen_inputs[1]
+    assert "drops data" in seen_inputs[1]["prior_findings"]