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_merge_sync_provenance(
    worktree_path: str,
    *,
    prior_head_sha: str | None,
    synced_head_sha: str | None,
) -> dict:
    """Observe whether ``synced_head_sha`` is a sanctioned merge-based branch sync
    that advanced the PR branch past ``prior_head_sha`` (#871).

    ``gitea_update_pr_branch_by_merge`` advances a PR branch by merging the base
    branch *into* the branch (``POST /pulls/{n}/update?style=merge``). The result
    is a merge commit ``M`` on the branch whose **first** parent is the prior
    branch head and whose second parent is the base tip. When the owning session
    then dies without the durable lock's recorded head being refreshed, the local
    worktree still sits at ``prior_head_sha`` while the live PR head is ``M``.

    Recovering that drift safely requires proving the remote head is *exactly*
    such a merge-sync — not a rewrite, rebase, force-push, or an unrelated
    commit. This is that server-side observation. It 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 (#871).

    Provenance is proven only when ALL hold:

    * both commits are present (a rewritten/force-moved prior head leaves the
      object graph and fails closed);
    * ``prior_head_sha`` is a strict ancestor of ``synced_head_sha`` (the branch
      history is preserved, never replaced);
    * ``synced_head_sha`` is a merge commit (two or more parents), i.e. a base
      merged in — a plain fast-forward of new direct commits is not a sync;
    * ``prior_head_sha`` is an ancestor of the merge's **first** parent, so the
      branch mainline (first-parent lineage) still reaches the prior head — a
      rebase/force-push that re-authored the branch side fails this.
    """
    path = (worktree_path or "").strip()
    prior = (prior_head_sha or "").strip()
    synced = (synced_head_sha or "").strip()
    result: dict = {
        "prior_head_sha": prior or None,
        "synced_head_sha": synced or None,
        "probe_ok": False,
        "prior_present": False,
        "synced_present": False,
        "prior_is_ancestor": False,
        "synced_is_merge": False,
        "first_parent_reaches_prior": False,
        "is_merge_sync": False,
        "first_parent_sha": None,
        "parent_count": None,
        "proof": None,
        "reasons": [],
    }
    if not path or not prior or not synced:
        result["reasons"].append(
            "merge-sync provenance probe requires a worktree path and both "
            "commit SHAs"
        )
        return result
    if prior == synced:
        result["reasons"].append(
            "prior and synced heads are identical; no branch sync occurred"
        )
        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

    def _is_ancestor(ancestor: str, descendant: str) -> bool | None:
        res = subprocess.run(
            ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
            capture_output=True,
            text=True,
            check=False,
        )
        if res.returncode == 0:
            return True
        if res.returncode == 1:
            return False
        return None  # failed probe — never a silent "no"

    try:
        result["prior_present"] = _present(prior)
        result["synced_present"] = _present(synced)
    except OSError as exc:  # git unavailable — fail closed, never assume
        result["reasons"].append(f"merge-sync provenance probe could not run: {exc}")
        return result

    if not result["prior_present"]:
        result["reasons"].append(
            f"prior head {prior} is not reachable in '{path}'; history may have "
            "been rewritten or force-moved"
        )
    if not result["synced_present"]:
        result["reasons"].append(
            f"synced head {synced} is not reachable in '{path}'"
        )
    if not (result["prior_present"] and result["synced_present"]):
        return result

    ancestor = _is_ancestor(prior, synced)
    if ancestor is None:
        result["reasons"].append(
            "ancestry probe failed; merge-sync provenance unproven"
        )
        return result
    result["prior_is_ancestor"] = bool(ancestor)
    if not ancestor:
        result["reasons"].append(
            f"prior head {prior} is not an ancestor of synced head {synced}; "
            "the branch history was not preserved (not a merge-based sync)"
        )
        return result

    parents_res = subprocess.run(
        ["git", "-C", path, "rev-list", "--parents", "-n", "1", synced],
        capture_output=True,
        text=True,
        check=False,
    )
    if parents_res.returncode != 0:
        result["reasons"].append(
            f"could not read parents of {synced}; merge-sync provenance unproven"
        )
        return result
    tokens = (parents_res.stdout or "").split()
    # tokens[0] is the commit itself; the rest are its parents.
    parents = tokens[1:]
    result["parent_count"] = len(parents)
    result["synced_is_merge"] = len(parents) >= 2
    if not result["synced_is_merge"]:
        result["probe_ok"] = True
        result["reasons"].append(
            f"synced head {synced} has {len(parents)} parent(s); a merge-based "
            "branch sync produces a merge commit (two or more parents)"
        )
        return result
    first_parent = parents[0]
    result["first_parent_sha"] = first_parent

    fp_reaches = _is_ancestor(prior, first_parent) if prior != first_parent else True
    if fp_reaches is None:
        result["reasons"].append(
            "first-parent ancestry probe failed; merge-sync provenance unproven"
        )
        return result
    result["first_parent_reaches_prior"] = bool(fp_reaches)
    result["probe_ok"] = True
    if not fp_reaches:
        result["reasons"].append(
            f"merge first parent {first_parent} does not reach prior head "
            f"{prior}; the branch mainline was re-authored (not a sanctioned sync)"
        )
        return result

    result["is_merge_sync"] = True
    result["proof"] = (
        f"synced head {synced} is a merge commit (parents={len(parents)}) whose "
        f"first-parent lineage reaches prior head {prior}; base merged into branch"
    )
    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,
    renewal_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.

    ``renewal_sanctioned`` waives base-equivalence on exactly the same grounds
    for the other proven-ownership case (#760): ``issue_lock_renewal`` has shown
    that an *expired* lease is being renewed by its exact recorded owner — same
    remote, org, repo, issue, operation, branch, realpath-normalized worktree,
    claimant username and profile — with the local head matching the remote head
    and any owning PR head. Such a branch carries committed work for the same
    reason a recovered one does, so it can never be base-equivalent either.

    Both waivers relax this one requirement and nothing else. Neither is
    caller-supplied: each is computed server-side from durable lock state plus
    live observation. With both False every precondition applies exactly as
    before.
    """
    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 or renewal_sanctioned:
        # Base-equivalence intentionally not evaluated: ownership was proven
        # against the durable lock record instead — by dead-session recovery
        # (#753) or by exact-owner renewal of an expired lease (#760). Every
        # other precondition above and below still applies; cleanliness in
        # particular is checked before this branch and is never waived.
        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,
        renewal_sanctioned=renewal_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,
    renewal_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,
        "renewal_sanctioned": renewal_sanctioned,
        # Either proven-ownership waiver relaxes base-equivalence; the two are
        # reported separately so an audit can tell which one applied.
        "base_equivalence_waived": bool(recovery_sanctioned or renewal_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