Port from code-yeongyu/oh-my-openagent#7151: worktree audit gains --json, --older-than, and external-tree visibility

omo's omo-agent-toolkit worktree-sweep (their PR #7151) added three
capabilities our hermes worktree command lacked:

- --json on list and prune: machine-readable audit/result payloads so
  scripts and agents can consume verdicts without scraping table output.
- --older-than DAYS: an age floor that only ever RESTRICTS reaping
  (young-but-reapable trees are kept); it never widens eligibility, so
  the existing safety invariants are untouched.
- External-tree visibility: linked worktrees registered outside
  .worktrees/ are now reported read-only in the audit (branch, locked,
  missing) instead of being invisible, and registrations whose
  directory has vanished are dropped via git worktree prune (metadata
  only, no files touched) during prune.

Not ported: omo's ancestor-of-default-branch merge test (our git cherry
patch-equivalence is strictly stronger under rebase/squash merges), and
their hardcoded external-root exclusion list (we exclude by location:
everything outside .worktrees/ is hands-off).

Tests: 9 new contracts in tests/hermes_cli/test_worktree_gc.py (age gate
restrict-only, external trees never reaped, stale-registration prune
dry-run/real, JSON shapes, negative --older-than rejected). Live E2E on
a scratch repo verified all three flags end to end.
This commit is contained in:
Teknium
2026-08-25 22:11:40 -07:00
parent cb3447b139
commit 65a6b6831e
5 changed files with 308 additions and 17 deletions

View File

