fix(cli): revalidate worktree cleanup gates before removal

Signed-off-by: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com>
(cherry picked from commit 86fad9bd6050a985c9119cf59932ddb5df0fff6a)

hermes_cli/worktree_gc.py

@@ -285,8 +285,14 @@ def reclaim_worktrees(
 ) -> List[str]:
     """Remove every reap-verdict tree from a frozen audit list — never re-globs inside the
     destructive loop, so trees created by concurrent sessions after the audit are out of scope."""
+    from hermes_cli import worktree_ops as _ops
+
     if records is None:
         records = audit_worktrees(repo_root, with_sizes=False)
+    merge_cache = {}
+    remote_heads = (
+        _ops._fetch_remote_branch_heads(repo_root)
+        if not dry_run and any(r.verdict == "reap-keep-branch" for r in records) else None)
     actions: List[str] = []
     for record in records:
         if record.verdict not in _REAP_VERDICTS:
@@ -296,6 +302,19 @@ def reclaim_worktrees(
             continue
 
         entry = Path(record.path)
+        # The frozen list bounds membership, not authority to discard later work.
+        # Another session may have changed a tree while earlier records were archived.
+        try:
+            verdict, reason, untracked = _classify_tree(
+                _ops, repo_root, entry, merge_cache, remote_heads)
+            branch = _git(["branch", "--show-current"], cwd=record.path, timeout=5)
+            if (verdict != record.verdict or untracked != record.untracked
+                    or branch.returncode != 0 or branch.stdout.strip() != record.branch):
+                actions.append(f"kept {record.name} (state changed since audit: {reason})")
+                continue
+        except Exception as exc:
+            actions.append(f"kept {record.name} (could not revalidate: {exc})")
+            continue
         if record.untracked:
             archive = _archive_untracked(entry, record.untracked)
             if archive is None:
@@ -307,7 +326,10 @@ def reclaim_worktrees(
         with contextlib.suppress(Exception):
             _git(["worktree", "unlock", record.path], cwd=repo_root, timeout=10)
         try:
-            remove_result = _git(["worktree", "remove", record.path, "--force"], cwd=repo_root, timeout=30)
+            remove_args = ["worktree", "remove", record.path]
+            if record.untracked:
+                remove_args.append("--force")  # only after the audited scratch was archived
+            remove_result = _git(remove_args, cwd=repo_root, timeout=30)
             if remove_result.returncode != 0:
                 actions.append(f"failed to remove {record.name}: {remove_result.stderr.strip()}")
                 continue

hermes_cli/worktree_ops.py

@@ -930,6 +930,17 @@ def _reap_prune_verdicts(repo_root: str, verdicts: list, stale_work_cutoff: floa
     preserved_stale: list = []
     kept_branches: set = set()
     for entry, mtime, force, verdict, lock_state in verdicts:
+        # Keep the original candidate set, but refresh its destructive safety gates:
+        # classification can precede removal by a long parallel sweep.
+        if verdict in {"reap", "reap-keep-branch"}:
+            try:
+                refreshed = _classify_prune_candidates(repo_root, [(entry, mtime, force)])
+                if not refreshed:
+                    continue
+                _, _, _, verdict, lock_state = refreshed[0]
+            except Exception as exc:
+                logger.debug("Could not revalidate worktree %s: %s", entry.name, exc)
+                continue
         reason = _PRESERVE_REASONS.get(verdict)
         if reason:
             if mtime <= stale_work_cutoff:
@@ -945,7 +956,7 @@ def _reap_prune_verdicts(repo_root: str, verdicts: list, stale_work_cutoff: floa
 
         try:
             branch = _git(["branch", "--show-current"], str(entry), timeout=5).stdout.strip()
-            remove_result = _git(["worktree", "remove", str(entry), "--force"], repo_root, timeout=15)
+            remove_result = _git(["worktree", "remove", str(entry)], repo_root, timeout=15)
             if remove_result.returncode != 0:
                 logger.debug("Failed to remove worktree %s: %s", entry.name, remove_result.stderr.strip())
                 continue

tests/hermes_cli/test_worktree_reclaim_revalidation.py

@@ -0,0 +1,68 @@
+"""An old cleanup verdict must not delete work added since the audit."""
+
+import os
+import subprocess
+import time
+
+import pytest
+
+
+@pytest.fixture
+def repository(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
+    monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull)
+    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
+    root = tmp_path / "repo"
+    _git(tmp_path, "init", "-b", "main", str(root))
+    _git(root, "config", "user.name", "Hermes test")
+    _git(root, "config", "user.email", "hermes-test@example.invalid")
+    (root / "README.md").write_text("initial\n", encoding="utf-8")
+    _git(root, "add", "README.md")
+    _git(root, "-c", "commit.gpgsign=false", "commit", "-m", "initial")
+    tree = root / ".worktrees" / "scratch"
+    _git(root, "worktree", "add", "-b", "scratch", str(tree))
+    return root, tree
+
+
+def _git(cwd, *args):
+    result = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, check=True)
+    return result.stdout.strip()
+
+
+def _add_work(tree, kind):
+    (tree / "README.md").write_text("new work after audit\n", encoding="utf-8")
+    if kind == "commit":
+        _git(tree, "add", "README.md")
+        _git(tree, "-c", "commit.gpgsign=false", "commit", "-m", "new unique work")
+
+
+@pytest.mark.parametrize("kind", ["edit", "commit"])
+def test_attended_reclaim_preserves_work_added_after_audit(repository, kind):
+    from hermes_cli.worktree_gc import audit_worktrees, reclaim_worktrees
+
+    root, tree = repository
+    records = audit_worktrees(str(root), with_sizes=False)
+    assert next(record for record in records if record.path == str(tree)).verdict == "reap"
+    _add_work(tree, kind)
+
+    reclaim_worktrees(str(root), records=records)
+
+    assert tree.is_dir()
+    assert (tree / "README.md").read_text(encoding="utf-8") == "new work after audit\n"
+    assert _git(root, "rev-parse", "--verify", "scratch")
+
+
+@pytest.mark.parametrize("kind", ["edit", "commit"])
+def test_startup_reclaim_preserves_work_added_after_classification(repository, kind):
+    from hermes_cli.worktree_ops import _classify_prune_candidates, _reap_prune_verdicts
+
+    root, tree = repository
+    verdicts = _classify_prune_candidates(str(root), [(tree, time.time() - 86400, False)])
+    assert verdicts[0][3] == "reap"
+    _add_work(tree, kind)
+
+    _reap_prune_verdicts(str(root), verdicts, stale_work_cutoff=0)
+
+    assert tree.is_dir()
+    assert (tree / "README.md").read_text(encoding="utf-8") == "new work after audit\n"
+    assert _git(root, "rev-parse", "--verify", "scratch")