Coverage for little_loops / worktree_utils.py: 16%

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 re 

11import shutil 

12import subprocess 

13from pathlib import Path 

14from typing import TYPE_CHECKING 

15 

16if TYPE_CHECKING: 

17 from little_loops.logger import Logger 

18 from little_loops.parallel.git_lock import GitLock 

19 

20 

21def setup_worktree( 

22 repo_path: Path, 

23 worktree_path: Path, 

24 branch_name: str, 

25 copy_files: list[str], 

26 logger: Logger, 

27 git_lock: GitLock, 

28 base_branch: str | None = None, 

29) -> None: 

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

31 

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

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

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

35 

36 Args: 

37 repo_path: Path to the main repository. 

38 worktree_path: Destination path for the new worktree. 

39 branch_name: Name of the new branch to create. 

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

41 logger: Logger instance. 

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

43 base_branch: Optional commit-ish to fork the new branch from. When None, 

44 the branch forks from the current HEAD of repo_path (existing behavior). 

45 When provided, validated before use; fails fast if unresolvable. 

46 

47 Raises: 

48 RuntimeError: If git worktree creation fails or base_branch does not resolve. 

49 """ 

50 if worktree_path.exists(): 

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

52 

53 if base_branch is not None: 

54 verify_result = git_lock.run( 

55 ["rev-parse", "--verify", base_branch], 

56 cwd=repo_path, 

57 timeout=10, 

58 ) 

59 if verify_result.returncode != 0: 

60 raise RuntimeError( 

61 f"Branch '{base_branch}' does not resolve: {verify_result.stderr}" 

62 ) 

63 

64 worktree_args = ["worktree", "add", "-b", branch_name, str(worktree_path)] 

65 if base_branch is not None: 

66 worktree_args.append(base_branch) 

67 

68 result = git_lock.run( 

69 worktree_args, 

70 cwd=repo_path, 

71 timeout=60, 

72 ) 

73 if result.returncode != 0: 

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

75 

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

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

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

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

80 subprocess.run( 

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

82 cwd=worktree_path, 

83 capture_output=True, 

84 ) 

85 

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

87 claude_dir = repo_path / ".claude" 

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

89 dest_claude_dir = worktree_path / ".claude" 

90 if dest_claude_dir.exists(): 

91 shutil.rmtree(dest_claude_dir) 

92 shutil.copytree(claude_dir, dest_claude_dir) 

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

94 

95 # Copy additional configured files 

96 for file_path in copy_files: 

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

98 continue # already covered by the copytree above 

99 src = repo_path / file_path 

100 if src.exists(): 

101 if src.is_dir(): 

102 logger.warning( 

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

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

105 ) 

106 continue 

107 dest = worktree_path / file_path 

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

109 shutil.copy2(src, dest) 

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

111 else: 

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

113 

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

115 

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

117 if worktree_path.exists(): 

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

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

120 

121 

122def cleanup_worktree( 

123 worktree_path: Path, 

124 repo_path: Path, 

125 logger: Logger, 

126 git_lock: GitLock, 

127 delete_branch: bool = True, 

128) -> None: 

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

130 

131 Args: 

132 worktree_path: Path to the worktree to remove. 

133 repo_path: Path to the main repository. 

134 logger: Logger instance. 

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

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

137 """ 

138 if not worktree_path.exists(): 

139 return 

140 

141 branch_name: str | None = None 

142 if delete_branch: 

143 branch_result = subprocess.run( 

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

145 cwd=worktree_path, 

146 capture_output=True, 

147 text=True, 

148 ) 

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

150 

151 git_lock.run(["worktree", "unlock", str(worktree_path)], cwd=repo_path, timeout=10) 

152 git_lock.run( 

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

154 cwd=repo_path, 

155 timeout=30, 

156 ) 

157 

158 if worktree_path.exists(): 

159 shutil.rmtree(worktree_path, ignore_errors=True) 

160 

161 if delete_branch and branch_name: 

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

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

164 

165 

166def _is_ll_worktree(name: str) -> bool: 

167 """Return True if the directory name matches an ll-managed worktree naming pattern. 

168 

169 Matches both ll-parallel worker dirs (``worker-<issue>-<timestamp>``) and 

170 ll-loop worktree dirs (``<YYYYMMDD>-<HHMMSS>-<safe-name>``). 

171 """ 

172 return name.startswith("worker-") or re.match(r"^\d{8}-\d{6}-", name) is not None