"""Composable final-report validator for reviewer and reconciliation handoffs (#327).

Single entry point ``assess_final_report_validator`` runs task-specific rule
modules and returns structured findings suitable for controller handoff
``Blocker`` / ``Safe next action`` fields.
"""

from __future__ import annotations

import inspect
import re
from typing import Any, Callable

import branch_cleanup_guard
import issue_acceptance_gate
import issue_lock_provenance
import merger_lease_adoption
import reviewer_handoff_consistency
import thread_state_ledger_validator
from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
from review_proofs import (
    HANDOFF_HEADING,
    assess_controller_handoff,
    assess_email_disclosure,
    assess_issue_filing_final_report,
    assess_mutation_ledger_report,
    assess_review_mutation_final_report,
    assess_validation_report,
)
from state_handoff_ledger import assess_state_handoff_ledger_report
from validation_status_vocabulary import assess_validation_status_vocabulary

FINAL_REPORT_TASK_KINDS = frozenset({
    "review_pr",
    "merge_pr",
    "reconcile_already_landed",
    "author_issue",
    "work_issue",
    "issue_filing",
    "issue_selection",
    "inventory",
    "controller_close",
})

_TASK_KIND_ALIASES = {
    "review": "review_pr",
    "review-merge-pr": "review_pr",
    "merge": "merge_pr",
    "merge_pr": "merge_pr",
    "reconcile-landed-pr": "reconcile_already_landed",
    "reconcile_already_landed": "reconcile_already_landed",
    "work-issue": "work_issue",
    "create_issue": "issue_filing",
    "close_issue": "controller_close",
    "close-issue": "controller_close",
    "controller-close": "controller_close",
}

_HANDOFF_ROLE_BY_TASK = {
    "review_pr": "review",
    "merge_pr": "merger",
    "reconcile_already_landed": None,
    "author_issue": "author",
    "work_issue": "author",
    "issue_filing": "issue_filing",
    "issue_selection": None,
    "inventory": "inventory",
    "controller_close": None,
}

