feat(loop): hot-file overlap guard + review-queue WIP limit

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

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

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)

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) ─────────────────────────────────────────
 
 

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) ───────────────────────────