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

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

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,
)
_ELIGIBLE_REVIEW_RE = re.compile(
    r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
    re.IGNORECASE,
)
_REVIEWED_LANDED_RE = re.compile(
    r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
    re.IGNORECASE | re.DOTALL,
)
_RECONCILE_STALE_FIELDS = (
    "pr number opened",
    "pinned reviewed head",
    "scratch worktree used",
    "review decision",
    "merge result",
)
_RECONCILE_REQUIRED_FIELDS = (
    "Selected PR",
    "Eligibility class",
    "Linked issue live status",
    "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 _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 and _GIT_FETCH_RE.search(text):
        return [
            validator_finding(
                "reviewer.git_fetch_readonly",
                "downgrade",
                "Git ref mutations",
                "git fetch mentioned without Git ref mutation classification",
                "record git fetch under Git ref mutations",
            )
        ]
    return []


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_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_RE.search(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_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) -> list[dict[str, str]]:
    text = (report_text or "").lower()
    findings = []
    for field in _RECONCILE_STALE_FIELDS:
        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_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",
    )


_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "review_pr": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        _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_structured,
        _rule_reviewer_linked_issue,
        _rule_reviewer_baseline_on_failure,
        _rule_reviewer_main_checkout_baseline,
        _rule_reviewer_main_checkout_path,
        _rule_reviewer_already_landed_eligible,
        _rule_reviewer_mutation_ledger,
        _rule_reviewer_review_mutation,
    ],
    "reconcile_already_landed": [
        _rule_reconcile_controller_handoff,
        _rule_shared_email_disclosure,
        _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,
        _rule_reviewer_vague_mutations_none,
    ],
    "work_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        _rule_reviewer_vague_mutations_none,
    ],
    "issue_filing": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
    ],
    "inventory": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        _rule_reconcile_pagination_proof,
    ],
    "issue_selection": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
    ],
}


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

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