feat: operator-only test-rung route to verify a rung works

README.md

@@ -63,6 +63,14 @@ board to a PR — or fork it as a starting point.
   coder/acceptance/test command ⇒ honest degrade to the single shot; missing
   `coder_solve_fusion_delegate` ⇒ the ladder simply stops at tree-search — see
   `coder_seam.py`.
+- **Rung diagnostic — `POST /api/plugins/project_board/features/{id}/test-rung`**
+  (operator-only, no `@tool` wrapper): runs exactly ONE named rung
+  (`greedy`/`best-of-k`/`tree-search`/`fusion`) against a feature's real acceptance
+  tests, in a throwaway worktree that's ALWAYS reaped — never promoted, no PR, no
+  board state touched. Verifying a specific rung — fusion especially, only
+  otherwise reached after three cheaper rungs fail — shouldn't require contriving a
+  task hard enough to fail its way there. `{"rung": "fusion"}` in the body; `coder`
+  optional (defaults to `project_board.coder`).
 - **Planning layer** — two reasoning subagents (`decompose` + `antagonist`) driven by
   the `decompose-project` skill: idea → outline → MADR ADRs → epics › milestones ›
   features, hardened by an adversary, with a per-epic human gate.
@@ -138,7 +146,9 @@ project_board:
 - **HTTP API** (`/plugins/project_board/*`): `epics`, `milestones`, `features`,
   `features/{id}/{ready,dep,block,unblock,ci}`, and `/webhook/pr` (the Done edge —
   a stable public URL GitHub posts to; ungated so GitHub, which can't send a bearer,
-  reaches it).
+  reaches it). `features/{id}/{cancel,test-rung}` and `DELETE features/{id}` are
+  **operator-only** — no `@tool` wrapper, so the board's own lead agent has no way
+  to call them.
 - **Watch it:** the **Board** console view (left-rail) at
   `/plugins/project_board/board` — Kanban + list, live-refreshing, served by the
   same router as the API (so the declared view path is genuinely mounted).

api.py