_LEGACY_WORKSPACE_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*workspace\s+mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
)
_VAGUE_MUTATIONS_NONE_RE = re.compile(
    r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
    re.IGNORECASE | re.MULTILINE,
)
_BARE_PYTEST_RE = re.compile(r"\b(?:pytest|python\s+-m\s+pytest)\b", re.IGNORECASE)
_VALIDATION_ENV_RE = re.compile(
    r"(?:working directory|cwd|executed (?:in|from)|command\s+path)\s*:|\bin\s+branches/",
    re.IGNORECASE,
)
_MAIN_CHECKOUT_RE = re.compile(
    r"(?:/Gitea-Tools(?:/|$)(?!branches/)|main checkout|shared checkout|control checkout)",
    re.IGNORECASE,
)
_SAME_AS_MASTER_RE = re.compile(
    r"same as master|same failures as master|matches master baseline",
    re.IGNORECASE,
)
_BASELINE_WORKTREE_RE = re.compile(
    r"baseline worktree",
    re.IGNORECASE,
)
_APPROVAL_CLAIM_RE = re.compile(
    r"(?:submitted\s+['\"]approve['\"]|review decision\s*:\s*approve|claims?\s+approve)",
    re.IGNORECASE,
)
_FULL_SUITE_FAIL_RE = re.compile(
    r"(?:full suite failed|full test suite failed|\d+\s+failed(?:,|\s))",
    re.IGNORECASE,
)
_ALREADY_LANDED_RE = re.compile(
    r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed",
    re.IGNORECASE,
)
_ALREADY_LANDED_GATE_FIRED_RE = re.compile(
    r"\b(?:fired|passed|true|yes|already_landed_reconcile_required)\b",
    re.IGNORECASE,
)
_ELIGIBLE_REVIEW_RE = re.compile(
    r"eligible for (?:review|merge)|ready for (?:review|merge)|"
    r"(?:next|oldest)\s+eligible\s+pr",
    re.IGNORECASE,
)
_APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
_MERGED_STATE_RE = re.compile(r"\bmerged\b", re.IGNORECASE)
_READY_TO_MERGE_STATE_RE = re.compile(r"\bready_to_merge\b|\bready to merge\b", re.IGNORECASE)
_NEGATED_STATE_RE = re.compile(
    r"\b(?:not|no|none|n/a|never)\b|request_changes|not attempted",
    re.IGNORECASE,
)
_REVIEWED_LANDED_RE = re.compile(
    r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
    re.IGNORECASE | re.DOTALL,
)
_TARGET_BRANCH_UP_TO_DATE_RE = re.compile(
    r"\btarget branch (?:is )?up[- ]to[- ]date\b",
    re.IGNORECASE,
)
_TARGET_BRANCH_SHA_RE = re.compile(
    r"target branch sha\s*:\s*[0-9a-f]{40}",
    re.IGNORECASE,
)
# #698: structured proof is rendered in several equivalent shapes —
# `workflow_hash: abc...`, `workflow_hash=abc...`, or JSON
# `"workflow_hash": "abc..."`. Recognize all of them; demanding one exact
# punctuation style rejects legitimate structured workflow-load proof.
_WORKFLOW_LOAD_HELPER_RE = re.compile(
    r"workflow[-_ ]load[-_ ]helper[-_ ]result\s*[:=]",
    re.IGNORECASE,
)
_WORKFLOW_LOAD_HASH_RE = re.compile(
    r"workflow[-_ ]load[-_ ]helper[-_ ]result[\s\S]{0,400}?"
    r"workflow[_ ]hash\"?\s*[:=]\s*\"?[0-9a-f]{12}",
    re.IGNORECASE,
)
_WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
    r"workflow[-_ ]load[-_ ]helper[-_ ]result[\s\S]{0,400}?"
    r"boundary[_ ]status\"?\s*[:=]\s*\"?(?:clean|violation)",
    re.IGNORECASE,
)
_WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
    r"(?:read|viewed|loaded)\s+(?:the\s+)?(?:canonical\s+)?(?:workflow|review-merge-pr\.md)",
    re.IGNORECASE,
)
_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
_RECONCILE_STALE_FIELDS = (
    "pr number opened",
    "pinned reviewed head",
    "scratch worktree used",
    "review decision",
    "merge result",
    "issue lock proof",
    "claim/comment status",
    "workspace mutations",
)
# Issue #307: reconciliation handoffs must carry the full canonical schema.
_RECONCILE_REQUIRED_FIELDS = (
    "Task",
    "Repo",
    "Role/profile",
    "Identity",
    "Selected PR",
    "PR live state",
    "Candidate head SHA",
    "Target branch",
    "Target branch SHA",
    "Ancestor proof",
    "Linked issue",
    "Linked issue live status",
    "Eligibility class",
    "Capabilities proven",
    "Missing capabilities",
    "PR comments posted",
    "Issue comments posted",
    "PRs closed",
    "Issues closed",
    "Git ref mutations",
    "Worktree mutations",
    "MCP/Gitea mutations",
    "External-state mutations",
    "Read-only diagnostics",
    "Blocker",
    "Safe next action",
    "No review/merge confirmation",
)
_LINKED_ISSUE_STATUS_RE = re.compile(
    r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_LINKED_ISSUE_NUMBER_RE = re.compile(
    r"linked issue\s*:\s*#?(\d+)",
    re.IGNORECASE,
)
_INVENTORY_COMPLETE_RE = re.compile(
    r"inventory\s+(?:complete|completeness\s*:\s*complete)",
    re.IGNORECASE,
)
_PAGINATION_PROOF_RE = re.compile(
    r"(?:pagination\s+complete|final page|no next page|total\s+(?:open\s+)?pr\s+count|traversed\s+all\s+pages)",
    re.IGNORECASE,
)
_GIT_FETCH_RE = re.compile(r"\bgit\s+fetch\b", re.IGNORECASE)
_CANONICAL_VALIDATION_REJECTED_RE = re.compile(
    r"canonical comment validation failed|"
    r"canonical_comment_validation|"
    r'"allowed"\s*:\s*false',
    re.IGNORECASE,
)
_COMMENT_POSTED_CLAIM_RE = re.compile(
    r"(?:issue comments posted\s*:\s+(?!none\b)\S|"
    r"pr comments posted\s*:\s+(?!none\b)\S|"
    r"comment_id\s*[:=]\s*\d+|"
    r"gitea comment (?:was )?posted|"
    r"posted (?:issue|pr|canonical) (?:state )?comment|"
    r"comment posted successfully|"
    r"mcp/gitea mutations\s*:\s*[^;\n]*comment posted|"
    r"reconciliation mutations\s*:\s*[^;\n]*comment posted)",
    re.IGNORECASE,
)
_READONLY_DIAG_RE = re.compile(
    r"read[- ]only diagnostics\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_MUTATION_CATEGORY_FIELDS = (
    "file edits by reviewer",
    "worktree/index mutations",
    "git ref mutations",
    "mcp/gitea mutations",
    "reconciliation mutations",
    "file edits by reconciler",
)


def _normalize_task_kind(task_kind: str | None) -> str:
    raw = (task_kind or "").strip().lower().replace(" ", "_")
    return _TASK_KIND_ALIASES.get(raw, raw)


def _iter_action_entries(action_log: list | None) -> list[dict]:
    """Yield only structured (dict) action-log entries (#698).

    Callers must never crash on malformed entries (strings, numbers, null)
    that reach the validator from LLM-composed or partially parsed logs;
    :func:`sanitize_action_log` reports them separately.
    """
    return [e for e in (action_log or []) if isinstance(e, dict)]


def sanitize_action_log(
    action_log: list | None,
) -> tuple[list[dict], list[dict[str, str]]]:
    """Split an action log into structured entries and sanitized findings (#698).

    Malformed entries become clear, sanitized ``warning`` findings — the
    offending value's content is never echoed back (only its position and
    type), so secrets or garbage in a broken log cannot leak into validation
    errors, and validation itself proceeds without secondary exceptions.
    """
    if action_log is None:
        return [], []
    if not isinstance(action_log, (list, tuple)):
        return [], [
            validator_finding(
                "shared.action_log_malformed",
                "downgrade",
                "Action log",
                "action_log is not a list of structured entries "
                f"(got {type(action_log).__name__}); it was ignored",
                "pass action_log as a list of dict entries",
            )
        ]
    entries: list[dict] = []
    findings: list[dict[str, str]] = []
    for index, entry in enumerate(action_log):
        if isinstance(entry, dict):
            entries.append(entry)
            continue
        findings.append(
            validator_finding(
                "shared.action_log_malformed",
                "downgrade",
                "Action log",
                f"action_log entry {index} is not a structured mapping "
                f"(got {type(entry).__name__}); the entry was ignored",
                "repair the malformed action_log entry or drop it before "
                "revalidating",
            )
        )
    return entries, findings


def validator_finding(
    rule_id: str,
    severity: str,
    field: str,
    reason: str,
    safe_next_action: str,
) -> dict[str, str]:
    """Structured finding for controller handoff consumption."""
    return {
        "rule_id": rule_id,
        "severity": severity,
        "field": field,
        "reason": reason,
        "safe_next_action": safe_next_action,
    }


def _findings_from_reasons(
    rule_id: str,
    reasons: list[str],
    *,
    field: str = "",
    severity: str = "downgrade",
    safe_next_action: str = "downgrade final report; repair missing proof fields",
) -> list[dict[str, str]]:
    return [
        validator_finding(rule_id, severity, field, reason, safe_next_action)
        for reason in reasons
    ]


def _handoff_fields(report_text: str) -> dict[str, str]:
    from review_proofs import _handoff_section_lines  # local import avoids cycle at load

    section = _handoff_section_lines(report_text) or []
    fields: dict[str, str] = {}
    for line in section:
        stripped = line.strip().lstrip("-*").strip()
        if ":" not in stripped:
            continue
        key, value = stripped.split(":", 1)
        fields[key.strip().lower()] = value.strip()
    return fields


def _already_landed_gate_fired(report_text: str) -> bool:
    text = report_text or ""
    fields = _handoff_fields(text)

    gate_value = fields.get("already-landed gate", "")
    if gate_value and _ALREADY_LANDED_GATE_FIRED_RE.search(gate_value):
        return True

    ancestor_value = fields.get("ancestor proof", "")
    if re.search(r"\bpassed\b", ancestor_value, re.IGNORECASE):
        return True

    eligibility_value = fields.get("eligibility class", "")
    if "already_landed" in eligibility_value.lower():
        return True

    if re.search(
        r"\bpr\b.{0,60}\balready[- ]landed\b|\balready[- ]landed\b.{0,60}\bpr\b",
        text,
        re.IGNORECASE | re.DOTALL,
    ):
        return True

    return bool(_ALREADY_LANDED_RE.search(text))


def _positive_review_state_terms(fields: dict[str, str]) -> list[str]:
    terms: list[str] = []
    for key, value in fields.items():
        lowered_key = key.lower()
        lowered_value = value.lower()
        if _NEGATED_STATE_RE.search(value):
            continue
        if lowered_key == "review decision" and _APPROVED_STATE_RE.search(value):
            terms.append("APPROVED")
        elif lowered_key == "merge result" and _MERGED_STATE_RE.search(value):
            terms.append("MERGED")
        elif _READY_TO_MERGE_STATE_RE.search(value):
            terms.append("READY_TO_MERGE")
        elif lowered_value in {"approved", "approve"}:
            terms.append("APPROVED")
        elif lowered_value == "merged":
            terms.append("MERGED")
        elif lowered_value == "ready_to_merge":
            terms.append("READY_TO_MERGE")
    return sorted(set(terms))


def _rule_shared_controller_handoff(
    report_text: str,
    task_kind: str,
    *,
    local_edits: bool = False,
) -> list[dict[str, str]]:
    role = _HANDOFF_ROLE_BY_TASK.get(task_kind)
    result = assess_controller_handoff(
        report_text,
        role=role,
        local_edits=local_edits,
    )
    verdict = result.get("verdict")
    if verdict == "complete":
        return []
    severity = "block" if verdict == "missing" else "downgrade"
    return _findings_from_reasons(
        "shared.controller_handoff",
        result.get("reasons") or [],
        field="Controller Handoff",
        severity=severity,
        safe_next_action=(
            "add a complete Controller Handoff section with task-appropriate fields"
            if verdict == "missing"
            else "fill missing controller handoff fields before submission"
        ),
    )


def _rule_shared_state_handoff_next_action(report_text: str) -> list[dict[str, str]]:
    result = assess_state_handoff_ledger_report(report_text)
    if result.get("complete"):
        return []
    findings: list[dict[str, str]] = []
    next_action = result.get("next_action") or {}
    contradictions = result.get("contradictions") or {}
    if next_action.get("block"):
        findings.extend(
            _findings_from_reasons(
                "shared.state_handoff_next_action",
                next_action.get("reasons") or [],
                field="Next action handoff",
                severity="block",
                safe_next_action=next_action.get("safe_next_action")
                or "add next-action handoff fields to Controller Handoff",
            )
        )
    if contradictions.get("block"):
        findings.extend(
            _findings_from_reasons(
                "shared.state_handoff_contradiction",
                contradictions.get("reasons") or [],
                field="State handoff",
                severity="block",
                safe_next_action=contradictions.get("safe_next_action")
                or "resolve contradictory state claims before submission",
            )
        )
    return findings


def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
    result = assess_email_disclosure(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.email_disclosure",
        result.get("reasons") or [],
        field="Identity",
        severity="downgrade",
        safe_next_action="use username / profile identity; remove personal email",
    )


def _rule_shared_canonical_comment_post_claim(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
) -> list[dict[str, str]]:
    """#496: reports must not claim comment posted when validator rejected."""
    text = report_text or ""
    rejected_in_report = bool(_CANONICAL_VALIDATION_REJECTED_RE.search(text))
    rejected_in_log = False
    if action_log:
        for entry in _iter_action_entries(action_log):
            validation = entry.get("canonical_comment_validation") or {}
            if validation.get("allowed") is False:
                rejected_in_log = True
                break
            result = entry.get("result") or {}
            nested = result.get("canonical_comment_validation") or {}
            if nested.get("allowed") is False:
                rejected_in_log = True
                break
    if not (rejected_in_report or rejected_in_log):
        return []
    if not _COMMENT_POSTED_CLAIM_RE.search(text):
        return []
    return [
        validator_finding(
            "shared.canonical_comment_post_claim",
            "block",
            "MCP/Gitea mutations",
            "report claims a Gitea comment was posted while canonical comment "
            "validation rejected the workflow comment",
            "do not claim comment posted when validator fail-closed; repair "
            "the canonical comment body and retry posting",
        )
    ]


def _rule_reviewer_legacy_workspace_mutations(
    report_text: str,
    *,
    mutations_observed: bool = False,
) -> list[dict[str, str]]:
    if not mutations_observed:
        return []
    if not _LEGACY_WORKSPACE_MUTATIONS_RE.search(report_text or ""):
        return []
    return [
        validator_finding(
            "reviewer.legacy_workspace_mutations",
            "block",
            "Workspace mutations",
            "legacy 'Workspace mutations' field used after observed mutations; "
            "use precise mutation categories instead",
            "replace Workspace mutations with file/worktree/git/MCP category fields",
        )
    ]


def _rule_reviewer_vague_mutations_none(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
    mutations_observed: bool = False,
) -> list[dict[str, str]]:
    # #698: infer review mutations only from authoritative evidence — an
    # entry proves a mutation only when it affirmatively records
    # performed=true and was not gated. Read-only diagnostics and pre-API
    # rejections (entries without a performed flag) are not mutations.
    performed = any(
        e.get("performed") is True and not e.get("gated_rejected")
        for e in _iter_action_entries(action_log)
    )
    if not (mutations_observed or performed):
        return []
    if not _VAGUE_MUTATIONS_NONE_RE.search(report_text or ""):
        return []
    return [
        validator_finding(
            "reviewer.vague_mutations_none",
            "block",
            "Mutations",
            "report claims 'Mutations: none' while mutations were observed",
            "list performed mutations under precise category fields",
        )
    ]


def _rule_reviewer_mutation_categories(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    lower = text.lower()
    if "mutations:" not in lower and "mutation categories" not in lower:
        return [
            validator_finding(
                "reviewer.mutation_categories",
                "downgrade",
                "Mutations",
                "reviewer handoff lacks precise mutation category fields",
                "report file edits, worktree/index, git ref, MCP/Gitea, and review mutations separately",
            )
        ]
    if any(term in lower for term in ("workspace mutations", "mutations: none")):
        return []
    if any(field in lower for field in _MUTATION_CATEGORY_FIELDS):
        return []
    vague_only = "mutations:" in lower and not any(
        marker in lower
        for marker in ("file edits", "worktree", "git ref", "mcp/gitea", "review mutation")
    )
    if vague_only:
        return [
            validator_finding(
                "reviewer.mutation_categories",
                "downgrade",
                "Mutations",
                "mutation section is vague; precise categories are required",
                "split mutations into file edits, worktree/index, git ref, MCP/Gitea, review",
            )
        ]
    return []


def _rule_reviewer_git_fetch_readonly(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
) -> list[dict[str, str]]:
    text = report_text or ""
    fetch_observed = any(
        _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
        for e in _iter_action_entries(action_log)
    ) or _GIT_FETCH_RE.search(text)
    if not fetch_observed:
        return []

    readonly_match = _READONLY_DIAG_RE.search(text)
    readonly_value = (readonly_match.group(1) if readonly_match else "").strip().lower()
    fields = _handoff_fields(text)
    git_ref_value = fields.get("git ref mutations", "").strip().lower()
    git_ref_reported = bool(git_ref_value) and git_ref_value not in {
        "none",
        "n/a",
        "no",
    }
    if git_ref_reported:
        return []
    if readonly_match and _GIT_FETCH_RE.search(readonly_value):
        return [
            validator_finding(
                "reviewer.git_fetch_readonly",
                "block",
                "Git ref mutations",
                "git fetch occurred but was classified only as read-only diagnostics",
                "classify git fetch under Git ref mutations, not read-only diagnostics",
            )
        ]
    if not git_ref_reported:
        # Issue #307: an observed git fetch (report text or action log)
        # must be listed under Git ref mutations.
        return [
            validator_finding(
                "reviewer.git_fetch_readonly",
                "downgrade",
                "Git ref mutations",
                "git fetch occurred without Git ref mutation classification",
                "record git fetch under Git ref mutations",
            )
        ]
    return []


def _rule_reviewer_validation_failure_history(
    report_text: str,
    *,
    validation_session: dict | None = None,
) -> list[dict[str, str]]:
    from reviewer_validation_failure_history import (
        assess_validation_failure_history_report,
    )

    session = validation_session or {}
    observed = session.get("observed_failures") or []
    if not observed:
        return []

    result = assess_validation_failure_history_report(
        report_text,
        validation_session=session,
    )
    if result.get("proven"):
        return []
    severity = "block" if result.get("violations") else "downgrade"
    return [
        validator_finding(
            "reviewer.validation_failure_history",
            severity,
            "Validation failure history",
            reason,
            result.get("safe_next_action")
            or "document every observed validation failure before claiming pass",
        )
        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
    ]


def _rule_reviewer_validation_cwd_proof(
    report_text: str,
    *,
    validation_session: dict | None = None,
) -> list[dict[str, str]]:
    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report

    session = validation_session or {}
    claims = (
        session.get("validation_ran")
        or session.get("command")
        or session.get("baseline_validation_ran")
    )
    if not claims and "validation command:" not in (report_text or "").lower():
        return []

    result = assess_validation_cwd_proof_report(
        report_text,
        validation_session=session,
    )
    if result.get("proven") or not result.get("claims_validation"):
        return []
    severity = "block" if result.get("violations") else "downgrade"
    return [
        validator_finding(
            "reviewer.validation_cwd_proof",
            severity,
            "Validation cwd/HEAD proof",
            reason,
            result.get("safe_next_action")
            or "document pwd, HEAD SHA, and explicit cwd before validation",
        )
        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
    ]


def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
    from pr_work_lease import assess_reviewer_stale_head_final_report

    result = assess_reviewer_stale_head_final_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "reviewer.stale_head_proof",
        result.get("reasons") or [],
        field="Stale-head proof",
        severity="block",
        safe_next_action=(
            "state reviewed head SHA, live head before approval/merge, and "
            "whether any push occurred during validation"
        ),
    )


def _rule_conflict_fix_classification_proof(report_text: str) -> list[dict[str, str]]:
    from conflict_fix_classification import (
        assess_conflict_fix_classification_final_report,
    )

    text = report_text or ""
    result = assess_conflict_fix_classification_final_report(text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "author.conflict_fix_classification_proof",
        result.get("reasons") or [],
        field="Conflict-fix classification",
        severity="block",
        safe_next_action=(
            "call gitea_assess_conflict_fix_classification, state the live head "
            "SHA, and state the classification before creating a conflict-fix worktree"
        ),
    )


def _rule_conflict_fix_push_proof(report_text: str) -> list[dict[str, str]]:
    from pr_work_lease import assess_conflict_fix_final_report

    text = report_text or ""
    if "conflict-fix" not in text.lower() and "conflict fix" not in text.lower():
        return []
    result = assess_conflict_fix_final_report(text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "author.conflict_fix_push_proof",
        result.get("reasons") or [],
        field="Conflict-fix push proof",
        severity="block",
        safe_next_action=(
            "state branch head before/after push, reviewer lease status, "
            "fast-forward status, and whether any reviewer was active"
        ),
    )


def _rule_worktree_cleanup_audit_proof(report_text: str) -> list[dict[str, str]]:
    from worktree_cleanup_audit import assess_cleanup_audit_final_report

    text = report_text or ""
    if "cleanup audit" not in text.lower() and "reconciliation table" not in text.lower():
        return []
    result = assess_cleanup_audit_final_report(text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "author.worktree_cleanup_audit_proof",
        result.get("reasons") or [],
        field="Worktree cleanup audit",
        severity="block",
        safe_next_action=(
            "include reconciliation table counts, disposition rows, and "
            "final git worktree list proof"
        ),
    )


def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _BARE_PYTEST_RE.search(text):
        return []
    if _VALIDATION_ENV_RE.search(text):
        return []
    return [
        validator_finding(
            "reviewer.validation_command_proof",
            "downgrade",
            "Validation",
            "bare pytest command without working-directory or executable path proof",
            "report exact validation command, cwd, and path proof for bare commands",
        )
    ]


def _rule_reviewer_linked_issue(
    report_text: str,
    *,
    linked_issue_lock: dict | None = None,
) -> list[dict[str, str]]:
    lock = linked_issue_lock or {}
    expected_issue = lock.get("issue_number")
    expected_pr = lock.get("pr_number")
    if expected_issue is None and expected_pr is None:
        return []

    text = report_text or ""
    findings: list[dict[str, str]] = []
    reported_issue = None
    issue_match = _LINKED_ISSUE_NUMBER_RE.search(text)
    if issue_match:
        reported_issue = int(issue_match.group(1))
    elif expected_issue is not None:
        if f"#{expected_issue}" not in text and f"issue {expected_issue}" not in text.lower():
            findings.append(
                validator_finding(
                    "reviewer.linked_issue_missing",
                    "downgrade",
                    "Linked issue",
                    f"linked issue #{expected_issue} not documented in final report",
                    "document the linked issue number matching the selected PR",
                )
            )

    if expected_issue is not None and reported_issue is not None and reported_issue != expected_issue:
        findings.append(
            validator_finding(
                "reviewer.linked_issue_mismatch",
                "block",
                "Linked issue",
                f"reported linked issue #{reported_issue} does not match "
                f"selected PR linked issue #{expected_issue}",
                "reconcile linked issue proof with the selected PR before reporting status",
            )
        )

    if expected_pr is not None:
        pr_tokens = (f"pr #{expected_pr}", f"pr{expected_pr}", f"#pr{expected_pr}")
        if not any(token in text.lower().replace(" ", "") for token in pr_tokens):
            if f"#{expected_pr}" not in text:
                findings.append(
                    validator_finding(
                        "reviewer.linked_pr_missing",
                        "downgrade",
                        "Selected PR",
                        f"selected PR #{expected_pr} not consistently referenced",
                        "reference the selected PR number in handoff and diagnostics",
                    )
                )
    return findings


def _rule_reviewer_baseline_on_failure(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not (_APPROVAL_CLAIM_RE.search(text) and _FULL_SUITE_FAIL_RE.search(text)):
        return []
    if _BASELINE_WORKTREE_RE.search(text) and "baseline worktree path" in text.lower():
        return []
    return [
        validator_finding(
            "reviewer.baseline_on_full_suite_failure",
            "block",
            "Baseline worktree",
            "approval claimed after full-suite failure without baseline worktree proof",
            "create a clean branches/ baseline worktree and report path plus results",
        )
    ]


def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _SAME_AS_MASTER_RE.search(text):
        return []
    if _BASELINE_WORKTREE_RE.search(text):
        return []
    return [
        validator_finding(
            "reviewer.main_checkout_baseline",
            "block",
            "Baseline worktree",
            "vague 'same as master' claim without clean baseline worktree proof",
            "validate against a clean branches/ baseline worktree; do not use main checkout",
        )
    ]


def _rule_reviewer_premerge_baseline_proof(report_text: str) -> list[dict[str, str]]:
    from premerge_baseline_proof import assess_premerge_baseline_proof

    result = assess_premerge_baseline_proof(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.premerge_baseline_proof",
        result.get("reasons") or [],
        field="Pre-merge baseline proof",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or (
            "prove the failure on the PR pre-merge base commit or a known-failure "
            "record predating the PR; current-master reproduction is not baseline proof"
        ),
    )


def _rule_reviewer_post_merge_validation(report_text: str) -> list[dict[str, str]]:
    """Block active approval on an already-merged/closed PR, and post-merge moot
    validation claimed without merged-state + merge-commit proof (#529)."""
    from post_merge_validation import assess_post_merge_validation

    result = assess_post_merge_validation(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.post_merge_validation",
        result.get("reasons") or [],
        field="Validation status",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or (
            "record post-merge moot validation instead of an active approval; "
            "cite PR state merged/closed and the merge commit SHA"
        ),
    )


def _rule_reviewer_validation_status_vocabulary(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
) -> list[dict[str, str]]:
    text = report_text or ""
    if not re.search(
        r"validation status|pr-head validation status|official validation status",
        text,
        re.IGNORECASE,
    ):
        return []
    result = assess_validation_status_vocabulary(
        text,
        command_log=action_log,
    )
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.validation_status_vocabulary",
        result.get("reasons") or [],
        field="Validation status",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "use a validation status that matches the proof path executed",
    )


def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if "baseline worktree path" not in text.lower():
        return []
    fields = _handoff_fields(text)
    baseline_path = ""
    for key, value in fields.items():
        if key.startswith("baseline worktree path"):
            baseline_path = value
            break
    if baseline_path and _MAIN_CHECKOUT_RE.search(baseline_path):
        return [
            validator_finding(
                "reviewer.main_checkout_path",
                "block",
                "Baseline worktree path",
                "baseline validation used main/shared checkout instead of branches/ worktree",
                "recreate baseline proof under branches/ and report that path",
            )
        ]
    return []


def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _already_landed_gate_fired(text):
        return []
    if not _ELIGIBLE_REVIEW_RE.search(text):
        return []
    return [
        validator_finding(
            "reviewer.already_landed_eligible_wording",
            "block",
            "Eligibility class",
            "already-landed PR described as eligible for review or merge",
            "classify as ALREADY_LANDED_RECONCILE_REQUIRED and stop normal review",
        )
    ]


def _rule_reviewer_already_landed_state(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _already_landed_gate_fired(text):
        return []

    terms = _positive_review_state_terms(_handoff_fields(text))
    if not terms:
        return []

    return [
        validator_finding(
            "reviewer.already_landed_review_state",
            "block",
            "Review decision",
            "already-landed gate fired but final report claims "
            f"{', '.join(terms)} state",
            "stop normal review/merge; classify as ALREADY_LANDED_RECONCILE_REQUIRED",
        )
    ]


def _rule_reviewer_target_branch_freshness(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
) -> list[dict[str, str]]:
    text = report_text or ""
    if not _TARGET_BRANCH_UP_TO_DATE_RE.search(text):
        return []

    fields = _handoff_fields(text)
    fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any(
        _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
        for e in _iter_action_entries(action_log)
    )
    target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any(
        "target branch" in key and "sha" in key and _FULL_SHA_RE.search(value)
        for key, value in fields.items()
    )
    if fetch_reported and target_sha_reported:
        return []

    missing = []
    if not fetch_reported:
        missing.append("fresh git fetch")
    if not target_sha_reported:
        missing.append("target branch SHA")
    return [
        validator_finding(
            "reviewer.target_branch_freshness_proof",
            "block",
            "Target branch SHA",
            "target branch up-to-date claim lacks " + " and ".join(missing),
            "fetch the target branch and report the exact target branch SHA before "
            "claiming it is up to date",
        )
    ]


def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]:
    from review_proofs import _handoff_section_lines

    if _handoff_section_lines(report_text) is None:
        return [
            validator_finding(
                "reconcile.controller_handoff",
                "block",
                "Controller Handoff",
                f"final report has no section titled exactly '{HANDOFF_HEADING}'",
                "add a reconciliation Controller Handoff section",
            )
        ]

    fields = _handoff_fields(report_text)
    missing = [
        name
        for name in _RECONCILE_REQUIRED_FIELDS
        if not (fields.get(name.lower()) or "").strip()
    ]
    if missing:
        return [
            validator_finding(
                "reconcile.controller_handoff",
                "downgrade",
                "Controller Handoff",
                f"reconciliation handoff missing required field: {field}",
                "complete the reconciliation handoff schema before submission",
            )
            for field in missing
        ]
    return []


def _rule_reconcile_stale_author_fields(
    report_text: str,
    *,
    session_pr_opened: bool = False,
) -> list[dict[str, str]]:
    text = (report_text or "").lower()
    findings = []
    for field in _RECONCILE_STALE_FIELDS:
        if field == "pr number opened" and session_pr_opened:
            # Issue #307: allowed only when a PR was actually opened
            # in this session.
            continue
        if f"{field}:" in text:
            findings.append(
                validator_finding(
                    "reconcile.stale_author_reviewer_fields",
                    "block",
                    field,
                    f"reconciliation handoff includes stale author/reviewer field '{field}'",
                    "use reconciliation schema only; remove author/reviewer review fields",
                )
            )
    return findings


def _rule_reconcile_eligible_reviewed(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _ALREADY_LANDED_RE.search(text):
        return []
    findings = []
    if _ELIGIBLE_REVIEW_RE.search(text):
        findings.append(
            validator_finding(
                "reconcile.eligible_reviewed_wording",
                "block",
                "Eligibility class",
                "already-landed PR described as eligible for review or merge",
                "report ALREADY_LANDED_RECONCILE_REQUIRED and reconciliation-only next steps",
            )
        )
    if _REVIEWED_LANDED_RE.search(text):
        findings.append(
            validator_finding(
                "reconcile.reviewed_landed_wording",
                "block",
                "Current status",
                "already-landed PR described as reviewed or approval-eligible",
                "use reconciliation status wording only",
            )
        )
    return findings


def _rule_reconcile_linked_issue_live(
    report_text: str,
    *,
    linked_issue_lock: dict | None = None,
) -> list[dict[str, str]]:
    lock = linked_issue_lock or {}
    text = report_text or ""
    status_match = _LINKED_ISSUE_STATUS_RE.search(text)
    if not status_match:
        return []

    status_value = status_match.group(1).strip().lower()
    if "not verified" in status_value:
        return []

    live_proof = lock.get("live_fetched") is True or lock.get("verified_live") is True
    if live_proof:
        return []

    if status_value not in {"", "none", "n/a", "unknown"}:
        return [
            validator_finding(
                "reconcile.linked_issue_live_status",
                "downgrade",
                "Linked issue live status",
                "linked issue status reported without live verification proof in this session",
                "fetch linked issue live or report 'not verified in this session'",
            )
        ]
    return []


_PR_CLOSE_NEGATIVE = frozenset({"", "none", "n/a", "na", "0", "no", "not closed", "not performed"})
_ANCESTOR_AFFIRMATIVE_RE = re.compile(
    r"ancestor|passed|true|verified|confirmed", re.IGNORECASE
)


def _reconciler_pr_close_performed(fields: dict[str, str], lock: dict) -> bool:
    """True when the report/session indicates a reconciler PR close happened."""
    if lock.get("pr_closed") is True:
        return True
    value = (fields.get("prs closed", "") or "").strip().lower()
    if value in _PR_CLOSE_NEGATIVE:
        return False
    # A closed PR is reported by number (e.g. "#99") or an affirmative result.
    return bool(re.search(r"#\s*\d+|\b(?:closed|success|done)\b", value))


def _rule_reconcile_close_proof(
    report_text: str,
    *,
    reconciler_close_lock: dict | None = None,
) -> list[dict[str, str]]:
    """#306: a reconciler PR close must carry exact proof fields.

    Read-only/comment-only reconciliations are untouched. Once a PR close is
    reported (via the ``PRs closed`` field or a session close lock), the
    handoff must prove the close capability, ancestor landing, PR close
    result, and the linked-issue result — narrative alone fails closed.
    """
    fields = _handoff_fields(report_text)
    lock = reconciler_close_lock or {}
    if not _reconciler_pr_close_performed(fields, lock):
        return []

    missing: list[str] = []
    capabilities = fields.get("capabilities proven", "")
    if "gitea.pr.close" not in capabilities.lower():
        missing.append("close capability proof (gitea.pr.close)")
    ancestor = fields.get("ancestor proof", "")
    if not _ANCESTOR_AFFIRMATIVE_RE.search(ancestor):
        missing.append("ancestor proof")
    prs_closed = (fields.get("prs closed", "") or "").strip().lower()
    if prs_closed in _PR_CLOSE_NEGATIVE and lock.get("pr_closed") is True:
        missing.append("PR close result")
    linked = fields.get("linked issue live status", "") or fields.get("issues closed", "")
    if not linked.strip():
        missing.append("linked issue result")

    if not missing:
        return []
    return [
        validator_finding(
            "reconcile.close_proof_fields",
            "block",
            "Reconciler close proof",
            "reconciler PR close reported without required proof field(s): "
            + ", ".join(missing),
            "include close capability proof, ancestor proof, PR close result, "
            "and linked issue result in the handoff",
        )
    ]


def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _INVENTORY_COMPLETE_RE.search(text):
        return []
    if _PAGINATION_PROOF_RE.search(text):
        return []
    return [
        validator_finding(
            "reconcile.inventory_pagination_proof",
            "downgrade",
            "PR inventory",
            "inventory completeness claimed without pagination or final-page proof",
            "include pagination complete, final page, or total open PR count evidence",
        )
    ]


def _rule_reviewer_validation_structured(
    validation_report: dict | None,
) -> list[dict[str, str]]:
    if not validation_report:
        return []
    result = assess_validation_report(validation_report)
    if result.get("verdict") == "strong":
        return []
    severity = "block" if result.get("verdict") == "invalid" else "downgrade"
    return _findings_from_reasons(
        "reviewer.validation_structured",
        result.get("reasons") or [],
        field="Validation",
        severity=severity,
        safe_next_action="provide exact validation command, output read proof, and counts",
    )


def _rule_reviewer_handoff_consistency(report_text: str) -> list[dict[str, str]]:
    result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(
        report_text
    )
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "reviewer.handoff_consistency",
        result.get("reasons") or [],
        field="Review mutations",
        severity="block",
        safe_next_action=(
            "rewrite reviewer handoff so narrative claims match the mutation "
            "ledger; use the blocked-review handoff template when submission "
            "was rejected"
        ),
    )


def _rule_reviewer_mutation_ledger(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
) -> list[dict[str, str]]:
    if not action_log:
        return []
    result = assess_mutation_ledger_report(report_text, action_log=action_log)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "reviewer.mutation_ledger",
        result.get("reasons") or [],
        field="File edits by reviewer",
        severity="block",
        safe_next_action="align mutation ledger with observed file edits",
    )


