Gitea-Tools/issue_lock_worktree.py

"""Issue-lock worktree validation (#249).

Author issue locks must validate the caller's own scratch clone (or declared
worktree path), not the shared MCP server working directory. A clean scratch at
``master``/``main`` must remain lockable while an unrelated session leaves the
shared dev worktree dirty or on a feature branch.
"""

from __future__ import annotations

import os
import subprocess

from reviewer_worktree import parse_dirty_tracked_files

AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
BASE_BRANCHES = frozenset({"master", "main"})


def resolve_author_worktree_path(
    explicit: str | None,
    project_root: str,
) -> str:
    """Resolve the author worktree path for lock/PR gates."""
    path = (explicit or "").strip()
    if not path:
        path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
    if not path:
        path = project_root
    return os.path.realpath(os.path.abspath(path))


def read_worktree_git_state(worktree_path: str) -> dict:
    """Read branch name and porcelain status from a git worktree."""
    path = (worktree_path or "").strip()
    if not path:
        return {"current_branch": None, "porcelain_status": ""}

    branch_res = subprocess.run(
        ["git", "-C", path, "branch", "--show-current"],
        capture_output=True,
        text=True,
        check=False,
    )
    current_branch = (branch_res.stdout or "").strip() or None

    status_res = subprocess.run(
        ["git", "-C", path, "status", "--porcelain"],
        capture_output=True,
        text=True,
        check=False,
    )
    return {
        "current_branch": current_branch,
        "porcelain_status": status_res.stdout or "",
    }


def assess_issue_lock_worktree(
    *,
    worktree_path: str,
    current_branch: str | None,
    porcelain_status: str,
    base_branches: frozenset[str] | None = None,
) -> dict:
    """Fail closed when lock preconditions are not met on the declared worktree."""
    bases = base_branches or BASE_BRANCHES
    reasons: list[str] = []
    path = (worktree_path or "").strip()
    if not path:
        reasons.append("worktree path not declared for issue lock; fail closed")
        return _assessment(False, reasons, path, None, [])

    branch = (current_branch or "").strip()
    dirty_files = parse_dirty_tracked_files(porcelain_status)

    if dirty_files:
        reasons.append(
            "tracked file edits exist before issue lock; "
            "lock must precede implementation work"
        )
    if not branch:
        reasons.append(
            "current branch unknown (detached HEAD?); issue lock must be taken "
            f"from base branch ({_base_list(bases)})"
        )
    elif branch not in bases:
        reasons.append(
            f"issue lock must be taken from base branch ({_base_list(bases)}), "
            f"not '{branch}'"
        )

    proven = not reasons
    return _assessment(proven, reasons, path, branch or None, dirty_files)


def format_issue_lock_worktree_error(assessment: dict) -> str:
    """Format a single fail-closed error for ``gitea_lock_issue``."""
    reasons = list(assessment.get("reasons") or [])
    if not reasons:
        reasons = ["issue lock worktree validation failed"]
    return "; ".join(reasons) + " (fail closed)"


def verify_pr_worktree_matches_lock(
    locked_worktree_path: str | None,
    declared_worktree_path: str | None,
    project_root: str,
) -> dict:
    """PR creation must use the same worktree the lock was validated against."""
    locked = (locked_worktree_path or "").strip()
    if not locked:
        return {"proven": True, "block": False, "reasons": []}

    declared = resolve_author_worktree_path(declared_worktree_path, project_root)
    locked_real = os.path.realpath(locked)
    declared_real = os.path.realpath(declared)
    if locked_real != declared_real:
        return {
            "proven": False,
            "block": True,
            "reasons": [
                f"PR worktree '{declared_real}' does not match locked worktree "
                f"'{locked_real}' (fail closed)"
            ],
            "locked_worktree_path": locked_real,
            "declared_worktree_path": declared_real,
        }
    return {
        "proven": True,
        "block": False,
        "reasons": [],
        "locked_worktree_path": locked_real,
        "declared_worktree_path": declared_real,
    }


def _base_list(bases: frozenset[str]) -> str:
    return "/".join(sorted(bases))


def _assessment(
    proven: bool,
    reasons: list[str],
    worktree_path: str,
    current_branch: str | None,
    dirty_files: list[str],
) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "worktree_path": worktree_path or None,
        "current_branch": current_branch,
        "dirty_files": dirty_files,
    }