Gitea-Tools/already_landed_reconcile.py

"""Already-landed open PR reconciliation gates (#310).

Reconciler workflows may close an open PR only when the PR head SHA is proven
an ancestor of a freshly fetched target branch. Arbitrary PR closure is denied.
"""

from __future__ import annotations

import subprocess
from typing import Any

from merged_cleanup_reconcile import extract_linked_issue, is_head_ancestor_of_ref

ELIGIBILITY_ALREADY_LANDED = "ALREADY_LANDED_RECONCILE_REQUIRED"
ELIGIBILITY_NOT_LANDED = "NOT_ALREADY_LANDED"
ELIGIBILITY_STALE_TARGET = "TARGET_BRANCH_UNVERIFIED"
ELIGIBILITY_PR_NOT_OPEN = "PR_NOT_OPEN"


def fetch_target_branch(project_root: str, remote: str, branch: str) -> dict[str, Any]:
    """Fetch *branch* from *remote* and return the resolved SHA."""
    ref = f"{remote}/{branch}"
    fetch = subprocess.run(
        ["git", "-C", project_root, "fetch", remote, branch],
        capture_output=True,
        text=True,
        check=False,
    )
    if fetch.returncode != 0:
        return {
            "success": False,
            "target_branch": branch,
            "target_ref": ref,
            "target_branch_sha": None,
            "reasons": [
                f"git fetch {remote} {branch} failed: "
                f"{(fetch.stderr or fetch.stdout or '').strip()}"
            ],
        }

    rev = subprocess.run(
        ["git", "-C", project_root, "rev-parse", ref],
        capture_output=True,
        text=True,
        check=False,
    )
    if rev.returncode != 0:
        return {
            "success": False,
            "target_branch": branch,
            "target_ref": ref,
            "target_branch_sha": None,
            "reasons": [
                f"git rev-parse {ref} failed: "
                f"{(rev.stderr or rev.stdout or '').strip()}"
            ],
        }

    return {
        "success": True,
        "target_branch": branch,
        "target_ref": ref,
        "target_branch_sha": (rev.stdout or "").strip(),
        "reasons": [],
        "git_fetch_command": f"git fetch {remote} {branch}",
    }


def assess_already_landed_reconciliation(
    *,
    pr: dict[str, Any],
    project_root: str,
    remote: str,
    target_branch: str,
    target_fetch: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return eligibility and proof for reconciling an open PR."""
    pr_number = int(pr.get("number") or 0)
    pr_state = (pr.get("state") or "").strip().lower()
    head_sha = (pr.get("head") or {}).get("sha") if isinstance(pr.get("head"), dict) else None
    if not head_sha and isinstance(pr.get("head"), str):
        head_sha = None
    head_ref = (pr.get("head") or {}).get("ref") if isinstance(pr.get("head"), dict) else pr.get("head")
    base_ref = (pr.get("base") or {}).get("ref") if isinstance(pr.get("base"), dict) else pr.get("base")
    title = pr.get("title") or ""
    body = pr.get("body") or ""

    fetch_result = target_fetch or fetch_target_branch(project_root, remote, target_branch)
    linked_issue = extract_linked_issue(title, body)

    result: dict[str, Any] = {
        "pr_number": pr_number,
        "pr_state": pr_state,
        "candidate_head_sha": head_sha,
        "head_ref": head_ref,
        "base_ref": base_ref or target_branch,
        "target_branch": target_branch,
        "target_branch_sha": fetch_result.get("target_branch_sha"),
        "linked_issue": linked_issue,
        "git_ref_mutations": [],
        "reasons": [],
    }
    if fetch_result.get("git_fetch_command"):
        result["git_ref_mutations"].append(fetch_result["git_fetch_command"])

    if pr_state != "open":
        result["eligibility_class"] = ELIGIBILITY_PR_NOT_OPEN
        result["ancestor_proof"] = None
        result["close_allowed"] = False
        result["reasons"].append(f"PR #{pr_number} state is {pr_state!r}, not open")
        return result

    if not fetch_result.get("success"):
        result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
        result["ancestor_proof"] = None
        result["close_allowed"] = False
        result["reasons"].extend(fetch_result.get("reasons") or [])
        return result

    target_ref = fetch_result.get("target_ref") or f"{remote}/{target_branch}"
    ancestor = is_head_ancestor_of_ref(project_root, head_sha, target_ref)
    result["ancestor_proof"] = ancestor

    if ancestor is None:
        result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
        result["close_allowed"] = False
        result["reasons"].append(
            f"ancestor check failed for head {head_sha!r} against {target_ref}"
        )
        return result

    if ancestor:
        result["eligibility_class"] = ELIGIBILITY_ALREADY_LANDED
        result["close_allowed"] = True
        return result

    result["eligibility_class"] = ELIGIBILITY_NOT_LANDED
    result["close_allowed"] = False
    result["reasons"].append(
        f"PR head {head_sha} is not an ancestor of {target_ref}"
    )
    return result