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(f"Branch '{base_branch}' does not resolve: {verify_result.stderr}") 

61 

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

63 if base_branch is not None: 

64 worktree_args.append(base_branch) 

65 

66 result = git_lock.run( 

67 worktree_args, 

68 cwd=repo_path, 

69 timeout=60, 

70 ) 

71 if result.returncode != 0: 

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

73 

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

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

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

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

78 subprocess.run( 

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

80 cwd=worktree_path, 

81 capture_output=True, 

82 ) 

83 

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

85 claude_dir = repo_path / ".claude" 

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

87 dest_claude_dir = worktree_path / ".claude" 

88 if dest_claude_dir.exists(): 

89 shutil.rmtree(dest_claude_dir) 

90 shutil.copytree(claude_dir, dest_claude_dir) 

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

92 

93 # Copy additional configured files 

94 for file_path in copy_files: 

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

96 continue # already covered by the copytree above 

97 src = repo_path / file_path 

98 if src.exists(): 

99 if src.is_dir(): 

100 logger.warning( 

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

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

103 ) 

104 continue 

105 dest = worktree_path / file_path 

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

107 shutil.copy2(src, dest) 

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

109 else: 

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

111 

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

113 

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

115 if worktree_path.exists(): 

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

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

118 

119 

120def cleanup_worktree( 

121 worktree_path: Path, 

122 repo_path: Path, 

123 logger: Logger, 

124 git_lock: GitLock, 

125 delete_branch: bool = True, 

126) -> None: 

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

128 

129 Args: 

130 worktree_path: Path to the worktree to remove. 

131 repo_path: Path to the main repository. 

132 logger: Logger instance. 

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

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

135 """ 

136 if not worktree_path.exists(): 

137 return 

138 

139 branch_name: str | None = None 

140 if delete_branch: 

141 branch_result = subprocess.run( 

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

143 cwd=worktree_path, 

144 capture_output=True, 

145 text=True, 

146 ) 

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

148 

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

150 git_lock.run( 

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

152 cwd=repo_path, 

153 timeout=30, 

154 ) 

155 

156 if worktree_path.exists(): 

157 shutil.rmtree(worktree_path, ignore_errors=True) 

158 

159 if delete_branch and branch_name: 

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

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

162 

163 

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

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

166 

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

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

169 """ 

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