fix(cli): widen startup worktree pruning to all .worktrees/ trees and detect squash-merged work (#69831)

The startup pruner only considered directories named hermes-* (the
hermes -w scratch trees), so salvage/review/port lanes created with raw
'git worktree add' accumulated forever — a real checkout reached 117
directories / 26 GB with trees dating back months. Two further leaks:
squash-merged branches' local commits stay unreachable from
refs/remotes/* forever, so the unpushed-commits guard preserved fully
merged scratch trees indefinitely; and preserved trees rotted silently
with no visibility.

- Pruner now covers every directory under .worktrees/ except kanban
  task trees (t_<hex>, owned by 'hermes kanban gc'). Named (non
  hermes-*) trees get a 3x timeline (72h soft / 9d hard) since they
  were created deliberately.
- New _worktree_commits_all_merged_upstream(): git-cherry
  patch-equivalence check against origin/HEAD|main|master, bounded at
  20 commits ahead, fails safe toward preserve. Lets the pruner reap
  trees whose every local-only commit already landed upstream via
  squash-merge/cherry-pick.
- Dirty guard now applies at every tier (previously the 24-72h tier
  skipped it — it only survived because the unpushed check usually
  caught the same trees).
- Trees preserved for unpushed/dirty reasons older than 7 days are
  listed in a single WARNING so in-flight work can't rot silently.
- tips.py text updated; 13 new behavior-contract tests.

cli.py

@@ -1744,6 +1744,66 @@ def _worktree_is_dirty(worktree_path: str, timeout: int = 10) -> bool:
         return True
 
 
+def _worktree_commits_all_merged_upstream(
+    worktree_path: str, timeout: int = 30, max_ahead: int = 20
+) -> bool:
+    """Return whether every local-only commit is patch-equivalent to a commit
+    already on the default upstream branch.
+
+    The dominant ``.worktrees/`` leak: a branch is pushed, its PR is
+    squash-merged (or cherry-picked), and the remote branch is deleted. The
+    local commits are then unreachable from ``refs/remotes/*`` forever, so the
+    unpushed-commits guard preserves the worktree indefinitely even though its
+    content is fully merged. ``git cherry`` detects patch-equivalence, letting
+    the pruner reap these.
+
+    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).
+    """
+    import subprocess
+
+    base = None
+    for candidate in ("origin/HEAD", "origin/main", "origin/master"):
+        try:
+            probe = subprocess.run(
+                ["git", "rev-parse", "--verify", "--quiet", candidate],
+                capture_output=True, text=True, timeout=timeout, cwd=worktree_path,
+            )
+            if probe.returncode == 0 and probe.stdout.strip():
+                base = candidate
+                break
+        except Exception:
+            return False
+    if base is None:
+        return False
+
+    try:
+        ahead = subprocess.run(
+            ["git", "rev-list", "--count", f"{base}..HEAD"],
+            capture_output=True, text=True, timeout=timeout, cwd=worktree_path,
+        )
+        if ahead.returncode != 0:
+            return False
+        count = int(ahead.stdout.strip() or "0")
+        if count == 0:
+            return True
+        if count > max_ahead:
+            return False
+
+        cherry = subprocess.run(
+            ["git", "cherry", base, "HEAD"],
+            capture_output=True, text=True, timeout=timeout, cwd=worktree_path,
+        )
+        if cherry.returncode != 0:
+            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)
+    except Exception:
+        return False
+
+
 def _worktree_lock_is_live(repo_root: str, worktree_path: str, timeout: int = 10):
     """Classify a worktree's git lock as live, dead, or absent.
 
@@ -1950,11 +2010,21 @@ def _run_checkpoint_auto_maintenance() -> None:
 def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
     """Remove stale worktrees and orphaned branches on startup.
 
-    Age-based tiers (aggressive cleanup keeps ``.worktrees/`` from growing
-    unbounded):
-    - Under max_age_hours (24h): skip — session may still be active.
-    - 24h–72h: remove if no unpushed commits.
-    - Over 72h: force remove regardless (nothing should sit this long).
+    Covers EVERY directory under ``.worktrees/`` except kanban task trees
+    (``t_<hex>`` — owned by the kanban dispatcher's own gc). Scratch trees
+    created by ``hermes -w`` (``hermes-*``) age out fast; named trees created
+    manually for salvage/review lanes age out on a slower schedule:
+
+    - ``hermes-*``: skip under 24h; reap 24h+ when clean and merged/pushed;
+      72h+ is the aggressive tier (still never deletes real work).
+    - named trees: same logic at 3x the timeline (72h soft / 9d hard).
+
+    Work-preservation guards (all tiers, any age):
+    - uncommitted changes (dirty) — never removed;
+    - unpushed commits — never removed, UNLESS every local-only commit is
+      patch-equivalent to a commit already on upstream (``git cherry``): the
+      squash-merged-PR case, which is the dominant ``.worktrees/`` leak since
+      those commits stay unreachable from ``refs/remotes/*`` forever.
 
     Lock handling (orthogonal to age): ``hermes -w`` locks each worktree with
     reason ``hermes pid=<pid>`` so a concurrent hermes process leaves an in-use
@@ -1967,9 +2037,14 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
     removal never orphans the branch (which would drop easy reachability of any
     commits still in the worktree).
 
+    Preserved-work visibility: trees skipped for unpushed/dirty reasons that
+    are older than 7 days are listed in a single WARNING so real in-flight
+    work can't rot silently.
+
     Also prunes orphaned ``hermes/*`` and ``pr-*`` local branches that
     have no corresponding worktree.
     """
+    import re
     import subprocess
     import time
 
@@ -1979,13 +2054,23 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         return
 
     now = time.time()
-    soft_cutoff = now - (max_age_hours * 3600)       # 24h default
-    hard_cutoff = now - (max_age_hours * 3 * 3600)   # 72h default
+    stale_work_cutoff = now - (7 * 24 * 3600)
+    preserved_stale: list = []
+    # Kanban task worktrees (<repo>/.worktrees/t_<hex>) have their own
+    # 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():
-        if not entry.is_dir() or not entry.name.startswith("hermes-"):
+        if not entry.is_dir() or kanban_re.match(entry.name):
             continue
 
+        # Scratch trees (hermes-*) age out on the default schedule; named
+        # trees (salvage/review lanes someone created deliberately) get 3x.
+        scratch = entry.name.startswith("hermes-")
+        tier_hours = max_age_hours if scratch else max_age_hours * 3
+        soft_cutoff = now - (tier_hours * 3600)
+        hard_cutoff = now - (tier_hours * 3 * 3600)
+
         # Check age
         try:
             mtime = entry.stat().st_mtime
@@ -1994,19 +2079,24 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         except Exception:
             continue
 
-        force = mtime <= hard_cutoff  # Over 72h — reap aggressively
+        force = mtime <= hard_cutoff  # Aggressive tier — reap clean trees
 
-        # Never delete real work, regardless of age. Unpushed commits and
-        # uncommitted changes may be a crashed session's in-flight work; the
-        # >72h tier reaps only abandoned *clean, fully-pushed* worktrees (the
-        # scratch trees that actually cause .worktrees/ bloat).
+        # 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
         if _worktree_has_unpushed_commits(str(entry), timeout=5):
-            continue  # Has unpushed commits or can't check — skip
-        if not force:
-            # 24h–72h tier is conservative: unpushed check above is enough.
-            pass
-        elif _worktree_is_dirty(str(entry), timeout=5):
-            continue  # >72h but dirty — preserve uncommitted work
+            # 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
 
         # Respect git-native session locks. A lock owned by a still-running
         # hermes process means the worktree is actively in use — never touch
@@ -2055,6 +2145,13 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         except Exception as e:
             logger.debug("Failed to prune worktree %s: %s", entry.name, e)
 
+    if preserved_stale:
+        logger.warning(
+            "Preserving %d worktree(s) older than 7 days with unmerged work "
+            "(push or remove them to reclaim disk): %s",
+            len(preserved_stale), ", ".join(sorted(preserved_stale)),
+        )
+
     _prune_orphaned_branches(repo_root)
 
 

hermes_cli/tips.py

@@ -297,7 +297,7 @@ TIPS = [
     "The gateway caches AIAgent instances per session — destroying this cache breaks Anthropic prompt caching.",
     "Any website can expose skills via /.well-known/skills/index.json — the skills hub discovers them automatically.",
     "The skills audit log at ~/.hermes/skills/.hub/audit.log tracks every install and removal operation.",
-    "Stale git worktrees are auto-cleaned: 24-72h old with no unpushed commits get pruned on startup.",
+    "Stale git worktrees are auto-cleaned on startup: clean, fully-merged trees get pruned; dirty or unpushed work is always preserved.",
     "Profiles scope Hermes state via HERMES_HOME; host tool subprocesses keep your real HOME unless terminal.home_mode is profile.",
     "HERMES_HOME_MODE env var (octal, e.g. 0701) sets custom directory permissions for web server traversal.",
     "Container mode: place .container-mode in HERMES_HOME and the host CLI auto-execs into the container.",

tests/cli/test_worktree.py

@@ -1146,3 +1146,181 @@ class TestWorktreeLockPredicate:
         import cli
         # Not a git repo -> git query fails -> must report "live" (never delete)
         assert cli._worktree_lock_is_live(str(tmp_path), str(tmp_path / "x")) == "live"
+
+
+class TestWidenedPruner:
+    """Behavior contracts for the widened pruner (#all-.worktrees coverage,
+    squash-merge escape hatch, kanban exclusion, preserved-work warning).
+
+    Previously only ``hermes-*`` directories were considered, so salvage/
+    review/port lanes created with raw ``git worktree add`` accumulated
+    forever (real incident: 117 dirs / 26 GB). And squash-merged branches'
+    local commits are unreachable from refs/remotes/* forever, so the
+    unpushed guard preserved fully-merged scratch trees indefinitely.
+    """
+
+    @staticmethod
+    def _age(path, hours):
+        import time
+        t = time.time() - (hours * 3600)
+        os.utime(path, (t, t))
+
+    @staticmethod
+    def _mk(repo, name, commit=False, dirty=False, age_h=100):
+        p = repo / ".worktrees" / name
+        (repo / ".worktrees").mkdir(exist_ok=True)
+        subprocess.run(
+            ["git", "worktree", "add", str(p), "-b", f"wt/{name}", "HEAD"],
+            cwd=repo, capture_output=True,
+        )
+        sha = None
+        if commit:
+            (p / "work.txt").write_text(f"work for {name}\n")
+            subprocess.run(["git", "add", "work.txt"], cwd=p, capture_output=True)
+            subprocess.run(["git", "commit", "-m", "wip"], cwd=p, capture_output=True)
+            sha = subprocess.run(
+                ["git", "rev-parse", "HEAD"], cwd=p,
+                capture_output=True, text=True,
+            ).stdout.strip()
+        if dirty:
+            (p / "dirty.txt").write_text("uncommitted")
+        TestWidenedPruner._age(p, age_h)
+        return p, sha
+
+    @staticmethod
+    def _merge_upstream(repo, sha):
+        """Simulate a squash-merge: land a patch-equivalent commit (different
+        SHA) on the branch refs/remotes/origin/main points at."""
+        # Distinct committer identity forces a distinct SHA even when the
+        # cherry-pick lands in the same second as the original commit.
+        subprocess.run(
+            ["git", "-c", "user.email=merger@test.com", "-c", "user.name=Merger",
+             "cherry-pick", sha],
+            cwd=repo, capture_output=True,
+        )
+        new_head = subprocess.run(
+            ["git", "rev-parse", "HEAD"], cwd=repo,
+            capture_output=True, text=True,
+        ).stdout.strip()
+        assert new_head != sha
+        subprocess.run(
+            ["git", "update-ref", "refs/remotes/origin/main", new_head],
+            cwd=repo, capture_output=True,
+        )
+
+    # -- named (non hermes-*) directories are now covered ------------------
+
+    def test_named_clean_stale_tree_is_reaped(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "salvage-12345", age_h=80)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert not wt.exists(), "clean named tree past 72h soft tier should be reaped"
+
+    def test_named_tree_gets_3x_grace(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "salvage-fresh", age_h=48)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists(), "named tree under 72h must be kept (3x scratch timeline)"
+
+    def test_named_dirty_tree_survives_any_age(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "salvage-dirty", dirty=True, age_h=24 * 30)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists(), "dirty named tree must never be reaped"
+
+    def test_named_unpushed_tree_survives_any_age(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "salvage-unpushed", commit=True, age_h=24 * 30)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists(), "unique unpushed work must never be reaped"
+
+    def test_kanban_task_tree_never_touched(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "t_0a1b2c3d", age_h=24 * 90)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists(), "kanban t_<hex> trees belong to kanban gc, not the pruner"
+
+    # -- squash-merge escape hatch ------------------------------------------
+
+    def test_squash_merged_tree_is_reaped(self, git_repo):
+        import cli
+        wt, sha = self._mk(git_repo, "hermes-merged", commit=True, age_h=100)
+        self._merge_upstream(git_repo, sha)
+        assert cli._worktree_has_unpushed_commits(str(wt)), (
+            "precondition: commit unreachable from remotes (the leak this fixes)"
+        )
+        cli._prune_stale_worktrees(str(git_repo))
+        assert not wt.exists(), (
+            "worktree whose commits are all patch-equivalent upstream is merged "
+            "work and should be reaped"
+        )
+
+    def test_partially_merged_tree_survives(self, git_repo):
+        import cli
+        wt, sha = self._mk(git_repo, "hermes-partial", commit=True, age_h=100)
+        self._merge_upstream(git_repo, sha)
+        # add a second, unmerged commit on top
+        (wt / "extra.txt").write_text("unique work\n")
+        subprocess.run(["git", "add", "extra.txt"], cwd=wt, capture_output=True)
+        subprocess.run(["git", "commit", "-m", "extra"], cwd=wt, capture_output=True)
+        self._age(wt, 100)
+        cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists(), "any non-equivalent local commit must preserve the tree"
+
+    # -- _worktree_commits_all_merged_upstream unit contracts ----------------
+
+    def test_merged_predicate_true_on_patch_equivalence(self, git_repo):
+        import cli
+        wt, sha = self._mk(git_repo, "hermes-eq", commit=True)
+        self._merge_upstream(git_repo, sha)
+        assert cli._worktree_commits_all_merged_upstream(str(wt)) is True
+
+    def test_merged_predicate_false_on_unique_work(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "hermes-uniq", commit=True)
+        assert cli._worktree_commits_all_merged_upstream(str(wt)) is False
+
+    def test_merged_predicate_true_at_zero_ahead(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "hermes-zero")
+        assert cli._worktree_commits_all_merged_upstream(str(wt)) is True
+
+    def test_merged_predicate_fails_safe_without_upstream(self, git_repo_no_remote):
+        import cli
+        repo = git_repo_no_remote
+        p = repo / ".worktrees" / "hermes-noremote"
+        (repo / ".worktrees").mkdir(exist_ok=True)
+        subprocess.run(
+            ["git", "worktree", "add", str(p), "-b", "wt/noremote", "HEAD"],
+            cwd=repo, capture_output=True,
+        )
+        assert cli._worktree_commits_all_merged_upstream(str(p)) is False
+
+    def test_merged_predicate_fails_safe_on_stale_base(self, git_repo):
+        import cli
+        wt, _ = self._mk(git_repo, "hermes-manyahead", commit=True)
+        assert cli._worktree_commits_all_merged_upstream(str(wt), max_ahead=0) is False
+
+    # -- preserved-work warning ----------------------------------------------
+
+    def test_preserved_stale_work_emits_warning(self, git_repo, caplog):
+        import logging
+        import cli
+        wt, _ = self._mk(git_repo, "salvage-old-work", commit=True, age_h=24 * 10)
+        with caplog.at_level(logging.WARNING, logger="cli"):
+            cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists()
+        assert any(
+            "salvage-old-work" in rec.getMessage() for rec in caplog.records
+        ), "worktree with >7d-old unmerged work should be named in a WARNING"
+
+    def test_no_warning_for_recent_work(self, git_repo, caplog):
+        import logging
+        import cli
+        wt, _ = self._mk(git_repo, "salvage-new-work", commit=True, age_h=100)
+        with caplog.at_level(logging.WARNING, logger="cli"):
+            cli._prune_stale_worktrees(str(git_repo))
+        assert wt.exists()
+        assert not any(
+            "salvage-new-work" in rec.getMessage() for rec in caplog.records
+        ), "under-7d preserved work should not warn"