Gitea-Tools/review_final_report_schema.py

"""Reviewer final-report schema verification before session output (#391).

Composes the composable ``assess_final_report_validator`` (#327) with
review-specific schema gates required before a reviewer session completes.
"""

from __future__ import annotations

import re
from typing import Any

from final_report_validator import (
    assess_final_report_validator,
    validator_finding,
    _handoff_fields,
)

_LEGACY_STALE_FIELDS = (
    "pinned reviewed head",
    "scratch worktree used",
    "workspace mutations",
)

_REVIEWED_HEAD_RE = re.compile(
    r"(?:pinned reviewed head|reviewed head sha)\s*:\s*([0-9a-f]{7,40})",
    re.IGNORECASE,
)
# #698: validation pass proof appears in several legitimate shapes —
# "Validation: pass", "Validation: focused 50 passed; full 2665 passed",
# or structured "validation_status: pass". Accept pass evidence anywhere in
# the Validation field's value, not only as its first token.
_VALIDATION_PASS_RE = re.compile(
    r"validation(?:_status)?\s*:[^\n]{0,300}?"
    r"(?:\bpass(?:ed)?\b|\bstrong\b|\bok\b|\bgreen\b|\d+\s+passed)",
    re.IGNORECASE,
)
_MERGED_CLAIM_RE = re.compile(
    r"\b(?:merged|merge result\s*:\s*merged|pr\s+#\d+\s+merged)\b",
    re.IGNORECASE,
)
_MERGE_RESULT_RE = re.compile(r"merge result\s*:", re.IGNORECASE)
_ANCESTRY_PROOF_RE = re.compile(
    r"(?:ancestor proof|target branch sha|merge commit)",
    re.IGNORECASE,
)
_ISSUE_CLOSED_RE = re.compile(
    r"(?:linked issue(?:\s+live)?\s+status\s*:\s*closed|issue\s+#\d+\s+closed)",
    re.IGNORECASE,
)
_LIVE_ISSUE_PROOF_RE = re.compile(
    r"gitea_view_issue|(?:issue|linked issue).{0,40}fetched live|live fetch proof",
    re.IGNORECASE,
)
_FULL_SUITE_PASS_RE = re.compile(
    r"(?:full suite passed|full test suite passed|\d+\s+passed,\s*0\s+failed)",
    re.IGNORECASE,
)
_TESTS_IGNORED_RE = re.compile(
    r"(?:ignored tests|tests?\s+ignored|skipped tests|not run|tests?\s+skipped)",
    re.IGNORECASE,
)
_BLOCKED_HANDOFF_RE = re.compile(
    r"(?:blocker\s*:|recovery handoff|infra_stop|capability stop|mutation blocked)",
    re.IGNORECASE,
)
_REPLAY_COMMAND_RE = re.compile(
    r"(?:gitea_submit_pr_review|gitea_merge_pr|gitea_review_pr|"
    r"submitted\s+['\"]approve['\"]|replay approve|replay merge)",
    re.IGNORECASE,
)
_PENDING_RE = re.compile(r"\bPENDING\b", re.IGNORECASE)
_APPROVED_RE = re.compile(
    r"(?:review decision\s*:\s*approve|submitted\s+approve|official review submitted)",
    re.IGNORECASE,
)
_DRY_RUN_RE = re.compile(r"dry[- ]run", re.IGNORECASE)
_FINDING_READY_RE = re.compile(
    r"(?:finding ready|ready to submit|not yet submitted)",
    re.IGNORECASE,
)
_NARRATIVE_APPROVE_RE = re.compile(
    r"(?:^|\n)\s*(?:##\s+)?(?:summary|verdict|recommendation)\b[^\n]*\bapprove\b",
    re.IGNORECASE | re.MULTILINE,
)
_HANDOFF_DECISION_RE = re.compile(
    r"review decision\s*:\s*(\w+)",
    re.IGNORECASE,
)


def _rule_legacy_stale_fields(report_text: str) -> list[dict[str, str]]:
    fields = _handoff_fields(report_text)
    findings: list[dict[str, str]] = []
    for stale in _LEGACY_STALE_FIELDS:
        if stale in fields and fields[stale].lower() not in {"", "none", "n/a"}:
            findings.append(
                validator_finding(
                    "reviewer.legacy_stale_field",
                    "block",
                    stale.title(),
                    f"legacy or stale handoff field '{stale}' must not appear in "
                    "canonical reviewer final reports",
                    "remove stale fields; use Candidate head SHA and precise "
                    "mutation categories instead",
                )
            )
    return findings


