diff --git a/apps/backend/core/worktree.py b/apps/backend/core/worktree.py index 596a3be4..c608674f 100644 --- a/apps/backend/core/worktree.py +++ b/apps/backend/core/worktree.py @@ -378,6 +378,66 @@ class WorktreeManager: return "auto-claude" return None + def _branch_exists(self, branch_name: str) -> bool: + """ + Check if a local branch exists in the repository. + + Uses git show-ref to specifically check for local branches, avoiding + false positives from tags or other refs with the same name. + + Args: + branch_name: The name of the branch to check (e.g., 'auto-claude/my-spec') + + Returns: + True if the local branch exists, False otherwise. + """ + result = self._run_git(["show-ref", "--verify", f"refs/heads/{branch_name}"]) + return result.returncode == 0 + + def _worktree_is_registered(self, worktree_path: Path) -> bool: + """ + Check if a worktree path is registered with git. + + This determines if git tracks the worktree even if the directory exists. + Useful for detecting orphaned worktree directories that need cleanup. + + Args: + worktree_path: The path to the worktree directory to check. + + Returns: + True if the worktree is registered with git, False otherwise. + """ + result = self._run_git(["worktree", "list", "--porcelain"]) + if result.returncode != 0: + return False + + # Parse porcelain output to get registered worktree paths + # Format: "worktree /path/to/worktree" for each worktree + registered_paths = set() + for line in result.stdout.split("\n"): + if line.startswith("worktree "): + parts = line.split(" ", 1) + if len(parts) == 2: + registered_paths.add(Path(parts[1])) + + # Check if worktree_path matches any registered path + # Use samefile() for accurate comparison on case-insensitive filesystems + resolved_path = worktree_path.resolve() + for registered_path in registered_paths: + # Try samefile first (handles case-insensitivity and symlinks) + try: + if resolved_path.exists() and registered_path.exists(): + if os.path.samefile(resolved_path, registered_path): + return True + except OSError: + pass + # Fallback to normalized case comparison for non-existent paths + if os.path.normcase(str(resolved_path)) == os.path.normcase( + str(registered_path) + ): + return True + return False + def _get_worktree_stats(self, spec_name: str) -> dict: """Get diff statistics for a worktree.""" worktree_path = self.get_worktree_path(spec_name) @@ -467,13 +527,25 @@ class WorktreeManager: def create_worktree(self, spec_name: str) -> WorktreeInfo: """ - Create a worktree for a spec. + Create a worktree for a spec (idempotent). + + This method is idempotent - calling it multiple times with the same spec_name + will succeed regardless of prior state. It handles: + - Existing valid worktrees (returns existing) + - Corrupted worktrees (force removes and recreates) + - Orphaned worktree references (prunes them) + - Stale worktree directories (cleans them up) + - Existing branches without worktrees (reuses the branch) + + Note: + This method is NOT thread-safe for concurrent calls with the same spec_name. + If concurrent access is needed, implement external locking. Args: spec_name: The spec folder name (e.g., "002-implement-memory") Returns: - WorktreeInfo for the created worktree + WorktreeInfo for the created or existing worktree Raises: WorktreeError: If a branch namespace conflict exists or worktree creation fails @@ -481,7 +553,11 @@ class WorktreeManager: worktree_path = self.get_worktree_path(spec_name) branch_name = self.get_branch_name(spec_name) - # Check for branch namespace conflict (e.g., 'auto-claude' blocking 'auto-claude/*') + # Step 1: Prune orphaned worktree references first + # This cleans up any stale references that might block operations + self._run_git(["worktree", "prune"]) + + # Step 2: Check for branch namespace conflict (e.g., 'auto-claude' blocking 'auto-claude/*') conflicting_branch = self._check_branch_namespace_conflict() if conflicting_branch: raise WorktreeError( @@ -494,14 +570,41 @@ class WorktreeManager: f" git branch -m {conflicting_branch} {conflicting_branch}-backup" ) - # Remove existing if present (from crashed previous run) - if worktree_path.exists(): - self._run_git(["worktree", "remove", "--force", str(worktree_path)]) + # Step 3: Check if worktree already exists and is valid + if worktree_path.exists() and self._worktree_is_registered(worktree_path): + # Worktree exists and is tracked by git - return existing (idempotent) + existing = self.get_worktree_info(spec_name) + if existing: + print( + f"Using existing worktree: {worktree_path.name} on branch {existing.branch}" + ) + return existing + else: + # Worktree is registered but corrupted (e.g., unreadable HEAD) + # Force remove the registration and let it be recreated + print(f"Removing corrupted worktree registration: {worktree_path.name}") + remove_result = self._run_git( + ["worktree", "remove", "--force", str(worktree_path)] + ) + if remove_result.returncode != 0: + raise WorktreeError( + f"Failed to remove corrupted worktree: {remove_result.stderr}" + ) - # Delete branch if it exists (from previous attempt) - self._run_git(["branch", "-D", branch_name]) + # Step 4: Handle stale worktree directory (exists but not registered with git) + if worktree_path.exists() and not self._worktree_is_registered(worktree_path): + print(f"Removing stale worktree directory: {worktree_path.name}") + shutil.rmtree(worktree_path, ignore_errors=True) + if worktree_path.exists(): + raise WorktreeError( + f"Failed to remove stale worktree directory: {worktree_path}\n" + f"This may be due to permission issues or file locks." + ) - # Fetch latest from remote to ensure we have the most up-to-date code + # Step 5: Check if branch already exists + branch_exists = self._branch_exists(branch_name) + + # Step 6: Fetch latest from remote to ensure we have the most up-to-date code # GitHub/remote is the source of truth, not the local branch fetch_result = self._run_git(["fetch", "origin", self.base_branch]) if fetch_result.returncode != 0: @@ -510,25 +613,31 @@ class WorktreeManager: ) print("Falling back to local branch...") - # Determine the start point for the worktree - # Prefer origin/{base_branch} (remote) over local branch to ensure we have latest code - remote_ref = f"origin/{self.base_branch}" - start_point = self.base_branch # Default to local branch - - # Check if remote ref exists and use it as the source of truth - check_remote = self._run_git(["rev-parse", "--verify", remote_ref]) - if check_remote.returncode == 0: - start_point = remote_ref - print(f"Creating worktree from remote: {remote_ref}") + # Step 7: Create the worktree + if branch_exists: + # Branch exists - attach worktree to existing branch (no -b flag) + print(f"Reusing existing branch: {branch_name}") + result = self._run_git(["worktree", "add", str(worktree_path), branch_name]) else: - print( - f"Remote ref {remote_ref} not found, using local branch: {self.base_branch}" - ) + # Branch doesn't exist - create new branch from remote or local base + # Determine the start point for the worktree + remote_ref = f"origin/{self.base_branch}" + start_point = self.base_branch # Default to local branch - # Create worktree with new branch from the start point (remote preferred) - result = self._run_git( - ["worktree", "add", "-b", branch_name, str(worktree_path), start_point] - ) + # Check if remote ref exists and use it as the source of truth + check_remote = self._run_git(["rev-parse", "--verify", remote_ref]) + if check_remote.returncode == 0: + start_point = remote_ref + print(f"Creating worktree from remote: {remote_ref}") + else: + print( + f"Remote ref {remote_ref} not found, using local branch: {self.base_branch}" + ) + + # Create worktree with new branch from the start point + result = self._run_git( + ["worktree", "add", "-b", branch_name, str(worktree_path), start_point] + ) if result.returncode != 0: raise WorktreeError( diff --git a/tests/test_worktree.py b/tests/test_worktree.py index a6b725d2..f8cb4101 100644 --- a/tests/test_worktree.py +++ b/tests/test_worktree.py @@ -29,7 +29,10 @@ class TestWorktreeManagerInitialization: manager = WorktreeManager(temp_git_repo) assert manager.project_dir == temp_git_repo - assert manager.worktrees_dir == temp_git_repo / ".auto-claude" / "worktrees" / "tasks" + assert ( + manager.worktrees_dir + == temp_git_repo / ".auto-claude" / "worktrees" / "tasks" + ) assert manager.base_branch is not None def test_init_prefers_main_over_current_branch(self, temp_git_repo: Path): @@ -37,7 +40,8 @@ class TestWorktreeManagerInitialization: # Create and switch to a new branch subprocess.run( ["git", "checkout", "-b", "feature-branch"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, ) # Even though we're on feature-branch, manager should prefer main @@ -49,11 +53,11 @@ class TestWorktreeManagerInitialization: # Delete main branch to force fallback subprocess.run( ["git", "checkout", "-b", "feature-branch"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, ) subprocess.run( - ["git", "branch", "-D", "main"], - cwd=temp_git_repo, capture_output=True + ["git", "branch", "-D", "main"], cwd=temp_git_repo, capture_output=True ) manager = WorktreeManager(temp_git_repo) @@ -113,6 +117,187 @@ class TestWorktreeCreation: # The test file should still be there (same worktree) assert (info2.path / "test-file.txt").exists() + def test_create_worktree_idempotent(self, temp_git_repo: Path): + """create_worktree succeeds when called twice with same spec name.""" + manager = WorktreeManager(temp_git_repo) + manager.setup() + + # First creation should succeed + info1 = manager.create_worktree("test-spec") + assert info1.path.exists() + assert info1.branch == "auto-claude/test-spec" + + # Create a file in the worktree to verify it's preserved + (info1.path / "test-file.txt").write_text("test content") + + # Second creation should also succeed (idempotent) + info2 = manager.create_worktree("test-spec") + + # Should return valid worktree info + assert info2.path.exists() + assert info2.branch == "auto-claude/test-spec" + # The test file should still be there (same worktree returned) + assert (info2.path / "test-file.txt").exists() + assert (info2.path / "test-file.txt").read_text() == "test content" + + def test_create_worktree_branch_exists_no_worktree(self, temp_git_repo: Path): + """create_worktree reuses existing branch when worktree is missing.""" + manager = WorktreeManager(temp_git_repo) + manager.setup() + + # Create initial worktree + info1 = manager.create_worktree("test-spec") + branch_name = info1.branch + assert info1.path.exists() + assert branch_name == "auto-claude/test-spec" + + # Remove worktree but keep the branch (delete_branch=False is default) + manager.remove_worktree("test-spec", delete_branch=False) + + # Verify worktree directory is gone + assert not info1.path.exists() + + # Verify branch still exists + result = subprocess.run( + ["git", "branch", "--list", branch_name], + cwd=temp_git_repo, + capture_output=True, + text=True, + ) + assert branch_name in result.stdout, ( + "Branch should still exist after worktree removal" + ) + + # Create worktree again - should succeed by reusing existing branch + info2 = manager.create_worktree("test-spec") + + # Should return valid worktree info with the same branch + assert info2.path.exists() + assert info2.branch == branch_name + assert info2.is_active is True + # README should exist (copied from base branch) + assert (info2.path / "README.md").exists() + + def test_create_worktree_stale_directory(self, temp_git_repo: Path): + """create_worktree cleans up stale directory and recreates worktree.""" + manager = WorktreeManager(temp_git_repo) + manager.setup() + + # Create a worktree normally + info = manager.create_worktree("test-spec") + worktree_path = info.path + branch_name = info.branch + assert worktree_path.exists() + + # Add a file to the worktree so we can verify it gets cleaned up + (worktree_path / "test-file.txt").write_text("test content") + + # Force-remove the worktree from git's tracking, but leave directory intact + # This simulates a stale state where directory exists but git doesn't track it + result = subprocess.run( + ["git", "worktree", "remove", "--force", str(worktree_path)], + cwd=temp_git_repo, + capture_output=True, + ) + assert result.returncode == 0, ( + f"Failed to force remove worktree: {result.stderr}" + ) + + # Recreate the directory manually to simulate stale state + # (git worktree remove also deletes the directory, so we recreate it) + worktree_path.mkdir(parents=True, exist_ok=True) + (worktree_path / "stale-file.txt").write_text("stale content") + + # Verify directory exists but is not tracked by git + assert worktree_path.exists() + wt_list_result = subprocess.run( + ["git", "worktree", "list", "--porcelain"], + cwd=temp_git_repo, + capture_output=True, + text=True, + ) + assert str(worktree_path) not in wt_list_result.stdout, ( + "Worktree should not be registered" + ) + + # Now create_worktree should clean up the stale directory and recreate successfully + info2 = manager.create_worktree("test-spec") + + # Should return valid worktree info + assert info2.path.exists() + assert info2.branch == branch_name + assert info2.is_active is True + # README should exist (from base branch) + assert (info2.path / "README.md").exists() + # Stale file should be gone (directory was cleaned up) + assert not (info2.path / "stale-file.txt").exists() + + def test_create_worktree_stale_directory_with_existing_branch( + self, temp_git_repo: Path + ): + """create_worktree handles stale directory when branch already exists.""" + manager = WorktreeManager(temp_git_repo) + manager.setup() + + # Create a worktree normally + info = manager.create_worktree("test-spec") + worktree_path = info.path + branch_name = info.branch + assert worktree_path.exists() + + # Unregister the worktree but KEEP the branch + # Use 'git worktree remove' which removes directory, then manually recreate stale dir + # But first we need to ensure the branch survives + result = subprocess.run( + ["git", "worktree", "remove", "--force", str(worktree_path)], + cwd=temp_git_repo, + capture_output=True, + ) + assert result.returncode == 0, f"Failed to remove worktree: {result.stderr}" + + # Verify branch still exists (git worktree remove doesn't delete branch) + result = subprocess.run( + ["git", "branch", "--list", branch_name], + cwd=temp_git_repo, + capture_output=True, + text=True, + ) + assert branch_name in result.stdout, ( + "Branch should still exist after worktree removal" + ) + + # Recreate stale directory manually (simulates orphaned directory) + worktree_path.mkdir(parents=True, exist_ok=True) + (worktree_path / "stale-file.txt").write_text("stale content") + + # Verify: directory exists, worktree NOT registered, branch EXISTS + assert worktree_path.exists() + wt_list_result = subprocess.run( + ["git", "worktree", "list", "--porcelain"], + cwd=temp_git_repo, + capture_output=True, + text=True, + ) + assert str(worktree_path) not in wt_list_result.stdout, ( + "Worktree should not be registered" + ) + + # Now create_worktree should: + # 1. Detect stale directory (not registered) + # 2. Clean up stale directory + # 3. Detect existing branch + # 4. Reuse existing branch (no -b flag) + info2 = manager.create_worktree("test-spec") + + # Should return valid worktree info with SAME branch (reused) + assert info2.path.exists() + assert info2.branch == branch_name + assert info2.is_active is True + # README should exist (from branch content) + assert (info2.path / "README.md").exists() + # Stale file should be gone (directory was cleaned up before worktree add) + assert not (info2.path / "stale-file.txt").exists() + class TestWorktreeRemoval: """Tests for removing worktrees.""" @@ -139,7 +324,9 @@ class TestWorktreeRemoval: # Verify branch is deleted result = subprocess.run( ["git", "branch", "--list", branch_name], - cwd=temp_git_repo, capture_output=True, text=True + cwd=temp_git_repo, + capture_output=True, + text=True, ) assert branch_name not in result.stdout @@ -158,7 +345,8 @@ class TestWorktreeCommitAndMerge: subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) subprocess.run( ["git", "commit", "-m", "Worker commit"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, ) # Merge worktree back to main @@ -167,7 +355,11 @@ class TestWorktreeCommitAndMerge: assert result is True # Verify file is in main branch - subprocess.run(["git", "checkout", manager.base_branch], cwd=temp_git_repo, capture_output=True) + subprocess.run( + ["git", "checkout", manager.base_branch], + cwd=temp_git_repo, + capture_output=True, + ) assert (temp_git_repo / "worker-file.txt").exists() def test_merge_worktree_already_on_target_branch(self, temp_git_repo: Path): @@ -176,17 +368,24 @@ class TestWorktreeCommitAndMerge: manager.setup() # Ensure we're on the base branch - result = subprocess.run(["git", "checkout", manager.base_branch], cwd=temp_git_repo, capture_output=True) + result = subprocess.run( + ["git", "checkout", manager.base_branch], + cwd=temp_git_repo, + capture_output=True, + ) assert result.returncode == 0, f"Checkout failed: {result.stderr}" # Create a worktree with changes worker_info = manager.create_worktree("worker-spec") (worker_info.path / "worker-file.txt").write_text("worker content") - result = subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) + result = subprocess.run( + ["git", "add", "."], cwd=worker_info.path, capture_output=True + ) assert result.returncode == 0, f"Git add failed: {result.stderr}" result = subprocess.run( ["git", "commit", "-m", "Worker commit"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, ) assert result.returncode == 0, f"Commit failed: {result.stderr}" @@ -206,13 +405,18 @@ class TestWorktreeCommitAndMerge: # Create a worktree with changes worker_info = manager.create_worktree("worker-spec") (worker_info.path / "worker-file.txt").write_text("worker content") - add_result = subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=worker_info.path, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Worker commit"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # First merge succeeds result = manager.merge_worktree("worker-spec", delete_after=False) @@ -222,7 +426,9 @@ class TestWorktreeCommitAndMerge: result = manager.merge_worktree("worker-spec", delete_after=False) assert result is True - def test_merge_worktree_already_up_to_date_with_no_commit(self, temp_git_repo: Path): + def test_merge_worktree_already_up_to_date_with_no_commit( + self, temp_git_repo: Path + ): """merge_worktree with no_commit=True succeeds when already up to date (ACS-226).""" manager = WorktreeManager(temp_git_repo) manager.setup() @@ -230,30 +436,44 @@ class TestWorktreeCommitAndMerge: # Create a worktree with changes worker_info = manager.create_worktree("worker-spec") (worker_info.path / "worker-file.txt").write_text("worker content") - add_result = subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=worker_info.path, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Worker commit"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # First merge with no_commit succeeds - result = manager.merge_worktree("worker-spec", no_commit=True, delete_after=False) + result = manager.merge_worktree( + "worker-spec", no_commit=True, delete_after=False + ) assert result is True # Commit the staged changes merge_commit_result = subprocess.run( ["git", "commit", "-m", "Merge commit"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert merge_commit_result.returncode == 0, ( + f"git commit failed: {merge_commit_result.stderr}" ) - assert merge_commit_result.returncode == 0, f"git commit failed: {merge_commit_result.stderr}" # Second merge should also succeed (already up to date) - result = manager.merge_worktree("worker-spec", no_commit=True, delete_after=False) + result = manager.merge_worktree( + "worker-spec", no_commit=True, delete_after=False + ) assert result is True - def test_merge_worktree_already_up_to_date_with_delete_after(self, temp_git_repo: Path): + def test_merge_worktree_already_up_to_date_with_delete_after( + self, temp_git_repo: Path + ): """merge_worktree with delete_after=True succeeds when already up to date (ACS-226).""" manager = WorktreeManager(temp_git_repo) manager.setup() @@ -262,13 +482,18 @@ class TestWorktreeCommitAndMerge: worker_info = manager.create_worktree("worker-spec") branch_name = worker_info.branch (worker_info.path / "worker-file.txt").write_text("worker content") - add_result = subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=worker_info.path, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Worker commit"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # First merge succeeds result = manager.merge_worktree("worker-spec", delete_after=False) @@ -284,9 +509,13 @@ class TestWorktreeCommitAndMerge: # Verify branch was deleted branch_list_result = subprocess.run( ["git", "branch", "--list", branch_name], - cwd=temp_git_repo, capture_output=True, text=True + cwd=temp_git_repo, + capture_output=True, + text=True, + ) + assert branch_name not in branch_list_result.stdout, ( + f"Branch {branch_name} should be deleted" ) - assert branch_name not in branch_list_result.stdout, f"Branch {branch_name} should be deleted" def test_merge_worktree_conflict_detection(self, temp_git_repo: Path): """merge_worktree correctly detects and handles merge conflicts.""" @@ -295,34 +524,49 @@ class TestWorktreeCommitAndMerge: # Create initial file on base branch (temp_git_repo / "shared.txt").write_text("base content") - add_result = subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=temp_git_repo, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Add shared file"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # Create worktree with conflicting change worker_info = manager.create_worktree("worker-spec") (worker_info.path / "shared.txt").write_text("worker content") - add_result = subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=worker_info.path, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Worker change"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # Make conflicting change on base branch (temp_git_repo / "shared.txt").write_text("base change") - add_result = subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=temp_git_repo, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Base change"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # Merge should detect conflict and fail result = manager.merge_worktree("worker-spec", delete_after=False) @@ -332,18 +576,25 @@ class TestWorktreeCommitAndMerge: # Check that MERGE_HEAD does not exist merge_head_result = subprocess.run( ["git", "rev-parse", "--verify", "MERGE_HEAD"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert merge_head_result.returncode != 0, ( + "MERGE_HEAD should not exist after abort" ) - assert merge_head_result.returncode != 0, "MERGE_HEAD should not exist after abort" # Verify git status shows no unmerged/conflict status codes git_status = subprocess.run( ["git", "status", "--porcelain"], - cwd=temp_git_repo, capture_output=True, text=True + cwd=temp_git_repo, + capture_output=True, + text=True, ) # Should have no output (clean working directory) assert git_status.returncode == 0 - assert not git_status.stdout.strip(), f"Expected clean status, got: {git_status.stdout}" + assert not git_status.stdout.strip(), ( + f"Expected clean status, got: {git_status.stdout}" + ) def test_merge_worktree_conflict_with_no_commit(self, temp_git_repo: Path): """merge_worktree with no_commit=True handles conflicts correctly.""" @@ -352,54 +603,78 @@ class TestWorktreeCommitAndMerge: # Create initial file on base branch (temp_git_repo / "shared.txt").write_text("base content") - add_result = subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=temp_git_repo, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Add shared file"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # Create worktree with conflicting change worker_info = manager.create_worktree("worker-spec") (worker_info.path / "shared.txt").write_text("worker content") - add_result = subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=worker_info.path, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Worker change"], - cwd=worker_info.path, capture_output=True + cwd=worker_info.path, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # Make conflicting change on base branch (temp_git_repo / "shared.txt").write_text("base change") - add_result = subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True) + add_result = subprocess.run( + ["git", "add", "."], cwd=temp_git_repo, capture_output=True + ) assert add_result.returncode == 0, f"git add failed: {add_result.stderr}" commit_result = subprocess.run( ["git", "commit", "-m", "Base change"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert commit_result.returncode == 0, ( + f"git commit failed: {commit_result.stderr}" ) - assert commit_result.returncode == 0, f"git commit failed: {commit_result.stderr}" # Merge with no_commit should detect conflict and fail - result = manager.merge_worktree("worker-spec", no_commit=True, delete_after=False) + result = manager.merge_worktree( + "worker-spec", no_commit=True, delete_after=False + ) assert result is False # Verify merge was aborted (no merge state exists) # Check that MERGE_HEAD does not exist merge_head_result = subprocess.run( ["git", "rev-parse", "--verify", "MERGE_HEAD"], - cwd=temp_git_repo, capture_output=True + cwd=temp_git_repo, + capture_output=True, + ) + assert merge_head_result.returncode != 0, ( + "MERGE_HEAD should not exist after abort" ) - assert merge_head_result.returncode != 0, "MERGE_HEAD should not exist after abort" # Verify git status shows no staged/unstaged changes git_status = subprocess.run( ["git", "status", "--porcelain"], - cwd=temp_git_repo, capture_output=True, text=True + cwd=temp_git_repo, + capture_output=True, + text=True, ) assert git_status.returncode == 0 - assert not git_status.stdout.strip(), f"Expected clean status, got: {git_status.stdout}" + assert not git_status.stdout.strip(), ( + f"Expected clean status, got: {git_status.stdout}" + ) class TestChangeTracking: @@ -433,8 +708,7 @@ class TestChangeTracking: (info.path / "README.md").write_text("modified") subprocess.run(["git", "add", "."], cwd=info.path, capture_output=True) subprocess.run( - ["git", "commit", "-m", "Changes"], - cwd=info.path, capture_output=True + ["git", "commit", "-m", "Changes"], cwd=info.path, capture_output=True ) summary = manager.get_change_summary("test-spec") @@ -452,8 +726,7 @@ class TestChangeTracking: (info.path / "added.txt").write_text("new file") subprocess.run(["git", "add", "."], cwd=info.path, capture_output=True) subprocess.run( - ["git", "commit", "-m", "Add file"], - cwd=info.path, capture_output=True + ["git", "commit", "-m", "Add file"], cwd=info.path, capture_output=True ) files = manager.get_changed_files("test-spec")