kin/core/worktree.py

"""
Kin — Git worktree management for isolated agent execution.

Each eligible agent step gets its own worktree in {project_path}/.kin_worktrees/
to prevent file-write conflicts between parallel or sequential agents.

All functions are defensive: never raise, always log warnings on error.
"""

import logging
import shutil
import subprocess
import time
from pathlib import Path

_logger = logging.getLogger("kin.worktree")


def _git(project_path: str) -> str:
    """Resolve git executable, preferring extended PATH."""
    try:
        from agents.runner import _build_claude_env
        env = _build_claude_env()
        found = shutil.which("git", path=env["PATH"])
        return found or "git"
    except Exception:
        return shutil.which("git") or "git"


def create_worktree(project_path: str, task_id: str, step_name: str = "step") -> str | None:
    """Create a git worktree for isolated agent execution.

    Creates: {project_path}/.kin_worktrees/{task_id}-{step_name}
    Branch name equals the worktree directory name.

    Returns the absolute worktree path, or None on any failure.
    """
    git = _git(project_path)
    safe_step = step_name.replace("/", "_").replace(" ", "_")
    branch_name = f"{task_id}-{safe_step}"
    worktrees_dir = Path(project_path) / ".kin_worktrees"
    worktree_path = worktrees_dir / branch_name

    try:
        worktrees_dir.mkdir(exist_ok=True)
        r = subprocess.run(
            [git, "worktree", "add", "-b", branch_name, str(worktree_path), "HEAD"],
            cwd=project_path,
            capture_output=True,
            text=True,
            timeout=30,
        )
        if r.returncode != 0:
            _logger.warning("git worktree add failed for %s: %s", branch_name, r.stderr.strip())
            return None
        _logger.info("Created worktree: %s", worktree_path)
        return str(worktree_path)
    except Exception as exc:
        _logger.warning("create_worktree error for %s: %s", branch_name, exc)
        return None


def merge_worktree(worktree_path: str, project_path: str, max_retries: int = 0, retry_delay_s: int = 15) -> dict:
    """Merge the worktree branch back into current HEAD of project_path.

    Branch name is derived from the worktree directory name.
    On conflict: aborts merge and returns success=False with conflict list.

    max_retries: number of retry attempts after the first failure (default 0 = no retry).
    retry_delay_s: seconds to wait between retry attempts.

    Returns {success: bool, conflicts: list[str], merged_files: list[str]}
    """
    git = _git(project_path)
    branch_name = Path(worktree_path).name

    try:
        # Stage all changes in the worktree before merge
        subprocess.run(
            [git, "-C", worktree_path, "add", "-A"],
            capture_output=True,
            text=True,
            timeout=30,
        )

        # Commit staged changes; ignore failure (nothing to commit = returncode 1)
        commit_result = subprocess.run(
            [git, "-C", worktree_path, "commit", "-m", f"kin: {branch_name}"],
            capture_output=True,
            text=True,
            timeout=30,
        )
        commit_had_changes = commit_result.returncode == 0

        for attempt in range(max_retries + 1):
            merge_result = subprocess.run(
                [git, "merge", "--no-ff", branch_name],
                cwd=project_path,
                capture_output=True,
                text=True,
                timeout=60,
            )
            if merge_result.returncode == 0:
                diff_result = subprocess.run(
                    [git, "diff", "HEAD~1", "HEAD", "--name-only"],
                    cwd=project_path,
                    capture_output=True,
                    text=True,
                    timeout=10,
                )
                merged_files = [
                    f.strip() for f in diff_result.stdout.splitlines() if f.strip()
                ]
                _logger.info("Merged worktree %s: %d files", branch_name, len(merged_files))
                return {"success": True, "conflicts": [], "merged_files": merged_files}

            # Merge failed — if commit was also empty (nothing to commit), treat as success
            if not commit_had_changes:
                _logger.info("Worktree %s: nothing to commit, skipping merge", branch_name)
                return {"success": True, "conflicts": [], "merged_files": []}

            # Merge failed — collect conflicts and abort
            conflict_result = subprocess.run(
                [git, "diff", "--name-only", "--diff-filter=U"],
                cwd=project_path,
                capture_output=True,
                text=True,
                timeout=10,
            )
            conflicts = [f.strip() for f in conflict_result.stdout.splitlines() if f.strip()]

            subprocess.run(
                [git, "merge", "--abort"],
                cwd=project_path,
                capture_output=True,
                timeout=10,
            )

            if attempt < max_retries:
                _logger.warning(
                    "KIN-139: merge conflict in %s (attempt %d/%d), retrying in %ds",
                    branch_name, attempt + 1, max_retries + 1, retry_delay_s,
                )
                time.sleep(retry_delay_s)
                continue

            _logger.warning("Merge conflict in worktree %s: %s", branch_name, conflicts)
            return {"success": False, "conflicts": conflicts, "merged_files": []}

    except Exception as exc:
        _logger.warning("merge_worktree error for %s: %s", branch_name, exc)
        return {"success": False, "conflicts": [], "merged_files": [], "error": str(exc)}


def cleanup_worktree(worktree_path: str, project_path: str) -> None:
    """Remove the git worktree and its branch. Never raises."""
    git = _git(project_path)
    branch_name = Path(worktree_path).name

    try:
        subprocess.run(
            [git, "-C", project_path, "worktree", "remove", "--force", worktree_path],
            capture_output=True,
            timeout=30,
        )
        subprocess.run(
            [git, "-C", project_path, "branch", "-D", branch_name],
            capture_output=True,
            timeout=10,
        )
        _logger.info("Cleaned up worktree: %s", worktree_path)
    except Exception as exc:
        _logger.warning("cleanup_worktree error for %s: %s", branch_name, exc)


def ensure_gitignore(project_path: str) -> None:
    """Ensure .kin_worktrees/ is in project's .gitignore. Never raises."""
    entry = ".kin_worktrees/"
    gitignore = Path(project_path) / ".gitignore"
    try:
        if gitignore.exists():
            content = gitignore.read_text()
            if entry not in content:
                with gitignore.open("a") as f:
                    f.write(f"\n{entry}\n")
        else:
            gitignore.write_text(f"{entry}\n")
    except Exception as exc:
        _logger.warning("ensure_gitignore error: %s", exc)