Gitea-Tools/pr_queue_cleanup.py

"""PR-only queue cleanup mode gates and report verifier (#390).

Cleanup mode dispatches exactly one canonical review run per PR. It forbids
author-side mutations (issue claiming, branch creation, implementation edits,
issue filing) and enforces the terminal-mutation chain: stop after
``REQUEST_CHANGES``; after ``APPROVED`` continue only to same-PR merge when
merge is explicitly authorized for that PR and merge gates pass; stop after
merge or a merge blocker. The next PR always requires a fresh run.
"""

from __future__ import annotations

import re

CLEANUP_WORKFLOW_PATH = "workflows/pr-queue-cleanup.md"

# Author-side resolver tasks that must never run inside cleanup mode.
CLEANUP_FORBIDDEN_TASKS = frozenset({
    "claim_issue",
    "mark_issue",
    "lock_issue",
    "create_issue",
    "create_branch",
    "push_branch",
    "create_pr",
    "commit_files",
    "gitea_commit_files",
    "address_pr_change_requests",
})

# Run-state outcomes for one cleanup dispatch.
STOP_AFTER_REQUEST_CHANGES = "STOP_AFTER_REQUEST_CHANGES"
STOP_AFTER_DECISION = "STOP_AFTER_DECISION"
CONTINUE_TO_SAME_PR_MERGE = "CONTINUE_TO_SAME_PR_MERGE"
STOP_APPROVED_NO_MERGE_AUTH = "STOP_APPROVED_NO_MERGE_AUTH"
STOP_APPROVED_MERGE_GATES_FAILED = "STOP_APPROVED_MERGE_GATES_FAILED"
STOP_AFTER_MERGE = "STOP_AFTER_MERGE"
STOP_AFTER_MERGE_BLOCKER = "STOP_AFTER_MERGE_BLOCKER"
STOP_GATE_NOT_PROVEN = "STOP_GATE_NOT_PROVEN"

_TERMINAL_DECISIONS = frozenset({"approved", "request_changes", "comment", "skip"})


def resolve_cleanup_run_state(
    decision: str | None,
    *,
    merge_authorized_for_pr: bool = False,
    merge_gates_passed: bool | None = None,
    merge_completed: bool = False,
    merge_blocker: bool = False,
) -> dict:
    """Resolve what one cleanup run may do after its terminal review decision.

    Fail closed: unknown decisions stop the run with no further mutation.
    """
    normalized = (decision or "").strip().lower()

    if normalized not in _TERMINAL_DECISIONS:
        return {
            "outcome": STOP_GATE_NOT_PROVEN,
            "further_mutation_allowed": False,
            "reasons": [
                f"unknown terminal decision {decision!r}; cleanup run stops "
                "(fail closed)"
            ],
        }

    if normalized == "request_changes":
        return {
            "outcome": STOP_AFTER_REQUEST_CHANGES,
            "further_mutation_allowed": False,
            "reasons": ["REQUEST_CHANGES is terminal in cleanup mode"],
        }

    if normalized in {"comment", "skip"}:
        return {
            "outcome": STOP_AFTER_DECISION,
            "further_mutation_allowed": False,
            "reasons": [f"{normalized} decision ends the cleanup run"],
        }

    # normalized == "approved"
    if merge_completed:
        return {
            "outcome": STOP_AFTER_MERGE,
            "further_mutation_allowed": False,
            "reasons": ["merge completed; cleanup run stops"],
        }
    if merge_blocker:
        return {
            "outcome": STOP_AFTER_MERGE_BLOCKER,
            "further_mutation_allowed": False,
            "reasons": ["merge blocker recorded; cleanup run stops"],
        }
    if not merge_authorized_for_pr:
        return {
            "outcome": STOP_APPROVED_NO_MERGE_AUTH,
            "further_mutation_allowed": False,
            "reasons": [
                "APPROVED without explicit per-PR merge authorization; "
                "cleanup run stops"
            ],
        }
    if merge_gates_passed is not True:
        return {
            "outcome": STOP_APPROVED_MERGE_GATES_FAILED,
            "further_mutation_allowed": False,
            "reasons": [
                "merge gates not proven passed; cleanup run stops before merge"
            ],
        }
    return {
        "outcome": CONTINUE_TO_SAME_PR_MERGE,
        "further_mutation_allowed": True,
        "reasons": [],
        "allowed_mutation": "merge same PR only",
    }


