perf(cli): cut hermes -w startup from ~14s to ~1.8s by parallelizing + caching the worktree prune

`_prune_stale_worktrees` runs synchronously before the banner on every
`hermes -w` launch and shells out to git several times per candidate
worktree. On a repo with dozens of accumulated worktrees this dominated
startup: measured 18.5s of a 20.6s cold start, 11.6s on warm repeats.

The `git cherry` patch-equivalence probe was the single largest cost
(9.4s of 11.5s across 24 trees). It is also pure waste on repeat runs: a
tree preserved because it holds unpushed work is re-diff-hashed on every
launch, forever, always reaching the same verdict. On this repo 19 of 24
aged trees were unreapable, so ~11s per launch bought zero reaps.

Two changes, both verdict-preserving:

- Split the loop into a stat-only age filter, a parallel read-only
  classification phase (thread pool, bounded to min(8, cpu_count)), and
  a serial mutation phase. Only reads are concurrent; unlock/remove/
  branch -D stay ordered, so log output and removal order are unchanged.
  A pool failure falls back to serial rather than blocking startup.
- Memoize `git cherry` verdicts to
  $HERMES_HOME/cache/worktree_merge_verdicts.json, keyed on the exact
  `(base_sha, head_sha, max_ahead)` range the verdict was computed from.
  Because that key is the complete input to the git call, a cache hit is
  identical to recomputation by construction: if either ref moves, the
  key changes and real git runs again. Bounded to 1000 entries, written
  atomically, and a corrupt/hand-edited cache degrades to recomputation.

Measured on a 44-worktree repo (24 aged candidates):

  _prune_stale_worktrees   before 11.5s   after 2.05s cold / 0.42s warm
  hermes -w to banner      before 13.9s   after 1.79s

Work-preservation is unchanged: all 44 worktrees survived, and every
dirty/unpushed/live-locked guard still fires. Verified the 24 real-tree
verdicts are byte-identical across serial, cold-cache, and warm-cache
runs.

Tests: 75/75 in tests/cli/test_worktree.py (67 existing + 8 new). The
new cache tests were sabotage-verified — swapping the exact-sha key for a
naive path-only key makes two of them fail by deleting a worktree that
had gained unmerged work, which is the data-loss case the key prevents.

cli.py

@@ -1760,8 +1760,68 @@ def _worktree_is_dirty(worktree_path: str, timeout: int = 10) -> bool:
         return True
 
 
