Gitea-Tools/final_report_validator.py

"""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 issue_lock_provenance
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 validation_status_vocabulary import assess_validation_status_vocabulary

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

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

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

_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,
)
_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)
_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 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_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_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]]:
    performed = any(
        e.get("performed") is not False and not e.get("gated_rejected")
        for e in (action_log or [])
    )
    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 (action_log or [])
    ) 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_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_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_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_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 (action_log or [])
    )
    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 []


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_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_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_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_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",
    )


_SHARED_ISSUE_LOCK_RULES = (
    _rule_shared_issue_lock_external_state,
    _rule_shared_manual_lock_pr_override,
    _rule_shared_author_reviewer_same_run,
)

_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "review_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_validation_command,
        _rule_reviewer_validation_failure_history,
        _rule_reviewer_validation_structured,
        _rule_reviewer_linked_issue,
        _rule_reviewer_baseline_on_failure,
        _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_mutation_ledger,
        _rule_reviewer_review_mutation,
        _rule_reviewer_post_merge_cleanup_proof,
        _rule_reviewer_stale_head_proof,
    ],
    "reconcile_already_landed": [
        _rule_reconcile_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_stale_author_fields,
        _rule_reconcile_eligible_reviewed,
        _rule_reconcile_linked_issue_live,
        _rule_reconcile_pagination_proof,
        _rule_reviewer_git_fetch_readonly,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
    ],
    "author_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
    ],
    "work_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
        _rule_conflict_fix_push_proof,
    ],
    "issue_filing": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
    "inventory": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_pagination_proof,
    ],
    "issue_selection": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
}


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,
) -> 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]] = []

    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,
    }

    for rule in _RULES_BY_TASK.get(normalized_kind, ()):
        findings.extend(_call_rule(rule, report_text, normalized_kind, rule_kwargs))

    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,
    }