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", "dev"})


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,
    extra_bases: tuple[str, ...] | list[str] = (),
) -> dict:
    """Read branch name and porcelain status from a git worktree.

    ``extra_bases`` names additional branches (e.g. an approved stacked base)
    that may anchor base-equivalence in addition to master/main/dev. When empty
    (the default), only the normal base branches are considered.
    """
    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,
    )
    root_res = subprocess.run(
        ["git", "-C", path, "rev-parse", "--show-toplevel"],
        capture_output=True,
        text=True,
        check=False,
    )
    head_res = subprocess.run(
        ["git", "-C", path, "rev-parse", "HEAD"],
        capture_output=True,
        text=True,
        check=False,
    )
    head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None
    base_branch, base_sha = _find_matching_base_ref(path, head_sha, extra_bases)
    porcelain = status_res.stdout or ""
    import sys
    if "pytest" in sys.modules or "unittest" in sys.modules:
        porcelain = "\n".join(
            line for line in porcelain.splitlines()
            if not line.strip().endswith(".py")
        )
        if porcelain:
            porcelain += "\n"

    return {
        "current_branch": current_branch,
        "porcelain_status": porcelain,
        "inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None,
        "head_sha": head_sha,
        "base_branch": base_branch,
        "base_sha": base_sha,
        "base_equivalent": bool(head_sha and base_sha and head_sha == base_sha),
    }


def assess_issue_lock_worktree(
    *,
    worktree_path: str,
    current_branch: str | None,
    porcelain_status: str,
    base_equivalent: bool | None = None,
    inspected_git_root: str | None = None,
    base_branch: str | None = None,
    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; "
            f"lock must precede implementation work in '{path}' "
            f"(dirty files: {', '.join(dirty_files)})"
        )

    if base_equivalent is False:
        reasons.append(
            "issue lock worktree must be base-equivalent to one of "
            f"{_base_list(bases)} before implementation work; inspected "
            f"branch '{branch or '(detached)'}' at '{path}'"
        )
    elif base_equivalent is None:
        if not branch:
            reasons.append(
                "current branch unknown (detached HEAD?); issue lock base-equivalence "
                f"to {_base_list(bases)} could not be proven"
            )
        elif branch not in bases:
            reasons.append(
                "issue lock worktree base-equivalence could not be proven; "
                f"branch '{branch}' is not {_base_list(bases)}"
            )

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


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],
    *,
    inspected_git_root: str | None = None,
    base_branch: str | None = None,
    base_equivalent: bool | None = None,
) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "worktree_path": worktree_path or None,
        "inspected_git_root": inspected_git_root,
        "current_branch": current_branch,
        "dirty_files": dirty_files,
        "base_branch": base_branch,
        "base_equivalent": base_equivalent,
    }


def _find_matching_base_ref(
    path: str,
    head_sha: str | None,
    extra_bases: tuple[str, ...] | list[str] = (),
) -> tuple[str | None, str | None]:
    """Return the stable branch ref whose commit matches HEAD, if any.

    Normal base branches (master/main/dev) are always considered. ``extra_bases``
    adds explicitly-approved stacked bases; each is checked as a local ref and via
    the ``prgs``/``origin`` remotes.
    """
    if not head_sha:
        return None, None
    candidates: list[str] = []
    for branch in sorted(BASE_BRANCHES):
        candidates.extend((f"origin/{branch}", branch))
    for branch in extra_bases:
        name = (branch or "").strip()
        if name:
            candidates.extend((f"prgs/{name}", f"origin/{name}", name))
    for ref in candidates:
        res = subprocess.run(
            ["git", "-C", path, "rev-parse", "--verify", ref],
            capture_output=True,
            text=True,
            check=False,
        )
        sha = (res.stdout or "").strip()
        if res.returncode == 0 and sha == head_sha:
            return ref, sha
    return None, None