From 016f0cc08062389c7013bed5554e4dbdab0a3b64 Mon Sep 17 00:00:00 2001 From: Josh Mabry Date: Fri, 12 Jun 2026 23:23:31 -0700 Subject: [PATCH] =?UTF-8?q?feat(loop):=20periodic=20health=20sweep=20?= =?UTF-8?q?=E2=80=94=20self-heal=20stale=20drives=20+=20orphaned=20worktre?= =?UTF-8?q?es?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A long-running loop accumulates two kinds of cruft that nothing reclaimed: features stuck in_progress after a drive died without finishing (holding a board slot), and feat- worktrees left behind by a missed reap. A periodic sweep (health_sweep_interval_s, default 300s; 0 disables) now self-heals both: - in_progress features with no live drive (not in self._inflight_files) → reconciled the same way boot recovery does (adopt an open PR → in_review, else reset → ready). Boot recovery and the sweep now share _reconcile_orphan. - feat- worktrees whose feature is gone or already done → reaped (in_review keeps its worktree for a CI-fail re-dispatch; a live drive's worktree is left alone). worktree.list_feature_worktrees enumerates the feat- dirs under worktrees_root. Second of the P1 concurrency-correctness ports. Tests: +6 (110 total) — list_feature_worktrees (dirs only, absent dir); _sweep reconciles a driveless in_progress (skips a driven one) and reaps done/missing worktrees (keeps in_review); _maybe_sweep rate-limit + disabled. Co-Authored-By: Claude Fable 5 --- loop.py | 75 ++++++++++++++++++++++++++++++++-------- protoagent.plugin.yaml | 3 ++ tests/test_loop.py | 78 ++++++++++++++++++++++++++++++++++++++++++ tests/test_worktree.py | 16 +++++++++ worktree.py | 12 +++++++ 5 files changed, 169 insertions(+), 15 deletions(-) diff --git a/loop.py b/loop.py index 3b17715..f176038 100644 --- a/loop.py +++ b/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-`` 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") diff --git a/protoagent.plugin.yaml b/protoagent.plugin.yaml index f28b340..10d7312 100644 --- a/protoagent.plugin.yaml +++ b/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- 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 diff --git a/tests/test_loop.py b/tests/test_loop.py index 8c23d90..19ee34f 100644 --- a/tests/test_loop.py +++ b/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 diff --git a/tests/test_worktree.py b/tests/test_worktree.py index b2edf63..c0e2c3f 100644 --- a/tests/test_worktree.py +++ b/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 ────────────────── diff --git a/worktree.py b/worktree.py index e3067e7..cebe307 100644 --- a/worktree.py +++ b/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-`` worktree dir under + ``/`` — 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``.