diff --git a/cli.py b/cli.py index d1cf26a2a64..a700f0ddded 100644 --- a/cli.py +++ b/cli.py @@ -992,19 +992,21 @@ def _setup_worktree(repo_root: str = None) -> Optional[Dict[str, str]]: def _worktree_has_unpushed_commits(worktree_path: str, timeout: int = 10) -> bool: """Return whether a worktree has commits not reachable from any remote branch. - If the repo has no configured remotes, treat it as having no "unpushed" - commits because there is no remote baseline to compare against. + ``git log HEAD --not --remotes`` compares against remote-tracking refs under + ``refs/remotes/*``. If a repo has no remote-tracking refs yet, there is no + usable remote baseline to compare against, so treat it as having no + "unpushed" commits. """ import subprocess try: - remotes = subprocess.run( - ["git", "remote"], + remote_refs = subprocess.run( + ["git", "for-each-ref", "--format=%(refname)", "refs/remotes"], capture_output=True, text=True, timeout=timeout, cwd=worktree_path, ) - if remotes.returncode != 0: + if remote_refs.returncode != 0: return True - if not remotes.stdout.strip(): + if not remote_refs.stdout.strip(): return False result = subprocess.run( diff --git a/tests/cli/test_worktree.py b/tests/cli/test_worktree.py index 3340deb3f0b..b139acf7d2f 100644 --- a/tests/cli/test_worktree.py +++ b/tests/cli/test_worktree.py @@ -69,6 +69,33 @@ def git_repo_no_remote(tmp_path): return repo +@pytest.fixture +def git_repo_remote_no_tracking(tmp_path): + """Create a temporary git repo with a remote but no remote-tracking refs.""" + repo = tmp_path / "test-repo-remote-no-tracking" + repo.mkdir() + subprocess.run(["git", "init"], cwd=repo, capture_output=True) + subprocess.run( + ["git", "config", "user.email", "test@test.com"], + cwd=repo, capture_output=True, + ) + subprocess.run( + ["git", "config", "user.name", "Test"], + cwd=repo, capture_output=True, + ) + (repo / "README.md").write_text("# Test Repo\n") + subprocess.run(["git", "add", "."], cwd=repo, capture_output=True) + subprocess.run( + ["git", "commit", "-m", "Initial commit"], + cwd=repo, capture_output=True, + ) + subprocess.run( + ["git", "remote", "add", "origin", "https://example.com/test-repo.git"], + cwd=repo, capture_output=True, + ) + return repo + + # --------------------------------------------------------------------------- # Lightweight reimplementations for testing (avoid importing cli.py) # --------------------------------------------------------------------------- @@ -115,22 +142,25 @@ def _setup_worktree(repo_root): def _has_unpushed_commits(worktree_path, timeout=10): """Test version of the worktree unpushed-commit helper.""" - remotes = subprocess.run( - ["git", "remote"], - capture_output=True, text=True, timeout=timeout, cwd=worktree_path, - ) - if remotes.returncode != 0: - return True - if not remotes.stdout.strip(): - return False + try: + remote_refs = subprocess.run( + ["git", "for-each-ref", "--format=%(refname)", "refs/remotes"], + capture_output=True, text=True, timeout=timeout, cwd=worktree_path, + ) + if remote_refs.returncode != 0: + return True + if not remote_refs.stdout.strip(): + return False - result = subprocess.run( - ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], - capture_output=True, text=True, timeout=timeout, cwd=worktree_path, - ) - if result.returncode != 0: + result = subprocess.run( + ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], + capture_output=True, text=True, timeout=timeout, cwd=worktree_path, + ) + if result.returncode != 0: + return True + return bool(result.stdout.strip()) + except Exception: return True - return bool(result.stdout.strip()) def _cleanup_worktree(info): @@ -305,6 +335,19 @@ class TestWorktreeCleanup: assert result is True assert not Path(info["path"]).exists() + def test_clean_worktree_removed_without_remote_tracking_refs( + self, git_repo_remote_no_tracking + ): + """Configured remotes without fetched refs should not block cleanup.""" + info = _setup_worktree(str(git_repo_remote_no_tracking)) + assert info is not None + assert Path(info["path"]).exists() + assert _has_unpushed_commits(info["path"], timeout=10) is False + + result = _cleanup_worktree(info) + assert result is True + assert not Path(info["path"]).exists() + def test_branch_deleted_on_cleanup(self, git_repo): info = _setup_worktree(str(git_repo)) branch = info["branch"] @@ -642,6 +685,50 @@ class TestStaleWorktreePruning: assert not Path(info["path"]).exists() + def test_prunes_old_clean_worktree_without_remote_tracking_refs( + self, git_repo_remote_no_tracking + ): + """Old clean worktrees with no fetched remote refs should be pruned.""" + import time + + info = _setup_worktree(str(git_repo_remote_no_tracking)) + assert info is not None + assert Path(info["path"]).exists() + + old_time = time.time() - (25 * 3600) + os.utime(info["path"], (old_time, old_time)) + + worktrees_dir = git_repo_remote_no_tracking / ".worktrees" + cutoff = time.time() - (24 * 3600) + + for entry in worktrees_dir.iterdir(): + if not entry.is_dir() or not entry.name.startswith("hermes-"): + continue + mtime = entry.stat().st_mtime + if mtime > cutoff: + continue + if _has_unpushed_commits(str(entry), timeout=5): + continue + + branch_result = subprocess.run( + ["git", "branch", "--show-current"], + capture_output=True, text=True, timeout=5, cwd=str(entry), + ) + branch = branch_result.stdout.strip() + subprocess.run( + ["git", "worktree", "remove", str(entry), "--force"], + capture_output=True, text=True, timeout=15, + cwd=str(git_repo_remote_no_tracking), + ) + if branch: + subprocess.run( + ["git", "branch", "-D", branch], + capture_output=True, text=True, timeout=10, + cwd=str(git_repo_remote_no_tracking), + ) + + assert not 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