From a605b954dfae88a9fe9793d8ced24e1eeb587ff9 Mon Sep 17 00:00:00 2001 From: Josh Mabry Date: Fri, 12 Jun 2026 21:51:42 -0700 Subject: [PATCH] feat(loop): merge-poll Done-edge fallback + parallel drives (max_concurrent>1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two autonomy improvements to the orchestration loop, both off-by-default-safe: 1. Merge-poll fallback for the single Done edge. The merge webhook stays the primary path, but many deployments (local / NAT / tailnet-only agents) have no public URL GitHub can post to, so a merged feature would sit in_review forever. The loop now polls each in_review PR via `gh pr view ... --json state` and runs the same idempotent record_merge for any that merged. On by default; rate-limited to merge_poll_interval_s (60s); only probes in_review PRs, so it's cheap. - worktree.pr_is_merged() — the probe (a gh failure → False, next poll retries). - worktree.reap_feature_worktree() — shared id→worktree/branch reap; the webhook reap in api.py now uses it too (DRY). 2. Lift the max_concurrent cap. The loop drove exactly one feature at a time; it now drives up to max_concurrent concurrently, each in its own worktree (the puller tops up to the cap each tick, re-checking promptly while busy). Per-feature in-flight worktrees are tracked (self._inflight) so shutdown cancels every drive and reaps every orphaned worktree. Default stays 1 (serial) — fully backward compatible. Config: max_concurrent (now honoured; floors at 1), merge_poll (default true), merge_poll_interval_s (default 60). Manifest + README updated; version → 0.3.0. Tests: +11 (72 total) — pr_is_merged, reap_feature_worktree, _spawn_ready cap + claim accounting, _poll_merges (merged-only + reap), poll rate-limit + off paths, and the _active→_inflight tracking. Co-Authored-By: Claude Fable 5 --- README.md | 8 ++- api.py | 4 +- loop.py | 135 ++++++++++++++++++++++++++++++----------- protoagent.plugin.yaml | 11 +++- pyproject.toml | 2 +- tests/test_loop.py | 126 +++++++++++++++++++++++++++++++++++++- tests/test_worktree.py | 32 ++++++++++ worktree.py | 17 ++++++ 8 files changed, 289 insertions(+), 46 deletions(-) diff --git a/README.md b/README.md index d6f66c8..75e8773 100644 --- a/README.md +++ b/README.md @@ -31,8 +31,10 @@ board to a PR — or fork it as a starting point. separate store, so the work graph can't drift out of sync. - **The loop** pulls the top-priority `ready` feature → creates a disposable `git worktree` off `origin/` → dispatches a coder (`acp` delegate) scoped to - it → commits/pushes → opens a PR → `in_review`. A **merge webhook** is the single - edge that sets `done` (and reaps the worktree). + it → commits/pushes → opens a PR → `in_review`. A **merge webhook** sets `done` + (and reaps the worktree); where GitHub can't reach a webhook URL, a **merge poll** + (`merge_poll`, on by default) runs the same idempotent Done edge. Set + `max_concurrent > 1` to build several features in parallel, each in its own worktree. - **DAG + gates** — `depends_on` are `blocks` edges; a dependent stays out of the puller until its blocker is **merged** (foundation merge-gate). The **Ready gate** requires a spec, EARS acceptance criteria, and explicit `files_to_modify`. @@ -76,6 +78,8 @@ project_board: repo: ~/dev/my-repo base_branch: main loop_enabled: false # flip true to start the background puller + max_concurrent: 1 # >1 builds features in parallel (each its own worktree) + merge_poll: true # poll merged PRs as a fallback to the webhook Done edge # webhook_secret: "..." # set before exposing /webhook/pr publicly ``` diff --git a/api.py b/api.py index ed2ef26..a242a8e 100644 --- a/api.py +++ b/api.py @@ -138,9 +138,7 @@ async def _webhook_pr(request: Request): try: from . import worktree - repo = store_kw["repo"] - wt = os.path.join(repo, worktrees_root, f"feat-{f['id']}") - await worktree.remove_worktree(repo, wt, f"feat/{f['id']}") + await worktree.reap_feature_worktree(store_kw["repo"], worktrees_root, f["id"]) except Exception: # noqa: BLE001 — reaping is best-effort; done is already set log.warning("[project_board] worktree reap for %s failed", f["id"], exc_info=True) log.info("[project_board] merge webhook → done: %s (%s)", f["id"], pr_url) diff --git a/loop.py b/loop.py index 11c3ea5..116195c 100644 --- a/loop.py +++ b/loop.py @@ -11,17 +11,21 @@ └──▶ in_review ──delegate_to(reviewer)──▶ (CI + review on the PR) │ merge webhook ▼ CI fail ▼ any failure ▼ - done in_progress (bounce) blocked (flag + reason) + /merge poll in_progress (bounce) blocked (flag + reason) + done -CI status + merge arrive out-of-band via the board API (``api.py``), not here. -Concurrency is capped at 1 (token + merge-integration cost); the cap is where -parallelism lands later. +CI status arrives out-of-band via the board API (``api.py``). ``done`` is set by +the merge webhook (``api.record_merge``) — or, when no public webhook URL is +reachable, by the loop's **merge poll** (``merge_poll``), which asks ``gh`` whether +each ``in_review`` PR has merged and runs the same idempotent Done edge. Up to +``max_concurrent`` features build concurrently, each in its own worktree. """ from __future__ import annotations import asyncio import logging +import time from . import worktree from .store import BoardError, escalation_enabled, get_store @@ -49,6 +53,14 @@ def __init__(self, cfg: dict): # and we never write redundant tier/attempt labels. self.coders = {str(k): str(v) for k, v in (self.cfg.get("coders") or {}).items()} self.escalation_on = escalation_enabled(self.cfg) + # Concurrency: drive up to `max_concurrent` features at once, each in its own + # worktree. 1 (the default) = serial — the safe default for token + merge- + # integration cost; raise it on a repo that parallelizes cleanly. + self.max_concurrent = max(1, int(self.cfg.get("max_concurrent", 1))) + # Merge poll: a fallback to the /webhook/pr Done edge for deployments with no + # public webhook URL. On by default (cheap; only probes `in_review` PRs). + self.merge_poll = bool(self.cfg.get("merge_poll", True)) + self.merge_poll_interval = float(self.cfg.get("merge_poll_interval_s", 60)) self._store_kw = dict( db=self.cfg.get("db_path") or None, repo=self.cfg.get("repo", "."), @@ -56,10 +68,12 @@ def __init__(self, cfg: dict): ) self._task: asyncio.Task | None = None self._stop = asyncio.Event() - # The in-flight build's worktree, so shutdown can reap it (a cancel mid-drive - # would otherwise orphan the worktree; the coder subprocess is reaped by - # dispatch_coder's finally). (repo, worktree_path, branch) or None. - self._active: tuple[str, str, str] | None = None + # The running drive tasks, and the worktrees they hold (fid → (repo, wt, + # branch)) so shutdown can reap any a cancel mid-drive would orphan; the coder + # subprocess itself is reaped by dispatch_coder's finally. + self._drives: set[asyncio.Task] = set() + self._inflight: dict[str, tuple[str, str, str]] = {} + self._last_poll = 0.0 # monotonic ts of the last merge poll def _store(self): return get_store(**self._store_kw) @@ -71,10 +85,12 @@ def start(self): return None self._task = asyncio.create_task(self._run(), name="project-board-loop") log.info( - "[project_board] loop started (coder=%s reviewer=%s every %ss)", + "[project_board] loop started (coder=%s reviewer=%s every %ss, max_concurrent=%d, merge_poll=%s)", self.coder_name, self.reviewer_name, self.interval, + self.max_concurrent, + self.merge_poll, ) return self._task @@ -86,11 +102,16 @@ async def stop(self): await self._task except (asyncio.CancelledError, Exception): # noqa: BLE001 pass - # Reap an interrupted build's worktree (a drive cancelled mid-flight leaves - # self._active set; a completed/blocked drive clears it). Best-effort. - act, self._active = self._active, None - if act: - repo, wt, branch = act + # Cancel any in-flight drives and await them out. A drive cancelled mid-flight + # can't run its own cleanup, so its worktree stays in self._inflight — reaped + # below. (A completed/blocked drive already popped itself.) + drives, self._drives = list(self._drives), set() + for t in drives: + t.cancel() + if drives: + await asyncio.gather(*drives, return_exceptions=True) + inflight, self._inflight = dict(self._inflight), {} + for fid, (repo, wt, branch) in inflight.items(): try: await worktree.remove_worktree(repo, wt, branch or "") log.info("[project_board] reaped in-flight worktree on shutdown: %s", wt) @@ -100,26 +121,68 @@ async def stop(self): # ── the puller ──────────────────────────────────────────────────────────── async def _run(self): while not self._stop.is_set(): + spawned = False try: - worked = await self._tick() + await self._maybe_poll_merges() + spawned = self._spawn_ready() except Exception: # noqa: BLE001 — a bad tick must never kill the loop log.exception("[project_board] loop tick failed") - worked = False - # If we did work, loop again immediately (drain Ready); else sleep. - if not worked: - try: - await asyncio.wait_for(self._stop.wait(), timeout=self.interval) - except asyncio.TimeoutError: - pass - - async def _tick(self) -> bool: - """Claim and drive at most one Ready feature. Returns True if it claimed - one (so the runner can drain), False if Ready was empty.""" - feature = self._store().claim_next_ready(assignee=self.coder_name) - if feature is None: - return False - await self._drive(feature) - return True + # Idle (nothing started, nothing running) → sleep the full interval. Busy + # → re-check soon so a freed concurrency slot refills and merges land + # promptly (the poll itself stays rate-limited by merge_poll_interval). + idle = not spawned and not self._drives + timeout = self.interval if idle else min(self.interval, 3.0) + try: + await asyncio.wait_for(self._stop.wait(), timeout=timeout) + except asyncio.TimeoutError: + pass + + def _spawn_ready(self) -> bool: + """Claim Ready features up to the concurrency cap and spawn a drive task for + each. Returns True if it started at least one (so the runner stays hot).""" + spawned = False + while len(self._drives) < self.max_concurrent: + feature = self._store().claim_next_ready(assignee=self.coder_name) + if feature is None: + break + task = asyncio.create_task(self._drive(feature), name=f"pb-drive-{feature['id']}") + self._drives.add(task) + task.add_done_callback(self._drives.discard) + spawned = True + return spawned + + # ── the merge poll (Done-edge fallback to the webhook) ───────────────────── + async def _maybe_poll_merges(self): + """Run the merge poll at most once per ``merge_poll_interval`` (and only when + enabled) — cheap, but no reason to hammer ``gh`` every busy tick.""" + if not self.merge_poll: + return + now = time.monotonic() + if now - self._last_poll < self.merge_poll_interval: + return + self._last_poll = now + await self._poll_merges() + + async def _poll_merges(self): + """Ask ``gh`` whether each ``in_review`` PR has merged and run the idempotent + Done edge for any that have — the fallback for deployments GitHub can't post + a webhook to (otherwise a merged feature would sit in_review forever).""" + store = self._store() + repo = self._store_kw["repo"] + for f in store.list_features(state="in_review"): + pr_url = f.get("pr_url") + if not pr_url: + continue + try: + if not await worktree.pr_is_merged(pr_url, cwd=repo): + continue + done = store.record_merge(pr_url=pr_url) + except Exception: # noqa: BLE001 — a poll error must never kill the loop + log.warning("[project_board] merge poll for %s failed", f["id"], exc_info=True) + continue + if done: + await worktree.reap_feature_worktree(repo, self.root, f["id"]) + log.info("[project_board] merge poll → done: %s (%s)", f["id"], pr_url) async def _drive(self, feature: dict): """Drive one feature ready→in_review (or →blocked). `done` is set later by @@ -143,7 +206,7 @@ async def _drive(self, feature: dict): return # Fresh worktree per attempt (a failed attempt may leave partial work). wt, branch = await worktree.create_worktree(repo, base, fid, self.root) - self._active = (repo, wt, branch) # track for shutdown reaping + self._inflight[fid] = (repo, wt, branch) # track for shutdown reaping try: 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]) @@ -163,7 +226,7 @@ async def _drive(self, feature: dict): store.flag_blocked(fid, str(exc)) if wt: await worktree.remove_worktree(repo, wt, branch or "") - self._active = None + self._inflight.pop(fid, None) return # Built + PR opened. The fleet PR-review pipeline reviews it on open; # only dispatch an explicit review when configured to (review_dispatch). @@ -173,18 +236,18 @@ async def _drive(self, feature: dict): await self._request_review(fid, pr_url) # Keep the worktree (a CI-fail bounce re-dispatches); reaping happens # on a terminal block above, and the coder subprocess is already reaped. - self._active = None # built OK — not an interrupted build to reap + self._inflight.pop(fid, None) # built OK — not an interrupted build to reap return except BoardError as exc: log.warning("[project_board] %s blocked (board): %s", fid, exc) store.flag_blocked(fid, str(exc)) - self._active = None + self._inflight.pop(fid, None) except Exception as exc: # noqa: BLE001 — unexpected; block, don't crash the loop log.exception("[project_board] %s unexpected failure", fid) store.flag_blocked(fid, f"unexpected: {type(exc).__name__}: {exc}") if wt: await worktree.remove_worktree(repo, wt, branch or "") - self._active = None + self._inflight.pop(fid, None) async def _request_review(self, fid: str, pr_url: str): """Hand the PR to the reviewer (an a2a delegate, e.g. quinn). Best-effort: diff --git a/protoagent.plugin.yaml b/protoagent.plugin.yaml index 5ee8320..ac53446 100644 --- a/protoagent.plugin.yaml +++ b/protoagent.plugin.yaml @@ -1,6 +1,6 @@ id: project_board name: Project Board (coding orchestration) -version: 0.2.0 +version: 0.3.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 @@ -40,7 +40,14 @@ config: worktrees_root: .worktrees loop_enabled: false # the background puller; off until a board exists loop_interval_s: 30 # how often the puller wakes - max_concurrent: 1 # worktree/coder concurrency cap (token + merge cost) + max_concurrent: 1 # worktree/coder concurrency cap (token + merge cost). + # >1 drives that many features in parallel, each in its + # own worktree; 1 = serial (the safe default). + merge_poll: true # poll merged PRs as a fallback to the /webhook/pr Done + # edge — for deployments GitHub can't post a webhook to + # (no public URL). The Done edge stays idempotent either + # way; set false to rely on the webhook alone. + merge_poll_interval_s: 60 # how often the loop polls in_review PRs for a merge db_path: "" # beads db path; blank → br auto-discovers .beads/*.db webhook_secret: "" # GitHub webhook HMAC secret for /webhook/pr (the Done # edge). Blank → signature NOT verified (dev only); set diff --git a/pyproject.toml b/pyproject.toml index 9cad52b..f58040c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "project-board" -version = "0.2.0" +version = "0.3.0" description = "Board-driven coding-orchestration plugin for protoAgent (beads board + ACP spawn loop + planning layer + console view)." requires-python = ">=3.11" diff --git a/tests/test_loop.py b/tests/test_loop.py index 2826fb3..6cc0e73 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -10,6 +10,8 @@ from __future__ import annotations +import asyncio + from project_board import worktree from project_board.loop import BoardLoop @@ -54,6 +56,8 @@ def test_config_defaults(): assert loop.review_dispatch is False assert loop.interval == 30 and loop.enabled is False assert loop.escalation_on is False # no coders map → single-coder mode + assert loop.max_concurrent == 1 # serial by default + assert loop.merge_poll is True and loop.merge_poll_interval == 60 def test_escalation_on_with_two_distinct_coders(): @@ -61,6 +65,11 @@ def test_escalation_on_with_two_distinct_coders(): assert loop.escalation_on is True +def test_max_concurrent_floors_at_one(): + assert BoardLoop({"max_concurrent": 0}).max_concurrent == 1 + assert BoardLoop({"max_concurrent": 4}).max_concurrent == 4 + + # ── the coder prompt (ProtoMaker discipline: name the files, demand the diff) ──── @@ -108,7 +117,7 @@ async def _open_pr(wt, branch, *, base, title, body): loop, store = await _drive_with(monkeypatch, open_pr=_open_pr) assert ("open_review", "bd-1", "https://example/pr/1") in store.calls - assert loop._active is None # a completed drive leaves nothing to reap + assert loop._inflight == {} # a completed drive leaves nothing to reap async def test_drive_blocks_on_an_empty_diff_with_a_single_coder(monkeypatch): @@ -119,7 +128,7 @@ async def _open_pr(wt, branch, *, base, title, body): # No escalation ladder (single coder) → a capability failure blocks immediately. assert "flag_blocked" in store.names() assert "open_review" not in store.names() - assert loop._active is None + assert loop._inflight == {} async def test_drive_blocks_when_the_coder_is_not_configured(monkeypatch): @@ -128,3 +137,116 @@ async def _open_pr(wt, branch, *, base, title, body): loop, store = await _drive_with(monkeypatch, open_pr=_open_pr, coder=None) assert store.names() == ["flag_blocked"] # blocked before any worktree work + + +# ── concurrency: _spawn_ready claims up to max_concurrent ──────────────────────── + + +class _ClaimStore: + """Yields a fixed sequence of features (then None) from claim_next_ready, and + counts the claims so a test can prove the cap stops the puller.""" + + def __init__(self, features): + self._queue = list(features) + self.claims = 0 + + def claim_next_ready(self, assignee=""): + self.claims += 1 + return self._queue.pop(0) if self._queue else None + + +async def test_spawn_ready_claims_up_to_max_concurrent(monkeypatch): + store = _ClaimStore([{"id": "bd-1"}, {"id": "bd-2"}, {"id": "bd-3"}]) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({"max_concurrent": 2}) + + release = asyncio.Event() + + async def _hold(feature): + await release.wait() # keep the drive task "running" so the slot stays taken + + monkeypatch.setattr(loop, "_drive", _hold) + spawned = loop._spawn_ready() + try: + assert spawned is True + assert len(loop._drives) == 2 # capped at max_concurrent + assert store.claims == 2 # stopped claiming once full (no 3rd claim) + finally: + release.set() + await asyncio.gather(*loop._drives, return_exceptions=True) + + +async def test_spawn_ready_is_false_when_nothing_ready(monkeypatch): + store = _ClaimStore([]) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({"max_concurrent": 2}) + assert loop._spawn_ready() is False + assert loop._drives == set() + + +# ── the merge poll (Done-edge fallback) ───────────────────────────────────────── + + +class _PollStore: + def __init__(self, in_review): + self._in_review = in_review + self.merged = [] + + def list_features(self, state=None): + return self._in_review if state == "in_review" else [] + + def record_merge(self, *, pr_url): + self.merged.append(pr_url) + return {"id": "x", "board_state": "done"} + + +async def test_poll_merges_runs_done_edge_for_merged_only(monkeypatch): + store = _PollStore( + [ + {"id": "bd-1", "pr_url": "https://example/pr/1"}, + {"id": "bd-2", "pr_url": "https://example/pr/2"}, + {"id": "bd-3", "pr_url": ""}, # no PR → skipped entirely + ] + ) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + + async def _is_merged(url, *, cwd="."): + return url.endswith("/1") # only PR 1 has merged + + reaped = [] + + async def _reap(repo, root, fid): + reaped.append(fid) + + monkeypatch.setattr(worktree, "pr_is_merged", _is_merged) + monkeypatch.setattr(worktree, "reap_feature_worktree", _reap) + + await BoardLoop({})._poll_merges() + assert store.merged == ["https://example/pr/1"] # the unmerged + PR-less ones skipped + assert reaped == ["bd-1"] # the merged feature's worktree is reaped + + +async def test_maybe_poll_is_rate_limited(monkeypatch): + loop = BoardLoop({"merge_poll": True, "merge_poll_interval_s": 60}) + calls = [] + + async def _poll(): + calls.append(1) + + monkeypatch.setattr(loop, "_poll_merges", _poll) + clock = {"t": 1000.0} + monkeypatch.setattr("project_board.loop.time.monotonic", lambda: clock["t"]) + + await loop._maybe_poll_merges() # first → polls + await loop._maybe_poll_merges() # immediately again → rate-limited, skipped + clock["t"] += 61 + await loop._maybe_poll_merges() # interval elapsed → polls again + assert len(calls) == 2 + + +async def test_merge_poll_off_never_polls(monkeypatch): + loop = BoardLoop({"merge_poll": False}) + called = [] + monkeypatch.setattr(loop, "_poll_merges", lambda: called.append(1)) + await loop._maybe_poll_merges() + assert called == [] # disabled → the poll is never reached diff --git a/tests/test_worktree.py b/tests/test_worktree.py index ca551e6..e745c20 100644 --- a/tests/test_worktree.py +++ b/tests/test_worktree.py @@ -127,3 +127,35 @@ async def test_create_worktree_falls_back_to_local_base_without_a_remote(monkeyp assert branch == "feat/bd-2" add = next(a for a in git.calls if a[:2] == ("worktree", "add")) assert add[-1] == "main" # fell back to the local branch + + +# ── pr_is_merged: the merge-poll probe ────────────────────────────────────────── + + +@pytest.mark.parametrize( + "gh_state,expected", + [ + ((0, "MERGED", ""), True), + ((0, "OPEN", ""), False), + ((0, "CLOSED", ""), False), # closed-unmerged is NOT done + ((1, "", "no pr found"), False), # a gh failure → False, the poll retries + ], +) +async def test_pr_is_merged(monkeypatch, gh_state, expected): + gh = FakeGh({"view": gh_state}) + _install(monkeypatch, FakeGit(), gh) + assert await worktree.pr_is_merged("https://example/pr/1", cwd="/repo") is expected + + +# ── reap_feature_worktree: the shared id → worktree/branch reap ────────────────── + + +async def test_reap_feature_worktree_computes_path_and_branch(monkeypatch): + calls = [] + + async def _remove(repo, wt, branch=""): + calls.append((repo, wt, branch)) + + monkeypatch.setattr(worktree, "remove_worktree", _remove) + await worktree.reap_feature_worktree("/repo", ".worktrees", "bd-7") + assert calls == [("/repo", "/repo/.worktrees/feat-bd-7", "feat/bd-7")] diff --git a/worktree.py b/worktree.py index 2b3657f..20a74eb 100644 --- a/worktree.py +++ b/worktree.py @@ -89,6 +89,14 @@ async def remove_worktree(repo: str, worktree: str, branch: str = "") -> None: await _git(repo, "branch", "-D", branch) +async def reap_feature_worktree(repo: str, worktrees_root: str, fid: str) -> None: + """Remove the worktree + branch a feature owns, by its id — the one place that + knows the ``feat-`` / ``feat/`` naming. Shared by the merge webhook and + the merge poll (both reap once a feature reaches ``done``).""" + wt = os.path.join(repo, worktrees_root, f"feat-{fid}") + await remove_worktree(repo, wt, f"feat/{fid}") + + async def dispatch_coder(coder, worktree: str, prompt: str, *, timeout: float | None = None) -> str: """Dispatch the coder (an ``acp`` Delegate) scoped to ``worktree``. @@ -171,3 +179,12 @@ async def open_pr(worktree: str, branch: str, *, base: str = "main", title: str, if vrc == 0 and vout.strip(): return vout.strip() raise WorktreeError(f"gh pr create failed: {err.strip()[:300]}") + + +async def pr_is_merged(pr_url: str, *, cwd: str = ".") -> bool: + """True iff the PR has merged — the merge poll's probe (a fallback to the + webhook for deployments with no public webhook URL). A non-zero ``gh`` / + transient failure returns False so the next poll simply retries; this never + raises into the loop.""" + rc, out, _err = await _gh("pr", "view", pr_url, "--json", "state", "--jq", ".state", cwd=cwd) + return rc == 0 and out.strip() == "MERGED"