@@ -17,9 +17,21 @@ def build_worktree_parser(subparsers) -> None:
"list", aliases=["ls", "audit"],
help="Classify every tree: age, size, verdict, reason (default action)")
worktree_list.add_argument("--repo", help="Repo root (default: current repo)")
worktree_list.add_argument(
"--json", action="store_true",
help="Machine-readable audit output (trees, external trees, branches)")
worktree_list.add_argument(
"--older-than", type=float, metavar="DAYS", dest="older_than",
help="Treat reapable trees younger than DAYS as keep")
worktree_prune = worktree_subparsers.add_parser(
"prune", help="Remove safe trees and delete fully-merged local branches")
worktree_prune.add_argument("--repo", help="Repo root (default: current repo)")
worktree_prune.add_argument(
"--json", action="store_true",
help="Machine-readable result (actions taken/planned, preserved trees)")
worktree_prune.add_argument(
"--older-than", type=float, metavar="DAYS", dest="older_than",
help="Only reap trees idle for at least DAYS days (safety gates still apply)")
worktree_prune.add_argument(
"--dry-run", action="store_true", help="Show the plan without changing anything")
worktree_prune.add_argument(

View File

@@ -1,8 +1,12 @@
"""``hermes worktree`` — audit (``list``) and reclaim (``prune [--dry-run] [--trees-only |
--branches-only]``) accumulated git worktrees/branches."""
"""``hermes worktree`` — audit (``list [--json] [--older-than DAYS]``) and reclaim
(``prune [--dry-run] [--json] [--older-than DAYS] [--trees-only | --branches-only]``)
accumulated git worktrees/branches. ``--json`` output is the only thing written to stdout in
that mode so scripts can consume it."""
from __future__ import annotations
import json
from dataclasses import asdict
from typing import Optional
@@ -13,19 +17,38 @@ def _fmt_size(size_mb: Optional[int]) -> str:
def _list(worktree_gc, repo_root: str, args) -> int:
records = worktree_gc.audit_worktrees(repo_root)
if not records:
older_than = getattr(args, "older_than", None)
records = worktree_gc.audit_worktrees(repo_root, older_than_days=older_than)
external = worktree_gc.audit_external_trees(repo_root)
branch_records = worktree_gc.audit_branches(repo_root)
if getattr(args, "json", False):
print(json.dumps({
"repo": repo_root,
"trees": [asdict(r) for r in records],
"external_trees": [asdict(r) for r in external],
"branches": [asdict(b) for b in branch_records],
}, indent=2))
return 0
if not records and not external:
print("No worktrees under .worktrees/ — nothing to reclaim.")
return 0
total_mb = sum(r.size_mb or 0 for r in records)
reapable_mb = sum(r.size_mb or 0 for r in records if r.verdict.startswith("reap"))
print(f"{'TREE':32} {'AGE':>6} {'SIZE':>6} {'VERDICT':13} REASON")
for r in sorted(records, key=lambda x: -(x.size_mb or 0)):
print(f"{r.name[:32]:32} {r.age_days:>5.1f}d {_fmt_size(r.size_mb):>6} {r.verdict:13} {r.reason}")
print(
f"\n{len(records)} tree(s), {_fmt_size(total_mb)} total — "
f"{_fmt_size(reapable_mb)} reclaimable now via `hermes worktree prune`.")
deletable = [b for b in worktree_gc.audit_branches(repo_root) if b.verdict == "delete"]
if records:
total_mb = sum(r.size_mb or 0 for r in records)
reapable_mb = sum(r.size_mb or 0 for r in records if r.verdict.startswith("reap"))
print(f"{'TREE':32} {'AGE':>6} {'SIZE':>6} {'VERDICT':13} REASON")
for r in sorted(records, key=lambda x: -(x.size_mb or 0)):
print(f"{r.name[:32]:32} {r.age_days:>5.1f}d {_fmt_size(r.size_mb):>6} {r.verdict:13} {r.reason}")
print(
f"\n{len(records)} tree(s), {_fmt_size(total_mb)} total — "
f"{_fmt_size(reapable_mb)} reclaimable now via `hermes worktree prune`.")
if external:
print(f"\n{len(external)} externally-registered worktree(s) (never touched by prune):")
for e in external:
state = "MISSING" if e.missing else ("locked" if e.locked else "ok")
print(f" {e.path} [{e.branch or '?'}] {state}")
if any(e.missing and not e.locked for e in external):
print(" Stale registrations (MISSING) are cleaned by `hermes worktree prune` (metadata only).")
deletable = [b for b in branch_records if b.verdict == "delete"]
if deletable:
print(f"{len(deletable)} local branch(es) fully merged/patch-equivalent upstream would also be deleted.")
return 0
@@ -33,19 +56,31 @@ def _list(worktree_gc, repo_root: str, args) -> int:
def _prune(worktree_gc, repo_root: str, args) -> int:
dry_run = bool(getattr(args, "dry_run", False))
as_json = bool(getattr(args, "json", False))
older_than = getattr(args, "older_than", None)
actions: list = []
kept: list = []
if not getattr(args, "branches_only", False):
tree_records = worktree_gc.audit_worktrees(repo_root, with_sizes=False)
actions += worktree_gc.prune_missing_registrations(repo_root, dry_run=dry_run)
tree_records = worktree_gc.audit_worktrees(repo_root, with_sizes=False, older_than_days=older_than)
actions += worktree_gc.reclaim_worktrees(repo_root, dry_run=dry_run, records=tree_records)
kept = [r for r in tree_records if r.verdict == "keep"
and "kanban" not in r.reason and "in use" not in r.reason]
if kept:
if kept and not as_json:
print(f"Preserved {len(kept)} tree(s) with real work:")
for r in kept:
print(f" {r.name}: {r.reason}")
if not getattr(args, "trees_only", False):
actions += worktree_gc.reclaim_branches(repo_root, dry_run=dry_run)
if as_json:
print(json.dumps({
"repo": repo_root,
"dry_run": dry_run,
"actions": actions,
"preserved": [asdict(r) for r in kept],
}, indent=2))
return 0
if actions:
for line in actions:
print(f" {line}")
@@ -67,6 +102,10 @@ def cmd_worktree(args) -> int:
if not repo_root:
print("Not inside a git repository (or pass --repo <path>).")
return 1
older_than = getattr(args, "older_than", None)
if older_than is not None and older_than < 0:
print("--older-than must be a non-negative number of days.")
return 1
action = getattr(args, "worktree_action", None) or "list"
handler = _ACTIONS.get(action)
if handler is None:

View File

@@ -55,6 +55,18 @@ def _run(cmd: list, timeout: int, cwd: Optional[str] = None) -> subprocess.Compl
timeout=timeout, cwd=cwd)
@dataclass
class ExternalTreeRecord:
"""A linked worktree registered on the repo but living OUTSIDE
``.worktrees/`` — created by hand or by another tool. Reported for
visibility only; the reclaim paths never touch these."""
path: str
branch: str # branch name, or "detached @<sha>" when detached
locked: bool
missing: bool # registered but the directory no longer exists
def _git(args: list, cwd: str, timeout: int = 15) -> subprocess.CompletedProcess:
"""Run git, translating timeouts into returncode 124. Every verdict fails safe toward "keep"
on nonzero, so a slow ``git cherry`` on a huge repo degrades to keep instead of aborting the
@@ -135,8 +147,89 @@ def _classify_tree(_ops, repo_root: str, entry: Path, merge_cache, remote_heads)
return "reap", "clean and fully merged/pushed", []
def audit_worktrees(repo_root: str, *, with_sizes: bool = True) -> List[TreeRecord]:
"""Classify every tree under ``.worktrees/`` without mutating anything."""
def audit_external_trees(repo_root: str) -> List[ExternalTreeRecord]:
"""List linked worktrees registered OUTSIDE ``.worktrees/``.
``hermes -w`` scratch trees all live under ``<repo>/.worktrees/``, but
``git worktree list --porcelain`` also knows about trees the user (or
another tool) registered elsewhere. Those are someone else's state, so
the reclaim paths never touch them — but hiding them entirely makes the
audit lie about what the repo is carrying. Report them read-only, and
flag registrations whose directory has vanished (safe to
``git worktree prune``).
"""
result = _git(["worktree", "list", "--porcelain"], cwd=repo_root, timeout=10)
if result.returncode != 0:
return []
managed_root = os.path.realpath(str(Path(repo_root) / ".worktrees"))
main_root = os.path.realpath(repo_root)
records: List[ExternalTreeRecord] = []
current: dict = {}
def _flush():
path = current.get("path")
if not path:
return
real = os.path.realpath(path)
if real == main_root:
return # the main checkout itself
if real == managed_root or real.startswith(managed_root + os.sep):
return # hermes-managed scratch tree — covered by audit_worktrees
branch = current.get("branch", "")
if not branch and current.get("head"):
branch = f"detached @{current['head'][:10]}"
records.append(ExternalTreeRecord(
path=path,
branch=branch,
locked=bool(current.get("locked")),
missing=not os.path.exists(path),
))
for line in result.stdout.splitlines():
line = line.rstrip()
if not line:
_flush()
current = {}
continue
if line.startswith("worktree "):
current["path"] = line[len("worktree "):]
elif line.startswith("branch refs/heads/"):
current["branch"] = line[len("branch refs/heads/"):]
elif line.startswith("HEAD "):
current["head"] = line[len("HEAD "):]
elif line == "locked" or line.startswith("locked "):
current["locked"] = True
_flush()
return records
def prune_missing_registrations(repo_root: str, *, dry_run: bool = False) -> List[str]:
"""Drop registrations whose directory no longer exists (any location).
The equivalent of a targeted ``git worktree prune``: purely
metadata-level, never removes files, so it is safe even for external
trees — a missing directory means there is nothing left to protect.
"""
stale = [r for r in audit_external_trees(repo_root) if r.missing and not r.locked]
if not stale:
return []
if dry_run:
return [f"would prune stale registration {r.path}" for r in stale]
result = _git(["worktree", "prune"], cwd=repo_root, timeout=15)
if result.returncode != 0:
return [f"failed to prune stale registrations: {result.stderr.strip()}"]
return [f"pruned stale registration {r.path}" for r in stale]
def audit_worktrees(repo_root: str, *, with_sizes: bool = True,
older_than_days: Optional[float] = None) -> List[TreeRecord]:
"""Classify every tree under ``.worktrees/`` without mutating anything.
``older_than_days`` only ever RESTRICTS: a reapable tree younger than the threshold is kept
("too recent"). It never widens eligibility — age alone can't doom a tree carrying unmerged work.
"""
from hermes_cli import worktree_ops as _ops
worktrees_dir = Path(repo_root) / ".worktrees"
if not worktrees_dir.exists():
@@ -164,6 +257,9 @@ def audit_worktrees(repo_root: str, *, with_sizes: bool = True) -> List[TreeReco
except Exception:
branch = ""
verdict, reason, untracked = _classify_tree(_ops, repo_root, entry, merge_cache, remote_heads)
if older_than_days is not None and verdict in _REAP_VERDICTS and age_days < older_than_days:
verdict, reason, untracked = "keep", (
f"reapable but only {age_days:.1f}d old (--older-than {older_than_days:g})"), []
records.append(TreeRecord(
name=entry.name, path=str(entry), branch=branch,
age_days=age_days, size_mb=_tree_size_mb(entry) if with_sizes else None,

View File

@@ -247,3 +247,138 @@ class TestBranchGC:
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

View File

@@ -68,12 +68,21 @@ explicitly:
```bash
hermes worktree list # audit: age, size, verdict, reason per tree
hermes worktree list --json # machine-readable audit (trees, external trees, branches)
hermes worktree prune # remove safe trees + delete merged branches
hermes worktree prune --dry-run # show the plan without changing anything
hermes worktree prune --older-than 7 # only reap trees idle for 7+ days
hermes worktree prune --trees-only # leave local branches alone
hermes worktree prune --branches-only # leave worktrees alone
```
Worktrees registered **outside** `.worktrees/` (created by hand or by another
tool) are reported read-only in `list` output and are never removed. The one
exception is metadata: registrations whose directory no longer exists are
dropped via `git worktree prune` (no files are touched). `--older-than DAYS`
only ever narrows what gets reaped — a tree carrying real work is kept at any
age regardless of the flag.
Inside a session, `/worktree prune [--dry-run]` does the same (and never
touches the tree the session is running in).