feat: require pagination proof for reconciliation inventory claims (Closes #308)

Add assess_reconciliation_inventory_proof to review_proofs:
already-landed reconciliation reports may not claim a complete queue
scan or that all already-landed PRs were found unless the report
carries the pagination proof fields (Requested page size, Pages
fetched, Returned PR count per page, Final page proof) and the
per-page gitea_list_prs pagination metadata proves the final page was
reached — inventory_complete, is_final_page with has_more=false, or a
returned count below the requested page size. Declared page counts and
per-page counts must match the evidence. Reports without a
completeness claim pass without proof. Fails closed on missing
evidence or malformed page metadata. Tests cover first page below
limit, first page exactly at the page limit, multi-page traversal to
the final page, missing pagination metadata, missing report fields,
page-count and per-page-count mismatches, and the no-claim case.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

final_report_validator.py

@@ -0,0 +1,878 @@
+"""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,
+    }

review_proofs.py

@@ -1052,6 +1052,122 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
     return None
 
 
+# ── Reconciliation inventory pagination proof (Issue #308) ───────────────────
+#
+# Already-landed reconciliation may not claim a complete queue scan (or
+# that every already-landed PR was found) unless the report carries the
+# pagination proof fields and the per-page gitea_list_prs metadata
+# proves the final page was reached or the page limit was not hit.
+
+_RECONCILE_COMPLETE_CLAIM_RE = re.compile(
+    r"complete queue scan|all already-landed|inventory complete|"
+    r"all open prs (?:were )?(?:inspected|scanned|listed)|found all",
+    re.IGNORECASE,
+)
+
+RECONCILIATION_INVENTORY_PROOF_FIELDS = (
+    "requested page size",
+    "pages fetched",
+    "returned pr count per page",
+    "final page proof",
+)
+
+
+def assess_reconciliation_inventory_proof(report_text, *,
+                                          pagination_evidence=None):
+    """#308: reconciliation completeness claims need pagination proof.
+
+    *pagination_evidence* is the list of per-page ``pagination`` dicts
+    from the session's ``gitea_list_prs`` calls, in fetch order. Reports
+    that do not claim inventory completeness pass without proof; any
+    completeness claim requires the report proof fields and evidence
+    that the final page was reached (``inventory_complete``,
+    ``is_final_page`` with ``has_more=false``, or a returned count below
+    the requested page size).
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
+    """
+    reasons = []
+    if not _RECONCILE_COMPLETE_CLAIM_RE.search(report_text or ""):
+        return {"complete": True, "downgraded": False, "reasons": []}
+
+    fields = _report_labeled_fields(report_text)
+    for required in RECONCILIATION_INVENTORY_PROOF_FIELDS:
+        if required not in fields:
+            reasons.append(
+                "reconciliation completeness claimed but report missing "
+                f"pagination proof field '{required}'"
+            )
+
+    if not pagination_evidence:
+        reasons.append(
+            "reconciliation completeness claimed without pagination "
+            "evidence from gitea_list_prs"
+        )
+        return {"complete": False, "downgraded": True, "reasons": reasons}
+
+    evidence_valid = True
+    for page in pagination_evidence:
+        if not isinstance(page, dict) or not all(
+            key in page for key in ("per_page", "returned_count")
+        ):
+            reasons.append(
+                "pagination evidence page missing 'per_page'/"
+                "'returned_count' metadata"
+            )
+            evidence_valid = False
+
+    if evidence_valid:
+        last = pagination_evidence[-1]
+        final_proven = (
+            last.get("inventory_complete") is True
+            or (last.get("is_final_page") is True
+                and last.get("has_more") is False)
+            or last["returned_count"] < last["per_page"]
+        )
+        if not final_proven:
+            reasons.append(
+                "pagination evidence does not prove final page (need "
+                "inventory_complete, is_final_page with has_more=false, "
+                "or a returned count below the requested page size)"
+            )
+
+        pages_fetched = fields.get("pages fetched")
+        if pages_fetched:
+            declared = re.findall(r"\d+", pages_fetched)
+            if declared and int(declared[0]) != len(pagination_evidence):
+                reasons.append(
+                    f"report 'Pages fetched' ({declared[0]}) does not match "
+                    f"pagination evidence ({len(pagination_evidence)} pages)"
+                )
+
+        counts_field = fields.get("returned pr count per page")
+        if counts_field:
+            declared_counts = [int(n) for n in re.findall(r"\d+", counts_field)]
+            actual_counts = [p["returned_count"] for p in pagination_evidence]
+            if declared_counts != actual_counts:
+                reasons.append(
+                    "report 'Returned PR count per page' "
+                    f"({declared_counts}) does not match pagination "
+                    f"evidence {actual_counts}"
+                )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+# ── Linked-issue consistency (Issue #314) ────────────────────────────────────
+#
+# Stale linked-issue text from a previous PR must not leak into a final
+# report: the "Linked issue status" field must match the issue(s) the
+# selected PR actually links, verified live this session, and no other
+# "Issue #N" mention may appear unless it is a verified linked issue.
+
+_ISSUE_MENTION_RE = re.compile(r"\bissue\s+#(\d+)", re.IGNORECASE)
+
 # ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
 #
 # A worktree reset/checkout/clean is a workspace mutation even when the
@@ -1092,6 +1208,68 @@ def _report_labeled_fields(report_text):
     return fields
 
 
+def _issue_numbers_mentioned(text):
+    """Return the set of ints referenced as 'Issue #N' in *text*."""
+    return {int(n) for n in _ISSUE_MENTION_RE.findall(text or "")}
+
+
+def assess_linked_issue_consistency(report_text, *, selected_pr=None,
+                                    linked_issues=None):
+    """#314: linked-issue status must match the selected PR, live-verified.
+
+    *linked_issues* is the list of issue numbers the selected PR was
+    live-verified to link this session, or None when no live verification
+    happened. Without live proof the report may only say the status was
+    not verified; with proof, the ``Linked issue status`` field must name
+    every linked issue and no stale issue number may appear anywhere in
+    the report.
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
+    """
+    fields = _report_labeled_fields(report_text)
+    status_value = fields.get("linked issue status")
+
+    reasons = []
+
+    if status_value is None:
+        reasons.append(
+            "final report missing 'Linked issue status' field for the "
+            "selected PR"
+        )
+    elif linked_issues is None:
+        if not status_value.strip().lower().startswith("not verified"):
+            reasons.append(
+                "linked issue status claimed without live proof; report "
+                "'Linked issue status: not verified in this session' or "
+                "verify the linked issue live"
+            )
+    else:
+        allowed = set(linked_issues)
+        status_mentions = _issue_numbers_mentioned(status_value)
+        for missing in sorted(allowed - status_mentions):
+            reasons.append(
+                f"linked issue #{missing} of selected PR "
+                f"#{selected_pr} not reported in 'Linked issue status'"
+            )
+        for stale in sorted(status_mentions - allowed):
+            reasons.append(
+                f"'Linked issue status' names issue #{stale}, which is "
+                f"not a live-verified linked issue of selected PR "
+                f"#{selected_pr}"
+            )
+        for stale in sorted(_issue_numbers_mentioned(report_text) - allowed):
+            reasons.append(
+                f"stale issue #{stale} mentioned in final report but not "
+                f"live-verified as linked to selected PR #{selected_pr}"
+            )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 def _field_claims_none(fields, label):
     """True when *label* is present and its value is empty or 'none...'."""
     value = fields.get(label)
@@ -1791,6 +1969,20 @@ HANDOFF_BASE_FIELDS = (
     ("Safety", ("safety",)),
 )
 
+# Issue #320: reviewer handoffs replace the legacy ambiguous
+# "Workspace mutations" field with these precise mutation categories.
+HANDOFF_REVIEW_MUTATION_FIELDS = (
+    ("File edits by reviewer", ("file edits by reviewer",)),
+    ("Worktree/index mutations", ("worktree/index mutations",)),
+    ("Git ref mutations", ("git ref mutations",)),
+    ("MCP/Gitea mutations", ("mcp/gitea mutations",)),
+    ("Review mutations", ("review mutations",)),
+    ("Merge mutations", ("merge mutations",)),
+    ("Cleanup mutations", ("cleanup mutations",)),
+    ("External-state mutations", ("external-state mutations",)),
+    ("Read-only diagnostics", ("read-only diagnostics",)),
+)
+
 HANDOFF_ROLE_FIELDS = {
     "review": (
         ("Selected PR", ("selected pr",)),
@@ -1806,7 +1998,7 @@ HANDOFF_ROLE_FIELDS = {
         ("Merge result", ("merge result",)),
         ("Linked issue status", ("linked issue status", "linked issue")),
         ("Cleanup status", ("cleanup status", "cleanup")),
-    ),
+    ) + HANDOFF_REVIEW_MUTATION_FIELDS,
     "author": (
         ("Selected issue", ("selected issue",)),
         ("Issue lock proof", ("issue lock proof", "lock before diff")),
@@ -1957,6 +2149,25 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
             field for field in required
             if field[0] not in ("Workspace mutations", "Mutations")
         ]
+    if role == "review":
+        # Issue #320: reviewer handoffs use the precise mutation categories
+        # in HANDOFF_REVIEW_MUTATION_FIELDS instead of the legacy ambiguous
+        # "Workspace mutations" field, which is rejected below.
+        required = [
+            field for field in required
+            if field[0] != "Workspace mutations"
+        ]
+        if any(label.startswith("workspace mutations") for label in labels):
+            return {
+                "verdict": "incomplete",
+                "downgraded": True,
+                "missing_fields": [],
+                "reasons": [
+                    "review handoff must not include legacy "
+                    "'Workspace mutations' field; report the precise "
+                    "mutation categories instead (issue #320)"
+                ],
+            }
     required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
 
     missing = []
@@ -3629,6 +3840,20 @@ def assess_email_disclosure(
     }
 
 
+def assess_reviewer_fallback_report(report_text, **kwargs):
+    """#324: block local Gitea fallbacks during normal reviewer workflows."""
+    from reviewer_fallback import assess_reviewer_fallback_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_final_report_validator(report_text, task_kind, **kwargs):
+    """#327: single entry point for task-specific final-report validation."""
+    from final_report_validator import assess_final_report_validator as _validate
+
+    return _validate(report_text, task_kind, **kwargs)
+
+
 _QUEUE_STATUS_REPORT_HINT = re.compile(
     r"queue[- ]status|selected pr:\s*none|no pr selected|"
     r"queue[- ]status[- ]only",
@@ -4012,6 +4237,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
     }
 
 
+def assess_prior_blocker_skip_proof(report_text, **kwargs):
+    """#318: require live blocker proof before skipping earlier open PRs."""
+    from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_non_mergeable_skip_proof(report_text, **kwargs):
     """#322: require conflict proof when skipping non-mergeable PRs."""
     from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess

reviewer_blocker_skip.py

@@ -0,0 +1,164 @@
+"""Prior-blocker skip proof verifier for reviewer queue reports (#318)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
+_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE)
+_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
+_BLOCKING_DECISION_RE = re.compile(
+    r"(?:blocking review decision|review decision|blocking decision)\s*:\s*request[_ ]changes|"
+    r"request[_ ]changes",
+    re.IGNORECASE,
+)
+_BLOCKING_HEAD_RE = re.compile(
+    r"(?:blocking review head(?:\s+sha)?|blocker head(?:\s+sha)?|review head at blocker)",
+    re.IGNORECASE,
+)
+_HEAD_CHANGED_RE = re.compile(
+    r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
+    r"head changed (?:after|since) (?:the )?blocker|head unchanged since blocker)",
+    re.IGNORECASE,
+)
+_BLOCKER_REASON_RE = re.compile(
+    r"(?:reason (?:it )?remains blocked|remains blocked because|blocking category)",
+    re.IGNORECASE,
+)
+_LIVE_BLOCKER_PROOF_RE = re.compile(
+    r"(?:blocker revalidated live|live proof|gitea_get_pr_review_feedback|"
+    r"gitea_view_pr|review feedback fetched|current review state)",
+    re.IGNORECASE,
+)
+_BLOCKER_UNVERIFIED_RE = re.compile(r"\bBLOCKER_STATUS_UNVERIFIED\b")
+
+
+def _pr_section(text: str, pr_number: int) -> str:
+    lines = (text or "").splitlines()
+    chunks: list[str] = []
+    capture = False
+    token = f"#{pr_number}"
+    for line in lines:
+        lower = line.lower()
+        if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""):
+            capture = True
+            chunks.append(line)
+            continue
+        if capture:
+            if _PR_NUMBER_RE.search(line) and token not in line.lower():
+                break
+            if line.strip() == "" and len(chunks) > 3:
+                break
+            chunks.append(line)
+    if chunks:
+        return "\n".join(chunks)
+    return text or ""
+
+
+def _has_head_sha(text: str, head_sha: str | None) -> bool:
+    if not head_sha:
+        return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text))
+    head = head_sha.strip().lower()
+    if head in text.lower():
+        return True
+    if len(head) >= 7 and head[:7] in text.lower():
+        return True
+    return bool(_FULL_SHA.search(text))
+
+
+def assess_prior_blocker_skip_proof(
+    report_text: str,
+    *,
+    skipped_prs: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Validate live blocker proof for skipped earlier open PRs (#318)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    assessments: list[dict[str, Any]] = []
+
+    for entry in skipped_prs or []:
+        pr_number = entry.get("pr_number")
+        if pr_number is None:
+            reasons.append("skipped PR entry missing pr_number")
+            continue
+        pr_number = int(pr_number)
+        section = _pr_section(text, pr_number)
+        verified = entry.get("blocker_verified")
+        head_changed = entry.get("head_changed_since_blocker")
+        blocking_head = (entry.get("blocking_review_head_sha") or "").strip() or None
+        current_head = (entry.get("head_sha") or "").strip() or None
+        skip_reason = (entry.get("skip_reason") or entry.get("blocking_decision") or "").lower()
+
+        item_reasons: list[str] = []
+
+        if head_changed is True:
+            item_reasons.append(
+                f"PR #{pr_number} head changed after blocker; it cannot be skipped "
+                "based on stale REQUEST_CHANGES"
+            )
+
+        if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower():
+            item_reasons.append(f"skipped PR #{pr_number} not documented in final report")
+
+        if "request_changes" in skip_reason or entry.get("blocking_decision") == "request_changes":
+            if verified is False:
+                if not _BLOCKER_UNVERIFIED_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} blocker proof unavailable; report must classify "
+                        "BLOCKER_STATUS_UNVERIFIED"
+                    )
+            else:
+                if not _BLOCKING_DECISION_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing blocking review decision proof"
+                    )
+                if not _has_head_sha(section, current_head):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing current head SHA"
+                    )
+                if blocking_head and not _BLOCKING_HEAD_RE.search(section):
+                    if blocking_head.lower() not in section.lower():
+                        item_reasons.append(
+                            f"PR #{pr_number} skip missing blocking review head SHA"
+                        )
+                if head_changed is False and not _HEAD_CHANGED_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing head-changed-since-blocker proof"
+                    )
+                if not _BLOCKER_REASON_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing reason-it-remains-blocked"
+                    )
+                if not _LIVE_BLOCKER_PROOF_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing live blocker proof for this session"
+                    )
+
+        proven = not item_reasons
+        assessments.append({
+            "pr_number": pr_number,
+            "proven": proven,
+            "classification": (
+                "BLOCKER_STATUS_UNVERIFIED"
+                if verified is False
+                else "BLOCKED_SKIPPED"
+            ),
+            "reasons": item_reasons,
+        })
+        reasons.extend(item_reasons)
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "downgraded": False,
+        "assessments": assessments,
+        "reasons": reasons,
+        "safe_next_action": (
+            "fetch live review feedback and document blocker proof for each skipped PR, "
+            "or classify BLOCKER_STATUS_UNVERIFIED"
+            if reasons
+            else "proceed"
+        ),
+    }

reviewer_fallback.py

@@ -0,0 +1,213 @@
+"""Reviewer local-fallback detection for normal PR review workflows (#324).
+
+Normal reviewer runs must use MCP tools. Reading profile secret files or
+running local Gitea helper scripts while MCP is available is a fail-closed
+violation. Explicit recovery mode may use local fallback only with full proof.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+LOCAL_GITEA_SCRIPT_NAMES = (
+    "list_prs.py",
+    "view_pr.py",
+    "create_pr.py",
+    "merge_pr.py",
+    "edit_pr.py",
+    "list_issues.py",
+    "delete_branch.py",
+    "create_issue.py",
+    "close_issue.py",
+    "review_pr.py",
+    "mark_issue.py",
+    "mirror_refs.sh",
+)
+
+PROFILE_SECRET_MARKERS = (
+    "profiles.json",
+    ".config/gitea-tools/profiles",
+    "gitea_auth.py",
+    "gitea_config.py",
+    "keychain",
+    "credential fill",
+    "token store",
+)
+
+_RECOVERY_MODE_RE = re.compile(
+    r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|"
+    r"mcp tools unavailable|no mcp path)\b",
+    re.IGNORECASE,
+)
+_FALLBACK_CLASSIFICATION_RE = re.compile(
+    r"\b(?:local fallback|fallback mode|used local gitea|ran local script)\b",
+    re.IGNORECASE,
+)
+_MCP_TOOL_USE_RE = re.compile(
+    r"\b(?:gitea_list_prs|gitea_view_pr|gitea_review_pr|gitea_merge_pr|"
+    r"gitea_resolve_task_capability|gitea_whoami|mcp tool)\b",
+    re.IGNORECASE,
+)
+_LOCAL_SCRIPT_RE = re.compile(
+    r"(?:^|[\s\"'`/])(?:" + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + r")",
+    re.IGNORECASE,
+)
+_PROFILE_ACCESS_RE = re.compile(
+    r"(?:read|open|inspect|cat|view|access(?:ed)?|loaded?)\s+(?:file\s+)?[`'\"]?"
+    r"[^`'\"]*profiles\.json|profiles\.json[`'\"]?\s+(?:read|opened|inspected|accessed)|"
+    r"~/?\.config/gitea-tools/profiles",
+    re.IGNORECASE,
+)
+_LOCAL_ENV_SCRIPT_RE = re.compile(
+    r"GITEA_MCP_(?:CONFIG|PROFILE)=.*\b(?:python\s+)?(?:list_prs|view_pr|create_pr|merge_pr)\.py",
+    re.IGNORECASE,
+)
+_RECOVERY_PROOF_FIELDS = (
+    ("identity proof", ("identity proof", "exact identity proof")),
+    ("profile proof", ("profile proof", "exact profile proof")),
+    ("repo proof", ("repo proof", "exact repo proof")),
+    ("capability proof", ("capability proof", "exact capability proof")),
+    ("mcp unavailable reason", (
+        "why mcp was unavailable",
+        "mcp unavailable",
+        "mcp unavailability",
+    )),
+)
+
+
+def _collect_observed_text(
+    report_text: str,
+    action_log: list[dict] | None,
+) -> str:
+    chunks = [report_text or ""]
+    for entry in action_log or []:
+        for key in ("command", "action", "detail", "path", "script"):
+            value = entry.get(key)
+            if value:
+                chunks.append(str(value))
+    return "\n".join(chunks)
+
+
+def _detect_profile_secret_access(text: str) -> list[str]:
+    reasons = []
+    lower = text.lower()
+    if _PROFILE_ACCESS_RE.search(text):
+        reasons.append("report or action log shows profiles.json/profile secret access")
+    for marker in PROFILE_SECRET_MARKERS:
+        if marker == "profiles.json":
+            continue
+        if marker in lower and "do not" not in lower and "must not" not in lower:
+            if any(verb in lower for verb in ("read ", "open ", "inspect ", "cat ", "loaded ")):
+                reasons.append(f"profile secret surface '{marker}' accessed during review")
+    return reasons
+
+
+def _detect_local_script_use(text: str) -> list[str]:
+    reasons = []
+    match = _LOCAL_SCRIPT_RE.search(text)
+    if match:
+        reasons.append(
+            f"local Gitea helper script invoked ({match.group(0).strip()})"
+        )
+    if _LOCAL_ENV_SCRIPT_RE.search(text):
+        reasons.append(
+            "local Gitea script run with GITEA_MCP_CONFIG/GITEA_MCP_PROFILE env overrides"
+        )
+    return reasons
+
+
+def _recovery_proof_missing(text: str) -> list[str]:
+    lower = text.lower()
+    missing = []
+    for label, aliases in _RECOVERY_PROOF_FIELDS:
+        if not any(alias in lower for alias in aliases):
+            missing.append(label)
+    if not _FALLBACK_CLASSIFICATION_RE.search(text):
+        missing.append("fallback classification")
+    if not _RECOVERY_MODE_RE.search(text):
+        missing.append("recovery mode declaration")
+    return missing
+
+
+def assess_reviewer_fallback_report(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+    recovery_mode: bool = False,
+    mcp_available: bool = True,
+    mcp_tools_used: bool | None = None,
+) -> dict[str, Any]:
+    """Fail closed when normal reviewer workflows use local Gitea fallbacks (#324)."""
+    text = _collect_observed_text(report_text, action_log)
+    lower = (report_text or "").lower()
+
+    profile_violations = _detect_profile_secret_access(text)
+    script_violations = _detect_local_script_use(text)
+    violations = profile_violations + script_violations
+
+    if mcp_tools_used is None:
+        mcp_tools_used = bool(_MCP_TOOL_USE_RE.search(report_text or ""))
+    elif mcp_tools_used is False and _MCP_TOOL_USE_RE.search(report_text or ""):
+        mcp_tools_used = True
+
+    reasons: list[str] = []
+    recovery_declared = recovery_mode or bool(_RECOVERY_MODE_RE.search(report_text or ""))
+
+    if violations and mcp_available and not recovery_declared:
+        reasons.extend(violations)
+        if mcp_tools_used:
+            reasons.append(
+                "MCP tools were available and used; local fallback/profile access is forbidden"
+            )
+        else:
+            reasons.append(
+                "MCP path is available; normal review must not use local Gitea fallbacks"
+            )
+
+    if violations and recovery_declared:
+        missing = _recovery_proof_missing(report_text or "")
+        if missing:
+            reasons.append(
+                "recovery-mode fallback missing proof fields: "
+                + ", ".join(missing)
+            )
+
+    if (
+        not violations
+        and recovery_declared
+        and _FALLBACK_CLASSIFICATION_RE.search(report_text or "")
+        and not recovery_mode
+    ):
+        missing = _recovery_proof_missing(report_text or "")
+        if missing:
+            reasons.append(
+                "report claims local fallback but recovery proof is incomplete: "
+                + ", ".join(missing)
+            )
+
+    proven = not reasons
+    blocked = bool(reasons) and not recovery_declared or any(
+        "recovery-mode fallback missing" in r or "recovery proof is incomplete" in r
+        for r in reasons
+    )
+    return {
+        "proven": proven,
+        "block": blocked or (bool(reasons) and mcp_available and not recovery_declared),
+        "downgraded": bool(reasons) and not blocked,
+        "recovery_mode": recovery_declared,
+        "mcp_available": mcp_available,
+        "mcp_tools_used": mcp_tools_used,
+        "profile_violations": profile_violations,
+        "script_violations": script_violations,
+        "reasons": reasons,
+        "safe_next_action": (
+            "use MCP reviewer tools only; do not read profiles.json or run local Gitea scripts"
+            if reasons and not recovery_declared
+            else (
+                "complete recovery-mode fallback proof before claiming local fallback"
+                if reasons
+                else "proceed with MCP tools"
+            )
+        ),
+    }

tests/test_final_report_validator.py

@@ -0,0 +1,273 @@
+"""Tests for composable final-report validator (#327)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import (  # noqa: E402
+    FINAL_REPORT_TASK_KINDS,
+    assess_final_report_validator,
+    validator_finding,
+)
+
+
+def _review_handoff(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: review PR #203",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: reviewer",
+        "- Identity: sysadmin / prgs-reviewer",
+        "- Issue/PR: #182 / PR #203",
+        "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Files changed: review_proofs.py",
+        "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Mutations: review only",
+        "- Workspace mutations: none",
+        "- File edits by reviewer: none",
+        "- Worktree/index mutations: none",
+        "- Git ref mutations: git fetch prgs master",
+        "- MCP/Gitea mutations: review submitted",
+        "- Current status: PR open",
+        "- Blockers: none",
+        "- Next: await author",
+        "- Safety: no self-review; no self-merge",
+        "- Selected PR: #203",
+        "- Reviewer eligibility: eligible",
+        "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Worktree path: branches/review-203",
+        "- Worktree dirty: clean",
+        "- Scratch worktree used: yes (branches/review-203)",
+        "- Unrelated local mutations: none",
+        "- Review decision: request_changes",
+        "- Merge result: not attempted",
+        "- Linked issue: #182",
+        "- Linked issue status: open",
+        "- Cleanup status: none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        text = text.replace(f"- {key}:", f"- {key}: {value}")
+    return text
+
+
+def _reconcile_handoff(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: reconcile already-landed PR #99",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: reconciler",
+        "- Identity: sysadmin / prgs-author",
+        "- Selected PR: #99",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- Linked issue: #98",
+        "- Linked issue live status: not verified in this session",
+        "- File edits by reconciler: none",
+        "- Git ref mutations: git fetch prgs master",
+        "- Reconciliation mutations: PR comment posted",
+        "- Current status: reconciliation complete",
+        "- Safe next action: none",
+        "- No review/merge confirmation: confirmed",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        if f"- {key}:" in text:
+            text = text.replace(f"- {key}:", f"- {key}: {value}")
+    return text
+
+
+class TestValidatorFinding(unittest.TestCase):
+    def test_finding_shape(self):
+        finding = validator_finding(
+            "reviewer.test",
+            "block",
+            "Validation",
+            "missing proof",
+            "repair validation proof",
+        )
+        self.assertEqual(finding["rule_id"], "reviewer.test")
+        self.assertEqual(finding["severity"], "block")
+        self.assertEqual(finding["field"], "Validation")
+
+
+class TestReviewPrRules(unittest.TestCase):
+    def test_clean_reviewer_report_passes(self):
+        result = assess_final_report_validator(
+            _review_handoff(),
+            "review_pr",
+            linked_issue_lock={"issue_number": 182, "pr_number": 203},
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(result["blocked"])
+        self.assertEqual(result["findings"], [])
+
+    def test_legacy_workspace_mutations_blocks(self):
+        report = _review_handoff() + "\n- Workspace mutations: none"
+        result = assess_final_report_validator(
+            report,
+            "review_pr",
+            mutations_observed=True,
+        )
+        self.assertEqual(result["grade"], "blocked")
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.legacy_workspace_mutations", rule_ids)
+
+    def test_vague_mutations_none_blocks(self):
+        report = _review_handoff().replace(
+            "- Mutations: review only",
+            "- Mutations: none",
+        )
+        result = assess_final_report_validator(
+            report,
+            "review_pr",
+            mutations_observed=True,
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reviewer.vague_mutations_none" for f in result["findings"])
+        )
+
+    def test_bare_pytest_without_cwd_downgrades(self):
+        report = _review_handoff().replace(
+            "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+            "- Validation: pytest passed",
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertEqual(result["grade"], "downgraded")
+        self.assertTrue(
+            any(f["rule_id"] == "reviewer.validation_command_proof" for f in result["findings"])
+        )
+
+    def test_linked_issue_mismatch_blocks(self):
+        report = _review_handoff().replace("- Linked issue: #182", "- Linked issue: #999")
+        result = assess_final_report_validator(
+            report,
+            "review_pr",
+            linked_issue_lock={"issue_number": 182, "pr_number": 203},
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reviewer.linked_issue_mismatch" for f in result["findings"])
+        )
+
+    def test_approval_after_full_suite_failure_blocks_without_baseline(self):
+        report = (
+            _review_handoff()
+            .replace("- Review decision: request_changes", "- Review decision: approve")
+            + "\nSubmitted 'approve' after full suite failed with 3 failed tests."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.baseline_on_full_suite_failure"
+                for f in result["findings"]
+            )
+        )
+
+    def test_same_as_master_without_baseline_blocks(self):
+        report = _review_handoff() + "\nFailures are same as master baseline."
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reviewer.main_checkout_baseline" for f in result["findings"])
+        )
+
+    def test_already_landed_eligible_wording_blocks(self):
+        report = (
+            _review_handoff()
+            + "\nPR is already landed and eligible for review next."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.already_landed_eligible_wording"
+                for f in result["findings"]
+            )
+        )
+
+    def test_git_fetch_must_not_be_readonly_only(self):
+        report = (
+            _review_handoff()
+            .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
+            + "\n- Read-only diagnostics: git fetch prgs master"
+        )
+        result = assess_final_report_validator(
+            report,
+            "review_pr",
+            action_log=[{"command": "git fetch prgs master", "performed": True}],
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reviewer.git_fetch_readonly" for f in result["findings"])
+        )
+
+
+class TestReconciliationRules(unittest.TestCase):
+    def test_clean_reconcile_report_passes(self):
+        result = assess_final_report_validator(
+            _reconcile_handoff(),
+            "reconcile_already_landed",
+        )
+        self.assertEqual(result["grade"], "A")
+
+    def test_stale_author_field_blocks(self):
+        report = _reconcile_handoff() + "\n- PR number opened: #12"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                for f in result["findings"]
+            )
+        )
+
+    def test_eligible_wording_for_landed_pr_blocks(self):
+        report = _reconcile_handoff() + "\nPR already landed and eligible for merge."
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+
+    def test_linked_issue_status_without_live_proof_downgrades(self):
+        report = _reconcile_handoff().replace(
+            "- Linked issue live status: not verified in this session",
+            "- Linked issue live status: closed",
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "downgraded")
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.linked_issue_live_status" for f in result["findings"])
+        )
+
+    def test_inventory_complete_requires_pagination_proof(self):
+        report = _reconcile_handoff() + "\nInventory complete for open PRs."
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "downgraded")
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.inventory_pagination_proof" for f in result["findings"])
+        )
+
+
+class TestEntryPoint(unittest.TestCase):
+    def test_unknown_task_kind_blocks(self):
+        result = assess_final_report_validator("report", "unknown_mode")
+        self.assertTrue(result["blocked"])
+        self.assertEqual(result["findings"][0]["rule_id"], "shared.unknown_task_kind")
+
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_final_report_validator as exported
+
+        self.assertTrue(callable(exported))
+        result = exported(_review_handoff(), "review_pr")
+        self.assertIn("grade", result)
+
+    def test_supported_task_kinds_include_review_and_reconcile(self):
+        self.assertIn("review_pr", FINAL_REPORT_TASK_KINDS)
+        self.assertIn("reconcile_already_landed", FINAL_REPORT_TASK_KINDS)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_linked_issue_consistency.py

@@ -0,0 +1,118 @@
+"""Tests for linked-issue consistency verifier (#314).
+
+Reviewer reports must verify the linked issue live for the selected PR;
+stale issue numbers from a previous PR must not leak into the final
+report, and linked-issue status must never be claimed without live proof.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_linked_issue_consistency  # noqa: E402
+
+
+class TestCorrectLinkedIssue(unittest.TestCase):
+    def test_matching_linked_issue_passes(self):
+        report = (
+            "Selected PR: 280\n"
+            "Linked issue status: Issue #275 open, closes on merge\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_multiple_linked_issues_all_reported_passes(self):
+        report = (
+            "Linked issue status: Issue #275 and Issue #276 both open\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275, 276]
+        )
+        self.assertTrue(result["complete"])
+
+    def test_multiple_linked_issues_one_missing_fails(self):
+        report = "Linked issue status: Issue #275 open\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275, 276]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("276" in r for r in result["reasons"])
+        )
+
+
+class TestStaleIssueLeak(unittest.TestCase):
+    def test_stale_issue_in_status_field_fails(self):
+        # #314 observed behavior: PR #280 links Issue #275 but the report
+        # carries Issue #260 from the previous PR.
+        report = "Linked issue status: Issue #260 open\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(any("260" in r for r in result["reasons"]))
+
+    def test_stale_issue_in_diagnostics_fails(self):
+        report = (
+            "Linked issue status: Issue #275 open\n"
+            "Read-only diagnostics: viewed Issue #260 status\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("stale" in r.lower() and "260" in r for r in result["reasons"])
+        )
+
+    def test_selected_pr_number_mention_is_not_stale(self):
+        # "PR #280" mentions must not be confused with issue mentions.
+        report = (
+            "Selected PR: 280\n"
+            "Validation: ran tests at PR #280 head\n"
+            "Linked issue status: Issue #275 open\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertTrue(result["complete"])
+
+
+class TestMissingProof(unittest.TestCase):
+    def test_status_claim_without_live_proof_fails(self):
+        report = "Linked issue status: Issue #275 open\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=None
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("live proof" in r.lower() for r in result["reasons"])
+        )
+
+    def test_unverified_wording_without_proof_passes(self):
+        report = (
+            "Linked issue status: not verified in this session\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=None
+        )
+        self.assertTrue(result["complete"])
+
+    def test_missing_field_fails(self):
+        report = "Selected PR: 280\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("linked issue status" in r.lower() for r in result["reasons"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -118,12 +118,30 @@ def test_start_issue_template_references_work_issue_workflow():
     assert "workflows/work-issue.md" in text
 
 
+def test_reviewer_fallback_verifier_exported():
+    from review_proofs import assess_reviewer_fallback_report
+
+    assert callable(assess_reviewer_fallback_report)
+
+
+def test_final_report_validator_exported():
+    from review_proofs import assess_final_report_validator
+
+    assert callable(assess_final_report_validator)
+
+
 def test_create_issue_final_report_verifier_exported():
     from review_proofs import assess_create_issue_final_report
 
     assert callable(assess_create_issue_final_report)
 
 
+def test_prior_blocker_skip_verifier_exported():
+    from review_proofs import assess_prior_blocker_skip_proof
+
+    assert callable(assess_prior_blocker_skip_proof)
+
+
 def test_non_mergeable_skip_verifier_exported():
     from review_proofs import assess_non_mergeable_skip_proof
 

tests/test_reconciliation_inventory_proof.py

@@ -0,0 +1,150 @@
+"""Tests for reconciliation PR-inventory pagination proof (#308).
+
+Already-landed reconciliation reports may not claim a complete queue
+scan (or that all already-landed PRs were found) unless the report
+carries pagination proof fields and the per-page evidence proves the
+final page was reached or the page limit was not hit.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_reconciliation_inventory_proof  # noqa: E402
+
+
+def _report(claim=True, fields=True, pages="1", counts="12",
+            final_proof="is_final_page=true, has_more=false"):
+    lines = []
+    if claim:
+        lines.append(
+            "Inventory scope: complete queue scan; all already-landed "
+            "PRs were found."
+        )
+    if fields:
+        lines += [
+            "- Requested page size: 50",
+            f"- Pages fetched: {pages}",
+            f"- Returned PR count per page: {counts}",
+            f"- Final page proof: {final_proof}",
+        ]
+    return "\n".join(lines) + "\n"
+
+
+def _page(page, returned, per_page=50, has_more=False, final=True,
+          inventory_complete=None):
+    d = {
+        "page": page,
+        "per_page": per_page,
+        "returned_count": returned,
+        "has_more": has_more,
+        "is_final_page": final,
+    }
+    if inventory_complete is not None:
+        d["inventory_complete"] = inventory_complete
+    return d
+
+
+class TestCompleteScanProven(unittest.TestCase):
+    def test_first_page_below_limit_passes(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(),
+            pagination_evidence=[_page(1, 12)],
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_multiple_pages_to_final_passes(self):
+        report = _report(
+            pages="3", counts="50, 50, 12",
+            final_proof="has_more=false on page 3",
+        )
+        evidence = [
+            _page(1, 50, has_more=True, final=False),
+            _page(2, 50, has_more=True, final=False),
+            _page(3, 12),
+        ]
+        result = assess_reconciliation_inventory_proof(
+            report, pagination_evidence=evidence
+        )
+        self.assertTrue(result["complete"])
+
+    def test_no_completeness_claim_needs_no_proof(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(claim=False, fields=False),
+            pagination_evidence=None,
+        )
+        self.assertTrue(result["complete"])
+
+
+class TestUnprovenScanFails(unittest.TestCase):
+    def test_first_page_exactly_at_limit_fails(self):
+        # 50 returned with per_page 50 and no final-page signal: the page
+        # limit was hit, so completeness is unproven.
+        evidence = [_page(1, 50, has_more=True, final=False)]
+        result = assess_reconciliation_inventory_proof(
+            _report(counts="50", final_proof="none"),
+            pagination_evidence=evidence,
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("final page" in r.lower() for r in result["reasons"])
+        )
+
+    def test_missing_pagination_evidence_fails(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(), pagination_evidence=None
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pagination" in r.lower() for r in result["reasons"])
+        )
+
+    def test_missing_report_fields_fails(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(fields=False),
+            pagination_evidence=[_page(1, 12)],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("requested page size" in r.lower()
+                for r in result["reasons"])
+        )
+
+    def test_page_count_mismatch_fails(self):
+        # Report claims 1 page; evidence shows 2 were fetched.
+        evidence = [
+            _page(1, 50, has_more=True, final=False),
+            _page(2, 3),
+        ]
+        result = assess_reconciliation_inventory_proof(
+            _report(pages="1", counts="50"),
+            pagination_evidence=evidence,
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pages fetched" in r.lower() for r in result["reasons"])
+        )
+
+    def test_per_page_count_mismatch_fails(self):
+        evidence = [_page(1, 12)]
+        result = assess_reconciliation_inventory_proof(
+            _report(counts="11"),
+            pagination_evidence=evidence,
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("count per page" in r.lower() for r in result["reasons"])
+        )
+
+    def test_evidence_page_missing_metadata_fails(self):
+        result = assess_reconciliation_inventory_proof(
+            _report(),
+            pagination_evidence=[{"page": 1}],
+        )
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_review_proofs.py

@@ -941,13 +941,18 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("Safety", result["missing_fields"])
 
     def test_review_role_requires_review_fields(self):
-        result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
+        # Issue #320: review handoffs must not carry the legacy
+        # 'Workspace mutations' line, so strip it from the shared base.
+        review_base = "\n".join(
+            line for line in self.BASE_HANDOFF.splitlines()
+            if not line.startswith("- Workspace mutations:"))
+        result = assess_controller_handoff(review_base, role="review")
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("Pinned reviewed head", result["missing_fields"])
         self.assertIn("Worktree path", result["missing_fields"])
         self.assertIn("Merge result", result["missing_fields"])
 
-        complete = self.BASE_HANDOFF + "\n" + "\n".join([
+        complete = review_base + "\n" + "\n".join([
             "- Selected PR: #999",
             "- Reviewer eligibility: passed",
             "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
@@ -959,6 +964,15 @@ class TestControllerHandoff(unittest.TestCase):
             "- Merge result: merged",
             "- Linked issue status: closed",
             "- Cleanup status: branch deleted",
+            "- File edits by reviewer: none",
+            "- Worktree/index mutations: created and removed review worktree",
+            "- Git ref mutations: git fetch prgs",
+            "- MCP/Gitea mutations: none",
+            "- Review mutations: one approve on #999",
+            "- Merge mutations: PR #999 merged",
+            "- Cleanup mutations: removed review worktree",
+            "- External-state mutations: none",
+            "- Read-only diagnostics: git log, git diff",
         ])
         result = assess_controller_handoff(complete, role="review")
         self.assertEqual(result["verdict"], "complete")
@@ -1081,6 +1095,144 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertEqual(res2["verdict"], "complete")
 
 
+class TestReviewHandoffPreciseMutationCategories(unittest.TestCase):
+    """Issue #320: reviewer handoffs drop legacy 'Workspace mutations'."""
+
+    REVIEW_BASE = "\n".join([
+        "## Controller Handoff",
+        "",
+        "- Task: review PR #999",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: reviewer",
+        "- Identity: jcwalker3 / prgs-reviewer",
+        "- Issue/PR: PR #999",
+        "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Files changed: review_proofs.py",
+        "- Validation: 700 passed, 6 skipped",
+        "- Mutations: one review submitted",
+        "- Current status: review complete",
+        "- Blockers: none",
+        "- Next: controller decides",
+        "- Safety: no self-review; no self-merge; no secrets",
+        "- Selected PR: #999",
+        "- Reviewer eligibility: passed",
+        "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Worktree path: /repo/branches/review-pr-999",
+        "- Worktree dirty: no",
+        "- Scratch worktree used: yes (/repo/branches/review-pr-999)",
+        "- Unrelated local mutations: none",
+        "- Review decision: approve",
+        "- Merge result: none",
+        "- Linked issue status: open",
+        "- Cleanup status: worktree removed",
+    ])
+
+    PRECISE_CATEGORIES = "\n".join([
+        "- File edits by reviewer: none",
+        "- Worktree/index mutations: created and removed review worktree",
+        "- Git ref mutations: git fetch prgs",
+        "- MCP/Gitea mutations: none",
+        "- Review mutations: one approve on #999",
+        "- Merge mutations: none",
+        "- Cleanup mutations: removed review worktree",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: git log, git diff",
+    ])
+
+    def _complete_handoff(self, **overrides):
+        text = self.REVIEW_BASE + "\n" + self.PRECISE_CATEGORIES
+        for old, new in overrides.items():
+            text = text.replace(old, new)
+        return text
+
+    def test_review_handoff_with_precise_categories_is_complete(self):
+        result = assess_controller_handoff(
+            self._complete_handoff(), role="review")
+        self.assertEqual(result["verdict"], "complete")
+        self.assertFalse(result["downgraded"])
+
+    def test_review_handoff_rejects_legacy_workspace_mutations_none(self):
+        text = self._complete_handoff() + "\n- Workspace mutations: none"
+        result = assess_controller_handoff(text, role="review")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertTrue(result["downgraded"])
+        self.assertTrue(any(
+            "workspace mutations" in reason.lower()
+            for reason in result["reasons"]
+        ))
+
+    def test_review_handoff_rejects_legacy_workspace_mutations_any_value(self):
+        text = (self._complete_handoff()
+                + "\n- Workspace mutations: edited review_proofs.py")
+        result = assess_controller_handoff(text, role="review")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertTrue(any(
+            "workspace mutations" in reason.lower()
+            for reason in result["reasons"]
+        ))
+
+    def test_review_handoff_missing_precise_categories_is_incomplete(self):
+        result = assess_controller_handoff(self.REVIEW_BASE, role="review")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("File edits by reviewer", result["missing_fields"])
+        self.assertIn("Worktree/index mutations", result["missing_fields"])
+        self.assertIn("Read-only diagnostics", result["missing_fields"])
+
+    def test_review_handoff_does_not_require_workspace_mutations(self):
+        result = assess_controller_handoff(
+            self._complete_handoff(), role="review")
+        self.assertNotIn("Workspace mutations", result["missing_fields"])
+
+    def test_request_changes_handoff_with_precise_categories_is_complete(self):
+        text = self._complete_handoff(**{
+            "- Review decision: approve": "- Review decision: request-changes",
+            "- Review mutations: one approve on #999":
+                "- Review mutations: one request_changes on #999",
+        })
+        result = assess_controller_handoff(text, role="review")
+        self.assertEqual(result["verdict"], "complete")
+
+    def test_merge_handoff_with_precise_categories_is_complete(self):
+        text = self._complete_handoff(**{
+            "- Merge result: none": "- Merge result: merged",
+            "- Merge mutations: none": "- Merge mutations: PR #999 merged",
+        })
+        result = assess_controller_handoff(text, role="review")
+        self.assertEqual(result["verdict"], "complete")
+
+    def test_fetch_only_handoff_with_precise_categories_is_complete(self):
+        text = self._complete_handoff(**{
+            "- Review decision: approve": "- Review decision: none",
+            "- Review mutations: one approve on #999": "- Review mutations: none",
+            "- Worktree/index mutations: created and removed review worktree":
+                "- Worktree/index mutations: none",
+            "- Cleanup mutations: removed review worktree":
+                "- Cleanup mutations: none",
+            "- Mutations: one review submitted": "- Mutations: fetch only",
+        })
+        result = assess_controller_handoff(text, role="review")
+        self.assertEqual(result["verdict"], "complete")
+
+    def test_blocked_handoff_with_precise_categories_is_complete(self):
+        text = self._complete_handoff(**{
+            "- Blockers: none": "- Blockers: infra_stop (registry unreachable)",
+            "- Review decision: approve": "- Review decision: none",
+            "- Review mutations: one approve on #999": "- Review mutations: none",
+        })
+        result = assess_controller_handoff(text, role="review")
+        self.assertEqual(result["verdict"], "complete")
+
+    def test_author_role_still_requires_workspace_mutations(self):
+        # Non-review roles keep the legacy contract until their own issues
+        # migrate them.
+        author = TestControllerHandoff.BASE_HANDOFF
+        stripped = "\n".join(
+            line for line in author.splitlines()
+            if not line.startswith("- Workspace mutations:"))
+        result = assess_controller_handoff(stripped, role="author")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("Workspace mutations", result["missing_fields"])
+
 
 class TestReviewMutationFinalReport(unittest.TestCase):
     """Final reports must list exactly one live review mutation."""

tests/test_reviewer_blocker_skip.py

@@ -0,0 +1,139 @@
+"""Tests for prior-blocker skip proof verifier (#318)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_blocker_skip import assess_prior_blocker_skip_proof  # noqa: E402
+
+HEAD_CURRENT = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+HEAD_BLOCKER = "1111111111111111111111111111111111111111"
+
+
+def _good_blocker_skip_report(pr_number: int) -> str:
+    return "\n".join([
+        f"Skipped PR #{pr_number} due to prior REQUEST_CHANGES.",
+        f"- PR number: #{pr_number}",
+        f"- Current head SHA: {HEAD_CURRENT}",
+        "- Blocking review decision: request_changes",
+        f"- Blocking review head SHA: {HEAD_CURRENT}",
+        "- Head unchanged since blocker",
+        "- Reason remains blocked: unresolved review feedback at same head",
+        "- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
+    ])
+
+
+class TestPriorBlockerSkipProof(unittest.TestCase):
+    def test_no_skips_passes(self):
+        result = assess_prior_blocker_skip_proof(
+            "Selected PR #281 for review.",
+            skipped_prs=[],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_blocked_unchanged_with_live_proof_passes(self):
+        report = _good_blocker_skip_report(280)
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_head_changed_after_blocker_blocks_skip(self):
+        result = assess_prior_blocker_skip_proof(
+            _good_blocker_skip_report(280),
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_BLOCKER,
+                "head_changed_since_blocker": True,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head changed" in r for r in result["reasons"]))
+
+    def test_missing_live_proof_blocks(self):
+        report = _good_blocker_skip_report(280).replace(
+            "- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
+            "",
+        )
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("live blocker proof" in r for r in result["reasons"]))
+
+    def test_blocker_unverified_classification_passes(self):
+        report = "\n".join([
+            "Skipped PR #280.",
+            "Classification: BLOCKER_STATUS_UNVERIFIED",
+            "Review feedback could not be fetched live in this session.",
+        ])
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "blocking_decision": "request_changes",
+                "blocker_verified": False,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unverified_without_classification_blocks(self):
+        result = assess_prior_blocker_skip_proof(
+            "Skipped PR #280 based on memory from last session.",
+            skipped_prs=[{
+                "pr_number": 280,
+                "blocking_decision": "request_changes",
+                "blocker_verified": False,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("BLOCKER_STATUS_UNVERIFIED" in r for r in result["reasons"]))
+
+    def test_missing_blocking_head_sha_blocks(self):
+        report = _good_blocker_skip_report(280).replace(
+            f"- Blocking review head SHA: {HEAD_CURRENT}",
+            "",
+        )
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_prior_blocker_skip_proof as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_fallback.py

@@ -0,0 +1,112 @@
+"""Tests for reviewer local-fallback verifier (#324)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_fallback import assess_reviewer_fallback_report  # noqa: E402
+
+
+class TestReviewerFallbackNormalMode(unittest.TestCase):
+    def test_clean_mcp_report_passes(self):
+        report = (
+            "Used gitea_list_prs and gitea_view_pr via MCP reviewer namespace.\n"
+            "Capability evidence: gitea_resolve_task_capability(review_pr)."
+        )
+        result = assess_reviewer_fallback_report(report, mcp_available=True)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_profiles_json_access_blocks(self):
+        report = (
+            "Inspected ~/.config/gitea-tools/profiles.json for reviewer credentials.\n"
+            "Then used gitea_view_pr."
+        )
+        result = assess_reviewer_fallback_report(report, mcp_available=True)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("profiles.json" in r for r in result["reasons"]))
+
+    def test_local_list_prs_script_blocks_when_mcp_available(self):
+        report = (
+            "Ran `python list_prs.py --remote prgs` with GITEA_MCP_CONFIG set.\n"
+            "MCP tools were also available."
+        )
+        result = assess_reviewer_fallback_report(
+            report,
+            mcp_available=True,
+            action_log=[{"command": "GITEA_MCP_PROFILE=prgs-reviewer python list_prs.py"}],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(result["script_violations"])
+
+    def test_action_log_profile_access_blocks(self):
+        result = assess_reviewer_fallback_report(
+            "Review complete via MCP.",
+            action_log=[{"path": "/Users/me/.config/gitea-tools/profiles.json", "action": "read"}],
+            mcp_available=True,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+
+class TestReviewerFallbackRecoveryMode(unittest.TestCase):
+    def _recovery_report(self, **extra):
+        base = "\n".join([
+            "Recovery mode: MCP tools unavailable in this session.",
+            "Local fallback: ran python view_pr.py after capability proof.",
+            "Identity proof: sysadmin / prgs-reviewer",
+            "Profile proof: prgs-reviewer",
+            "Repo proof: Scaled-Tech-Consulting/Gitea-Tools",
+            "Capability proof: gitea.pr.review allowed for recovery command",
+            "Why MCP was unavailable: reviewer MCP namespace failed to start",
+        ])
+        return base + ("\n" + extra.get("append", "") if extra.get("append") else "")
+
+    def test_recovery_fallback_with_proof_passes(self):
+        result = assess_reviewer_fallback_report(
+            self._recovery_report(),
+            action_log=[{"command": "python view_pr.py --remote prgs"}],
+            mcp_available=False,
+            recovery_mode=True,
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_recovery_fallback_without_proof_blocks(self):
+        report = (
+            "Recovery mode engaged.\n"
+            "Local fallback: python list_prs.py\n"
+        )
+        result = assess_reviewer_fallback_report(
+            report,
+            mcp_available=False,
+            recovery_mode=True,
+            action_log=[{"command": "python list_prs.py"}],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("recovery" in r.lower() for r in result["reasons"])
+        )
+
+    def test_mcp_unavailable_allows_recovery_attempt(self):
+        result = assess_reviewer_fallback_report(
+            self._recovery_report(),
+            mcp_available=False,
+            recovery_mode=True,
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestReviewerFallbackExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reviewer_fallback_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()