def _rule_shared_issue_lock_external_state(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_issue_lock_external_state_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.issue_lock_external_state",
        result.get("reasons") or [],
        field="External-state mutations",
        severity="block",
        safe_next_action=(
            "disclose gitea_issue_lock.json read/write/delete under "
            "External-state mutations; never claim none after lock seeding"
        ),
    )


def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_manual_lock_pr_without_override(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.manual_lock_pr_override",
        result.get("reasons") or [],
        field="External-state mutations",
        severity="block",
        safe_next_action=(
            "use gitea_lock_issue or #442 adoption instead of manual lock seeding; "
            "if operator override was authorized, cite override proof"
        ),
    )


def _rule_shared_manual_lease_proof_handoff(report_text: str) -> list[dict[str, str]]:
    """Block merge/review handoffs that claim unsafe lease seeding (#535)."""
    result = merger_lease_adoption.assess_manual_lease_proof_handoff(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.manual_lease_proof_handoff",
        result.get("reasons") or [],
        field="Merge mutations",
        severity="block",
        safe_next_action=(
            "use gitea_adopt_merger_pr_lease or gitea_acquire_reviewer_pr_lease; "
            "cite lease_proof_source / adoption_comment_id; never claim manual "
            "seeded/injected lease proof"
        ),
    )


