feat(loop): periodic health sweep — self-heal stale drives + orphaned worktrees

loop.py

@@ -71,6 +71,9 @@ def __init__(self, cfg: dict):
         # 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))
+        # Health sweep: periodic self-heal (reclaim slots from dead drives, reap
+        # orphaned worktrees). 0 disables it.
+        self.sweep_interval = float(self.cfg.get("health_sweep_interval_s", 300))
         self._store_kw = dict(
             db=self.cfg.get("db_path") or None,
             repo=self.cfg.get("repo", "."),
@@ -87,6 +90,7 @@ def __init__(self, cfg: dict):
         # (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
+        self._last_sweep = 0.0  # monotonic ts of the last health sweep
 
     def _store(self):
         return get_store(**self._store_kw)
@@ -134,27 +138,67 @@ async def stop(self):
                 log.warning("[project_board] worktree reap on shutdown failed: %s", wt, exc_info=True)
 
     # ── crash recovery (runs once, before the puller claims new work) ──────────
+    async def _reconcile_orphan(self, fid: str):
+        """A claimed feature with no live drive: if its PR actually got opened (a crash
+        between ``open_pr`` and ``open_review``) adopt it → ``in_review``; otherwise
+        reset it to ``ready`` for a clean rebuild (a stale worktree is cleaned when the
+        puller re-claims it). Shared by boot recovery and the health sweep."""
+        store = self._store()
+        pr_url = await worktree.pr_url_for_branch(f"feat/{fid}", cwd=self._store_kw["repo"])
+        if pr_url:
+            store.open_review(fid, pr_url=pr_url)
+            log.info("[project_board] %s already had a PR → in_review (%s)", fid, pr_url)
+        else:
+            store.requeue(fid)
+            log.info("[project_board] %s reset to ready (no PR — rebuild fresh)", fid)
+
     async def _recover(self):
-        """Reconcile features the previous run left mid-drive. A drive doesn't survive
-        a restart, so every ``in_progress`` feature is orphaned (claimed, no PR yet):
-        if its PR actually got opened (a crash between ``open_pr`` and ``open_review``)
-        adopt it → ``in_review``; otherwise reset it to ``ready`` for a clean rebuild
-        (any stale worktree is cleaned when the puller re-claims it). ``in_review``
-        features are NOT touched — they have a PR and the webhook/poll resolves them."""
+        """On boot, reconcile every ``in_progress`` feature the previous run left
+        mid-drive (a drive doesn't survive a restart). ``in_review`` features are NOT
+        touched — they have a PR and the webhook/poll resolves them."""
+        for f in self._store().list_features(state="in_progress"):
+            try:
+                await self._reconcile_orphan(f["id"])
+            except Exception:  # noqa: BLE001 — recovery is best-effort, per feature
+                log.warning("[project_board] recovery for %s failed", f["id"], exc_info=True)
+
+    # ── periodic health sweep (self-heal during the run) ───────────────────────
+    async def _maybe_sweep(self):
+        """Run the health sweep at most once per ``health_sweep_interval`` (0 = off)."""
+        if not self.sweep_interval:
+            return
+        now = time.monotonic()
+        if now - self._last_sweep < self.sweep_interval:
+            return
+        self._last_sweep = now
+        await self._sweep()
+
+    async def _sweep(self):
+        """Self-heal: (a) reset ``in_progress`` features that no live drive owns (a
+        drive died without finishing) — same reconcile as boot recovery; (b) reap
+        ``feat-<id>`` worktrees whose feature is gone or already ``done`` (a missed
+        reap). Best-effort; a per-item failure never stops the sweep or the loop."""
         store = self._store()
         repo = self._store_kw["repo"]
         for f in store.list_features(state="in_progress"):
             fid = f["id"]
+            if fid in self._inflight_files:
+                continue  # a live drive owns it
             try:
-                pr_url = await worktree.pr_url_for_branch(f"feat/{fid}", cwd=repo)
-                if pr_url:
-                    store.open_review(fid, pr_url=pr_url)
-                    log.info("[project_board] recovery: %s already had a PR → in_review (%s)", fid, pr_url)
-                else:
-                    store.requeue(fid)
-                    log.info("[project_board] recovery: %s reset to ready (no PR — rebuild fresh)", fid)
-            except Exception:  # noqa: BLE001 — recovery is best-effort, per feature
-                log.warning("[project_board] recovery for %s failed", fid, exc_info=True)
+                log.info("[project_board] sweep: %s in_progress with no live drive", fid)
+                await self._reconcile_orphan(fid)
+            except Exception:  # noqa: BLE001
+                log.warning("[project_board] sweep reconcile for %s failed", fid, exc_info=True)
+        for fid in worktree.list_feature_worktrees(repo, self.root):
+            if fid in self._inflight_files:
+                continue  # a live drive owns this worktree
+            try:
+                f = store.get_feature(fid)
+                if f is None or f["board_state"] == "done":
+                    await worktree.reap_feature_worktree(repo, self.root, fid)
+                    log.info("[project_board] sweep: reaped orphaned worktree feat-%s", fid)
+            except Exception:  # noqa: BLE001
+                log.warning("[project_board] sweep reap for %s failed", fid, exc_info=True)
 
     # ── the puller ────────────────────────────────────────────────────────────
     async def _run(self):
@@ -166,6 +210,7 @@ async def _run(self):
             spawned = False
             try:
                 await self._maybe_poll_merges()
+                await self._maybe_sweep()
                 spawned = self._spawn_ready()
             except Exception:  # noqa: BLE001 — a bad tick must never kill the loop
                 log.exception("[project_board] loop tick failed")

protoagent.plugin.yaml

@@ -57,6 +57,9 @@ config:
                             # (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
+  health_sweep_interval_s: 300  # periodic self-heal: reclaim slots from dead drives
+                            # (in_progress with no live drive → ready) + reap orphaned
+                            # feat-<id> worktrees (feature gone/done). 0 disables it.
   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

tests/test_loop.py

@@ -433,3 +433,81 @@ async def _pr_url(branch, *, cwd="."):
     # bd-1 errored and was skipped; bd-2 still recovered.
     assert ("requeue", "bd-2") in store.calls
     assert all(c[1] != "bd-1" for c in store.calls)
+
+
+# ── periodic health sweep ───────────────────────────────────────────────────────
+
+
+class _SweepStore:
+    def __init__(self, in_progress=(), features=None):
+        self._in_progress = list(in_progress)
+        self._features = features or {}  # fid -> board_state
+        self.requeued = []
+
+    def list_features(self, state=None):
+        return [{"id": f} for f in self._in_progress] if state == "in_progress" else []
+
+    def requeue(self, fid):
+        self.requeued.append(fid)
+
+    def open_review(self, fid, *, pr_url):
+        pass
+
+    def get_feature(self, fid):
+        st = self._features.get(fid)
+        return {"id": fid, "board_state": st} if st else None
+
+
+async def test_sweep_reconciles_in_progress_with_no_live_drive(monkeypatch):
+    store = _SweepStore(in_progress=["bd-1", "bd-2"])
+    monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store)
+    monkeypatch.setattr(worktree, "list_feature_worktrees", lambda repo, root: [])
+
+    async def _no_pr(branch, *, cwd="."):
+        return ""
+
+    monkeypatch.setattr(worktree, "pr_url_for_branch", _no_pr)
+    loop = BoardLoop({})
+    loop._inflight_files = {"bd-2": {"a.py"}}  # bd-2 has a live drive → skip
+    await loop._sweep()
+    assert store.requeued == ["bd-1"]  # bd-1 (no PR, no drive) reset; bd-2 left alone
+
+
+async def test_sweep_reaps_orphaned_worktrees(monkeypatch):
+    store = _SweepStore(features={"bd-done": "done", "bd-rev": "in_review"})
+    monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store)
+    monkeypatch.setattr(worktree, "list_feature_worktrees", lambda repo, root: ["bd-done", "bd-rev", "bd-gone"])
+    reaped = []
+
+    async def _reap(repo, root, fid):
+        reaped.append(fid)
+
+    monkeypatch.setattr(worktree, "reap_feature_worktree", _reap)
+    await BoardLoop({})._sweep()
+    # done + missing feature → reaped; in_review keeps its worktree (CI-fail re-dispatch).
+    assert set(reaped) == {"bd-done", "bd-gone"}
+
+
+async def test_maybe_sweep_is_rate_limited(monkeypatch):
+    loop = BoardLoop({"health_sweep_interval_s": 300})
+    calls = []
+
+    async def _sweep():
+        calls.append(1)
+
+    monkeypatch.setattr(loop, "_sweep", _sweep)
+    clock = {"t": 1000.0}
+    monkeypatch.setattr("project_board.loop.time.monotonic", lambda: clock["t"])
+    await loop._maybe_sweep()  # first → sweeps
+    await loop._maybe_sweep()  # immediately → rate-limited
+    clock["t"] += 301
+    await loop._maybe_sweep()  # interval elapsed → sweeps again
+    assert len(calls) == 2
+
+
+async def test_sweep_off_when_interval_zero(monkeypatch):
+    loop = BoardLoop({"health_sweep_interval_s": 0})
+    called = []
+    monkeypatch.setattr(loop, "_sweep", lambda: called.append(1))
+    await loop._maybe_sweep()
+    assert called == []  # disabled → never sweeps