@@ -242,4 +242,84 @@ async def _delete(fid: str, body: dict = Body(default={})):
         cancel to keep a visible, reopenable audit lane; use delete to leave no trace."""
         return _guard(lambda: store().delete_feature(fid, str((body or {}).get("reason", ""))))
 
+    # ── coder.solve() rung diagnostic (ADR 0064) — OPERATOR ONLY, deliberately no
+    #    @tool wrapper: same boundary this router already draws around cancel/
+    #    block/delete — the board's own lead agent has no tool to reach this.
+    @router.post("/features/{fid}/test-rung")
+    async def _test_rung(fid: str, body: dict = Body(...)):
+        """Run exactly ONE named rung of coder.solve() against this feature's REAL
+        acceptance tests, in a throwaway worktree that's ALWAYS reaped — never
+        promoted, no PR opened, no board state touched. For verifying a rung
+        actually works (fusion especially — otherwise only reached after three
+        cheaper rungs fail) without contriving a task that fails its way there.
+
+        Body: ``{"rung": "greedy"|"best-of-k"|"tree-search"|"fusion", "coder": "<delegate
+        name>"}`` (``coder`` optional, defaults to ``project_board.coder``)."""
+        rung = str(body.get("rung", "")).strip()
+        if rung not in ("greedy", "best-of-k", "tree-search", "fusion"):
+            raise HTTPException(400, "rung must be one of: greedy, best-of-k, tree-search, fusion")
+
+        f = _guard(lambda: store().get_feature(fid))
+        if f is None:
+            raise HTTPException(404, f"unknown feature {fid!r}")
+        if not str(f.get("acceptance_criteria") or "").strip():
+            raise HTTPException(400, f"feature {fid!r} has no acceptance_criteria — nothing to verify a rung against")
+
+        from . import coder_seam
+
+        if coder_seam._import_solve() is None:
+            raise HTTPException(400, "the `coder` plugin isn't installed/enabled on this host")
+
+        test_cmd = (
+            str((cfg or {}).get("coder_solve_test_cmd") or "").strip()
+            or str((cfg or {}).get("local_gate_cmd") or "").strip()
+        )
+        if not test_cmd:
+            raise HTTPException(400, "no coder_solve_test_cmd or local_gate_cmd configured — nothing to run tests with")
+
+        coder_name = str(body.get("coder") or (cfg or {}).get("coder", "proto"))
+        coder = coder_seam.resolve_delegate(coder_name, "acp")
+        if coder is None:
+            raise HTTPException(400, f"acp delegate {coder_name!r} not found — check `delegates:`")
+
+        fusion_delegate = None
+        if rung == "fusion":
+            fusion_name = str((cfg or {}).get("coder_solve_fusion_delegate") or "").strip()
+            if not fusion_name:
+                raise HTTPException(400, "rung='fusion' requires project_board.coder_solve_fusion_delegate")
+            fusion_delegate = coder_seam.resolve_delegate(fusion_name, "openai")
+            if fusion_delegate is None:
+                raise HTTPException(400, f"openai delegate {fusion_name!r} not found — check `delegates:`")
+
+        task = (
+            f"# {f.get('title', '')}\n\n"
+            f"## Task\n{f.get('spec', '')}\n\n"
+            f"## Files to create / modify\n"
+            + ("\n".join(f"- {p}" for p in (f.get("files_to_modify") or [])) or "(none listed)")
+            + f"\n\n## Acceptance criteria (definition of done)\n{f.get('acceptance_criteria', '')}\n"
+        )
+
+        try:
+            result = await coder_seam.test_rung(
+                rung=rung,
+                task=task,
+                coder=coder,
+                repo=(cfg or {}).get("repo", "."),
+                base=(cfg or {}).get("base_branch", "main"),
+                root=(cfg or {}).get("worktrees_root", ".worktrees"),
+                fid=fid,
+                dispatch_timeout=float((cfg or {}).get("coder_timeout_s", 1800)) or None,
+                test_cmd=test_cmd,
+                test_timeout=float((cfg or {}).get("coder_solve_test_timeout_s", 300)),
+                budget=max(1, int((cfg or {}).get("coder_solve_budget", 6))),
+                k=max(1, int((cfg or {}).get("coder_solve_k", 3))),
+                tree_depth=max(0, int((cfg or {}).get("coder_solve_tree_depth", 2))),
+                fusion_delegate=fusion_delegate,
+                fusion_k=max(1, int((cfg or {}).get("coder_solve_fusion_k", 2))),
+                files_to_modify=f.get("files_to_modify") or [],
+            )
+        except Exception as exc:  # noqa: BLE001 — surface as a 400, not a raw 500
+            raise HTTPException(400, f"test-rung failed: {exc}") from exc
+        return result
+
     return router

coder_seam.py

@@ -46,7 +46,15 @@
 context lines is a common failure mode `git apply` doesn't forgive. Only reached
 when a ``fusion_delegate`` is configured (an ``openai``-type Delegate, already
 resolved by the caller) — absent that, ``solve()`` gets ``fusion_generate=None``
-and stops at tree-search exactly as before (honest degrade, unchanged)."""
+and stops at tree-search exactly as before (honest degrade, unchanged).
+
+**``test_rung`` (operator-only diagnostic).** Verifying a specific rung — fusion
+especially, only otherwise reached after three cheaper rungs fail — shouldn't
+require contriving a task hard enough to fail its way there. ``test_rung`` runs
+ONE named rung once against a feature's real acceptance tests in a throwaway
+worktree that's ALWAYS reaped, win or lose — never promoted, no PR. Exposed via
+api.py's ``test-rung`` route with no ``@tool`` wrapper, so it's operator-only,
+not something the board's own lead agent can reach for itself."""
 
 from __future__ import annotations
 
@@ -61,6 +69,24 @@
 
 log = logging.getLogger("protoagent.plugins.project_board")
 
+
+def resolve_delegate(name: str, expect_type: str):
+    """Look up a live delegate by name from the delegates registry. Returns the
+    Delegate or None (not configured / wrong type / plugin disabled). Shared by
+    ``loop.py`` (coder/reviewer/fusion resolution in the real dispatch path) and
+    ``api.py`` (the operator-only test-rung route) — one lookup, not two copies."""
+    try:
+        from plugins.delegates.registry import DelegateRegistry
+        from plugins.delegates.store import merged_delegates
+
+        d = DelegateRegistry(merged_delegates()).get(name)
+    except Exception:  # noqa: BLE001 — delegates plugin may be disabled
+        return None
+    if d is None or d.type != expect_type:
+        return None
+    return d
+
+
 # ``### path/to/file.py`` header, then a fenced block (any/no language hint) holding
 # that file's COMPLETE new content. Deliberately simple/strict: a fusion completion
 # that doesn't follow the format parses to no files, which just fails verify() like
@@ -475,3 +501,85 @@ async def dispatch(
     )
     result_text = f"[coder.solve rung={result.rung} gens={result.gens_spent}] {result.note}"
     return canon_wt, canon_branch, result_text
+
+
+async def test_rung(
+    *,
+    rung: str,
+    task: str,
+    coder,
+    repo: str,
+    base: str,
+    root: str,
+    fid: str,
+    dispatch_timeout: float | None,
+    test_cmd: str,
+    test_timeout: float,
+    budget: int = 10,
+    k: int = 3,
+    tree_depth: int = 2,
+    fusion_delegate=None,
+    fusion_k: int = 2,
+    files_to_modify: list[str] | None = None,
+    _solve=None,
+    _budget_cls=None,
+    _verdict_cls=None,
+    _fusion_dispatch=None,
+) -> dict:
+    """Operator-only diagnostic (ADR 0064): run exactly ONE named rung of
+    ``coder.solve()`` against a feature's REAL acceptance tests, in a throwaway
+    worktree that is ALWAYS reaped afterward — never promoted, no PR opened, no
+    board state touched. For verifying a rung actually works (especially fusion,
+    only otherwise reached after three cheaper rungs fail) without contriving a
+    task hard enough to fail its way there.
+
+    Deliberately separate from ``dispatch()``: that function's contract (promote
+    the winner, raise ``SolveExhausted`` on exhaustion) is shaped for the board's
+    real per-feature build — mixing test semantics into it would risk the real
+    dispatch path. This is exposed to operators only via api.py's ``test-rung``
+    route, which carries NO ``@tool`` wrapper — the board's own lead agent has no
+    way to call this itself (see api.py's docstring for the same boundary the
+    plugin already draws around ``/features/{id}/cancel`` etc.)."""
+    if _solve is not None:
+        solve, Budget, Verdict = _solve, _budget_cls, _verdict_cls
+    else:
+        from plugins.coder.solve import Budget, Verdict, solve
+
+    adapter = _WorktreeSolveAdapter(
+        repo=repo,
+        base=base,
+        root=root,
+        fid=f"{fid}.test",
+        coder=coder,
+        dispatch_timeout=dispatch_timeout,
+        test_cmd=test_cmd,
+        test_timeout=test_timeout,
+        verdict_cls=Verdict,
+        fusion_delegate=fusion_delegate,
+        files_to_modify=files_to_modify,
+        _fusion_dispatch=_fusion_dispatch,
+    )
+    try:
+        result = await solve(
+            task,
+            generate=adapter.generate,
+            verify=adapter.verify,
+            budget=Budget(budget),
+            k=k,
+            tree_depth=tree_depth,
+            fusion_generate=adapter.generate_fusion if fusion_delegate is not None else None,
+            fusion_k=fusion_k,
+            force_rung=rung,
+        )
+    finally:
+        # ALWAYS reap — pass or fail, this is a diagnostic run, never a real build.
+        for wt, branch in adapter.candidates:
+            await worktree.remove_worktree(repo, wt, branch)
+    return {
+        "rung": result.rung,
+        "passed": result.passed,
+        "gens_spent": result.gens_spent,
+        "candidates_tried": result.candidates_tried,
+        "note": result.note,
+        "verdict_output": result.verdict.output if result.verdict else "",
+    }

loop.py

@@ -891,17 +891,10 @@ def _use_coder_solve(self, feature: dict) -> bool:
 
     def _resolve_delegate(self, name: str, expect_type: str):
         """Look up a live delegate by name from the delegates registry. Returns the
-        Delegate or None (not configured / wrong type / plugin disabled)."""
-        try:
-            from plugins.delegates.registry import DelegateRegistry
-            from plugins.delegates.store import merged_delegates
-
-            d = DelegateRegistry(merged_delegates()).get(name)
-        except Exception:  # noqa: BLE001 — delegates plugin may be disabled
-            return None
-        if d is None or d.type != expect_type:
-            return None
-        return d
+        Delegate or None (not configured / wrong type / plugin disabled). Thin
+        wrapper — the real lookup is shared with api.py's test-rung route via
+        ``coder_seam.resolve_delegate``."""
+        return coder_seam.resolve_delegate(name, expect_type)
 
     async def _run_fixups(self, wt: str) -> None:
         """Run the repo's auto-fix command (``format_cmd``, e.g.

protoagent.plugin.yaml

@@ -1,6 +1,6 @@
 id: project_board
 name: Project Board (coding orchestration)
-version: 0.28.0
+version: 0.29.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

pyproject.toml

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