def _rule_shared_issue_acceptance_gate(report_text: str) -> list[dict[str, str]]:
    result = issue_acceptance_gate.validate_final_report_issue_acceptance(report_text)
    if not result.get("applicable") or result.get("valid"):
        return []
    return _findings_from_reasons(
        "shared.issue_acceptance_gate",
        result.get("reasons") or [],
        field="Controller acceptance",
        severity="block",
        safe_next_action=(
            "add Controller Issue Acceptance proof or state that controller "
            "acceptance is pending; do not claim issue complete from PR merge alone"
        ),
    )


def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_author_reviewer_same_run_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.author_reviewer_same_run",
        result.get("reasons") or [],
        field="Review mutations",
        severity="block",
        safe_next_action=(
            "split author PR creation and reviewer approval across separate "
            "sessions and handoffs"
        ),
    )


def _rule_shared_raw_branch_delete_bypass(report_text: str) -> list[dict[str, str]]:
    result = branch_cleanup_guard.assess_raw_branch_delete_report(report_text)
    if not result["block"]:
        return []
    return _findings_from_reasons(
        "shared.raw_branch_delete_bypass",
        result["reasons"],
        field="Cleanup mutations",
        severity="block",
        safe_next_action=result["safe_next_action"],
    )
def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
    """#403: require structured workflow-load helper result, not file-view narrative."""
    if not report_text.strip():
        return []
    findings: list[dict[str, str]] = []
    has_helper = bool(_WORKFLOW_LOAD_HELPER_RE.search(report_text))
    has_hash = bool(_WORKFLOW_LOAD_HASH_RE.search(report_text))
    has_boundary = bool(_WORKFLOW_LOAD_BOUNDARY_RE.search(report_text))
    has_narrative_only = bool(_WORKFLOW_FILE_VIEW_NARRATIVE_RE.search(report_text))

    if has_narrative_only and not has_helper:
        findings.append(validator_finding(
            "reviewer.workflow_load_boundary",
            "block",
            "Workflow-load helper result",
            (
                "canonical workflow file-view narrative without structured "
                "gitea_load_review_workflow helper result"
            ),
            (
                "include Workflow-load helper result with workflow_hash and "
                "boundary_status from gitea_load_review_workflow"
            ),
        ))
        return findings

    if has_helper and (not has_hash or not has_boundary):
        missing = []
        if not has_hash:
            missing.append("workflow_hash")
        if not has_boundary:
            missing.append("boundary_status")
        findings.append(validator_finding(
            "reviewer.workflow_load_boundary",
            "block",
            "Workflow-load helper result",
            (
                "workflow-load helper result incomplete; missing "
                + ", ".join(missing)
            ),
            (
                "copy workflow_load_helper_result fields from "
                "gitea_load_review_workflow into the final report"
            ),
        ))
    return findings


