Coverage for little_loops / worktree_utils.py: 0%

1"""Shared worktree setup and cleanup utilities. 

2 

3Used by ll-parallel, ll-sprint, and ll-loop to create and remove isolated git 

4worktrees with consistent file-copy behavior. 

5""" 

6 

7from __future__ import annotations 

8 

9import os 

10import shutil 

11import subprocess 

12from pathlib import Path 

13from typing import TYPE_CHECKING 

14 

15if TYPE_CHECKING: 

16 from little_loops.logger import Logger 

17 from little_loops.parallel.git_lock import GitLock 

18 

19 

20def setup_worktree( 

21 repo_path: Path, 

22 worktree_path: Path, 

23 branch_name: str, 

24 copy_files: list[str], 

25 logger: Logger, 

26 git_lock: GitLock, 

27) -> None: 

28 """Create a git worktree on a new branch and copy essential files. 

29 

30 Copies the .claude/ directory (for project root detection by Claude Code) 

31 and any additional files listed in copy_files. Writes a session marker so 

32 orphan-cleanup routines can identify this process's worktrees. 

33 

34 Args: 

35 repo_path: Path to the main repository. 

36 worktree_path: Destination path for the new worktree. 

37 branch_name: Name of the new branch to create. 

38 copy_files: File paths (relative to repo_path) to copy into the worktree. 

39 logger: Logger instance. 

40 git_lock: Thread-safe git lock for serializing repo operations. 

41 

42 Raises: 

43 RuntimeError: If git worktree creation fails. 

44 """ 

45 if worktree_path.exists(): 

46 cleanup_worktree(worktree_path, repo_path, logger, git_lock, delete_branch=True) 

47 

48 result = git_lock.run( 

49 ["worktree", "add", "-b", branch_name, str(worktree_path)], 

50 cwd=repo_path, 

51 timeout=60, 

52 ) 

53 if result.returncode != 0: 

54 raise RuntimeError(f"Failed to create worktree: {result.stderr}") 

55 

56 # Copy git identity so commits inside the worktree have the right author 

57 for config_key in ["user.email", "user.name"]: 

58 value_result = git_lock.run(["config", config_key], cwd=repo_path) 

59 if value_result.returncode == 0 and value_result.stdout.strip(): 

60 subprocess.run( 

61 ["git", "config", config_key, value_result.stdout.strip()], 

62 cwd=worktree_path, 

63 capture_output=True, 

64 ) 

65 

66 # Copy .claude/ to establish project root for Claude Code (BUG-007) 

67 claude_dir = repo_path / ".claude" 

68 if claude_dir.exists() and claude_dir.is_dir(): 

69 dest_claude_dir = worktree_path / ".claude" 

70 if dest_claude_dir.exists(): 

71 shutil.rmtree(dest_claude_dir) 

72 shutil.copytree(claude_dir, dest_claude_dir) 

73 logger.info("Copied .claude/ directory to worktree") 

74 

75 # Copy additional configured files 

76 for file_path in copy_files: 

77 if file_path.startswith(".claude/"): 

78 continue # already covered by the copytree above 

79 src = repo_path / file_path 

80 if src.exists(): 

81 if src.is_dir(): 

82 logger.warning( 

83 f"Skipping '{file_path}' in copy_files: " 

84 "is a directory (use symlinks or copytree for directories)" 

85 ) 

86 continue 

87 dest = worktree_path / file_path 

88 dest.parent.mkdir(parents=True, exist_ok=True) 

89 shutil.copy2(src, dest) 

90 logger.info(f"Copied {file_path} to worktree") 

91 else: 

92 logger.debug(f"Skipped {file_path} (not found in main repo)") 

93 

94 logger.info(f"Created worktree at {worktree_path} on branch {branch_name}") 

95 

96 # Write session marker for orphan cleanup (BUG-579) 

97 if worktree_path.exists(): 

98 marker_path = worktree_path / f".ll-session-{os.getpid()}" 

99 marker_path.write_text(str(os.getpid())) 

100 

101 

102def cleanup_worktree( 

103 worktree_path: Path, 

104 repo_path: Path, 

105 logger: Logger, 

106 git_lock: GitLock, 

107 delete_branch: bool = True, 

108) -> None: 

109 """Remove a git worktree and optionally its associated branch. 

110 

111 Args: 

112 worktree_path: Path to the worktree to remove. 

113 repo_path: Path to the main repository. 

114 logger: Logger instance. 

115 git_lock: Thread-safe git lock for serializing repo operations. 

116 delete_branch: If True, detect and delete the worktree's branch after removal. 

117 """ 

118 if not worktree_path.exists(): 

119 return 

120 

121 branch_name: str | None = None 

122 if delete_branch: 

123 branch_result = subprocess.run( 

124 ["git", "rev-parse", "--abbrev-ref", "HEAD"], 

125 cwd=worktree_path, 

126 capture_output=True, 

127 text=True, 

128 ) 

129 branch_name = branch_result.stdout.strip() if branch_result.returncode == 0 else None 

130 

131 git_lock.run( 

132 ["worktree", "remove", "--force", str(worktree_path)], 

133 cwd=repo_path, 

134 timeout=30, 

135 ) 

136 

137 if worktree_path.exists(): 

138 shutil.rmtree(worktree_path, ignore_errors=True) 

139 

140 if delete_branch and branch_name: 

141 git_lock.run(["branch", "-D", branch_name], cwd=repo_path, timeout=10) 

142 logger.info(f"Deleted branch {branch_name}")