feat(loop): Max-Mode N-parallel coders — build N ways, judge, ship the best (v0.20.0) (#51)

The best-of-N judge (`_judge_candidates`) landed in #22; this wires the N-parallel
dispatch that feeds it. When `max_mode_n > 1`, a from-scratch build fans out N coders
on the same feature instead of one:

  • create N throwaway candidate worktrees (`feat-<id>.c<k>`) off the same base
  • dispatch the coder into ALL of them concurrently (asyncio.gather), each keeping
    its own coder_timeout watchdog + finally teardown; return_exceptions so one
    candidate timing out/erroring leaves an empty tree the judge skips, not a dead batch
  • `_judge_candidates` picks the winning diff
  • PROMOTE the winner into the canonical `feat-<id>` worktree / `feat/<id>` branch
    (git worktree move + branch -m, uncommitted changes preserved) so the rest of the
    lifecycle — CI-fail bounce, crash recovery, reaping — is unchanged; reap the losers
  • all-empty → NoChangesError, which the loop escalates/blocks like a single coder
    that produced nothing

Gating: only a FROM-SCRATCH build fans out. A carried-forward re-dispatch (CI bounce /
goal-fix / gate-fix — all signalled by _ci_feedback) FIXES the existing diff with one
coder, so it must not re-fan-out N. keep_wt fix retries stay single-candidate too.

Default off (max_mode_n=1). Bound N × max_concurrent to the host. Isolation stays on
git worktrees (container-use deferred to its own platform ADR).

Closes #21

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mabry1985

loop.py

@@ -604,20 +604,31 @@ async def _drive(self, feature: dict):
                 if coder is None:
                     store.flag_blocked(fid, f"coder delegate {coder_name!r} not configured/enabled")
                     return
-                # A goal-fix retry REUSES the worktree — the implementation is already
-                # there, the coder just adds what the reviewer flagged (usually tests).
-                # Rebuilding on a fresh worktree throws the impl away (the coder then
-                # spends its budget re-implementing and never reaches the tests — the
-                # bd-2fd/bd-3cj block). Otherwise: a fresh worktree per attempt.
-                if keep_wt and wt is not None:
-                    keep_wt = False  # consume the reuse
-                else:
-                    wt, branch = await worktree.create_worktree(repo, base, fid, self.root)
-                self._inflight[fid] = (repo, wt, branch)  # track for shutdown reaping
                 try:
-                    result = await worktree.dispatch_coder(
-                        coder, wt, prompt, timeout=self.coder_timeout or None
-                    )  # reaps subprocess; CoderTimeout if it overruns
+                    # How this attempt gets its worktree + coder result:
+                    #  • keep_wt  → REUSE the kept worktree (impl intact), one re-dispatch.
+                    #    A goal-fix/gate-fix retry must not throw the implementation away —
+                    #    the coder only ADDS what the reviewer flagged (usually tests); a
+                    #    fresh rebuild makes it re-implement and never reach the tests (the
+                    #    bd-2fd/bd-3cj block).
+                    #  • max-mode → N parallel candidates, judge, promote the winner (#21).
+                    #    ONLY for a from-scratch build: a carried-forward re-dispatch (a CI
+                    #    bounce / goal-fix / gate-fix — all signalled by _ci_feedback) FIXES
+                    #    the existing diff with one coder, so it must NOT re-fan-out N.
+                    #  • otherwise → one fresh worktree, one dispatch.
+                    if keep_wt and wt is not None:
+                        keep_wt = False  # consume the reuse
+                        self._inflight[fid] = (repo, wt, branch)
+                        result = await worktree.dispatch_coder(coder, wt, prompt, timeout=self.coder_timeout or None)
+                    elif self.max_mode_n > 1 and not self._ci_feedback.get(fid):
+                        wt, branch, result = await self._dispatch_max_mode(feature, coder, prompt, repo, base, fid)
+                        self._inflight[fid] = (repo, wt, branch)
+                    else:
+                        wt, branch = await worktree.create_worktree(repo, base, fid, self.root)
+                        self._inflight[fid] = (repo, wt, branch)  # track for shutdown reaping
+                        result = await worktree.dispatch_coder(
+                            coder, wt, prompt, timeout=self.coder_timeout or None
+                        )  # reaps subprocess; CoderTimeout if it overruns
                     # Goal-verification gate: confirm the diff meets the acceptance
                     # criteria before opening a PR. A gap is a capability failure (the
                     # coder didn't deliver) → escalate/block, don't open the PR.
@@ -940,6 +951,54 @@ async def _judge_candidates(self, feature: dict, base: str, worktrees: list[str]
                 return idx
         return nonempty[0]  # judge unclear → first non-empty candidate
 
+    async def _dispatch_max_mode(
+        self, feature: dict, coder, prompt: str, repo: str, base: str, fid: str
+    ) -> tuple[str, str, str]:
+        """Max-Mode (#21): build the feature N ways in parallel and ship the best diff.
+
+        Creates ``max_mode_n`` throwaway candidate worktrees off the same base (suffixed
+        ``feat-<id>.c<k>`` so none collides with the canonical name), dispatches the coder
+        into ALL of them concurrently — each keeps its own ``coder_timeout`` watchdog +
+        ``finally`` subprocess teardown (``dispatch_coder``), and ``return_exceptions``
+        means one candidate timing out / erroring leaves an empty tree the judge skips
+        rather than sinking the batch. ``_judge_candidates`` (the best-of-N judge that
+        landed in #22) picks the winning index; the winner is PROMOTED into the canonical
+        ``feat-<id>`` worktree / ``feat/<id>`` branch (so the rest of the lifecycle is
+        unchanged) and the losers are reaped. All-empty → ``NoChangesError``, which
+        ``_drive`` escalates/blocks exactly like a single coder that produced nothing.
+
+        Returns (canonical_wt, canonical_branch, winner_reply). The fan-out is bounded by
+        ``max_concurrent`` × ``max_mode_n`` coders; size those to the host."""
+        n = self.max_mode_n
+        cand_ids = [f"{fid}.c{i}" for i in range(n)]
+        # Create the N worktrees sequentially (git serializes worktree-list writes); the
+        # slow part — the coder dispatch — is what we then fan out in parallel.
+        cands: list[tuple[str, str]] = []
+        for cid in cand_ids:
+            cands.append(await worktree.create_worktree(repo, base, cid, self.root))
+        log.info("[project_board] %s max-mode: dispatching %d parallel candidates", fid, n)
+        results = await asyncio.gather(
+            *(worktree.dispatch_coder(coder, wt, prompt, timeout=self.coder_timeout or None) for wt, _b in cands),
+            return_exceptions=True,
+        )
+        for i, r in enumerate(results):
+            if isinstance(r, Exception):
+                log.info("[project_board] %s max-mode candidate %d failed (skipped): %s", fid, i, r)
+        idx = await self._judge_candidates(feature, base, [wt for wt, _b in cands])
+        if idx is None:
+            for cid in cand_ids:
+                await worktree.reap_feature_worktree(repo, self.root, cid)
+            raise worktree.NoChangesError(f"max-mode: all {n} candidates produced no diff")
+        log.info("[project_board] %s max-mode: candidate %d/%d wins → promoting", fid, idx, n)
+        win_wt, win_branch = cands[idx]
+        canon_wt, canon_branch = await worktree.promote_worktree(repo, win_wt, win_branch, fid, self.root)
+        # Reap the losers (the winner was moved out of its candidate name by promote).
+        for i, cid in enumerate(cand_ids):
+            if i != idx:
+                await worktree.reap_feature_worktree(repo, self.root, cid)
+        winner_reply = results[idx] if not isinstance(results[idx], Exception) else ""
+        return canon_wt, canon_branch, winner_reply
+
     async def _fetch_kg_lessons(self, feature: dict) -> str:
         """Query the knowledge graph (via graph.sdk) for lessons relevant to THIS
         feature — the read half of the flywheel. Builds the query from the feature's

protoagent.plugin.yaml

@@ -1,6 +1,6 @@
 id: project_board
 name: Project Board (coding orchestration)
-version: 0.19.1
+version: 0.20.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
@@ -53,10 +53,13 @@ config:
                             # opens ANYWAY (CI + ci_fix_max stay the backstop — a flaky or
                             # misconfigured gate never blocks good work).
   local_gate_timeout_s: 600 # hard cap on one gate run; a timeout degrades to "pass" (CI gates).
-  max_mode_n: 1             # best-of-N "Max-Mode" (borrowed from MiMo-Code). 1 = off. >1 runs
-                            # N coders on the same feature; a low-temp judge picks the diff that
-                            # best satisfies acceptance_criteria. The N-parallel dispatch is WIP
-                            # (#21); the judge (_judge_candidates) ships + is unit-tested now.
+  max_mode_n: 1             # best-of-N "Max-Mode" (borrowed from MiMo-Code). 1 = off. >1 builds a
+                            # feature N ways in parallel (N candidate worktrees, all dispatched at
+                            # once), a low-temp judge picks the diff that best satisfies
+                            # acceptance_criteria, the winner is promoted to the canonical branch
+                            # and the losers reaped (#21). Costs N× tokens — gate it to hard work;
+                            # only a FROM-SCRATCH build fans out (a CI/goal-fix re-dispatch fixes
+                            # the existing diff with one coder). Bound N × max_concurrent to the host.
   repo: "."                 # repo the coders work in (worktrees branch off it)
   base_branch: main         # base branch worktrees/PRs target
   worktrees_root: .worktrees

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "project-board"
-version = "0.19.1"
+version = "0.20.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

@@ -70,6 +70,12 @@ def test_max_concurrent_floors_at_one():
     assert BoardLoop({"max_concurrent": 4}).max_concurrent == 4
 
 
+def test_max_mode_n_parsing():
+    assert BoardLoop({}).max_mode_n == 1  # off by default
+    assert BoardLoop({"max_mode_n": 5}).max_mode_n == 5
+    assert BoardLoop({"max_mode_n": 0}).max_mode_n == 1  # floors at 1 (never < 1)
+
+
 # ── the coder prompt (ProtoMaker discipline: name the files, demand the diff) ────
 
 
@@ -163,12 +169,19 @@ async def _boom(*a, **k):
 # ── _drive: the state machine ───────────────────────────────────────────────────
 
 
-async def _drive_with(monkeypatch, *, open_pr, coder=object(), dispatch=None, cfg=None, gate=None):
+async def _drive_with(
+    monkeypatch, *, open_pr, coder=object(), dispatch=None, cfg=None, gate=None, judge=None, seed=None
+):
     """Run _drive over FEATURE with the worktree helpers + delegate stubbed.
-    Returns the FakeLoopStore so the test can assert the recorded transitions."""
+    Returns the FakeLoopStore so the test can assert the recorded transitions.
+
+    ``judge`` stubs ``_judge_candidates`` (Max-Mode best-of-N); ``seed`` is a callable
+    run on the loop before the drive (e.g. to pre-seed _ci_feedback for a CI-bounce test)."""
     store = FakeLoopStore()
     store.creates = []  # fids create_worktree was called for (a goal-fix retry reuses, so won't re-create)
     store.removes = []  # worktrees remove_worktree was called for
+    store.reaps = []  # fids reap_feature_worktree was called for (Max-Mode loser teardown)
+    store.promotes = []  # (src_wt, src_branch, fid) the Max-Mode winner was promoted with
     monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store)
 
     async def _create(repo, base, fid, root):
@@ -182,15 +195,28 @@ async def _remove(repo, wt, branch=""):
         store.removes.append(wt)
         return None
 
+    async def _reap(repo, root, fid):
+        store.reaps.append(fid)
+
+    async def _promote(repo, src_wt, src_branch, fid, root=".worktrees"):
+        store.promotes.append((src_wt, src_branch, fid))
+        return ("/wt/feat-" + fid, "feat/" + fid)
+
     monkeypatch.setattr(worktree, "create_worktree", _create)
     monkeypatch.setattr(worktree, "dispatch_coder", dispatch or _default_dispatch)
     monkeypatch.setattr(worktree, "open_pr", open_pr)
     monkeypatch.setattr(worktree, "remove_worktree", _remove)
+    monkeypatch.setattr(worktree, "reap_feature_worktree", _reap)
+    monkeypatch.setattr(worktree, "promote_worktree", _promote)
 
     loop = BoardLoop(cfg or {"coder": "proto"})
     monkeypatch.setattr(loop, "_resolve_delegate", lambda name, expect: coder)
     if gate is not None:
         monkeypatch.setattr(loop, "_run_local_gate", gate)
+    if judge is not None:
+        monkeypatch.setattr(loop, "_judge_candidates", judge)
+    if seed is not None:
+        seed(loop)
     await loop._drive(FEATURE)
     return loop, store
 
@@ -204,6 +230,91 @@ async def _open_pr(wt, branch, *, base, title, body):
     assert loop._inflight == {}  # a completed drive leaves nothing to reap
 
 
+# ── Max-Mode: N parallel candidates → judge → promote winner → ship (#21) ────────
+
+
+async def test_drive_max_mode_fans_out_and_ships_the_winner(monkeypatch):
+    """max_mode_n=3 → 3 candidate worktrees built + dispatched in parallel, the judge
+    picks one, the winner is promoted to the canonical name, the losers are reaped, and
+    ONLY the winner's PR opens (on the canonical branch)."""
+    opened = []
+
+    async def _open_pr(wt, branch, *, base, title, body):
+        opened.append((wt, branch))
+        return "https://example/pr/7"
+
+    dispatched = []
+
+    async def _dispatch(c, wt, prompt, *, timeout=None):
+        dispatched.append(wt)
+        return f"reply from {wt}"
+
+    async def _judge(feature, base, worktrees):
+        assert len(worktrees) == 3  # the judge sees every candidate
+        return 2  # candidate index 2 wins
+
+    loop, store = await _drive_with(
+        monkeypatch,
+        open_pr=_open_pr,
+        dispatch=_dispatch,
+        judge=_judge,
+        cfg={"coder": "proto", "max_mode_n": 3},
+    )
+    # Three candidate worktrees, suffixed so none collides with the canonical name.
+    assert store.creates == ["bd-1.c0", "bd-1.c1", "bd-1.c2"]
+    assert len(dispatched) == 3  # all three coders ran
+    # The winner (c2) is promoted to canonical; the two losers are reaped (winner is not).
+    assert store.promotes == [("/wt/feat-bd-1.c2", "feat/bd-1.c2", "bd-1")]
+    assert set(store.reaps) == {"bd-1.c0", "bd-1.c1"}
+    # Only the winner's PR opens, on the canonical branch.
+    assert opened == [("/wt/feat-bd-1", "feat/bd-1")]
+    assert ("open_review", "bd-1", "https://example/pr/7") in store.calls
+    assert loop._inflight == {}
+
+
+async def test_drive_max_mode_all_empty_reaps_all_and_blocks(monkeypatch):
+    """Every candidate empty → judge returns None → all candidates reaped, no PR, and
+    the feature blocks (NoChangesError, single coder with no ladder)."""
+
+    async def _open_pr(wt, branch, *, base, title, body):
+        raise AssertionError("no PR should open when every candidate is empty")
+
+    async def _judge(feature, base, worktrees):
+        return None  # nothing to ship
+
+    loop, store = await _drive_with(
+        monkeypatch,
+        open_pr=_open_pr,
+        judge=_judge,
+        cfg={"coder": "proto", "max_mode_n": 3},
+    )
+    assert set(store.reaps) == {"bd-1.c0", "bd-1.c1", "bd-1.c2"}  # every candidate torn down
+    assert store.promotes == []  # nothing promoted
+    assert "flag_blocked" in store.names()
+
+
+async def test_drive_max_mode_skips_fanout_on_a_carried_forward_fix(monkeypatch):
+    """A re-dispatch carrying _ci_feedback (a CI bounce / goal-fix / gate-fix) FIXES the
+    existing diff with ONE coder — Max-Mode must not re-fan-out N candidates."""
+
+    async def _open_pr(wt, branch, *, base, title, body):
+        return "https://example/pr/9"
+
+    async def _judge(feature, base, worktrees):
+        raise AssertionError("the judge must not run on a single-candidate carried-forward fix")
+
+    loop, store = await _drive_with(
+        monkeypatch,
+        open_pr=_open_pr,
+        judge=_judge,
+        cfg={"coder": "proto", "max_mode_n": 3},
+        seed=lambda lp: lp._ci_feedback.__setitem__("bd-1", "CI failed: lint"),
+    )
+    assert store.creates == ["bd-1"]  # one canonical worktree, NOT N suffixed candidates
+    assert store.promotes == [] and store.reaps == []
+    assert ("open_review", "bd-1", "https://example/pr/9") in store.calls
+
+
 async def test_drive_local_gate_failure_redispatches_then_opens(monkeypatch):
     """A pre-PR gate failure re-dispatches the SAME tier with the output injected,
     REUSING the worktree (one create), then opens the PR once the gate passes."""

tests/test_worktree.py

@@ -159,6 +159,33 @@ async def test_create_worktree_falls_back_to_local_base_without_a_remote(monkeyp
     assert add[-1] == "main"  # fell back to the local branch
 
 
+# ── promote_worktree: the Max-Mode winner takes the canonical name (#21) ─────────
+
+
+async def test_promote_worktree_moves_dir_and_renames_branch(monkeypatch):
+    git = FakeGit()
+    monkeypatch.setattr(worktree, "_git", git)
+    canon_wt, canon_branch = await worktree.promote_worktree(
+        "/repo", "/repo/.worktrees/feat-bd-1.c2", "feat/bd-1.c2", "bd-1", root=".worktrees"
+    )
+    assert canon_branch == "feat/bd-1"
+    assert canon_wt.endswith("/.worktrees/feat-bd-1")
+    move = next(a for a in git.calls if a[:2] == ("worktree", "move"))
+    assert move[2].endswith("/.worktrees/feat-bd-1.c2") and move[3].endswith("/.worktrees/feat-bd-1")
+    rename = next(a for a in git.calls if a[:1] == ("branch",) and "-m" in a)
+    assert rename == ("branch", "-m", "feat/bd-1.c2", "feat/bd-1")
+
+
+async def test_promote_worktree_is_a_noop_when_already_canonical(monkeypatch):
+    git = FakeGit()
+    monkeypatch.setattr(worktree, "_git", git)
+    _wt, branch = await worktree.promote_worktree(
+        "/repo", "/repo/.worktrees/feat-bd-1", "feat/bd-1", "bd-1", root=".worktrees"
+    )
+    assert branch == "feat/bd-1"
+    assert not git.ran("move") and not [a for a in git.calls if a[:1] == ("worktree",)]
+
+
 # ── pr_is_merged: the merge-poll probe ──────────────────────────────────────────
 
 

worktree.py

@@ -125,6 +125,38 @@ async def reap_feature_worktree(repo: str, worktrees_root: str, fid: str) -> Non
     await remove_worktree(repo, wt, f"feat/{fid}")
 
 
+async def promote_worktree(
+    repo: str, src_wt: str, src_branch: str, fid: str, root: str = ".worktrees"
+) -> tuple[str, str]:
+    """Promote a Max-Mode candidate worktree to the canonical ``feat-<id>`` /
+    ``feat/<id>`` name (#21). The N candidates build in throwaway ``feat-<id>.c<k>``
+    worktrees; the winner has to take over the canonical name so the rest of the
+    lifecycle — the CI-fail bounce, crash recovery (``pr_url_for_branch(feat/<id>)``),
+    and reaping (``reap_feature_worktree(<id>)``) — all of which key off the canonical
+    names — works unchanged.
+
+    Moves the worktree dir and renames its branch IN PLACE, so the coder's still-
+    uncommitted changes ride along (verified: ``git worktree move`` + ``branch -m``
+    preserve the dirty tree). Idempotently clears a stale canonical worktree/branch
+    first so ``move`` has a free destination. A winner already at the canonical path is
+    a no-op. Returns (canonical_path, canonical_branch)."""
+    canon_branch = f"feat/{fid}"
+    canon_rel = os.path.join(root, f"feat-{fid}")
+    canon_path = os.path.join(repo, canon_rel)
+    if os.path.abspath(src_wt) == os.path.abspath(canon_path):
+        return os.path.abspath(canon_path), canon_branch
+    # Free the destination: drop any stale canonical worktree/branch leftover.
+    await _git(repo, "worktree", "remove", "--force", canon_rel)
+    await _git(repo, "branch", "-D", canon_branch)
+    rc, _o, err = await _git(repo, "worktree", "move", os.path.abspath(src_wt), os.path.abspath(canon_path))
+    if rc != 0:
+        raise WorktreeError(f"worktree move failed: {err.strip()[:200]}")
+    rc, _o, err = await _git(canon_path, "branch", "-m", src_branch, canon_branch)
+    if rc != 0:
+        raise WorktreeError(f"branch rename failed: {err.strip()[:200]}")
+    return os.path.abspath(canon_path), canon_branch
+
+
 def list_feature_worktrees(repo: str, worktrees_root: str) -> list[str]:
     """The feature ids that currently have a ``feat-<id>`` worktree dir under
     ``<repo>/<worktrees_root>`` — for the health sweep's orphan check. Sync (a quick