From d549a7762f41445d95ba39ac8197bceea3c1af0f Mon Sep 17 00:00:00 2001 From: Josh Mabry Date: Fri, 12 Jun 2026 23:19:06 -0700 Subject: [PATCH] feat(loop): hot-file overlap guard + review-queue WIP limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Now that max_concurrent>1 runs features in parallel, two new back-pressure gates in _spawn_ready keep concurrency correct: - Hot-file guard: skip a ready candidate whose files_to_modify overlap an in-flight build's files — two parallel coders editing the same file are a guaranteed merge conflict. _spawn_ready now peeks the ready_queue (priority order) and claims a specific non-conflicting feature via the new store.claim(fid) instead of always taking the top; the files each drive holds are tracked in self._inflight_files and released by the drive's done-callback (so a deferred candidate can claim next). - Review-queue WIP limit (max_pending_reviews, default 5): pause claiming new work while that many PRs already await review, so the loop can't pile up PRs faster than they merge (flooding CI / reviewers). 0 = unlimited. store.claim(fid) atomically claims a chosen ready feature (vs claim_next_ready's top-of-queue), returning None if it changed state or lost the `br --claim` race. First of the P1 concurrency-correctness ports. Tests: +6 (104 total) — store.claim (specific / not-ready / claim-race); _spawn_ready caps at max_concurrent, defers a file-conflicting candidate, respects the review WIP limit, and releases files when a drive finishes. Co-Authored-By: Claude Fable 5 --- loop.py | 51 +++++++++++++++++++--- protoagent.plugin.yaml | 7 ++- store.py | 16 +++++++ tests/test_loop.py | 97 ++++++++++++++++++++++++++++++++++-------- tests/test_store.py | 27 ++++++++++++ 5 files changed, 172 insertions(+), 26 deletions(-) diff --git a/loop.py b/loop.py index 107319d..3b17715 100644 --- a/loop.py +++ b/loop.py @@ -58,6 +58,10 @@ def __init__(self, cfg: dict): # 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))) + # Review-queue WIP limit: pause new claims when this many PRs already await + # review, so the loop can't pile up PRs faster than they merge (flooding CI / + # reviewers). 0 = unlimited. + self.max_pending_reviews = int(self.cfg.get("max_pending_reviews", 5)) # Stuck-drive watchdog: hard cap on a single coder dispatch (the only # otherwise-unbounded await in a drive — git/gh calls already self-time-out). # 0 disables it. A timeout reaps the coder subprocess and is a capability @@ -79,6 +83,9 @@ def __init__(self, cfg: dict): # subprocess itself is reaped by dispatch_coder's finally. self._drives: set[asyncio.Task] = set() self._inflight: dict[str, tuple[str, str, str]] = {} + # files_to_modify of each in-flight feature, for the hot-file overlap guard + # (don't run two parallel coders that edit the same file → sure conflict). + self._inflight_files: dict[str, set[str]] = {} self._last_poll = 0.0 # monotonic ts of the last merge poll def _store(self): @@ -173,19 +180,49 @@ async def _run(self): 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).""" + """Claim Ready features up to the concurrency cap and spawn a drive for each, + with two back-pressure gates: pause when too many PRs already await review + (``max_pending_reviews``), and skip a candidate whose ``files_to_modify`` + overlap an in-flight build (the hot-file guard — two parallel coders editing + the same file are a guaranteed merge conflict). Returns True if it started at + least one drive (so the runner stays hot).""" + if len(self._drives) >= self.max_concurrent: + return False + store = self._store() + # Review-queue WIP limit — don't claim new work while the review queue is full. + if self.max_pending_reviews and len(store.list_features(state="in_review")) >= self.max_pending_reviews: + return False spawned = False - while len(self._drives) < self.max_concurrent: - feature = self._store().claim_next_ready(assignee=self.coder_name) - if feature is None: + busy = set().union(*self._inflight_files.values()) if self._inflight_files else set() + for candidate in store.ready_queue(): # priority order, dep-unblocked + if len(self._drives) >= self.max_concurrent: break - task = asyncio.create_task(self._drive(feature), name=f"pb-drive-{feature['id']}") + if candidate.get("board_state") != "ready" or candidate.get("blocked"): + continue # a blocked-flagged feature can carry the `ready` label too + files = set(candidate.get("files_to_modify") or []) + if files & busy: + continue # would edit a file an in-flight build owns → defer a tick + claimed = store.claim(candidate["id"], assignee=self.coder_name) + if claimed is None: + continue # raced / no longer ready + self._inflight_files[claimed["id"]] = files + task = asyncio.create_task(self._drive(claimed), name=f"pb-drive-{claimed['id']}") self._drives.add(task) - task.add_done_callback(self._drives.discard) + task.add_done_callback(self._make_drive_done_cb(claimed["id"])) + busy |= files spawned = True return spawned + def _make_drive_done_cb(self, fid: str): + """A drive task's done-callback: drop it from the running set and release the + files it held (so a deferred file-conflicting candidate can be claimed next).""" + + def _cb(task: asyncio.Task): + self._drives.discard(task) + self._inflight_files.pop(fid, None) + + return _cb + # ── 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 diff --git a/protoagent.plugin.yaml b/protoagent.plugin.yaml index 1ead67c..f28b340 100644 --- a/protoagent.plugin.yaml +++ b/protoagent.plugin.yaml @@ -46,7 +46,12 @@ config: # 0 disables the cap. 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). + # own worktree; 1 = serial (the safe default). Parallel + # builds whose files_to_modify overlap are serialized (the + # hot-file guard) to avoid sure merge conflicts. + max_pending_reviews: 5 # back-pressure: pause claiming new work while this many + # PRs already await review (don't pile up faster than they + # merge). 0 = unlimited. 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 diff --git a/store.py b/store.py index 08ad22a..2f5609b 100644 --- a/store.py +++ b/store.py @@ -191,6 +191,22 @@ def claim_next_ready(self, assignee: str = "") -> dict | None: self._run("update", fid, "--assignee", assignee) return self.get_feature(fid) + def claim(self, fid: str, assignee: str = "") -> dict | None: + """Atomically claim a SPECIFIC ready feature → `in_progress` (vs + ``claim_next_ready``, which takes the top of the queue). The loop uses this to + skip a candidate whose files overlap an in-flight build. Returns the feature, + or None if it's no longer claimable (changed state, or lost the claim race).""" + f = self.get_feature(fid) + if f is None or f["board_state"] != "ready": + return None + try: + self._run("update", fid, "--claim", "--remove-label", LABEL_READY) + except BoardError: + return None # raced — `br --claim` rejects an already-assigned bead + if assignee: + self._run("update", fid, "--assignee", assignee) + return self.get_feature(fid) + # ── In Progress → In Review ─────────────────────────────────────────────── def open_review(self, fid: str, *, pr_url: str) -> dict: f = self._require(fid) diff --git a/tests/test_loop.py b/tests/test_loop.py index 4d47ad6..8c23d90 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -216,37 +216,59 @@ async def _open_pr(wt, branch, *, base, title, body): 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.""" + """A peekable ready queue + atomic claim(fid), mirroring the store API _spawn_ready + now uses. Records claims so a test can prove the caps/gates stop the puller.""" - def __init__(self, features): - self._queue = list(features) - self.claims = 0 + def __init__(self, features, in_review=0): + self._features = [dict(f) for f in features] + self._in_review = in_review + self.claimed = [] - def claim_next_ready(self, assignee=""): - self.claims += 1 - return self._queue.pop(0) if self._queue else None + def ready_queue(self): + return [f for f in self._features if f["id"] not in self.claimed] + def claim(self, fid, assignee=""): + if fid in self.claimed: + return None + self.claimed.append(fid) + return next((f for f in self._features if f["id"] == fid), 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}) + def list_features(self, state=None): + return [{"id": f"rev-{i}"} for i in range(self._in_review)] if state == "in_review" else [] + +def _ready(fid, files): + return {"id": fid, "board_state": "ready", "files_to_modify": files} + + +async def _hold_drives(loop, monkeypatch): + """Replace _drive with a coroutine that blocks, so spawned tasks stay 'running'. + Returns a finalizer the test calls to release + await them.""" release = asyncio.Event() async def _hold(feature): - await release.wait() # keep the drive task "running" so the slot stays taken + await release.wait() monkeypatch.setattr(loop, "_drive", _hold) - spawned = loop._spawn_ready() + + async def _finish(): + release.set() + await asyncio.gather(*loop._drives, return_exceptions=True) + + return _finish + + +async def test_spawn_ready_claims_up_to_max_concurrent(monkeypatch): + store = _ClaimStore([_ready("bd-1", ["a.py"]), _ready("bd-2", ["b.py"]), _ready("bd-3", ["c.py"])]) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({"max_concurrent": 2}) + finish = await _hold_drives(loop, monkeypatch) try: - assert spawned is True + assert loop._spawn_ready() is True assert len(loop._drives) == 2 # capped at max_concurrent - assert store.claims == 2 # stopped claiming once full (no 3rd claim) + assert store.claimed == ["bd-1", "bd-2"] # stopped claiming once full finally: - release.set() - await asyncio.gather(*loop._drives, return_exceptions=True) + await finish() async def test_spawn_ready_is_false_when_nothing_ready(monkeypatch): @@ -257,6 +279,45 @@ async def test_spawn_ready_is_false_when_nothing_ready(monkeypatch): assert loop._drives == set() +async def test_spawn_ready_skips_a_file_conflicting_candidate(monkeypatch): + # bd-1 + bd-2 both touch shared.py; bd-3 touches other.py. + store = _ClaimStore([_ready("bd-1", ["shared.py"]), _ready("bd-2", ["shared.py"]), _ready("bd-3", ["other.py"])]) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({"max_concurrent": 3}) + finish = await _hold_drives(loop, monkeypatch) + try: + loop._spawn_ready() + # bd-1 claimed; bd-2 deferred (overlaps bd-1's file); bd-3 claimed (disjoint). + assert store.claimed == ["bd-1", "bd-3"] + assert loop._inflight_files == {"bd-1": {"shared.py"}, "bd-3": {"other.py"}} + finally: + await finish() + + +async def test_spawn_ready_respects_the_review_wip_limit(monkeypatch): + store = _ClaimStore([_ready("bd-1", ["a.py"])], in_review=5) # already at the cap + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({"max_concurrent": 2, "max_pending_reviews": 5}) + assert loop._spawn_ready() is False + assert store.claimed == [] # paused: too many PRs await review + + +async def test_drive_done_releases_its_files(monkeypatch): + store = _ClaimStore([_ready("bd-1", ["a.py"])]) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({"max_concurrent": 1}) + + async def _quick(feature): + return None + + monkeypatch.setattr(loop, "_drive", _quick) + loop._spawn_ready() + await asyncio.gather(*list(loop._drives), return_exceptions=True) + await asyncio.sleep(0) # let the done-callbacks run + assert loop._inflight_files == {} # files released when the drive finished + assert loop._drives == set() + + # ── the merge poll (Done-edge fallback) ───────────────────────────────────────── diff --git a/tests/test_store.py b/tests/test_store.py index eecde46..256f19f 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -190,6 +190,33 @@ def test_claim_next_ready_returns_none_when_empty(make_board): assert b.claim_next_ready() is None +def test_claim_claims_a_specific_ready_feature(make_board, monkeypatch): + br = Br() + b = make_board(br) + monkeypatch.setattr(b, "get_feature", lambda fid: {"id": fid, "board_state": "ready"}) + claimed = b.claim("bd-5", assignee="proto") + assert claimed["id"] == "bd-5" + assert ("update", "bd-5", "--claim", "--remove-label", "ready") in br.calls + assert ("update", "bd-5", "--assignee", "proto") in br.calls + + +def test_claim_returns_none_when_not_ready(make_board, monkeypatch): + b = make_board(Br()) + monkeypatch.setattr(b, "get_feature", lambda fid: {"id": fid, "board_state": "in_progress"}) + assert b.claim("bd-5") is None + + +def test_claim_returns_none_on_a_claim_race(make_board, monkeypatch): + def run_impl(*args, want_json=False): + if args and args[0] == "update" and "--claim" in args: + raise BoardError("already assigned to agent") + return [] if want_json else "" + + b = make_board(run_impl) + monkeypatch.setattr(b, "get_feature", lambda fid: {"id": fid, "board_state": "ready"}) + assert b.claim("bd-5") is None # br --claim rejected → lost the race + + # ── invariant #2: the single Done edge (record_merge) ───────────────────────────