Coverage for little_loops / worktree_utils.py: 98%

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) -> None: 

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

30 

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

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

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

34 

35 Args: 

36 repo_path: Path to the main repository. 

37 worktree_path: Destination path for the new worktree. 

38 branch_name: Name of the new branch to create. 

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

40 logger: Logger instance. 

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

42 

43 Raises: 

44 RuntimeError: If git worktree creation fails. 

45 """ 

46 if worktree_path.exists(): 

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

48 

49 result = git_lock.run( 

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

51 cwd=repo_path, 

52 timeout=60, 

53 ) 

54 if result.returncode != 0: 

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

56 

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

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

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

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

61 subprocess.run( 

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

63 cwd=worktree_path, 

64 capture_output=True, 

65 ) 

66 

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

68 claude_dir = repo_path / ".claude" 

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

70 dest_claude_dir = worktree_path / ".claude" 

71 if dest_claude_dir.exists(): 

72 shutil.rmtree(dest_claude_dir) 

73 shutil.copytree(claude_dir, dest_claude_dir) 

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

75 

76 # Copy additional configured files 

77 for file_path in copy_files: 

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

79 continue # already covered by the copytree above 

80 src = repo_path / file_path 

81 if src.exists(): 

82 if src.is_dir(): 

83 logger.warning( 

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

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

86 ) 

87 continue 

88 dest = worktree_path / file_path 

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

90 shutil.copy2(src, dest) 

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

92 else: 

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

94 

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

96 

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

98 if worktree_path.exists(): 

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

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

101 

102 

103def cleanup_worktree( 

104 worktree_path: Path, 

105 repo_path: Path, 

106 logger: Logger, 

107 git_lock: GitLock, 

108 delete_branch: bool = True, 

109) -> None: 

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

111 

112 Args: 

113 worktree_path: Path to the worktree to remove. 

114 repo_path: Path to the main repository. 

115 logger: Logger instance. 

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

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

118 """ 

119 if not worktree_path.exists(): 

120 return 

121 

122 branch_name: str | None = None 

123 if delete_branch: 

124 branch_result = subprocess.run( 

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

126 cwd=worktree_path, 

127 capture_output=True, 

128 text=True, 

129 ) 

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

131 

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

133 git_lock.run( 

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

135 cwd=repo_path, 

136 timeout=30, 

137 ) 

138 

139 if worktree_path.exists(): 

140 shutil.rmtree(worktree_path, ignore_errors=True) 

141 

142 if delete_branch and branch_name: 

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

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

145 

146 

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

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

149 

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

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

152 """ 

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