65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from django.utils.text import slugify
|
|
|
|
from control_plane.projects.models import Task, Worktree
|
|
|
|
|
|
class WorktreeManager:
|
|
def create_for_task(self, task: Task, repository_path: Path, base_ref: str = "main") -> Worktree:
|
|
repository_path = repository_path.resolve()
|
|
self.validate_clean_repository(repository_path)
|
|
branch_name = f"artifex/task-{task.id}"
|
|
worktree_path = repository_path.parent / f"{repository_path.name}-{slugify(str(task.id))}"
|
|
subprocess.run(
|
|
["git", "worktree", "add", "-b", branch_name, str(worktree_path), base_ref],
|
|
cwd=repository_path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return Worktree.objects.create(
|
|
task=task,
|
|
repository_path=str(repository_path),
|
|
worktree_path=str(worktree_path),
|
|
branch_name=branch_name,
|
|
base_ref=base_ref,
|
|
)
|
|
|
|
def validate_clean_repository(self, repository_path: Path) -> None:
|
|
status = subprocess.run(
|
|
["git", "status", "--short"],
|
|
cwd=repository_path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if status.stdout.strip():
|
|
raise RuntimeError("Repository must be clean before creating an autonomous worktree")
|
|
|
|
def validate_clean_worktree(self, worktree: Worktree) -> None:
|
|
status = subprocess.run(
|
|
["git", "status", "--short"],
|
|
cwd=worktree.worktree_path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if status.stdout.strip():
|
|
raise RuntimeError("Worktree has uncommitted changes")
|
|
|
|
def cleanup(self, worktree: Worktree) -> None:
|
|
path = Path(worktree.worktree_path)
|
|
if path.exists():
|
|
subprocess.run(
|
|
["git", "worktree", "remove", str(path)],
|
|
cwd=worktree.repository_path,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
worktree.status = "CLEANED"
|
|
worktree.save(update_fields=["status", "updated_at"])
|