From 28ab420302c05a2e9075bbbcb04c525f4a721c77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E7=B6=A0BG?= Date: Fri, 17 Apr 2026 11:58:30 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(cli):=20handle=20no-remote?= =?UTF-8?q?=20worktree=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cli.py | 53 ++++++++++------- tests/cli/test_worktree.py | 116 ++++++++++++++++++++++++++++++++----- 2 files changed, 133 insertions(+), 36 deletions(-) diff --git a/cli.py b/cli.py index d9192020b64..d1cf26a2a64 100644 --- a/cli.py +++ b/cli.py @@ -989,6 +989,35 @@ def _setup_worktree(repo_root: str = None) -> Optional[Dict[str, str]]: return info +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. + """ + import subprocess + + try: + 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 + + 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 + + def _cleanup_worktree(info: Dict[str, str] = None) -> None: """Remove a worktree and its branch on exit. @@ -1011,18 +1040,7 @@ def _cleanup_worktree(info: Dict[str, str] = None) -> None: if not Path(wt_path).exists(): return - # 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: - result = subprocess.run( - ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], - capture_output=True, text=True, timeout=10, cwd=wt_path, - ) - has_unpushed = bool(result.stdout.strip()) - except Exception: - has_unpushed = True # Assume unpushed on error — don't delete + has_unpushed = _worktree_has_unpushed_commits(wt_path, timeout=10) if has_unpushed: print(f"\n\033[33m⚠ Worktree has unpushed commits, keeping: {wt_path}\033[0m") @@ -1170,15 +1188,8 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None: 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 + if _worktree_has_unpushed_commits(str(entry), timeout=5): + continue # Has unpushed commits or can't check — skip # Safe to remove try: diff --git a/tests/cli/test_worktree.py b/tests/cli/test_worktree.py index fece9cf6be9..3340deb3f0b 100644 --- a/tests/cli/test_worktree.py +++ b/tests/cli/test_worktree.py @@ -33,9 +33,12 @@ def git_repo(tmp_path): ["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, + ) # 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. + # "pushed" when a remote is configured. subprocess.run( ["git", "update-ref", "refs/remotes/origin/main", "HEAD"], cwd=repo, capture_output=True, @@ -43,6 +46,29 @@ def git_repo(tmp_path): return repo +@pytest.fixture +def git_repo_no_remote(tmp_path): + """Create a temporary git repo with no configured remotes.""" + repo = tmp_path / "test-repo-no-remote" + 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, + ) + return repo + + # --------------------------------------------------------------------------- # Lightweight reimplementations for testing (avoid importing cli.py) # --------------------------------------------------------------------------- @@ -87,6 +113,26 @@ 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 + + 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()) + + def _cleanup_worktree(info): """Test version of _cleanup_worktree. @@ -100,14 +146,7 @@ def _cleanup_worktree(info): if not Path(wt_path).exists(): return - # Check for unpushed commits - result = subprocess.run( - ["git", "log", "--oneline", "HEAD", "--not", "--remotes"], - capture_output=True, text=True, timeout=10, cwd=wt_path, - ) - has_unpushed = bool(result.stdout.strip()) - - if has_unpushed: + if _has_unpushed_commits(wt_path, timeout=10): return False # Did not clean up — has unpushed commits subprocess.run( @@ -255,6 +294,17 @@ class TestWorktreeCleanup: assert result is False # Kept — has unpushed commits assert Path(info["path"]).exists() + def test_clean_worktree_removed_without_remote(self, git_repo_no_remote): + """Clean worktrees in repos without remotes should still be removed.""" + info = _setup_worktree(str(git_repo_no_remote)) + 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"] @@ -548,14 +598,50 @@ class TestStaleWorktreePruning: os.utime(info["path"], (old_time, old_time)) # 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_unpushed = bool(result.stdout.strip()) + has_unpushed = _has_unpushed_commits(info["path"]) assert has_unpushed # Has unpushed commits → not pruned in soft tier assert Path(info["path"]).exists() + def test_prunes_old_clean_worktree_without_remote(self, git_repo_no_remote): + """Old clean worktrees in repos without remotes should not be kept.""" + import time + + info = _setup_worktree(str(git_repo_no_remote)) + 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_no_remote / ".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_no_remote), + ) + if branch: + subprocess.run( + ["git", "branch", "-D", branch], + capture_output=True, text=True, timeout=10, cwd=str(git_repo_no_remote), + ) + + 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