tests/test_worktree.py

@@ -162,6 +162,22 @@ async def test_pr_url_for_branch_found_and_absent(monkeypatch):
     assert await worktree.pr_url_for_branch("feat/bd-9", cwd="/repo") == ""
 
 
+# ── list_feature_worktrees: the health sweep's orphan enumeration ───────────────
+
+
+def test_list_feature_worktrees(tmp_path):
+    root = tmp_path / "wt"
+    (root / "feat-bd-1").mkdir(parents=True)
+    (root / "feat-bd-2").mkdir()
+    (root / "other-dir").mkdir()  # not a feat- dir
+    (root / "feat-stray-file").write_text("x")  # a file, not a worktree dir
+    assert set(worktree.list_feature_worktrees(str(tmp_path), "wt")) == {"bd-1", "bd-2"}
+
+
+def test_list_feature_worktrees_absent_dir(tmp_path):
+    assert worktree.list_feature_worktrees(str(tmp_path), "does-not-exist") == []
+
+
 # ── reap_feature_worktree: the shared id → worktree/branch reap ──────────────────
 
 

worktree.py

@@ -104,6 +104,18 @@ async def reap_feature_worktree(repo: str, worktrees_root: str, fid: str) -> Non
     await remove_worktree(repo, wt, f"feat/{fid}")
 
 
+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
+    dir listing); returns ``[]`` if the dir is absent."""
+    base = os.path.join(repo, worktrees_root)
+    try:
+        names = os.listdir(base)
+    except OSError:
+        return []
+    return [n[len("feat-") :] for n in names if n.startswith("feat-") and os.path.isdir(os.path.join(base, n))]
+
+
 async def dispatch_coder(coder, worktree: str, prompt: str, *, timeout: float | None = None) -> str:
     """Dispatch the coder (an ``acp`` Delegate) scoped to ``worktree``.