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,
    *,
    session_lock_worktree: str | None = None,
) -> str:
    """Resolve the author worktree path for lock/PR gates.

    #618: prefer explicit path, then env, then the active issue lock worktree.
    Does not invent a branches/ worktree. Falling back to *project_root* is
    retained only for lock-time bootstrap when the process itself is already
    under branches/ or no binding exists yet (callers still fail closed via
    preflight / durable resolution before mutation).
    """
    path = (explicit or "").strip()
    if not path:
        path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
    if not path:
        path = (os.environ.get("GITEA_ACTIVE_WORKTREE") or "").strip()
    if not path:
        path = (session_lock_worktree or "").strip()
    if not path:
        path = project_root
    return os.path.realpath(os.path.abspath(path))


def read_head_ancestry(
    worktree_path: str,
    *,
    ancestor_sha: str | None,
    descendant_sha: str | None,
) -> dict:
    """Observe whether ``descendant_sha`` strictly descends from ``ancestor_sha`` (#768).

    Server-side git observation for dead-session lock recovery. The recovering
    author's only reachable clean-worktree state is one commit *ahead* of the
    head recorded at lock time, so recovery needs to know whether that commit
    extends the recorded head or replaces it.

    Reports facts only; the disposition lives in ``issue_lock_recovery``. Every
    field is read from git in the declared worktree — nothing here is supplied
    by, or reachable from, an MCP caller (#768 AC6).

    ``ancestor_present`` proves the recorded head is still reachable, which is
    what separates an honest fast-forward from a rewritten or force-moved
    history: a rewritten recorded head leaves the object graph and the probe
    fails closed.
    """
    path = (worktree_path or "").strip()
    ancestor = (ancestor_sha or "").strip()
    descendant = (descendant_sha or "").strip()
    result: dict = {
        "ancestor_sha": ancestor or None,
        "descendant_sha": descendant or None,
        "probe_ok": False,
        "ancestor_present": False,
        "descendant_present": False,
        "is_ancestor": False,
        "is_strict_descendant": False,
        "proof": None,
        "reasons": [],
    }
    if not path or not ancestor or not descendant:
        result["reasons"].append(
            "ancestry probe requires a worktree path and both commit SHAs"
        )
        return result

    def _present(sha: str) -> bool:
        res = subprocess.run(
            ["git", "-C", path, "rev-parse", "--verify", "--quiet", f"{sha}^{{commit}}"],
            capture_output=True,
            text=True,
            check=False,
        )
        return res.returncode == 0

    try:
        result["ancestor_present"] = _present(ancestor)
        result["descendant_present"] = _present(descendant)
    except OSError as exc:  # git unavailable — fail closed, never assume
        result["reasons"].append(f"ancestry probe could not run: {exc}")
        return result

    if not result["ancestor_present"]:
        result["reasons"].append(
            f"recorded head {ancestor} is not reachable in '{path}'; history may "
            "have been rewritten or force-moved"
        )
    if not result["descendant_present"]:
        result["reasons"].append(
            f"local head {descendant} is not reachable in '{path}'"
        )
    if not (result["ancestor_present"] and result["descendant_present"]):
        return result

    probe = subprocess.run(
        ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
        capture_output=True,
        text=True,
        check=False,
    )
    # 0 = is an ancestor, 1 = is not. Anything else is a failed probe, not a "no".
    if probe.returncode not in (0, 1):
        result["reasons"].append(
            f"ancestry probe failed with exit {probe.returncode}; ancestry unproven"
        )
        return result

    result["probe_ok"] = True
    result["is_ancestor"] = probe.returncode == 0
    result["is_strict_descendant"] = result["is_ancestor"] and ancestor != descendant
    result["proof"] = (
        f"git -C <worktree> merge-base --is-ancestor {ancestor} {descendant} "
        f"-> exit {probe.returncode}"
    )
    if not result["is_ancestor"]:
        result["reasons"].append(
            f"local head {descendant} does not descend from recorded head {ancestor}"
        )
    elif not result["is_strict_descendant"]:
        result["reasons"].append(
            f"local head {descendant} equals the recorded head; no descendant "
            "recovery is involved"
        )
    return result


