| 223 | |
| 224 | # -- WorktreeManager: create/list/run/remove git worktrees + lifecycle index -- |
| 225 | class WorktreeManager: |
| 226 | def __init__(self, repo_root: Path, tasks: TaskManager, events: EventBus): |
| 227 | self.repo_root = repo_root |
| 228 | self.tasks = tasks |
| 229 | self.events = events |
| 230 | self.dir = repo_root / ".worktrees" |
| 231 | self.dir.mkdir(parents=True, exist_ok=True) |
| 232 | self.index_path = self.dir / "index.json" |
| 233 | if not self.index_path.exists(): |
| 234 | self.index_path.write_text(json.dumps({"worktrees": []}, indent=2)) |
| 235 | self.git_available = self._is_git_repo() |
| 236 | |
| 237 | def _is_git_repo(self) -> bool: |
| 238 | try: |
| 239 | r = subprocess.run( |
| 240 | ["git", "rev-parse", "--is-inside-work-tree"], |
| 241 | cwd=self.repo_root, |
| 242 | capture_output=True, |
| 243 | text=True, |
| 244 | timeout=10, |
| 245 | ) |
| 246 | return r.returncode == 0 |
| 247 | except Exception: |
| 248 | return False |
| 249 | |
| 250 | def _run_git(self, args: list[str]) -> str: |
| 251 | if not self.git_available: |
| 252 | raise RuntimeError("Not in a git repository. worktree tools require git.") |
| 253 | r = subprocess.run( |
| 254 | ["git", *args], |
| 255 | cwd=self.repo_root, |
| 256 | capture_output=True, |
| 257 | text=True, |
| 258 | timeout=120, |
| 259 | ) |
| 260 | if r.returncode != 0: |
| 261 | msg = (r.stdout + r.stderr).strip() |
| 262 | raise RuntimeError(msg or f"git {' '.join(args)} failed") |
| 263 | return (r.stdout + r.stderr).strip() or "(no output)" |
| 264 | |
| 265 | def _load_index(self) -> dict: |
| 266 | return json.loads(self.index_path.read_text()) |
| 267 | |
| 268 | def _save_index(self, data: dict): |
| 269 | self.index_path.write_text(json.dumps(data, indent=2)) |
| 270 | |
| 271 | def _find(self, name: str) -> dict | None: |
| 272 | idx = self._load_index() |
| 273 | for wt in idx.get("worktrees", []): |
| 274 | if wt.get("name") == name: |
| 275 | return wt |
| 276 | return None |
| 277 | |
| 278 | def _validate_name(self, name: str): |
| 279 | if not re.fullmatch(r"[A-Za-z0-9._-]{1,40}", name or ""): |
| 280 | raise ValueError( |
| 281 | "Invalid worktree name. Use 1-40 chars: letters, numbers, ., _, -" |
| 282 | ) |
no outgoing calls
no test coverage detected