feat(loop): crash recovery on boot — resume features stranded mid-drive

README.md

@@ -35,6 +35,11 @@ board to a PR — or fork it as a starting point.
   (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.
+- **Resilience** — every `await` in a drive is bounded (a coder dispatch is hard-capped
+  by `coder_timeout_s`); **transient** failures (rate-limit / network / merge-conflict)
+  retry with backoff while **capability** failures (no diff / timeout) escalate a tier
+  or block; and on restart the loop **recovers** features stranded mid-build (adopt an
+  already-opened PR → `in_review`, else reset → `ready`).
 - **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`.

loop.py

@@ -126,8 +126,35 @@ async def stop(self):
             except Exception:  # noqa: BLE001 — teardown must not raise out of shutdown
                 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 _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."""
+        store = self._store()
+        repo = self._store_kw["repo"]
+        for f in store.list_features(state="in_progress"):
+            fid = f["id"]
+            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)
+
     # ── the puller ────────────────────────────────────────────────────────────
     async def _run(self):
+        try:
+            await self._recover()
+        except Exception:  # noqa: BLE001 — recovery must never stop the loop from starting
+            log.exception("[project_board] crash recovery failed")
         while not self._stop.is_set():
             spawned = False
             try:

tests/test_loop.py

@@ -323,3 +323,52 @@ async def test_merge_poll_off_never_polls(monkeypatch):
     monkeypatch.setattr(loop, "_poll_merges", lambda: called.append(1))
     await loop._maybe_poll_merges()
     assert called == []  # disabled → the poll is never reached
+
+
+# ── crash recovery on boot ──────────────────────────────────────────────────────
+
+
+class _RecoverStore:
+    def __init__(self, in_progress):
+        self._in_progress = in_progress
+        self.calls = []
+
+    def list_features(self, state=None):
+        return self._in_progress if state == "in_progress" else []
+
+    def open_review(self, fid, *, pr_url):
+        self.calls.append(("open_review", fid, pr_url))
+
+    def requeue(self, fid):
+        self.calls.append(("requeue", fid))
+
+
+async def test_recover_adopts_an_open_pr_else_resets_to_ready(monkeypatch):
+    store = _RecoverStore([{"id": "bd-1"}, {"id": "bd-2"}])
+    monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store)
+
+    async def _pr_url(branch, *, cwd="."):
+        return "https://example/pr/1" if branch == "feat/bd-1" else ""
+
+    monkeypatch.setattr(worktree, "pr_url_for_branch", _pr_url)
+    await BoardLoop({})._recover()
+    # bd-1 already had a PR (crash between open_pr and open_review) → adopt → in_review.
+    assert ("open_review", "bd-1", "https://example/pr/1") in store.calls
+    # bd-2 has no PR → reset to ready for a clean rebuild.
+    assert ("requeue", "bd-2") in store.calls
+
+
+async def test_recover_is_resilient_to_a_per_feature_error(monkeypatch):
+    store = _RecoverStore([{"id": "bd-1"}, {"id": "bd-2"}])
+    monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store)
+
+    async def _pr_url(branch, *, cwd="."):
+        if branch == "feat/bd-1":
+            raise RuntimeError("gh exploded")
+        return ""
+
+    monkeypatch.setattr(worktree, "pr_url_for_branch", _pr_url)
+    await BoardLoop({})._recover()  # must not raise
+    # 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)

tests/test_worktree.py

@@ -152,6 +152,16 @@ async def test_pr_is_merged(monkeypatch, gh_state, expected):
     assert await worktree.pr_is_merged("https://example/pr/1", cwd="/repo") is expected
 
 
+# ── pr_url_for_branch: the crash-recovery probe ─────────────────────────────────
+
+
+async def test_pr_url_for_branch_found_and_absent(monkeypatch):
+    _install(monkeypatch, FakeGit(), FakeGh({"view": (0, "https://example/pr/9", "")}))
+    assert await worktree.pr_url_for_branch("feat/bd-9", cwd="/repo") == "https://example/pr/9"
+    _install(monkeypatch, FakeGit(), FakeGh({"view": (1, "", "no pull requests found")}))
+    assert await worktree.pr_url_for_branch("feat/bd-9", cwd="/repo") == ""
+
+
 # ── reap_feature_worktree: the shared id → worktree/branch reap ──────────────────
 
 

worktree.py

@@ -201,3 +201,11 @@ async def pr_is_merged(pr_url: str, *, cwd: str = ".") -> bool:
     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"
+
+
+async def pr_url_for_branch(branch: str, *, cwd: str = ".") -> str:
+    """The URL of the PR whose head is ``branch``, or ``""`` if there is none — used
+    by crash recovery to tell a feature that already opened a PR (and just needs
+    adopting → in_review) from one that needs a fresh rebuild."""
+    rc, out, _err = await _gh("pr", "view", branch, "--json", "url", "--jq", ".url", cwd=cwd)
+    return out.strip() if rc == 0 else ""