+# Upper bound on retained `git cherry` verdict entries (see
+# _save_worktree_merge_cache). Each entry is ~90 bytes, so this caps the cache
+# near 90 KB even on a repo that churns thousands of worktree branches.
+_WORKTREE_MERGE_CACHE_MAX = 1000
+
+
+def _worktree_merge_cache_path() -> Path:
+    """Path of the patch-equivalence verdict cache (profile-aware)."""
+    return get_hermes_home() / "cache" / "worktree_merge_verdicts.json"
+
+
+def _load_worktree_merge_cache() -> Dict[str, bool]:
+    """Load the ``git cherry`` verdict cache. Missing/corrupt cache = empty."""
+    try:
+        raw = json.loads(
+            _worktree_merge_cache_path().read_text(encoding="utf-8")
+        )
+    except Exception:
+        return {}
+    if not isinstance(raw, dict):
+        return {}
+    entries = raw.get("verdicts")
+    if not isinstance(entries, dict):
+        return {}
+    # Only keep well-formed bool verdicts — a hand-edited or partially written
+    # cache must never inject a non-bool into the prune decision.
+    return {k: v for k, v in entries.items() if isinstance(v, bool)}
+
+
+def _save_worktree_merge_cache(verdicts: Dict[str, bool]) -> None:
+    """Persist the verdict cache atomically. Best-effort — never raises.
+
+    Bounded to the most recent ``_WORKTREE_MERGE_CACHE_MAX`` entries so the
+    file can't grow without limit across thousands of sessions.
+    """
+    path = _worktree_merge_cache_path()
+    tmp = None
+    try:
+        items = list(verdicts.items())
+        if len(items) > _WORKTREE_MERGE_CACHE_MAX:
+            items = items[-_WORKTREE_MERGE_CACHE_MAX:]
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(f".{os.getpid()}.tmp")
+        tmp.write_text(
+            json.dumps({"version": 1, "verdicts": dict(items)}),
+            encoding="utf-8",
+        )
+        os.replace(str(tmp), str(path))
+    except Exception as e:
+        logger.debug("Could not persist worktree merge cache: %s", e)
+        if tmp is not None:
+            try:
+                tmp.unlink()
+            except Exception:
+                pass
+
+
 def _worktree_commits_all_merged_upstream(
-    worktree_path: str, timeout: int = 30, max_ahead: int = 20
+    worktree_path: str,
+    timeout: int = 30,
+    max_ahead: int = 20,
+    cache: Optional[Dict[str, bool]] = None,
 ) -> bool:
     """Return whether every local-only commit is patch-equivalent to a commit
     already on the default upstream branch.
@@ -1776,6 +1836,15 @@ def _worktree_commits_all_merged_upstream(
     Bounded: skips (returns False) when the branch is more than ``max_ahead``
     commits ahead — a stale-base tree, too expensive to diff-hash and unlikely
     to be a merged scratch branch. Fails SAFE toward False (preserve).
+
+    ``git cherry`` diff-hashes every commit in the range, which on a large repo
+    costs ~0.2-1.0s per worktree — and a tree preserved for unpushed work is
+    re-tested on *every* startup, forever, always reaching the same answer. When
+    *cache* is provided, the verdict is memoized against
+    ``(base_sha, head_sha, max_ahead)``: the exact inputs ``git cherry``
+    consumes. A cache hit is therefore identical to recomputation by
+    construction — if either ref moves the key changes and the real git call
+    runs again.
     """
     import subprocess
 