def _rule_reviewed_head_without_validation(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _REVIEWED_HEAD_RE.search(text):
        return []
    if _VALIDATION_PASS_RE.search(text):
        return []
    return [
        validator_finding(
            "reviewer.reviewed_head_without_validation",
            "block",
            "Pinned reviewed head",
            "reviewed/pinned head SHA reported without validation pass proof",
            "document validation command, cwd, and pass result before pinning head SHA",
        )
    ]


def _rule_merged_without_proof(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _MERGED_CLAIM_RE.search(text):
        return []
    if _MERGE_RESULT_RE.search(text) and _ANCESTRY_PROOF_RE.search(text):
        return []
    return [
        validator_finding(
            "reviewer.merged_without_proof",
            "block",
            "Merge result",
            "merged outcome claimed without merge result and target ancestry proof",
            "include Merge result, target branch SHA, and ancestor proof before claiming merged",
        )
    ]


def _rule_issue_closed_without_live_proof(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _ISSUE_CLOSED_RE.search(text):
        return []
    fields = _handoff_fields(text)
    status_value = fields.get("linked issue live status", "")
    readonly_value = fields.get("read-only diagnostics", "")
    proof_blob = f"{status_value} {readonly_value}".lower()
    if _LIVE_ISSUE_PROOF_RE.search(proof_blob):
        return []
    return [
        validator_finding(
            "reviewer.issue_closed_without_live_proof",
            "block",
            "Linked issue",
            "issue closed claimed without live linked-issue verification proof",
            "fetch linked issue live (gitea_view_issue) and record Linked issue live status",
        )
    ]


def _rule_full_suite_pass_ignored_tests(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not (_FULL_SUITE_PASS_RE.search(text) and _TESTS_IGNORED_RE.search(text)):
        return []
    if re.search(r"explicitly\s+(?:ignored|skipped)", text, re.IGNORECASE):
        return []
    return [
        validator_finding(
            "reviewer.full_suite_pass_ignored_tests",
            "block",
            "Validation",
            "full suite pass claimed while tests were ignored or skipped without "
            "explicit disclosure",
            "state which tests were ignored/skipped or downgrade validation verdict",
        )
    ]


def _rule_blocked_handoff_replay(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _BLOCKED_HANDOFF_RE.search(text):
        return []
    if not _REPLAY_COMMAND_RE.search(text):
        return []
    return [
        validator_finding(
            "reviewer.blocked_handoff_replay",
            "block",
            "Safe next action",
            "blocked recovery handoff includes direct approve or merge replay commands",
            "restart the full workflow after the blocker clears; do not replay mutations",
        )
    ]


def _rule_narrative_handoff_drift(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    narrative_approve = bool(_NARRATIVE_APPROVE_RE.search(text))
    decision_match = _HANDOFF_DECISION_RE.search(text)
    if not narrative_approve or not decision_match:
        return []
    handoff_decision = decision_match.group(1).strip().lower()
    if handoff_decision in {"approve", "approved"}:
        return []
    return [
        validator_finding(
            "reviewer.narrative_handoff_drift",
            "block",
            "Review decision",
            f"narrative approves PR but controller handoff says '{handoff_decision}'",
            "align narrative summary with controller handoff Review decision field",
        )
    ]


def _rule_review_state_ambiguous(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    findings: list[dict[str, str]] = []
    if _PENDING_RE.search(text) and _APPROVED_RE.search(text):
        findings.append(
            validator_finding(
                "reviewer.pending_vs_approved",
                "block",
                "Review decision",
                "report conflates PENDING review state with APPROVED outcome",
                "distinguish official submitted review from pending or ready-to-submit state",
            )
        )
    if _DRY_RUN_RE.search(text) and _APPROVED_RE.search(text):
        if "dry-run only" not in text.lower():
            findings.append(
                validator_finding(
                    "reviewer.dry_run_vs_submitted",
                    "block",
                    "Review mutations",
                    "dry-run language mixed with official approve/submitted claims",
                    "mark dry-run explicitly or document the live review mutation proof",
                )
            )
    if _FINDING_READY_RE.search(text) and _APPROVED_RE.search(text):
        findings.append(
            validator_finding(
                "reviewer.finding_ready_vs_submitted",
                "downgrade",
                "Review decision",
                "finding-ready wording combined with submitted-approve claims",
                "state whether review was submitted live or only prepared for submission",
            )
        )
    return findings


_SCHEMA_RULES = (
    _rule_legacy_stale_fields,
    _rule_reviewed_head_without_validation,
    _rule_merged_without_proof,
    _rule_issue_closed_without_live_proof,
    _rule_full_suite_pass_ignored_tests,
    _rule_blocked_handoff_replay,
    _rule_narrative_handoff_drift,
    _rule_review_state_ambiguous,
)


def _merge_validator_results(
    base: dict[str, Any],
    extra_findings: list[dict[str, str]],
) -> dict[str, Any]:
    findings = list(base.get("findings") or []) + extra_findings
    blocked = base.get("blocked") or any(f["severity"] == "block" for f in extra_findings)
    downgraded = base.get("downgraded") or any(
        f["severity"] == "downgrade" for f in extra_findings
    )
    grade = base.get("grade", "A")
    if blocked:
        grade = "blocked"
    elif downgraded and grade == "A":
        grade = "downgraded"
    reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
    safe_next = base.get("safe_next_action") or "none"
    if extra_findings:
        safe_next = extra_findings[0].get("safe_next_action", safe_next)
    return {
        **base,
        "grade": grade,
        "blocked": blocked,
        "downgraded": downgraded,
        "findings": findings,
        "reasons": reasons,
        "safe_next_action": safe_next,
        "complete": grade == "A",
        "schema_rules_applied": len(_SCHEMA_RULES),
    }


def assess_review_final_report_schema(
    report_text: str,
    *,
    review_decision_lock: dict | None = None,
    linked_issue_lock: dict | None = None,
    validation_report: dict | None = None,
    action_log: list[dict] | None = None,
    mutations_observed: bool = False,
    local_edits: bool = False,
    validation_session: dict | None = None,
) -> dict[str, Any]:
    """Validate reviewer final report text before session completion (#391)."""
    base = assess_final_report_validator(
        report_text,
        "review_pr",
        review_decision_lock=review_decision_lock,
        linked_issue_lock=linked_issue_lock,
        validation_report=validation_report,
        action_log=action_log,
        mutations_observed=mutations_observed,
        local_edits=local_edits,
        validation_session=validation_session,
    )
    extra: list[dict[str, str]] = []
    for rule in _SCHEMA_RULES:
        extra.extend(rule(report_text))
    return _merge_validator_results(base, extra)