diff --git a/cli.py b/cli.py index 01ea17ff2..f00e6b7fe 100644 --- a/cli.py +++ b/cli.py @@ -760,7 +760,10 @@ def _setup_worktree(repo_root: str = None) -> Optional[Dict[str, str]]: def _cleanup_worktree(info: Dict[str, str] = None) -> None: """Remove a worktree and its branch on exit. - If the worktree has uncommitted changes, warn and keep it. + Preserves the worktree only if it has unpushed commits (real work + that hasn't been pushed to any remote). Uncommitted changes alone + (untracked files, test artifacts) are not enough to keep it — agent + work lives in commits/PRs, not the working tree. """ global _active_worktree info = info or _active_worktree @@ -776,23 +779,27 @@ def _cleanup_worktree(info: Dict[str, str] = None) -> None: if not Path(wt_path).exists(): return - # Check for uncommitted changes + # Check for unpushed commits — commits reachable from HEAD but not + # from any remote branch. These represent real work the agent did + # but didn't push. + has_unpushed = False try: - status = subprocess.run( - ["git", "status", "--porcelain"], + result = subprocess.run( + ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], capture_output=True, text=True, timeout=10, cwd=wt_path, ) - has_changes = bool(status.stdout.strip()) + has_unpushed = bool(result.stdout.strip()) except Exception: - has_changes = True # Assume dirty on error — don't delete + has_unpushed = True # Assume unpushed on error — don't delete - if has_changes: - print(f"\n\033[33m⚠ Worktree has uncommitted changes, keeping: {wt_path}\033[0m") - print(f" To clean up manually: git worktree remove {wt_path}") + if has_unpushed: + print(f"\n\033[33m⚠ Worktree has unpushed commits, keeping: {wt_path}\033[0m") + print(f" To clean up manually: git worktree remove --force {wt_path}") _active_worktree = None return - # Remove worktree + # Remove worktree (even if working tree is dirty — uncommitted + # changes without unpushed commits are just artifacts) try: subprocess.run( ["git", "worktree", "remove", wt_path, "--force"], @@ -801,7 +808,7 @@ def _cleanup_worktree(info: Dict[str, str] = None) -> None: except Exception as e: logger.debug("Failed to remove worktree: %s", e) - # Delete the branch (only if it was never pushed / has no upstream) + # Delete the branch try: subprocess.run( ["git", "branch", "-D", branch], @@ -815,19 +822,27 @@ def _cleanup_worktree(info: Dict[str, str] = None) -> None: def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None: - """Remove worktrees older than max_age_hours that have no uncommitted changes. + """Remove stale worktrees and orphaned branches on startup. - Runs silently on startup to clean up after crashed/killed sessions. + Age-based tiers: + - 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). + + Also prunes orphaned ``hermes/*`` and ``pr-*`` local branches that + have no corresponding worktree. """ import subprocess import time worktrees_dir = Path(repo_root) / ".worktrees" if not worktrees_dir.exists(): + _prune_orphaned_branches(repo_root) return now = time.time() - cutoff = now - (max_age_hours * 3600) + soft_cutoff = now - (max_age_hours * 3600) # 24h default + hard_cutoff = now - (max_age_hours * 3 * 3600) # 72h default for entry in worktrees_dir.iterdir(): if not entry.is_dir() or not entry.name.startswith("hermes-"): @@ -836,21 +851,24 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None: # Check age try: mtime = entry.stat().st_mtime - if mtime > cutoff: + if mtime > soft_cutoff: continue # Too recent — skip except Exception: continue - # Check for uncommitted changes - try: - status = subprocess.run( - ["git", "status", "--porcelain"], - capture_output=True, text=True, timeout=5, cwd=str(entry), - ) - if status.stdout.strip(): - continue # Has changes — skip - except Exception: - continue # Can't check — skip + force = mtime <= hard_cutoff # Over 72h — force remove + + if not force: + # 24h–72h tier: only remove if no unpushed commits + try: + result = subprocess.run( + ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], + capture_output=True, text=True, timeout=5, cwd=str(entry), + ) + if result.stdout.strip(): + continue # Has unpushed commits — skip + except Exception: + continue # Can't check — skip # Safe to remove try: @@ -869,10 +887,81 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None: ["git", "branch", "-D", branch], capture_output=True, text=True, timeout=10, cwd=repo_root, ) - logger.debug("Pruned stale worktree: %s", entry.name) + logger.debug("Pruned stale worktree: %s (force=%s)", entry.name, force) except Exception as e: logger.debug("Failed to prune worktree %s: %s", entry.name, e) + _prune_orphaned_branches(repo_root) + + +def _prune_orphaned_branches(repo_root: str) -> None: + """Delete local ``hermes/hermes-*`` and ``pr-*`` branches with no worktree. + + These are auto-generated by ``hermes -w`` sessions and PR review + workflows respectively. Once their worktree is gone they serve no + purpose and just accumulate. + """ + import subprocess + + try: + result = subprocess.run( + ["git", "branch", "--format=%(refname:short)"], + capture_output=True, text=True, timeout=10, cwd=repo_root, + ) + if result.returncode != 0: + return + all_branches = [b.strip() for b in result.stdout.strip().split("\n") if b.strip()] + except Exception: + return + + # Collect branches that are actively checked out in a worktree + active_branches: set = set() + try: + wt_result = subprocess.run( + ["git", "worktree", "list", "--porcelain"], + capture_output=True, text=True, timeout=10, cwd=repo_root, + ) + for line in wt_result.stdout.split("\n"): + if line.startswith("branch refs/heads/"): + active_branches.add(line.split("branch refs/heads/", 1)[-1].strip()) + except Exception: + return # Can't determine active branches — bail + + # Also protect the currently checked-out branch and main + try: + head_result = subprocess.run( + ["git", "branch", "--show-current"], + capture_output=True, text=True, timeout=5, cwd=repo_root, + ) + current = head_result.stdout.strip() + if current: + active_branches.add(current) + except Exception: + pass + active_branches.add("main") + + orphaned = [ + b for b in all_branches + if b not in active_branches + and (b.startswith("hermes/hermes-") or b.startswith("pr-")) + ] + + if not orphaned: + return + + # Delete in batches + for i in range(0, len(orphaned), 50): + batch = orphaned[i:i + 50] + try: + subprocess.run( + ["git", "branch", "-D"] + batch, + capture_output=True, text=True, timeout=30, cwd=repo_root, + ) + except Exception as e: + logger.debug("Failed to prune orphaned branches: %s", e) + + logger.debug("Pruned %d orphaned branches", len(orphaned)) + # ============================================================================ # ASCII Art & Branding # ============================================================================ diff --git a/tests/cli/test_worktree.py b/tests/cli/test_worktree.py index f545baa39..fece9cf6b 100644 --- a/tests/cli/test_worktree.py +++ b/tests/cli/test_worktree.py @@ -33,6 +33,13 @@ def git_repo(tmp_path): ["git", "commit", "-m", "Initial commit"], cwd=repo, capture_output=True, ) + # Add a fake remote ref so cleanup logic sees the initial commit as + # "pushed". Without this, `git log HEAD --not --remotes` treats every + # commit as unpushed and cleanup refuses to delete worktrees. + subprocess.run( + ["git", "update-ref", "refs/remotes/origin/main", "HEAD"], + cwd=repo, capture_output=True, + ) return repo @@ -81,7 +88,11 @@ def _setup_worktree(repo_root): def _cleanup_worktree(info): - """Test version of _cleanup_worktree.""" + """Test version of _cleanup_worktree. + + Preserves the worktree only if it has unpushed commits. + Dirty working tree alone is not enough to keep it. + """ wt_path = info["path"] branch = info["branch"] repo_root = info["repo_root"] @@ -89,15 +100,15 @@ def _cleanup_worktree(info): if not Path(wt_path).exists(): return - # Check for uncommitted changes - status = subprocess.run( - ["git", "status", "--porcelain"], + # Check for unpushed commits + result = subprocess.run( + ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], capture_output=True, text=True, timeout=10, cwd=wt_path, ) - has_changes = bool(status.stdout.strip()) + has_unpushed = bool(result.stdout.strip()) - if has_changes: - return False # Did not clean up + if has_unpushed: + return False # Did not clean up — has unpushed commits subprocess.run( ["git", "worktree", "remove", wt_path, "--force"], @@ -204,20 +215,45 @@ class TestWorktreeCleanup: assert result is True assert not Path(info["path"]).exists() - def test_dirty_worktree_kept(self, git_repo): + def test_dirty_worktree_cleaned_when_no_unpushed(self, git_repo): + """Dirty working tree without unpushed commits is cleaned up. + + Agent sessions typically leave untracked files / artifacts behind. + Since all real work is in pushed commits, these don't warrant + keeping the worktree. + """ info = _setup_worktree(str(git_repo)) assert info is not None - # Make uncommitted changes + # Make uncommitted changes (untracked file) (Path(info["path"]) / "new-file.txt").write_text("uncommitted") subprocess.run( ["git", "add", "new-file.txt"], cwd=info["path"], capture_output=True, ) + # The git_repo fixture already has a fake remote ref so the initial + # commit is seen as "pushed". No unpushed commits → cleanup proceeds. result = _cleanup_worktree(info) - assert result is False - assert Path(info["path"]).exists() # Still there + assert result is True # Cleaned up despite dirty working tree + assert not Path(info["path"]).exists() + + def test_worktree_with_unpushed_commits_kept(self, git_repo): + """Worktree with unpushed commits is preserved.""" + info = _setup_worktree(str(git_repo)) + assert info is not None + + # Make a commit that is NOT on any remote + (Path(info["path"]) / "work.txt").write_text("real work") + subprocess.run(["git", "add", "work.txt"], cwd=info["path"], capture_output=True) + subprocess.run( + ["git", "commit", "-m", "agent work"], + cwd=info["path"], capture_output=True, + ) + + result = _cleanup_worktree(info) + assert result is False # Kept — has unpushed commits + assert Path(info["path"]).exists() def test_branch_deleted_on_cleanup(self, git_repo): info = _setup_worktree(str(git_repo)) @@ -367,7 +403,7 @@ class TestMultipleWorktrees: lines = [l for l in result.stdout.strip().splitlines() if l.strip()] assert len(lines) == 11 - # Cleanup all + # Cleanup all (git_repo fixture has a fake remote ref so cleanup works) for info in worktrees: # Discard changes first so cleanup works subprocess.run( @@ -492,33 +528,77 @@ class TestStaleWorktreePruning: assert not pruned assert Path(info["path"]).exists() - def test_keeps_dirty_old_worktree(self, git_repo): - """Old worktrees with uncommitted changes should NOT be pruned.""" + def test_keeps_old_worktree_with_unpushed_commits(self, git_repo): + """Old worktrees (24-72h) with unpushed commits should NOT be pruned.""" import time info = _setup_worktree(str(git_repo)) assert info is not None - # Make it dirty - (Path(info["path"]) / "dirty.txt").write_text("uncommitted") + # Make an unpushed commit + (Path(info["path"]) / "work.txt").write_text("real work") + subprocess.run(["git", "add", "work.txt"], cwd=info["path"], capture_output=True) subprocess.run( - ["git", "add", "dirty.txt"], + ["git", "commit", "-m", "agent work"], cwd=info["path"], capture_output=True, ) - # Make it old + # Make it old (25h — in the 24-72h soft tier) old_time = time.time() - (25 * 3600) os.utime(info["path"], (old_time, old_time)) - # Check if it would be pruned - status = subprocess.run( - ["git", "status", "--porcelain"], + # Check for unpushed commits (simulates prune logic) + result = subprocess.run( + ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], capture_output=True, text=True, cwd=info["path"], ) - has_changes = bool(status.stdout.strip()) - assert has_changes # Should be dirty → not pruned + has_unpushed = bool(result.stdout.strip()) + assert has_unpushed # Has unpushed commits → not pruned in soft tier assert Path(info["path"]).exists() + def test_force_prunes_very_old_worktree(self, git_repo): + """Worktrees older than 72h should be force-pruned regardless.""" + import time + + info = _setup_worktree(str(git_repo)) + assert info is not None + + # Make an unpushed commit (would normally protect it) + (Path(info["path"]) / "work.txt").write_text("stale work") + subprocess.run(["git", "add", "work.txt"], cwd=info["path"], capture_output=True) + subprocess.run( + ["git", "commit", "-m", "old agent work"], + cwd=info["path"], capture_output=True, + ) + + # Make it very old (73h — beyond the 72h hard threshold) + old_time = time.time() - (73 * 3600) + os.utime(info["path"], (old_time, old_time)) + + # Simulate the force-prune tier check + hard_cutoff = time.time() - (72 * 3600) + mtime = Path(info["path"]).stat().st_mtime + assert mtime <= hard_cutoff # Should qualify for force removal + + # Actually remove it (simulates _prune_stale_worktrees force path) + branch_result = subprocess.run( + ["git", "branch", "--show-current"], + capture_output=True, text=True, timeout=5, cwd=info["path"], + ) + branch = branch_result.stdout.strip() + + subprocess.run( + ["git", "worktree", "remove", info["path"], "--force"], + capture_output=True, text=True, timeout=15, cwd=str(git_repo), + ) + if branch: + subprocess.run( + ["git", "branch", "-D", branch], + capture_output=True, text=True, timeout=10, cwd=str(git_repo), + ) + + assert not Path(info["path"]).exists() + class TestEdgeCases: """Test edge cases for robustness.""" @@ -611,6 +691,133 @@ class TestTerminalCWDIntegration: assert result.stdout.strip() == "true" +class TestOrphanedBranchPruning: + """Test cleanup of orphaned hermes/* and pr-* branches.""" + + def test_prunes_orphaned_hermes_branch(self, git_repo): + """hermes/hermes-* branches with no worktree should be deleted.""" + # Create a branch that looks like a worktree branch but has no worktree + subprocess.run( + ["git", "branch", "hermes/hermes-deadbeef", "HEAD"], + cwd=str(git_repo), capture_output=True, + ) + + # Verify it exists + result = subprocess.run( + ["git", "branch", "--list", "hermes/hermes-deadbeef"], + capture_output=True, text=True, cwd=str(git_repo), + ) + assert "hermes/hermes-deadbeef" in result.stdout + + # Simulate _prune_orphaned_branches logic + result = subprocess.run( + ["git", "branch", "--format=%(refname:short)"], + capture_output=True, text=True, cwd=str(git_repo), + ) + all_branches = [b.strip() for b in result.stdout.strip().split("\n") if b.strip()] + + wt_result = subprocess.run( + ["git", "worktree", "list", "--porcelain"], + capture_output=True, text=True, cwd=str(git_repo), + ) + active_branches = {"main"} + for line in wt_result.stdout.split("\n"): + if line.startswith("branch refs/heads/"): + active_branches.add(line.split("branch refs/heads/", 1)[-1].strip()) + + orphaned = [ + b for b in all_branches + if b not in active_branches + and (b.startswith("hermes/hermes-") or b.startswith("pr-")) + ] + assert "hermes/hermes-deadbeef" in orphaned + + # Delete them + if orphaned: + subprocess.run( + ["git", "branch", "-D"] + orphaned, + capture_output=True, text=True, cwd=str(git_repo), + ) + + # Verify gone + result = subprocess.run( + ["git", "branch", "--list", "hermes/hermes-deadbeef"], + capture_output=True, text=True, cwd=str(git_repo), + ) + assert "hermes/hermes-deadbeef" not in result.stdout + + def test_prunes_orphaned_pr_branch(self, git_repo): + """pr-* branches should be deleted during pruning.""" + subprocess.run( + ["git", "branch", "pr-1234", "HEAD"], + cwd=str(git_repo), capture_output=True, + ) + subprocess.run( + ["git", "branch", "pr-5678", "HEAD"], + cwd=str(git_repo), capture_output=True, + ) + + result = subprocess.run( + ["git", "branch", "--format=%(refname:short)"], + capture_output=True, text=True, cwd=str(git_repo), + ) + all_branches = [b.strip() for b in result.stdout.strip().split("\n") if b.strip()] + + active_branches = {"main"} + orphaned = [ + b for b in all_branches + if b not in active_branches and b.startswith("pr-") + ] + assert "pr-1234" in orphaned + assert "pr-5678" in orphaned + + subprocess.run( + ["git", "branch", "-D"] + orphaned, + capture_output=True, text=True, cwd=str(git_repo), + ) + + # Verify gone + result = subprocess.run( + ["git", "branch", "--format=%(refname:short)"], + capture_output=True, text=True, cwd=str(git_repo), + ) + remaining = result.stdout.strip() + assert "pr-1234" not in remaining + assert "pr-5678" not in remaining + + def test_preserves_active_worktree_branch(self, git_repo): + """Branches with active worktrees should NOT be pruned.""" + info = _setup_worktree(str(git_repo)) + assert info is not None + + result = subprocess.run( + ["git", "worktree", "list", "--porcelain"], + capture_output=True, text=True, cwd=str(git_repo), + ) + active_branches = set() + for line in result.stdout.split("\n"): + if line.startswith("branch refs/heads/"): + active_branches.add(line.split("branch refs/heads/", 1)[-1].strip()) + + assert info["branch"] in active_branches # Protected + + def test_preserves_main_branch(self, git_repo): + """main branch should never be pruned.""" + result = subprocess.run( + ["git", "branch", "--format=%(refname:short)"], + capture_output=True, text=True, cwd=str(git_repo), + ) + all_branches = [b.strip() for b in result.stdout.strip().split("\n") if b.strip()] + active_branches = {"main"} + + orphaned = [ + b for b in all_branches + if b not in active_branches + and (b.startswith("hermes/hermes-") or b.startswith("pr-")) + ] + assert "main" not in orphaned + + class TestSystemPromptInjection: """Test that the agent gets worktree context in its system prompt.""" @@ -625,7 +832,7 @@ class TestSystemPromptInjection: f"{info['path']}. Your branch is `{info['branch']}`. " f"Changes here do not affect the main working tree or other agents. " f"Remember to commit and push your changes, and create a PR if appropriate. " - f"The original repo is at {info['repo_root']}.]" + f"The original repo is at {info['repo_root']}.]\n" ) assert info["path"] in wt_note