Files
hermes-agent/tests/hermes_cli/test_worktree_gc.py
teknium1 0a3e792942 fix(worktree): judge no-remote repos against the local trunk instead of reaping everything
`hermes worktree prune` (and the startup/cron pruner) classified every clean tree in a
repository with no remote as "clean and fully merged/pushed" and force-deleted its branch,
even when the branch carried commits that exist nowhere else. `audit_branches` returned []
in the same repos, so a unique local-only branch was invisible to the audit as well.

Root cause: `_worktree_has_unpushed_commits` answered False when `refs/remotes` was empty
("nothing to be unpushed against") and both consumers — `worktree_gc._classify_tree` and
`worktree_ops._classify_prune_candidates` — read False as "safe to reap".

The preceding commit (#111897) flips that branch to True, which is safe but also means a
no-remote repo can never reclaim anything (a tree sitting at trunk, or squash-merged into
it, stays "unpushed" forever because `_worktree_commits_all_merged_upstream` finds no
origin/* base). This commit replaces the unconditional True with a real baseline:

- `_worktree_local_trunk`: `main`/`master`, else the branch checked out in the main
  worktree; None when no trunk exists.
- `_worktree_merge_base_ref`: origin/HEAD|origin/main|origin/master, falling back to the
  local trunk ONLY when the repo has no remote-tracking refs at all. Single resolver used
  by `_worktree_commits_all_merged_upstream` and `worktree_gc.audit_branches`.
- `_worktree_has_unpushed_commits`: with no remote refs, `git log HEAD --not <trunk>`;
  no trunk -> True (preserve), matching the docstring's fail-safe promise.

Net effect in a no-remote repo: unique work is kept ("unpushed commits not found
upstream"), trees at/merged into the local trunk still reap, branch audit reports unique
branches as keep and merged ones as delete, and `git branch -D` can only run on a branch
whose every commit is reachable from or patch-equivalent to the trunk.

Tests: the salvaged no-remote keep test now uses a shared `local_repo` fixture; a control
test proves trunk-merged trees still reclaim and the branch audit reports in the same
repo; `test_merged_predicate_fails_safe_without_upstream` now pins both halves of the
contract (trunk resolves -> merged; no trunk at all -> False/preserve).
2026-09-15 18:27:06 -07:00

444 lines
19 KiB
Python

"""Behavior contracts for hermes_cli.worktree_gc (attended reclaim).
Each guard gets its own contract against a REAL git repo fixture (no mocks —
the entire value of these tests is exercising actual git verdicts):
- clean + fully merged tree → reap
- untracked-only dirt → reap-archive (files archived, then removed)
- tracked modifications → keep, any age
- unique unpushed commits → keep
- patch-equivalent commits (rebase/squash-merge leak) → reap
- live-locked tree → keep
- kanban t_<hex> tree → keep (owned by kanban gc)
- branch GC: merged branch deleted, unique-commit branch kept,
checked-out branch kept, protected names kept
- reclaim operates ONLY on the frozen audit list (concurrent-session trap)
"""
import os
import subprocess
from pathlib import Path
import pytest
from hermes_cli import worktree_gc
def _git(args, cwd, env=None):
e = dict(os.environ)
e.update({
"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t",
})
if env:
e.update(env)
result = subprocess.run(
["git", *args], capture_output=True, text=True, cwd=str(cwd), env=e,
)
assert result.returncode == 0, f"git {args} failed: {result.stderr}"
return result.stdout.strip()
@pytest.fixture
def repo(tmp_path, monkeypatch):
"""origin (bare) + clone with .worktrees/, HOME redirected for archives."""
monkeypatch.setenv("HOME", str(tmp_path / "home"))
(tmp_path / "home").mkdir()
origin = tmp_path / "origin.git"
origin.mkdir()
_git(["init", "--bare", "-b", "main"], origin)
clone = tmp_path / "repo"
_git(["clone", str(origin), str(clone)], tmp_path)
(clone / "README.md").write_text("hello\n")
_git(["add", "."], clone)
_git(["commit", "-m", "init"], clone)
_git(["push", "origin", "main"], clone)
# origin/HEAD so upstream resolution works like a real clone.
_git(["remote", "set-head", "origin", "main"], clone)
(clone / ".worktrees").mkdir()
return clone
@pytest.fixture
def local_repo(tmp_path, monkeypatch):
"""``git init`` repo with NO remote at all (the #111895 shape), HOME redirected."""
monkeypatch.setenv("HOME", str(tmp_path / "home"))
(tmp_path / "home").mkdir()
repo = tmp_path / "local-repo"
_git(["init", "-b", "main", str(repo)], tmp_path)
(repo / "README.md").write_text("hello\n")
_git(["add", "."], repo)
_git(["commit", "-m", "init"], repo)
(repo / ".worktrees").mkdir()
return repo
def _add_worktree(repo_path, name, branch=None):
tree = repo_path / ".worktrees" / name
branch = branch or f"hermes/{name}"
_git(["worktree", "add", str(tree), "-b", branch], repo_path)
return tree, branch
def _verdict(records, name):
match = [record for record in records if record.name == name]
assert match, f"no record for {name}"
return match[0]
class TestAuditVerdicts:
def test_clean_merged_tree_reaps(self, repo):
_add_worktree(repo, "hermes-clean")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
assert _verdict(records, "hermes-clean").verdict == "reap"
def test_tracked_modifications_keep(self, repo):
tree, _ = _add_worktree(repo, "hermes-dirty")
(tree / "README.md").write_text("edited\n")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
record = _verdict(records, "hermes-dirty")
assert record.verdict == "keep"
assert "tracked" in record.reason
def test_untracked_only_is_reap_archive(self, repo):
tree, _ = _add_worktree(repo, "hermes-scratch")
(tree / "PR_BODY_DRAFT.md").write_text("draft\n")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
record = _verdict(records, "hermes-scratch")
assert record.verdict == "reap-archive"
assert record.untracked == ["PR_BODY_DRAFT.md"]
def test_unique_unpushed_commits_keep(self, repo):
tree, _ = _add_worktree(repo, "hermes-work")
(tree / "new.py").write_text("x = 1\n")
_git(["add", "."], tree)
_git(["commit", "-m", "unique work"], tree)
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
record = _verdict(records, "hermes-work")
assert record.verdict == "keep"
assert "unpushed" in record.reason
def test_unique_commits_without_a_remote_keep_tree_and_branch(self, local_repo):
"""No remote: unique commits are judged against the local trunk, never "merged/pushed"."""
tree, branch = _add_worktree(local_repo, "hermes-local-work")
(tree / "new.py").write_text("x = 1\n")
_git(["add", "."], tree)
_git(["commit", "-m", "unique local work"], tree)
records = worktree_gc.audit_worktrees(str(local_repo), with_sizes=False)
record = _verdict(records, "hermes-local-work")
assert record.verdict == "keep"
assert "unpushed" in record.reason
assert worktree_gc.reclaim_worktrees(str(local_repo), records=records) == []
assert tree.exists()
assert _git(["rev-parse", "--verify", branch], local_repo)
def test_without_a_remote_local_trunk_is_the_baseline(self, local_repo):
"""Control for the no-remote guard: work merged into the local ``main`` still reclaims
(tree and squash-merged branch), while a branch holding unique commits — with or without
a worktree — is reported and kept by the branch audit instead of vanishing."""
clean_tree, _ = _add_worktree(local_repo, "hermes-clean")
squashed_tree, _ = _add_worktree(local_repo, "hermes-squashed")
(squashed_tree / "k.py").write_text("k = 1\n")
_git(["add", "."], squashed_tree)
_git(["commit", "-m", "squash me"], squashed_tree)
_git(["merge", "--squash", "hermes/hermes-squashed"], local_repo)
_git(["commit", "-m", "squash merge"], local_repo)
_git(["branch", "orphan-unique", "main"], local_repo)
_git(["checkout", "-q", "orphan-unique"], local_repo)
(local_repo / "o.py").write_text("o = 1\n")
_git(["add", "."], local_repo)
_git(["commit", "-m", "unique, no worktree"], local_repo)
_git(["checkout", "-q", "main"], local_repo)
records = worktree_gc.audit_worktrees(str(local_repo), with_sizes=False)
assert _verdict(records, "hermes-clean").verdict == "reap"
assert _verdict(records, "hermes-squashed").verdict == "reap"
worktree_gc.reclaim_worktrees(str(local_repo), records=records)
assert not clean_tree.exists() and not squashed_tree.exists()
by_name = {b.name: b for b in worktree_gc.audit_branches(str(local_repo))}
assert by_name["orphan-unique"].verdict == "keep"
assert "unique" in by_name["orphan-unique"].reason
assert by_name["main"].verdict == "keep"
assert "hermes/hermes-squashed" not in by_name # branch -D ran only on merged work
def test_patch_equivalent_commits_reap(self, repo):
"""The squash/rebase-merge leak: local commit unreachable from any
remote ref but patch-equivalent to an upstream commit → merged work."""
tree, _ = _add_worktree(repo, "hermes-merged")
(tree / "feat.py").write_text("y = 2\n")
_git(["add", "."], tree)
_git(["commit", "-m", "feat"], tree)
sha = _git(["rev-parse", "HEAD"], tree)
# "Merge" it to main with a DIFFERENT committer so the cherry-pick
# produces a distinct sha (same-second identical-committer cherry
# picks can produce the identical sha — pitfall from the skill).
_git(["cherry-pick", sha], repo,
env={"GIT_COMMITTER_NAME": "other", "GIT_COMMITTER_EMAIL": "o@o"})
_git(["push", "origin", "main"], repo)
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
assert _verdict(records, "hermes-merged").verdict == "reap"
def test_live_locked_tree_keeps(self, repo):
tree, _ = _add_worktree(repo, "hermes-live")
_git(["worktree", "lock", str(tree),
"--reason", f"hermes pid={os.getpid()}"], repo)
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
record = _verdict(records, "hermes-live")
assert record.verdict == "keep"
assert "in use" in record.reason
def test_kanban_tree_untouched(self, repo):
_add_worktree(repo, "t_deadbeef", branch="kanban/t_deadbeef")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
record = _verdict(records, "t_deadbeef")
assert record.verdict == "keep"
assert "kanban" in record.reason
class TestReclaim:
def test_reap_removes_tree_and_branch(self, repo):
tree, branch = _add_worktree(repo, "hermes-clean")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
actions = worktree_gc.reclaim_worktrees(str(repo), records=records)
assert any("removed hermes-clean" in a for a in actions)
assert not tree.exists()
probe = subprocess.run(
["git", "rev-parse", "--verify", "--quiet", branch],
capture_output=True, text=True, cwd=str(repo),
)
assert probe.returncode != 0, "branch should be gone with its tree"
def test_untracked_files_archived_under_the_active_profile_home(self, repo, tmp_path, monkeypatch):
"""The archive follows the active Hermes home (a named profile here), never ~/.hermes."""
native_home = tmp_path / "native"
profile_home = tmp_path / "root" / "profiles" / "work"
profile_home.mkdir(parents=True)
monkeypatch.setattr(Path, "home", lambda: native_home)
monkeypatch.setenv("HERMES_HOME", str(profile_home))
tree, _ = _add_worktree(repo, "hermes-scratch")
(tree / "NOTES.md").write_text("important scribbles\n")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
worktree_gc.reclaim_worktrees(str(repo), records=records)
assert not tree.exists()
archived = list((profile_home / "archive" / "worktree-prune").rglob("NOTES.md"))
assert archived, "untracked file must be archived under the profile home, not destroyed"
assert archived[0].read_text() == "important scribbles\n"
assert not (native_home / ".hermes").exists()
def test_dry_run_changes_nothing(self, repo):
tree, _ = _add_worktree(repo, "hermes-clean")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
actions = worktree_gc.reclaim_worktrees(
str(repo), dry_run=True, records=records
)
assert any("would remove" in a for a in actions)
assert tree.exists()
def test_frozen_list_ignores_trees_created_after_audit(self, repo):
"""Concurrent-session trap: a tree created between audit and reclaim
must be out of scope by construction."""
_add_worktree(repo, "hermes-old")
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
late_tree, _ = _add_worktree(repo, "hermes-late")
worktree_gc.reclaim_worktrees(str(repo), records=records)
assert late_tree.exists(), "tree created after the audit must survive"
def test_dead_locked_tree_is_unlocked_and_reaped(self, repo):
tree, _ = _add_worktree(repo, "hermes-zombie")
_git(["worktree", "lock", str(tree),
"--reason", "hermes pid=999999999"], repo)
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
assert _verdict(records, "hermes-zombie").verdict == "reap"
worktree_gc.reclaim_worktrees(str(repo), records=records)
assert not tree.exists()
class TestBranchGC:
def test_merged_branch_deleted_any_name(self, repo):
"""Branch GC is content-gated, not name-gated: any fully-merged local
branch is safe to delete regardless of prefix."""
_git(["branch", "salv-12345", "main"], repo)
_git(["branch", "feat/some-old-thing", "main"], repo)
records = worktree_gc.audit_branches(str(repo))
by_name = {record.name: record for record in records}
assert by_name["salv-12345"].verdict == "delete"
assert by_name["feat/some-old-thing"].verdict == "delete"
worktree_gc.reclaim_branches(str(repo), records=records)
out = _git(["branch", "--format=%(refname:short)"], repo)
assert "salv-12345" not in out
assert "feat/some-old-thing" not in out
def test_unique_commit_branch_kept(self, repo):
_git(["checkout", "-b", "feat/real-work"], repo)
(repo / "wip.py").write_text("z = 3\n")
_git(["add", "."], repo)
_git(["commit", "-m", "wip"], repo)
_git(["checkout", "main"], repo)
records = worktree_gc.audit_branches(str(repo))
by_name = {record.name: record for record in records}
assert by_name["feat/real-work"].verdict == "keep"
assert "unique" in by_name["feat/real-work"].reason
def test_patch_equivalent_branch_deleted(self, repo):
"""Rebase-merged PR branch: SHAs differ from main but every commit is
patch-equivalent — the dominant branch leak."""
_git(["checkout", "-b", "fix/landed"], repo)
(repo / "fix.py").write_text("a = 4\n")
_git(["add", "."], repo)
_git(["commit", "-m", "fix"], repo)
sha = _git(["rev-parse", "HEAD"], repo)
_git(["checkout", "main"], repo)
_git(["cherry-pick", sha], repo,
env={"GIT_COMMITTER_NAME": "other", "GIT_COMMITTER_EMAIL": "o@o"})
_git(["push", "origin", "main"], repo)
records = worktree_gc.audit_branches(str(repo))
by_name = {record.name: record for record in records}
assert by_name["fix/landed"].verdict == "delete"
assert "patch-equivalent" in by_name["fix/landed"].reason
def test_checked_out_and_protected_kept(self, repo):
_tree, branch = _add_worktree(repo, "hermes-active")
records = worktree_gc.audit_branches(str(repo))
by_name = {record.name: record for record in records}
assert by_name["main"].verdict == "keep"
assert by_name[branch].verdict == "keep"
class TestOlderThanGate:
def test_young_reapable_tree_kept_under_older_than(self, repo):
_add_worktree(repo, "hermes-young")
records = worktree_gc.audit_worktrees(
str(repo), with_sizes=False, older_than_days=7,
)
record = _verdict(records, "hermes-young")
assert record.verdict == "keep"
assert "older-than" in record.reason
def test_aged_reapable_tree_still_reaps(self, repo):
import os as _os
import time as _time
tree, _ = _add_worktree(repo, "hermes-old")
old = _time.time() - 10 * 86400
_os.utime(tree, (old, old))
records = worktree_gc.audit_worktrees(
str(repo), with_sizes=False, older_than_days=7,
)
assert _verdict(records, "hermes-old").verdict == "reap"
def test_older_than_never_widens_eligibility(self, repo):
"""A tree with real work stays keep at ANY age — the age gate only
restricts, it can never doom unmerged/dirty work."""
import os as _os
import time as _time
tree, _ = _add_worktree(repo, "hermes-old-work")
(tree / "README.md").write_text("edited\n")
old = _time.time() - 30 * 86400
_os.utime(tree, (old, old))
records = worktree_gc.audit_worktrees(
str(repo), with_sizes=False, older_than_days=7,
)
record = _verdict(records, "hermes-old-work")
assert record.verdict == "keep"
assert "tracked" in record.reason
class TestExternalTrees:
def test_external_tree_reported_never_reaped(self, repo, tmp_path):
ext = tmp_path / "elsewhere-tree"
_git(["worktree", "add", str(ext), "-b", "ext/branch"], repo)
(ext / "WIP.txt").write_text("outside work\n")
external = worktree_gc.audit_external_trees(str(repo))
paths = [record.path for record in external]
assert any("elsewhere-tree" in p for p in paths)
record = [r for r in external if "elsewhere-tree" in r.path][0]
assert record.branch == "ext/branch"
assert not record.missing
# The managed audit + reclaim never see or touch it.
records = worktree_gc.audit_worktrees(str(repo), with_sizes=False)
assert all("elsewhere-tree" not in r.name for r in records)
worktree_gc.reclaim_worktrees(str(repo), records=records)
assert ext.exists() and (ext / "WIP.txt").exists()
def test_managed_trees_not_reported_as_external(self, repo):
_add_worktree(repo, "hermes-managed")
external = worktree_gc.audit_external_trees(str(repo))
assert all("hermes-managed" not in r.path for r in external)
def test_missing_registration_flagged_and_pruned(self, repo, tmp_path):
import shutil as _shutil
ext = tmp_path / "vanished-tree"
_git(["worktree", "add", str(ext), "-b", "ext/vanished"], repo)
_shutil.rmtree(ext)
external = worktree_gc.audit_external_trees(str(repo))
record = [r for r in external if "vanished-tree" in r.path][0]
assert record.missing
planned = worktree_gc.prune_missing_registrations(str(repo), dry_run=True)
assert any("vanished-tree" in line for line in planned)
# Dry-run changed nothing.
assert any(
r.missing for r in worktree_gc.audit_external_trees(str(repo))
)
done = worktree_gc.prune_missing_registrations(str(repo))
assert any("pruned" in line for line in done)
assert all(
"vanished-tree" not in r.path
for r in worktree_gc.audit_external_trees(str(repo))
)
class TestCmdWorktreeJson:
def _ns(self, repo, action, **kw):
import argparse
return argparse.Namespace(
repo=str(repo), worktree_action=action,
json=True, older_than=kw.get("older_than"),
dry_run=kw.get("dry_run", False),
trees_only=kw.get("trees_only", False),
branches_only=kw.get("branches_only", False),
)
def test_list_json_shape(self, repo, capsys):
import json
from hermes_cli.worktree_cmd import cmd_worktree
_add_worktree(repo, "hermes-json")
assert cmd_worktree(self._ns(repo, "list")) == 0
payload = json.loads(capsys.readouterr().out)
assert set(payload) == {"repo", "trees", "external_trees", "branches"}
names = [t["name"] for t in payload["trees"]]
assert "hermes-json" in names
tree = [t for t in payload["trees"] if t["name"] == "hermes-json"][0]
assert {"verdict", "reason", "age_days", "branch"} <= set(tree)
def test_prune_dry_run_json(self, repo, capsys):
import json
from hermes_cli.worktree_cmd import cmd_worktree
_add_worktree(repo, "hermes-json-prune")
assert cmd_worktree(self._ns(repo, "prune", dry_run=True)) == 0
payload = json.loads(capsys.readouterr().out)
assert payload["dry_run"] is True
assert any("hermes-json-prune" in a for a in payload["actions"])
# dry-run: tree still present
assert (repo / ".worktrees" / "hermes-json-prune").exists()
def test_negative_older_than_rejected(self, repo, capsys):
from hermes_cli.worktree_cmd import cmd_worktree
assert cmd_worktree(self._ns(repo, "prune", older_than=-1)) == 1