def read_recorded_base(
    worktree_path: str,
    *,
    head_sha: str | None,
    extra_bases: tuple[str, ...] | list[str] = (),
    base_branches: frozenset[str] | None = None,
) -> dict:
    """Observe the base commit an unpublished claim was branched from (#772).

    A published claim records its base implicitly: the remote branch head is the
    thing recovery measures against. An unpublished claim has no remote ref, so
    the base must be observed here, server-side, as the merge-base between the
    worktree HEAD and the base branch it was cut from.

    Reports facts only; the disposition lives in ``issue_lock_recovery``. Every
    field is read from git in the declared worktree — nothing is supplied by, or
    reachable from, an MCP caller, so a caller cannot nominate a base that would
    make unrelated history look like a descendant (#772 AC1/AC4).

    A HEAD with no common ancestor in any base branch yields ``probe_ok`` with no
    ``base_sha``: unrelated history is reported as exactly that, never as a base.
    """
    path = (worktree_path or "").strip()
    head = (head_sha or "").strip()
    bases = base_branches or BASE_BRANCHES
    candidates = [*extra_bases, *sorted(bases)]
    result: dict = {
        "base_branch": None,
        "base_sha": None,
        "head_sha": head or None,
        "probe_ok": False,
        "candidates": candidates,
        "reasons": [],
    }
    if not path or not head:
        result["reasons"].append(
            "recorded-base probe requires a worktree path and a HEAD sha"
        )
        return result

    probed_any = False
    for candidate in candidates:
        name = (candidate or "").strip()
        if not name:
            continue
        probe = subprocess.run(
            ["git", "-C", path, "merge-base", name, head],
            capture_output=True,
            text=True,
            check=False,
        )
        if probe.returncode not in (0, 1):
            # 0 = merge base found, 1 = no common ancestor. Anything else is a
            # failed probe (missing ref, broken repo) — try the next candidate.
            continue
        probed_any = True
        merge_base = (probe.stdout or "").strip()
        if probe.returncode == 0 and merge_base:
            result["base_branch"] = name
            result["base_sha"] = merge_base
            result["probe_ok"] = True
            return result

    result["probe_ok"] = probed_any
    if probed_any:
        result["reasons"].append(
            f"HEAD {head} shares no common ancestor with any of "
            f"{_base_list(bases)}; history is unrelated to this repository's base"
        )
    else:
        result["reasons"].append(
            f"recorded-base probe could not run against any of {_base_list(bases)} "
            f"in '{path}'"
        )
    return result


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)
    return {
        "current_branch": current_branch,
        "porcelain_status": status_res.stdout or "",
        "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,
    recovery_sanctioned: bool = False,
) -> dict:
    """Fail closed when lock preconditions are not met on the declared worktree.

    ``recovery_sanctioned`` is set only when ``issue_lock_recovery`` has already
    proven, from the durable lock itself, that this is a dead-session recovery of
    an existing claim (#753): same issue, branch, worktree, author, and head, with
    the recording process dead. In that one case the base-equivalence requirement
    is waived, because a branch that already carries the work is ahead of its base
    by construction and could never satisfy it. Every other precondition —
    notably worktree cleanliness — still applies unchanged, and brand-new issue
    claims keep the full base-equivalence requirement.
    """
    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 recovery_sanctioned:
        # Base-equivalence intentionally not evaluated: ownership was proven
        # against the durable lock record instead (#753).
        pass
    elif 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,
        recovery_sanctioned=recovery_sanctioned,
    )


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,
    recovery_sanctioned: bool = False,
) -> 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,
        "recovery_sanctioned": recovery_sanctioned,
        "base_equivalence_waived": bool(recovery_sanctioned),
    }


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