@@ -1795,6 +1864,27 @@ def _worktree_commits_all_merged_upstream(
         return False
 
     try:
+        # Resolve both endpoints to shas up front. These are the complete
+        # inputs to the range below, so they form an exact cache key. Cheap
+        # (~1ms) relative to the diff-hashing `git cherry` they guard.
+        cache_key = None
+        if cache is not None:
+            revs = subprocess.run(
+                ["git", "rev-parse", f"{base}^{{commit}}", "HEAD^{commit}"],
+                capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, cwd=worktree_path,
+            )
+            if revs.returncode == 0:
+                shas = revs.stdout.split()
+                if len(shas) == 2:
+                    cache_key = f"{shas[0]}..{shas[1]}:{max_ahead}"
+                    if cache_key in cache:
+                        return cache[cache_key]
+
+        def _memo(verdict: bool) -> bool:
+            if cache is not None and cache_key is not None:
+                cache[cache_key] = verdict
+            return verdict
+
         ahead = subprocess.run(
             ["git", "rev-list", "--count", f"{base}..HEAD"],
             capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, cwd=worktree_path,
@@ -1803,9 +1893,9 @@ def _worktree_commits_all_merged_upstream(
             return False
         count = int(ahead.stdout.strip() or "0")
         if count == 0:
-            return True
+            return _memo(True)
         if count > max_ahead:
-            return False
+            return _memo(False)
 
         cherry = subprocess.run(
             ["git", "cherry", base, "HEAD"],
@@ -1815,7 +1905,7 @@ def _worktree_commits_all_merged_upstream(
             return False
         lines = [ln for ln in cherry.stdout.splitlines() if ln.strip()]
         # "-" = patch-equivalent commit exists upstream; "+" = unique local work
-        return bool(lines) and all(ln.startswith("-") for ln in lines)
+        return _memo(bool(lines) and all(ln.startswith("-") for ln in lines))
     except Exception:
         return False
 
@@ -2074,6 +2164,21 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
 
     Also prunes orphaned ``hermes/*`` and ``pr-*`` local branches that
     have no corresponding worktree.
+
+    Performance: this runs on the startup path of every ``hermes -w`` session,
+    and each candidate tree costs several git subprocesses (the ``git cherry``
+    patch-equivalence probe dominates at ~0.2-1.0s on a large repo). With
+    dozens of accumulated worktrees the serial version added ~11-18s of latency
+    before the banner. Two changes keep the decisions byte-identical while
+    removing nearly all of that:
+
+    1. The read-only classification of each tree (dirty / unpushed / merged /
+       lock state) is independent per tree, so it runs on a thread pool. Only
+       the mutating phase (unlock, remove, branch -D) stays serial and ordered.
+    2. ``git cherry`` verdicts are memoized on disk keyed by the exact
+       ``(base_sha, head_sha)`` range they were computed from, so a tree
+       preserved for unpushed work is not re-diff-hashed on every subsequent
+       startup.
     """
     import re
     import subprocess
@@ -2091,7 +2196,11 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
     # dispatcher-driven lifecycle (hermes kanban gc) — never touch them here.
     kanban_re = re.compile(r"^t_[0-9a-f]+$")
 
-    for entry in worktrees_dir.iterdir():
+    # ── Phase 1: age filter (no subprocesses) ───────────────────────────────
+    # Cheap stat-only pass so the thread pool below is sized to the trees that
+    # actually need git work, not to everything on disk.
+    candidates: list = []
+    for entry in sorted(worktrees_dir.iterdir()):
         if not entry.is_dir() or kanban_re.match(entry.name):
             continue
 
@@ -2102,7 +2211,6 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         soft_cutoff = now - (tier_hours * 3600)
         hard_cutoff = now - (tier_hours * 3 * 3600)
 
-        # Check age
         try:
             mtime = entry.stat().st_mtime
             if mtime > soft_cutoff:
@@ -2110,24 +2218,42 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         except Exception:
             continue
 
-        force = mtime <= hard_cutoff  # Aggressive tier — reap clean trees
+        candidates.append((entry, mtime, mtime <= hard_cutoff))
 
+    if not candidates:
+        _prune_orphaned_branches(repo_root)
+        return
+
+    # ── Phase 2: classify in parallel (read-only git queries) ───────────────
+    # Every check here is a read-only git query against a distinct worktree, so
+    # they are safe to run concurrently (git takes no repo-wide lock for these,
+    # and each has its own index). Verdicts are collected and applied serially
+    # below so removal order and log output stay deterministic.
+    merge_cache = _load_worktree_merge_cache()
+    cache_size_before = len(merge_cache)
+    cache_lock = threading.Lock()
+
+    def _classify(item):
+        entry, mtime, force = item
         # Never delete real work, regardless of age or tier. Uncommitted
         # changes and unpushed commits may be a crashed session's in-flight
         # work; only clean, fully-merged/pushed trees (the scratch trees that
         # actually cause .worktrees/ bloat) are ever reaped.
         if _worktree_is_dirty(str(entry), timeout=5):
-            if mtime <= stale_work_cutoff:
-                preserved_stale.append(f"{entry.name} (uncommitted changes)")
-            continue
+            return (entry, mtime, force, "dirty", None)
         if _worktree_has_unpushed_commits(str(entry), timeout=5):
             # Squash-merge escape hatch: commits unreachable from any remote
             # ref but patch-equivalent to upstream commits are merged work,
             # not unpushed work.
-            if not _worktree_commits_all_merged_upstream(str(entry), timeout=30):
-                if mtime <= stale_work_cutoff:
-                    preserved_stale.append(f"{entry.name} (unpushed commits)")
-                continue
+            with cache_lock:
+                snapshot = dict(merge_cache)
+            merged = _worktree_commits_all_merged_upstream(
+                str(entry), timeout=30, cache=snapshot
+            )
+            with cache_lock:
+                merge_cache.update(snapshot)
+            if not merged:
+                return (entry, mtime, force, "unpushed", None)
 
         # Respect git-native session locks. A lock owned by a still-running
         # hermes process means the worktree is actively in use — never touch
@@ -2136,8 +2262,42 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         # otherwise dead-locked worktrees pile up indefinitely.
         lock_state = _worktree_lock_is_live(repo_root, str(entry), timeout=5)
         if lock_state == "live":
+            return (entry, mtime, force, "locked-live", None)
+        return (entry, mtime, force, "reap", lock_state)
+
+    # Bounded pool: enough to hide git's per-process startup latency without
+    # spawning dozens of concurrent git processes on a small machine.
+    workers = max(1, min(8, (os.cpu_count() or 4), len(candidates)))
+    try:
+        if workers > 1:
+            with concurrent.futures.ThreadPoolExecutor(
+                max_workers=workers, thread_name_prefix="hermes-wt-prune"
+            ) as pool:
+                verdicts = list(pool.map(_classify, candidates))
+        else:
+            verdicts = [_classify(c) for c in candidates]
+    except Exception as e:
+        # Never let a pool failure block startup — fall back to serial.
+        logger.debug("Parallel worktree classification failed (%s); serial", e)
+        verdicts = [_classify(c) for c in candidates]
+
+    if len(merge_cache) != cache_size_before:
+        _save_worktree_merge_cache(merge_cache)
+
+    # ── Phase 3: mutate serially (unlock / remove / branch -D) ──────────────
+    for entry, mtime, force, verdict, lock_state in verdicts:
+        if verdict == "dirty":
+            if mtime <= stale_work_cutoff:
+                preserved_stale.append(f"{entry.name} (uncommitted changes)")
+            continue
+        if verdict == "unpushed":
+            if mtime <= stale_work_cutoff:
+                preserved_stale.append(f"{entry.name} (unpushed commits)")
+            continue
+        if verdict == "locked-live":
             logger.debug("Skipping live-locked worktree: %s", entry.name)
             continue
+
         if lock_state == "dead":
             try:
                 subprocess.run(

tests/cli/test_worktree.py

@@ -1324,3 +1324,200 @@ class TestWidenedPruner:
         assert not any(
             "salvage-new-work" in rec.getMessage() for rec in caplog.records
         ), "under-7d preserved work should not warn"
+
+
+class TestMergeVerdictCache:
+    """The ``git cherry`` patch-equivalence probe is memoized on disk because it
+    dominates ``hermes -w`` startup (~0.2-1.0s per worktree, re-run on every
+    launch for every tree preserved as unpushed).
+
+    The invariant that makes caching safe: the verdict is a pure function of the
+    ``(base_sha, head_sha)`` range, so a cache entry is only reused while BOTH
+    endpoints are unchanged. These tests pin that relationship rather than any
+    specific timing or cache contents.
+    """
+
+    _age = staticmethod(TestWidenedPruner._age)
+    _mk = staticmethod(TestWidenedPruner._mk)
+    _merge_upstream = staticmethod(TestWidenedPruner._merge_upstream)
+
+    def test_cache_hit_matches_uncached_verdict(self, git_repo):
+        """A cached verdict must equal what the real git call returns."""
+        import cli
+        wt, sha = self._mk(git_repo, "hermes-cachehit", commit=True)
+        self._merge_upstream(git_repo, sha)
+
+        uncached = cli._worktree_commits_all_merged_upstream(str(wt))
+        cache = {}
+        cold = cli._worktree_commits_all_merged_upstream(str(wt), cache=cache)
+        warm = cli._worktree_commits_all_merged_upstream(str(wt), cache=cache)
+
+        assert cache, "verdict should have been memoized"
+        assert uncached is cold is warm is True
+
+    def test_cache_records_negative_verdicts_too(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "hermes-cacheneg", commit=True)
+        cache = {}
+        assert cli._worktree_commits_all_merged_upstream(str(wt), cache=cache) is False
+        assert cli._worktree_commits_all_merged_upstream(str(wt), cache=cache) is False
+        assert set(cache.values()) == {False}
+
+    def test_new_commit_invalidates_cached_verdict(self, git_repo):
+        """Moving HEAD must not reuse the old entry — the key includes head_sha.
+
+        This is the guard against the cache turning a 'merged, reapable' verdict
+        into a stale approval to delete a tree that has since gained real work.
+        """
+        import cli
+        wt, sha = self._mk(git_repo, "hermes-moves", commit=True)
+        self._merge_upstream(git_repo, sha)
+
+        cache = {}
+        assert cli._worktree_commits_all_merged_upstream(str(wt), cache=cache) is True
+        key_after_merge = set(cache)
+
+        # New local-only work lands in the worktree.
+        (wt / "more.txt").write_text("unmerged work\n")
+        subprocess.run(["git", "add", "more.txt"], cwd=wt, capture_output=True)
+        subprocess.run(["git", "commit", "-m", "new work"], cwd=wt, capture_output=True)
+
+        assert cli._worktree_commits_all_merged_upstream(str(wt), cache=cache) is False
+        assert set(cache) != key_after_merge, "moved HEAD must produce a new key"
+
+    def test_cached_tree_with_new_work_is_still_preserved(self, git_repo):
+        """End-to-end: a warm cache must never let the pruner delete new work."""
+        import cli
+        wt, sha = self._mk(git_repo, "hermes-warmsafe", commit=True)
+        self._merge_upstream(git_repo, sha)
+
+        # Warm the on-disk cache with the 'fully merged' verdict.
+        cli._prune_stale_worktrees(str(git_repo))
+        assert not wt.exists(), "merged tree should be reaped on the cold pass"
+
+        # Recreate the same-named tree, now carrying unmerged work.
+        wt2, _ = self._mk(git_repo, "hermes-warmsafe", commit=True)
+        (wt2 / "precious.txt").write_text("do not delete\n")
+        subprocess.run(["git", "add", "precious.txt"], cwd=wt2, capture_output=True)
+        subprocess.run(["git", "commit", "-m", "precious"], cwd=wt2, capture_output=True)
+        self._age(wt2, 100)
+
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt2.exists(), "warm cache must not authorize deleting unmerged work"
+
+    def test_corrupt_cache_file_is_ignored(self, git_repo, monkeypatch, tmp_path):
+        """A garbage cache must degrade to recomputation, not crash startup."""
+        import json
+        import cli
+        bad = tmp_path / "worktree_merge_verdicts.json"
+        bad.write_text("{not json at all")
+        monkeypatch.setattr(cli, "_worktree_merge_cache_path", lambda: bad)
+        assert cli._load_worktree_merge_cache() == {}
+
+        # Non-bool verdicts must be dropped rather than fed into the decision.
+        bad.write_text(json.dumps({"version": 1, "verdicts": {"a..b:20": "yes"}}))
+        assert cli._load_worktree_merge_cache() == {}
+
+        wt, _ = self._mk(git_repo, "hermes-corrupt", commit=True)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists(), "unmerged work survives a corrupt cache"
+
+    def test_cache_is_bounded(self, monkeypatch, tmp_path):
+        """The cache file must not grow without limit across sessions."""
+        import cli
+        path = tmp_path / "verdicts.json"
+        monkeypatch.setattr(cli, "_worktree_merge_cache_path", lambda: path)
+        monkeypatch.setattr(cli, "_WORKTREE_MERGE_CACHE_MAX", 10)
+
+        cli._save_worktree_merge_cache({f"sha{i}..sha{i}:20": True for i in range(50)})
+        assert len(cli._load_worktree_merge_cache()) == 10
+
+
+class TestPruneParallelEquivalence:
+    """Classification runs on a thread pool, so verdicts must not depend on
+    concurrency: the same mixed board must produce the same survivors whether
+    the pool has 1 worker or many."""
+
+    _age = staticmethod(TestWidenedPruner._age)
+    _mk = staticmethod(TestWidenedPruner._mk)
+    _merge_upstream = staticmethod(TestWidenedPruner._merge_upstream)
+
+    def _board(self, git_repo, tag=""):
+        """A board covering every verdict branch: reapable, dirty, unpushed."""
+        names = {}
+        for i in range(3):
+            n = f"hermes-merged{tag}{i}"
+            wt, sha = self._mk(git_repo, n, commit=True)
+            self._merge_upstream(git_repo, sha)
+            names[n] = wt
+        for i in range(3):
+            n = f"hermes-unpushed{tag}{i}"
+            wt, _ = self._mk(git_repo, n, commit=True)
+            names[n] = wt
+        for i in range(2):
+            n = f"hermes-dirty{tag}{i}"
+            wt, _ = self._mk(git_repo, n, dirty=True)
+            names[n] = wt
+        n = f"hermes-fresh{tag}"
+        wt, _ = self._mk(git_repo, n, commit=True, age_h=1)
+        names[n] = wt
+        # Every tree must really exist, otherwise the survivor comparison below
+        # is vacuous (a failed `git worktree add` would look like a reap).
+        for name, p in names.items():
+            assert p.exists(), f"fixture worktree {name} was not created"
+        return names
+
+    @staticmethod
+    def _kinds(survivors):
+        """Reduce survivor names to their kind, dropping the phase tag."""
+        out = set()
+        for n in survivors:
+            for kind in ("merged", "unpushed", "dirty", "fresh"):
+                if n.startswith(f"hermes-{kind}"):
+                    out.add(kind)
+        return out
+
+    def test_single_and_multi_worker_agree(self, git_repo, monkeypatch):
+        import cli
+
+        # Phase A — force the serial path (pool sized to 1 worker).
+        board = self._board(git_repo, tag="a")
+        monkeypatch.setattr(cli.os, "cpu_count", lambda: 1)
+        cli._prune_stale_worktrees(str(git_repo))
+        serial = self._kinds({n for n, p in board.items() if p.exists()})
+
+        # Phase B — an independent board (distinct branch names so creation
+        # can't collide with phase A's leftover refs) run through a real pool.
+        try:
+            cli._worktree_merge_cache_path().unlink()
+        except Exception:
+            pass
+        board2 = self._board(git_repo, tag="b")
+        monkeypatch.setattr(cli.os, "cpu_count", lambda: 8)
+        cli._prune_stale_worktrees(str(git_repo))
+        parallel = self._kinds({n for n, p in board2.items() if p.exists()})
+
+        assert parallel == serial, (
+            f"pool width changed the outcome: serial={serial} parallel={parallel}"
+        )
+        # Sanity: the board exercised both outcomes, so the equality above is
+        # not comparing two empty (or two identical-because-nothing-ran) sets.
+        assert serial == {"dirty", "unpushed", "fresh"}, serial
+
+    def test_pool_failure_falls_back_to_serial(self, git_repo, monkeypatch):
+        """A ThreadPoolExecutor failure must not block startup."""
+        import cli
+
+        wt, sha = self._mk(git_repo, "hermes-poolfail", commit=True)
+        self._merge_upstream(git_repo, sha)
+
+        class _Boom:
+            def __init__(self, *a, **kw):
+                raise RuntimeError("cannot start thread")
+
+        monkeypatch.setattr(cli.concurrent.futures, "ThreadPoolExecutor", _Boom)
+        monkeypatch.setattr(cli.os, "cpu_count", lambda: 8)
+
+        cli._prune_stale_worktrees(str(git_repo))
+        assert not wt.exists(), "serial fallback must still reap the merged tree"
+