def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
    from audit_reconciliation_mode import assess_audit_reconciliation_report

    result = assess_audit_reconciliation_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "reconcile.audit_cleanup_boundary",
        result.get("reasons") or [],
        field="Audit/cleanup phase",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "separate audit from authorized cleanup and classify mutations",
    )


def _rule_reviewer_review_mutation(
    report_text: str,
    *,
    review_decision_lock: dict | None = None,
) -> list[dict[str, str]]:
    if not review_decision_lock:
        return []
    result = assess_review_mutation_final_report(report_text, review_decision_lock)
    if result.get("complete"):
        return []
    return _findings_from_reasons(
        "reviewer.review_mutation_proof",
        result.get("reasons") or [],
        field="Review mutations",
        severity="downgrade",
        safe_next_action="document exactly one live review mutation or authorized correction flow",
    )


def _rule_shared_two_comment_workflow(report_text: str) -> list[dict[str, str]]:
    """#507: tagged Controller Handoff must pair with Thread State Ledger."""
    return thread_state_ledger_validator.findings_for_final_report(report_text)


def _rule_shared_canonical_state_update(report_text: str) -> list[dict[str, str]]:
    from canonical_state_comments import validate_final_report_state_update

    result = validate_final_report_state_update(report_text)
    if not result.get("applicable") or result.get("valid"):
        return []
    return [
        validator_finding(
            "shared.canonical_state_update",
            "block",
            "Canonical state update",
            reason,
            "include a canonical state block with STATE, WHO_IS_NEXT, NEXT_ACTION, and NEXT_PROMPT",
        )
        for reason in (result.get("reasons") or ["invalid canonical state update"])
    ]


