electron-rare/Aperant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto-claude: 162-fix-worktree-error-on-repeated-task-starts (#1453)

Browse Source 
* auto-claude: subtask-1-1 - Add _branch_exists() helper method to check if a branch exists

* auto-claude: subtask-1-2 - Add _worktree_is_registered() helper to check if worktree is tracked by git

This helper method uses 'git worktree list --porcelain' to determine if a
worktree path is registered with git. This is useful for detecting orphaned
worktree directories that need cleanup during idempotent worktree creation.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* auto-claude: subtask-1-3 - Refactor create_worktree() to be idempotent

- Run git worktree prune first to clean orphaned references
- Check if worktree already exists and is valid (return existing)
- Handle stale worktree directories (cleanup before recreation)
- Reuse existing branches without -b flag when branch exists
- Only use -b flag when creating new branch

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* auto-claude: subtask-1-4 - Add test_create_worktree_idempotent test case

* auto-claude: subtask-1-5 - Add test_create_worktree_branch_exists_no_worktree

Add test case that verifies create_worktree() correctly reuses an existing
branch when the worktree directory is missing. The test:
1. Creates a worktree to establish the branch
2. Removes the worktree but keeps the branch (delete_branch=False)
3. Verifies the branch still exists
4. Calls create_worktree() again - should succeed by reusing the branch
5. Verifies the worktree is recreated with the same branch name

* auto-claude: subtask-1-6 - Add test_create_worktree_stale_directory test case

Add test that verifies create_worktree() correctly handles the stale
directory scenario where a worktree directory exists on disk but is
not registered with git. The test:
1. Creates a worktree normally
2. Force-removes git tracking but recreates the directory
3. Verifies create_worktree() cleans up stale directory and recreates

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* style: Fix ruff formatting in worktree.py

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Handle edge cases in idempotent worktree creation

- Handle corrupted worktrees (registered but unreadable) by force removing
  and recreating them (NEW-001)
- Add thread-safety documentation to create_worktree docstring (NEW-002)
- Add defensive check for malformed porcelain output parsing (NEW-003)
- Use os.path.samefile() for accurate path comparison on case-insensitive
  filesystems like macOS HFS+/APFS and Windows NTFS (NEW-004)
- Add test for stale directory with existing branch scenario (NEW-005)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Add error handling for worktree cleanup operations

- NCR-001: Check if stale directory still exists after rmtree and raise
  WorktreeError with clear message about permission issues or file locks
- NCR-002: Check return code of corrupted worktree removal and raise
  WorktreeError if force remove fails
- NCR-003: Use git show-ref --verify refs/heads/{branch} instead of
  git rev-parse to specifically check for local branches, avoiding
  false positives from tags or other refs with the same name

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Andy
2026-01-25 11:31:40 +01:00
committed by GitHub
parent 31f116db52
commit b955badf7f
2 changed files with 467 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/core/worktree.py
+135 -26
View File
@@ -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(
tests/test_worktree.py
+332 -59
View File
@@ -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")