def check_cleanup_task_allowed(task: str) -> tuple[bool, list[str]]:
    """Fail closed on any author-side mutation task inside cleanup mode."""
    normalized = (task or "").strip().lower()
    if normalized in CLEANUP_FORBIDDEN_TASKS:
        return False, [
            f"task '{normalized}' is forbidden in PR-only cleanup mode: "
            "no issue claiming, branch creation, implementation edits, or "
            "issue filing"
        ]
    return True, []


_SELECTED_PR_RE = re.compile(
    r"^\s*[-*]?\s*selected pr\s*:\s*#?(\d+)", re.IGNORECASE | re.MULTILINE
)
_NEXT_SUGGESTED_RE = re.compile(
    r"^\s*[-*]?\s*next suggested pr\s*:", re.IGNORECASE | re.MULTILINE
)
_TERMINAL_MUTATION_RE = re.compile(
    r"^\s*[-*]?\s*(?:review (?:decision|verdict)|terminal (?:review )?"
    r"(?:decision|mutation))\s*:\s*"
    r"(approved|request_changes|request changes|merged|comment)",
    re.IGNORECASE | re.MULTILINE,
)
_PAGINATION_HINT_RE = re.compile(
    r"inventory_complete|final page|has_more\s*[=:]\s*false|total[_ ]count",
    re.IGNORECASE,
)
_MERGE_RESULT_RE = re.compile(
    r"^\s*[-*]?\s*merge result\s*:\s*(?!none\b|not attempted\b)\S",
    re.IGNORECASE | re.MULTILINE,
)
_MERGE_AUTHORIZED_RE = re.compile(
    r"^\s*[-*]?\s*merge authorized(?: for pr)?\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
)
_ISSUE_MUTATION_CLAIM_RE = re.compile(
    r"^\s*[-*]?\s*issue mutations\s*:\s*(?!none\b)\S",
    re.IGNORECASE | re.MULTILINE,
)
_BRANCH_MUTATION_CLAIM_RE = re.compile(
    r"^\s*[-*]?\s*branch mutations\s*:\s*(?!none\b)\S",
    re.IGNORECASE | re.MULTILINE,
)


def assess_pr_queue_cleanup_report(report_text: str) -> dict:
    """Validate a PR-only cleanup run report (single dispatch, fail closed)."""
    reasons: list[str] = []
    text = report_text or ""

    if CLEANUP_WORKFLOW_PATH not in text:
        reasons.append(
            f"report must cite the canonical cleanup workflow "
            f"({CLEANUP_WORKFLOW_PATH})"
        )

    selected = _SELECTED_PR_RE.findall(text)
    if len(selected) == 0:
        reasons.append("report must name exactly one Selected PR")
    elif len(set(selected)) > 1:
        reasons.append(
            "cleanup run selected multiple PRs "
            f"({', '.join(sorted(set(selected)))}); one canonical review per "
            "PR per run"
        )

    terminal = _TERMINAL_MUTATION_RE.findall(text)
    if len(terminal) > 1:
        reasons.append(
            "multiple terminal review mutations reported in one cleanup run"
        )

    if not _PAGINATION_HINT_RE.search(text):
        reasons.append(
            "report missing PR inventory pagination proof "
            "(inventory_complete / final page / total_count)"
        )

    if not _NEXT_SUGGESTED_RE.search(text):
        reasons.append(
            "report must include 'Next suggested PR' (without continuing to it)"
        )

    if _MERGE_RESULT_RE.search(text) and not _MERGE_AUTHORIZED_RE.search(text):
        reasons.append(
            "merge reported without explicit per-PR merge authorization "
            "('Merge authorized: true')"
        )

    if _ISSUE_MUTATION_CLAIM_RE.search(text):
        reasons.append("issue mutations are forbidden in PR-only cleanup mode")
    if _BRANCH_MUTATION_CLAIM_RE.search(text):
        reasons.append("branch mutations are forbidden in PR-only cleanup mode")

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "safe_next_action": (
            "proceed" if proven else
            "fix the cleanup report: one PR, one terminal mutation, pagination "
            "proof, next-suggested-PR field, and no issue/branch mutations"
        ),
    }