def _rule_reviewer_mutation_capability_proof(report_text: str) -> list[dict[str, str]]:
    from reviewer_mutation_capability_proof import assess_mutation_capability_proof

    result = assess_mutation_capability_proof(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.mutation_capability_proof",
        result.get("reasons") or [],
        field="Capabilities proven",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "document exact per-mutation capability proof before each mutation",
    )


def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
    result = assess_post_merge_cleanup_proof(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.post_merge_cleanup_proof",
        result.get("reasons") or [],
        field="Cleanup status",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "report CLEANUP_SKIPPED with blocker or full cleanup checklist",
    )


def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, str]]:
    result = assess_mcp_native_cleanup_proof(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "shared.mcp_native_cleanup_proof",
        result.get("reasons") or [],
        field="Cleanup mutations",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or "use authorized reconciler MCP cleanup tools; never raw scripts",
    )


_SHARED_ISSUE_LOCK_RULES = (
    _rule_shared_issue_lock_external_state,
    _rule_shared_manual_lock_pr_override,
    _rule_shared_manual_lease_proof_handoff,
    _rule_shared_author_reviewer_same_run,
    _rule_shared_raw_branch_delete_bypass,
    _rule_shared_canonical_state_update,
)

_SHARED_TWO_COMMENT_RULES = (
    _rule_shared_two_comment_workflow,
)
_SHARED_CLEANUP_PROOF_RULES = (
    _rule_shared_mcp_native_cleanup_proof,
)

_SHARED_CANONICAL_COMMENT_RULES = (
    _rule_shared_canonical_comment_post_claim,
)

_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "review_pr": [
        _rule_shared_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
        _rule_reviewer_mutation_categories,
        _rule_reviewer_git_fetch_readonly,
        _rule_reviewer_validation_command,
        _rule_reviewer_validation_failure_history,
        _rule_reviewer_validation_cwd_proof,
        _rule_reviewer_validation_structured,
        _rule_reviewer_linked_issue,
        _rule_reviewer_baseline_on_failure,
        _rule_reviewer_premerge_baseline_proof,
        _rule_reviewer_post_merge_validation,
        _rule_reviewer_validation_status_vocabulary,
        _rule_reviewer_main_checkout_baseline,
        _rule_reviewer_main_checkout_path,
        _rule_reviewer_already_landed_eligible,
        _rule_reviewer_already_landed_state,
        _rule_reviewer_target_branch_freshness,
        _rule_reviewer_handoff_consistency,
        _rule_reviewer_workflow_load_boundary,
        _rule_reviewer_mutation_ledger,
        _rule_reviewer_review_mutation,
        _rule_reviewer_mutation_capability_proof,
        _rule_reviewer_post_merge_cleanup_proof,
        *_SHARED_CLEANUP_PROOF_RULES,
        _rule_reviewer_stale_head_proof,
    ],
    "merge_pr": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
        _rule_reviewer_mutation_categories,
        _rule_reviewer_git_fetch_readonly,
        _rule_reviewer_linked_issue,
        _rule_reviewer_stale_head_proof,
    ],
    "reconcile_already_landed": [
        _rule_reconcile_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
        *_SHARED_CLEANUP_PROOF_RULES,
        _rule_reconcile_stale_author_fields,
        _rule_reconcile_eligible_reviewed,
        _rule_reconcile_linked_issue_live,
        _rule_reconcile_close_proof,
        _rule_reconcile_pagination_proof,
        _rule_reviewer_premerge_baseline_proof,
        _rule_reviewer_git_fetch_readonly,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
        _rule_audit_reconciliation_boundary,
    ],
    "author_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
    ],
    "work_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_shared_issue_acceptance_gate,
        _rule_reviewer_vague_mutations_none,
        _rule_conflict_fix_classification_proof,
        _rule_conflict_fix_push_proof,
        _rule_worktree_cleanup_audit_proof,
    ],
    "issue_filing": [
        _rule_shared_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
    "inventory": [
        _rule_shared_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_pagination_proof,
    ],
    "issue_selection": [
        _rule_shared_controller_handoff,
        _rule_shared_state_handoff_next_action,
        _rule_shared_email_disclosure,
        *_SHARED_TWO_COMMENT_RULES,
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
    # Controller issue closure (#529): a closure report must not bury an
    # unproven non-zero suite exit as an "expected pre-existing failure".
    # Kept intentionally narrow so a closure pre-check does not demand the
    # full reviewer/author handoff schema.
    "controller_close": [
        _rule_reviewer_premerge_baseline_proof,
    ],
}


def _aggregate_grade(findings: list[dict[str, str]]) -> tuple[str, bool, bool]:
    blocked = any(f["severity"] == "block" for f in findings)
    downgraded = any(f["severity"] == "downgrade" for f in findings)
    if blocked:
        return "blocked", True, downgraded
    if downgraded:
        return "downgraded", False, True
    return "A", False, False


def _primary_safe_next_action(findings: list[dict[str, str]]) -> str:
    for severity in ("block", "downgrade", "warning"):
        for finding in findings:
            if finding["severity"] == severity:
                return finding["safe_next_action"]
    return "proceed"


def _call_rule(
    rule: Callable[..., list[dict[str, str]]],
    report_text: str,
    task_kind: str,
    rule_kwargs: dict[str, Any],
) -> list[dict[str, str]]:
    if rule is _rule_shared_controller_handoff:
        return rule(
            report_text,
            task_kind,
            local_edits=bool(rule_kwargs.get("local_edits")),
        )
    if rule is _rule_reviewer_validation_structured:
        return rule(rule_kwargs.get("validation_report"))
    if rule is _rule_reconcile_controller_handoff:
        return rule(report_text)

    bound: dict[str, Any] = {}
    for name, param in inspect.signature(rule).parameters.items():
        if name == "report_text":
            bound[name] = report_text
        elif name in rule_kwargs:
            bound[name] = rule_kwargs[name]
        elif param.default is inspect.Parameter.empty and param.kind != inspect.Parameter.VAR_KEYWORD:
            continue
    return rule(**bound)


def assess_final_report_validator(
    report_text: str,
    task_kind: 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,
    issue_filing_lock: dict | None = None,
    session_pr_opened: bool = False,
    validation_session: dict | None = None,
    reconciler_close_lock: dict | None = None,
) -> dict[str, Any]:
    """Validate final-report text against task-specific proof rules (#327).

    Returns structured findings plus a composite grade suitable for workflow
    controllers. Optional *proof_locks* carry live session facts used to fail
    closed on stale or contradictory report fields.
    """
    normalized_kind = _normalize_task_kind(task_kind)
    if normalized_kind not in FINAL_REPORT_TASK_KINDS:
        return {
            "task_kind": normalized_kind,
            "grade": "blocked",
            "blocked": True,
            "downgraded": False,
            "findings": [
                validator_finding(
                    "shared.unknown_task_kind",
                    "block",
                    "Task",
                    f"unknown task kind '{task_kind}'",
                    "use a supported task kind from FINAL_REPORT_TASK_KINDS",
                )
            ],
            "checks": {},
            "reasons": [f"unknown task kind '{task_kind}'"],
            "safe_next_action": "set task_kind to a supported workflow mode",
            "complete": False,
        }

    checks: dict[str, Any] = {}
    findings: list[dict[str, str]] = []

    # #698: malformed action_log data must never crash validation with a
    # secondary exception; malformed entries surface as sanitized findings.
    sanitized_action_log, action_log_findings = sanitize_action_log(action_log)
    action_log = sanitized_action_log
    findings.extend(action_log_findings)

    if normalized_kind == "issue_filing" and issue_filing_lock is not None:
        checks["issue_filing"] = assess_issue_filing_final_report(
            report_text,
            **issue_filing_lock,
        )
        if checks["issue_filing"].get("downgraded"):
            findings.extend(
                _findings_from_reasons(
                    "issue_filing.composite",
                    checks["issue_filing"].get("reasons") or [],
                    field="Issue filing",
                    severity="downgrade",
                )
            )

    rule_kwargs = {
        "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,
        "session_pr_opened": session_pr_opened,
        "validation_session": validation_session,
        "reconciler_close_lock": reconciler_close_lock,
    }

    for rule in _RULES_BY_TASK.get(normalized_kind, ()):
        try:
            findings.extend(
                _call_rule(rule, report_text, normalized_kind, rule_kwargs)
            )
        except Exception as exc:  # #698: fail closed with a sanitized error
            findings.append(
                validator_finding(
                    "shared.validator_rule_error",
                    "block",
                    "Validator",
                    f"validator rule '{getattr(rule, '__name__', 'unknown')}' "
                    f"failed with {type(exc).__name__} (details withheld; "
                    "sanitized)",
                    "file a validator defect with the rule name; do not "
                    "bypass final-report validation",
                )
            )

    grade, blocked, downgraded = _aggregate_grade(findings)
    reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
    return {
        "task_kind": normalized_kind,
        "grade": grade,
        "blocked": blocked,
        "downgraded": downgraded,
        "findings": findings,
        "checks": checks,
        "reasons": reasons,
        "safe_next_action": _primary_safe_next_action(findings),
        "complete": grade == "A",
        "handoff_heading": HANDOFF_HEADING,
    }
