fix: resolve conflicts for PR #352

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

gitea_mcp_server.py

@@ -1764,6 +1764,51 @@ def record_live_review_mutation(pr_number: int, action: str, review_id: int | No
     _save_review_decision_lock(lock)
 
 
+def terminal_review_hard_stop_reasons(pr_number: int, operation: str) -> list[str]:
+    """Session hard-stop after a terminal live review mutation (#332).
+
+    After a terminal verdict is consumed in this run, the only permitted
+    continuation is the merge sequence for the same PR that was approved.
+    A REQUEST_CHANGES (or an approval of a different PR) blocks every
+    further review/mark-ready/merge mutation. An operator-approved
+    correction (#211) re-opens the review path only — never a cross-PR
+    merge.
+
+    *operation* is one of 'merge', 'mark_ready', or 'review'.
+    Returns [] when the operation may proceed.
+    """
+    lock = _load_review_decision_lock()
+    if lock is None:
+        return []
+    terminals = [
+        m for m in (lock.get("live_mutations") or [])
+        if m.get("action") in _TERMINAL_REVIEW_ACTIONS
+    ]
+    if not terminals:
+        return []
+    last = terminals[-1]
+    if (
+        operation == "merge"
+        and last.get("action") == "approve"
+        and last.get("pr_number") == pr_number
+    ):
+        return []
+    if operation in ("mark_ready", "review") and lock.get("correction_authorized"):
+        return []
+    if last.get("action") == "approve":
+        guidance = (
+            f"only the merge sequence for approved PR "
+            f"#{last.get('pr_number')} may continue"
+        )
+    else:
+        guidance = "the session must stop and produce a final report"
+    return [
+        "terminal review mutation already consumed in this run "
+        f"({last.get('action')} on PR #{last.get('pr_number')}); {guidance} "
+        "(fail closed, #332)"
+    ]
+
+
 # Patterns scrubbed from any surfaced error text so a credential can never leak.
 _SECRET_PREFIXES = ("token ", "Basic ")
 
@@ -2179,6 +2224,9 @@ def gitea_mark_final_review_decision(
         }
     org = resolved_org
     repo = resolved_repo
+    hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
+    if hard_stop:
+        return {"marked_ready": False, "reasons": hard_stop}
     if lock.get("live_mutations") and not lock.get("correction_authorized"):
         return {
             "marked_ready": False,
@@ -2196,6 +2244,33 @@ def gitea_mark_final_review_decision(
                 f"{sorted(_REVIEW_ACTIONS)}"
             ],
         }
+    if action == "request_changes":
+        # Duplicate request-changes suppression (#332): an unresolved
+        # REQUEST_CHANGES at the current head must not be duplicated.
+        feedback = gitea_get_pr_review_feedback(
+            pr_number, remote=remote, org=org, repo=repo)
+        if not feedback.get("success"):
+            return {
+                "marked_ready": False,
+                "reasons": [
+                    "could not verify existing request-changes state for "
+                    f"PR #{pr_number}; refusing to submit a possibly "
+                    "duplicate REQUEST_CHANGES (fail closed, #332)"
+                ],
+            }
+        if (
+            feedback.get("has_blocking_change_requests")
+            and not feedback.get("review_feedback_stale")
+        ):
+            return {
+                "marked_ready": False,
+                "reasons": [
+                    f"PR #{pr_number} already has an unresolved "
+                    "REQUEST_CHANGES at the current head SHA; do not "
+                    "duplicate the request-changes review (fail closed, "
+                    "#332)"
+                ],
+            }
     lock["final_review_decision_ready"] = True
     lock["ready_pr_number"] = pr_number
     lock["ready_action"] = action
@@ -2765,6 +2840,14 @@ def gitea_merge_pr(
         )
         return result
 
+    # Gate 2b — session hard-stop after a terminal review mutation (#332):
+    # merge may only continue for the same PR that was just approved.
+    # Local in-process check; zero API calls.
+    hard_stop = terminal_review_hard_stop_reasons(pr_number, "merge")
+    if hard_stop:
+        reasons.extend(hard_stop)
+        return result
+
     # Gate 3 — reuse #14 eligibility (identity + profile + merge-allowed +
     # author + self-merge block + open + mergeable/unknown fail-closed).
     # Read-only GETs only.

review_proofs.py

@@ -1052,6 +1052,125 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
     return None
 
 
+# ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
+#
+# A worktree reset/checkout/clean is a workspace mutation even when the
+# reviewer edited no files by hand. Final reports must therefore never say
+# "Workspace mutations: none" once a worktree mutation occurred, and every
+# observed worktree / git ref mutation must appear under its precise
+# category field.
+
+WORKTREE_MUTATION_MARKERS = (
+    "reset",
+    "checkout",
+    "switch",
+    "restore",
+    "clean",
+    "stash",
+    "merge",
+    "rebase",
+    "cherry-pick",
+    "worktree add",
+    "worktree remove",
+    "worktree prune",
+)
+
+GIT_REF_MUTATION_MARKERS = (
+    "fetch",
+    "pull",
+)
+
+
+def _report_labeled_fields(report_text):
+    """Return {lowercase label: value} for every 'Label: value' line."""
+    fields = {}
+    for line in (report_text or "").splitlines():
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            k, v = stripped.split(":", 1)
+            fields[k.strip().lower()] = v.strip()
+    return fields
+
+
+def _field_claims_none(fields, label):
+    """True when *label* is present and its value is empty or 'none...'."""
+    value = fields.get(label)
+    if value is None:
+        return False
+    value = value.strip().lower()
+    return not value or value.startswith("none")
+
+
+def _classify_git_commands(observed_commands):
+    """Split observed git commands into worktree and ref mutations."""
+    worktree_cmds = []
+    ref_cmds = []
+    for cmd in observed_commands or []:
+        lowered = " ".join((cmd or "").lower().split())
+        if "git" not in lowered:
+            continue
+        if any(marker in lowered for marker in WORKTREE_MUTATION_MARKERS):
+            worktree_cmds.append(cmd)
+        elif any(marker in lowered for marker in GIT_REF_MUTATION_MARKERS):
+            ref_cmds.append(cmd)
+    return worktree_cmds, ref_cmds
+
+
+def assess_workspace_mutation_consistency(report_text, observed_commands=None):
+    """#313: reject 'Workspace mutations: none' after worktree mutations.
+
+    *observed_commands* is the optional list of git commands the workflow
+    actually ran. Even without it, a report that itself lists a non-none
+    ``Worktree mutations`` value while claiming ``Workspace mutations:
+    none`` is internally contradictory and fails.
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed on any
+    contradiction or unreported observed mutation.
+    """
+    fields = _report_labeled_fields(report_text)
+    worktree_cmds, ref_cmds = _classify_git_commands(observed_commands)
+
+    reported_worktree = fields.get("worktree mutations")
+    reported_worktree_nonnone = bool(
+        reported_worktree and not reported_worktree.strip().lower().startswith("none")
+    )
+
+    reasons = []
+
+    workspace_none = _field_claims_none(fields, "workspace mutations")
+    if workspace_none and (worktree_cmds or reported_worktree_nonnone):
+        detail = worktree_cmds[0] if worktree_cmds else reported_worktree
+        reasons.append(
+            "report claims 'Workspace mutations: none' but worktree "
+            f"mutations occurred ({detail}); use precise categories such as "
+            "'File edits by reviewer: none' plus 'Worktree mutations: <cmd>'"
+        )
+
+    if worktree_cmds and (
+        "worktree mutations" not in fields
+        or _field_claims_none(fields, "worktree mutations")
+    ):
+        reasons.append(
+            "observed worktree mutation not reported under 'Worktree "
+            f"mutations': {worktree_cmds[0]}"
+        )
+
+    if ref_cmds and (
+        "git ref mutations" not in fields
+        or _field_claims_none(fields, "git ref mutations")
+    ):
+        reasons.append(
+            "observed git ref mutation not reported under 'Git ref "
+            f"mutations': {ref_cmds[0]}"
+        )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
                          justification=None):
     """Assess reviewer/author role separation for blind queue workflows.
@@ -1232,7 +1351,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
                        report_text=None, review_decision_lock=None,
                        controller_handoff=None, capability_proof=None,
                        sweep_proof=None, worktree_proof=None,
-                       validation_environment=None):
+                       validation_environment=None,
+                       queue_ordering=None,
+                       baseline_validation=None, project_root=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -1275,6 +1396,44 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reasons": [],
             "violations": [],
         }
+    if queue_ordering is None and report_text:
+        queue_ordering_proof = assess_queue_ordering_proof(
+            report_text=report_text,
+        )
+    elif queue_ordering is not None:
+        queue_ordering_proof = assess_queue_ordering_proof(
+            report_text=report_text,
+            queue_ordering=queue_ordering,
+        )
+    else:
+        queue_ordering_proof = {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "violations": [],
+        }
+    queue_status_report = (
+        assess_queue_status_report(report_text)
+        if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
+        else {"proven": True, "block": False, "reasons": [], "violations": []}
+    )
+    proof_wording = (
+        assess_proof_wording(report_text)
+        if report_text
+        else {"proven": True, "block": False, "reasons": [], "violations": []}
+    )
+    if baseline_validation is None and report_text:
+        baseline_validation = assess_reviewer_baseline_validation_proof(
+            report_text=report_text,
+            project_root=project_root,
+        )
+    elif baseline_validation is None:
+        baseline_validation = {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "violations": [],
+        }
 
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
@@ -1409,6 +1568,26 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "validation environment proof missing or failed (#311)"
         )
         downgrade_reasons.extend(validation_env_proof.get("reasons", []))
+    if not queue_ordering_proof.get("proven"):
+        downgrade_reasons.append(
+            "queue ordering proof missing or failed (#321)"
+        )
+        downgrade_reasons.extend(queue_ordering_proof.get("reasons", []))
+    if not proof_wording.get("proven"):
+        downgrade_reasons.append(
+            "unsupported proof wording in final report (#330)"
+        )
+        downgrade_reasons.extend(proof_wording.get("reasons", []))
+    if not queue_status_report.get("proven"):
+        downgrade_reasons.append(
+            "queue-status report violates loaded workflow proof gates (#339)"
+        )
+        downgrade_reasons.extend(queue_status_report.get("reasons", []))
+    if not baseline_validation.get("proven"):
+        downgrade_reasons.append(
+            "reviewer baseline validation proof missing or failed (#325)"
+        )
+        downgrade_reasons.extend(baseline_validation.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -1421,10 +1600,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         and live_state_proven
         and worktree_proven
         and validation_env_proof.get("proven")
+        and queue_ordering_proof.get("proven")
+        and baseline_validation.get("proven")
     )
 
     violations = []
     violations.extend(validation_env_proof.get("violations", []))
+    violations.extend(queue_ordering_proof.get("violations", []))
+    violations.extend(baseline_validation.get("violations", []))
     if merge_performed and not merge_allowed:
         violations.append(
             "merge was performed/claimed although the proofs did not allow "
@@ -1482,6 +1665,21 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "validation_environment_violations": list(
             validation_env_proof.get("violations") or []
         ),
+        "queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
+        "queue_ordering_policy": queue_ordering_proof.get("policy"),
+        "queue_ordering_violations": list(
+            queue_ordering_proof.get("violations") or []
+        ),
+        "proof_wording_proven": bool(proof_wording.get("proven")),
+        "proof_wording_violations": list(proof_wording.get("violations") or []),
+        "queue_status_report_proven": bool(queue_status_report.get("proven")),
+        "queue_status_violations": list(
+            queue_status_report.get("violations") or []
+        ),
+        "baseline_validation_proven": bool(baseline_validation.get("proven")),
+        "baseline_validation_violations": list(
+            baseline_validation.get("violations") or []
+        ),
     }
 
 
@@ -1623,6 +1821,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",)),
@@ -1638,7 +1850,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")),
@@ -1789,6 +2001,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 = []
@@ -3234,6 +3465,340 @@ def assess_validation_environment_proof(
     }
 
 
+
+
+# ---------------------------------------------------------------------------
+# Reviewer queue ordering proof (#321)
+# ---------------------------------------------------------------------------
+
+_QUEUE_SELECTION_CLAIM_RE = re.compile(
+    r"\b(?:next|oldest)\s+eligible\s+pr\b",
+    re.IGNORECASE,
+)
+
+_ORDERING_POLICY_LINE_RE = re.compile(
+    r"(?:queue\s+ordering\s+policy|selection\s+policy|ordering\s+policy)"
+    r"\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+_SKIPPED_PR_LINE_RE = re.compile(
+    r"\bskipped\s+pr\s*#?(\d+)\b",
+    re.IGNORECASE,
+)
+
+
+def _parse_ordering_policy(report_text: str) -> str:
+    for line in (report_text or "").splitlines():
+        match = _ORDERING_POLICY_LINE_RE.search(line)
+        if match:
+            return match.group(1).strip().rstrip(".")
+    return ""
+
+
+def _skipped_pr_numbers(report_text: str) -> set[int]:
+    skipped: set[int] = set()
+    for match in _SKIPPED_PR_LINE_RE.finditer(report_text or ""):
+        try:
+            skipped.add(int(match.group(1)))
+        except ValueError:
+            continue
+    return skipped
+
+
+def assess_queue_ordering_proof(
+    *,
+    report_text: str | None = None,
+    queue_ordering: dict | None = None,
+) -> dict:
+    """#321: reviewer queue selection must prove ordering before claiming next PR."""
+    text = report_text or ""
+    lower = text.lower()
+    proof = queue_ordering or {}
+    reasons: list[str] = []
+    violations: list[str] = []
+
+    claims_selection = bool(_QUEUE_SELECTION_CLAIM_RE.search(text))
+    selected = proof.get("selected_pr_number")
+    if selected is None and "selected pr #" in lower:
+        for token in lower.split():
+            if token.startswith("#") and token[1:].isdigit():
+                selected = int(token[1:])
+                break
+
+    if not claims_selection and selected is None:
+        return {
+            "proven": True,
+            "block": False,
+            "claims_selection": False,
+            "policy": None,
+            "reasons": [],
+            "violations": [],
+            "safe_next_action": "proceed",
+        }
+
+    policy = (proof.get("policy") or _parse_ordering_policy(text)).strip()
+    if not policy:
+        reasons.append(
+            "queue selection claimed without stated ordering policy (#321)"
+        )
+
+    policy_lower = policy.lower()
+    api_numbers: list[int] = []
+    for n in proof.get("api_pr_numbers") or []:
+        if isinstance(n, int):
+            api_numbers.append(n)
+        elif isinstance(n, str) and n.isdigit():
+            api_numbers.append(int(n))
+    skipped_entries = list(proof.get("skipped_earlier") or [])
+    skipped_numbers = {
+        int(entry["pr_number"])
+        for entry in skipped_entries
+        if isinstance(entry, dict) and entry.get("pr_number") is not None
+    }
+    skipped_numbers.update(_skipped_pr_numbers(text))
+
+    if selected is not None and api_numbers:
+        if "oldest" in policy_lower and "number" in policy_lower:
+            earlier_actionable = [
+                n for n in api_numbers
+                if n < int(selected) and n not in skipped_numbers
+            ]
+            for pr_number in earlier_actionable:
+                entry = next(
+                    (
+                        e for e in skipped_entries
+                        if isinstance(e, dict)
+                        and int(e.get("pr_number", -1)) == pr_number
+                    ),
+                    None,
+                )
+                reason = (entry or {}).get("reason", "").strip()
+                if not reason and f"skipped pr #{pr_number}" not in lower:
+                    violations.append(
+                        f"earlier actionable PR #{pr_number} skipped without "
+                        "proof-backed reason (#321)"
+                    )
+                    reasons.append(
+                        f"oldest-first policy requires skip reasoning for "
+                        f"earlier PR #{pr_number} (#321)"
+                    )
+            if api_numbers and api_numbers[0] != min(api_numbers):
+                if (
+                    "api order" not in lower
+                    and "sorted" not in lower
+                    and "newest-first" not in lower
+                    and not skipped_entries
+                ):
+                    reasons.append(
+                        "API response order differs from oldest-first "
+                        "selection but report does not prove explicit sort "
+                        "or skip reasoning (#321)"
+                    )
+
+        if "priority" in policy_lower:
+            override = (proof.get("priority_override") or "").strip()
+            if not override and "priority" not in lower:
+                reasons.append(
+                    "priority-label ordering policy requires override "
+                    "explanation in report (#321)"
+                )
+
+    proven = not reasons and not violations
+    return {
+        "proven": proven,
+        "block": bool(violations),
+        "claims_selection": True,
+        "policy": policy or None,
+        "selected_pr_number": selected,
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": (
+            "state queue ordering policy and proof-backed skip reasoning "
+            "for every earlier actionable PR"
+            if not proven
+            else "proceed"
+        ),
+    }
+
+
+
+# ---------------------------------------------------------------------------
+# Reviewer baseline validation (#325)
+# ---------------------------------------------------------------------------
+
+_TEST_VALIDATION_COMMAND_RE = re.compile(
+    r"(?:^|\s)(?:venv/)?(?:bin/)?(?:python\s+-m\s+)?"
+    r"(?:pytest|make\s+test|npm\s+test|cargo\s+test|go\s+test\b)",
+    re.IGNORECASE,
+)
+
+_PREEXISTING_FAILURE_CLAIM_RE = re.compile(
+    r"(?:\bsame\s+as\s+master\b|"
+    r"\bpre-?existing(?:\s+(?:on\s+)?master)?\b|"
+    r"\bfailures?\s+(?:are|is)\s+pre-?existing\b|"
+    r"\bmaster\s+also\s+fails?\b|"
+    r"\bfull-?suite\s+failures?\s+(?:are|is)\s+pre-?existing\b)",
+    re.IGNORECASE,
+)
+
+_REPORT_VALIDATION_CWD_RE = re.compile(
+    r"(?:working\s+directory|cwd|directory)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").rstrip("/")
+
+
+def _path_under_branches(path: str, project_root: str | None = None) -> bool:
+    """True when *path* is inside the project's ``branches/`` directory."""
+    normalized = _normalize_path(path)
+    if not normalized:
+        return False
+    if "/branches/" in f"{normalized}/":
+        return True
+    if normalized.endswith("/branches"):
+        return True
+    if project_root:
+        root = _normalize_path(project_root)
+        if normalized.startswith(f"{root}/"):
+            rel = normalized[len(root) + 1 :]
+            return rel == "branches" or rel.startswith("branches/")
+    return False
+
+
+def _is_test_validation_command(command: str) -> bool:
+    return bool(_TEST_VALIDATION_COMMAND_RE.search((command or "").strip()))
+
+
+def _is_full_sha(value: str | None) -> bool:
+    return bool(value and _FULL_SHA.match((value or "").strip()))
+
+
+def assess_reviewer_baseline_validation_proof(
+    *,
+    validation_runs: list[dict] | None = None,
+    baseline_proof: dict | None = None,
+    report_text: str | None = None,
+    project_root: str | None = None,
+) -> dict:
+    """#325: reviewer validation and baseline comparison must use ``branches/``.
+
+    *validation_runs* entries: ``command``, ``working_directory`` (or ``cwd``),
+    optional ``project_root``.
+
+    *baseline_proof* keys when claiming pre-existing master failures:
+    ``worktree_path``, ``baseline_target_sha``, ``pr_head_sha``,
+    ``baseline_failures``, ``pr_failures``, ``failure_signatures_match``,
+    ``clean_before``, ``clean_after``.
+    """
+    reasons: list[str] = []
+    violations: list[str] = []
+    text = report_text or ""
+
+    runs = list(validation_runs or [])
+    pending_command = None
+    for raw_line in text.splitlines():
+        line = raw_line.strip().lstrip("-*").strip()
+        lower = line.lower()
+        if lower.startswith("validation command:"):
+            pending_command = line.split(":", 1)[1].strip()
+            continue
+        cwd_match = _REPORT_VALIDATION_CWD_RE.search(line)
+        if cwd_match:
+            cwd = cwd_match.group(1).strip().rstrip(",.;")
+            command = pending_command or line
+            if _is_test_validation_command(command):
+                runs.append(
+                    {
+                        "command": command,
+                        "working_directory": cwd,
+                        "project_root": project_root,
+                    }
+                )
+            pending_command = None
+
+    for run in runs:
+        command = (run.get("command") or "").strip()
+        if not command or not _is_test_validation_command(command):
+            continue
+        cwd = (run.get("working_directory") or run.get("cwd") or "").strip()
+        root = run.get("project_root") or project_root
+        if not cwd:
+            reasons.append(
+                "test validation command stated without working directory; "
+                "cannot prove branches-only execution (#325)"
+            )
+            continue
+        if not _path_under_branches(cwd, root):
+            violations.append(
+                f"test validation ran outside branches/ worktree: cwd={cwd!r}"
+            )
+            reasons.append(
+                "reviewer workflow must not run test suites in the main "
+                f"checkout; cwd {cwd!r} is not under branches/ (#325)"
+            )
+
+    claims_preexisting = bool(_PREEXISTING_FAILURE_CLAIM_RE.search(text))
+    if claims_preexisting:
+        proof = baseline_proof or {}
+        worktree = (proof.get("worktree_path") or "").strip()
+        if not worktree or not _path_under_branches(worktree, project_root):
+            reasons.append(
+                "pre-existing master failure claimed without a baseline "
+                "worktree path under branches/ (#325)"
+            )
+        if not _is_full_sha(proof.get("baseline_target_sha")):
+            reasons.append(
+                "pre-existing master failure claimed without "
+                "baseline_target_sha proof (#325)"
+            )
+        if not _is_full_sha(proof.get("pr_head_sha")):
+            reasons.append(
+                "pre-existing master failure claimed without pr_head_sha "
+                "proof (#325)"
+            )
+        baseline_failures = proof.get("baseline_failures")
+        pr_failures = proof.get("pr_failures")
+        if baseline_failures is None or pr_failures is None:
+            reasons.append(
+                "pre-existing master failure claimed without baseline and "
+                "PR failure listings (#325)"
+            )
+        if proof.get("failure_signatures_match") is not True:
+            reasons.append(
+                "pre-existing master failure claimed without proven matching "
+                "failure signatures (#325)"
+            )
+        if proof.get("clean_before") is not True:
+            reasons.append(
+                "baseline worktree clean-before validation not proven (#325)"
+            )
+        if proof.get("clean_after") is not True:
+            reasons.append(
+                "baseline worktree clean-after validation not proven (#325)"
+            )
+
+    proven = not reasons and not violations
+    return {
+        "proven": proven,
+        "block": bool(violations),
+        "claims_preexisting": claims_preexisting,
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": (
+            "create a clean baseline worktree under branches/, e.g. "
+            "branches/baseline-master-pr<N>, and rerun validation there"
+            if not proven
+            else "proceed"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+
 # ---------------------------------------------------------------------------
 # Identity disclosure (#305)
 # ---------------------------------------------------------------------------
@@ -3315,6 +3880,280 @@ def assess_email_disclosure(
     }
 
 
+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",
+    re.I,
+)
+
+_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
+
+_NOT_APPLICABLE_VALUE = re.compile(
+    r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b",
+    re.I,
+)
+
+_PRIOR_PROOF_LABEL = re.compile(
+    r"prior (?:blocker|proof|request[- ]changes|feedback)|"
+    r"head sha unchanged since|prior blocker reused|labeled as prior",
+    re.I,
+)
+
+_PAGINATION_FINALITY_EVIDENCE = re.compile(
+    r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
+    r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
+    r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
+    r"total_count\s*:|pagination.*(?:final|complete)|"
+    r"inventory pagination proof:",
+    re.I,
+)
+
+_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
+    r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
+    r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|"
+    r"(?:complete|exhaustive).*(?:default )?page[- ]?size|"
+    r"page[- ]?size assumption",
+    re.I,
+)
+
+_PROOF_WORDING_RULES: tuple[dict, ...] = (
+    {
+        "id": "live_proof",
+        "pattern": re.compile(r"\blive (?:blocker )?proof\b", re.I),
+        "session_keys": ("live_session_proof", "session_proof_commands"),
+        "text_evidence": re.compile(
+            r"current[- ]session|ran in (?:the )?current session|"
+            r"proof (?:command|tool).*(?:current|this) session|"
+            r"revalidated in (?:the )?current session",
+            re.I,
+        ),
+        "allow_prior_label": True,
+    },
+    {
+        "id": "inventory_complete",
+        "pattern": re.compile(
+            r"\binventory (?:is )?complete\b|\binventory completeness\b",
+            re.I,
+        ),
+        "session_keys": ("pagination_complete", "inventory_complete"),
+        "text_evidence": _PAGINATION_FINALITY_EVIDENCE,
+        "allow_prior_label": False,
+    },
+    {
+        "id": "same_as_master",
+        "pattern": re.compile(r"\bsame as master\b", re.I),
+        "session_keys": ("baseline_worktree_proof", "clean_baseline_proof"),
+        "text_evidence": re.compile(
+            r"baseline worktree|clean baseline|diff base.*master|"
+            r"base[- ]equivalent.*master|worktree proof",
+            re.I,
+        ),
+        "allow_prior_label": False,
+    },
+    {
+        "id": "no_file_edits",
+        "pattern": re.compile(
+            r"\b(?:no file edits|file edits none|workspace mutations none)\b",
+            re.I,
+        ),
+        "session_keys": ("no_file_edits_proven", "mutation_ledger_clean"),
+        "text_evidence": re.compile(
+            r"mutation ledger|worktree mutations:\s*none|"
+            r"file edits:\s*none|tracked file edits:\s*none|"
+            r"no tracked (?:file )?edits",
+            re.I,
+        ),
+        "allow_prior_label": False,
+    },
+)
+
+_QUEUE_STATUS_GATES_NO_PR = (
+    "already-landed gate",
+    "author-safety result",
+    "merge preflight",
+)
+
+_REVIEW_WORKTREE_DETAIL_FIELDS = (
+    "review worktree dirty before validation",
+    "review worktree dirty after validation",
+    "review worktree head state",
+)
+
+
+def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]:
+    """Parse ``- Field: value`` lines from the Controller Handoff section."""
+    section = _handoff_section_lines(report_text)
+    if section is None:
+        return {}
+    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 _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool:
+    selected = fields.get("selected pr", "")
+    if selected and _NOT_APPLICABLE_VALUE.search(selected):
+        return True
+    if re.search(r"\bnone\b", selected, re.I):
+        return True
+    if re.search(
+        r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none",
+        report_text or "",
+        re.I,
+    ):
+        return True
+    return not selected
+
+
+def assess_proof_wording(
+    report_text: str | None,
+    *,
+    session_evidence: dict | None = None,
+) -> dict:
+    """Issue #330: reject unsupported proof phrases in final reports.
+
+    Forbidden wording such as ``inventory complete``, ``live proof``, and
+    ``same as master`` is allowed only when matching current-session evidence
+    appears in the report text or in *session_evidence*.
+    """
+    text = report_text or ""
+    evidence = dict(session_evidence or {})
+    violations: list[str] = []
+    flagged_rules: list[str] = []
+
+    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
+        if not _PAGINATION_FINALITY_EVIDENCE.search(text) and not (
+            evidence.get("pagination_complete") or evidence.get("inventory_complete")
+        ):
+            violations.append(
+                "inventory pagination assumed from default page size without "
+                "final-page/no-next-page/total-count/traversal proof"
+            )
+            flagged_rules.append("default_page_size_assumption")
+
+    for rule in _PROOF_WORDING_RULES:
+        if not rule["pattern"].search(text):
+            continue
+        if rule.get("allow_prior_label") and _PRIOR_PROOF_LABEL.search(text):
+            continue
+        if any(evidence.get(key) for key in rule.get("session_keys") or ()):
+            continue
+        if rule["text_evidence"].search(text):
+            continue
+        violations.append(
+            f"forbidden proof phrase ({rule['id']}) without matching evidence"
+        )
+        flagged_rules.append(rule["id"])
+
+    proven = not violations
+    return {
+        "proven": proven,
+        "block": not proven,
+        "violations": violations,
+        "flagged_rules": flagged_rules,
+        "reasons": violations,
+    }
+
+
+def assess_queue_status_report(
+    report_text: str | None,
+    *,
+    session_evidence: dict | None = None,
+) -> dict:
+    """Issue #339: reject contradictory reviewer queue-status-only reports."""
+    text = report_text or ""
+    fields = _controller_handoff_field_map(text)
+    violations: list[str] = []
+
+    proof = assess_proof_wording(text, session_evidence=session_evidence)
+    violations.extend(proof.get("violations", []))
+
+    if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search(
+        text
+    ):
+        violations.append(
+            "prior diagnostic worktree simulation used as current conflict proof"
+        )
+
+    queue_only = _queue_status_only_run(fields, text)
+    if queue_only:
+        for gate in _QUEUE_STATUS_GATES_NO_PR:
+            value = fields.get(gate, "")
+            if value and _GATE_PASSED_VALUE.search(value):
+                if not _NOT_APPLICABLE_VALUE.search(value):
+                    violations.append(
+                        f"{gate} cannot be 'passed' when no PR is selected (#339)"
+                    )
+
+        worktree_used = fields.get("review worktree used", "")
+        if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I):
+            for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS:
+                detail_value = fields.get(detail_field, "")
+                if (
+                    detail_value
+                    and not _NOT_APPLICABLE_VALUE.search(detail_value)
+                ):
+                    violations.append(
+                        f"{detail_field} must be 'not applicable' or 'none' "
+                        "when no review worktree was created"
+                    )
+
+    blockers = fields.get("blockers", "")
+    if blockers and re.search(r"\bnone\b", blockers, re.I):
+        if re.search(
+            r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|"
+            r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)",
+            text,
+            re.I,
+        ):
+            violations.append(
+                "Blockers: none contradicts report stating all PRs are "
+                "conflicted, blocked, or unverified"
+            )
+
+    if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I):
+        has_skip_proof = bool(
+            re.search(
+                r"current[- ]session|prior blocker reused|conflict proof:|"
+                r"gitea_view_pr|review[- ]feedback proof|unchanged head",
+                text,
+                re.I,
+            )
+            or (session_evidence or {}).get("skip_proof_per_pr")
+        )
+        if not has_skip_proof:
+            violations.append(
+                "skipped PRs listed without current-session or labeled prior proof"
+            )
+
+    if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations:
+        violations.append(
+            "canonical workflow cited but mandatory queue-status proof gates violated"
+        )
+
+    deduped = list(dict.fromkeys(violations))
+    proven = not deduped
+    return {
+        "proven": proven,
+        "block": not proven,
+        "queue_status_only": queue_only,
+        "violations": deduped,
+        "reasons": deduped,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Git ref mutations (#297)
 # ---------------------------------------------------------------------------
@@ -3429,3 +4268,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
         "ref_mutating_commands": ref_commands,
         "fetch_targets": fetch_targets,
     }
+
+
+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
+
+    return _assess(report_text, **kwargs)

reviewer_mergeability_skip.py

@@ -0,0 +1,159 @@
+"""Non-mergeable PR skip conflict-proof verifier for reviewer reports (#322)."""
+
+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)
+_MERGEABILITY_FALSE_RE = re.compile(
+    r"(?:mergeable\s*:\s*false|not mergeable|non[- ]mergeable|mergeability\s*:\s*false|"
+    r"mergeability result\s*:\s*false)",
+    re.IGNORECASE,
+)
+_CONFLICT_PROOF_RE = re.compile(
+    r"(?:merge-tree|git merge-tree|gitea_view_pr|gitea_check_pr_eligibility|"
+    r"conflict proof|mergeability tool|merge simulation|conflicting files?)",
+    re.IGNORECASE,
+)
+_CONFLICTING_FILES_RE = re.compile(
+    r"(?:conflicting files?|conflict files?)\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_HEAD_CHANGED_RE = re.compile(
+    r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
+    r"head changed since|head unchanged since)",
+    re.IGNORECASE,
+)
+_MERGEABILITY_UNVERIFIED_RE = re.compile(
+    r"\bMERGEABILITY_UNVERIFIED\b",
+)
+
+
+def _pr_section(text: str, pr_number: int) -> str:
+    """Return report lines likely describing one skipped PR."""
+    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_non_mergeable_skip_proof(
+    report_text: str,
+    *,
+    skipped_prs: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Validate skipped non-mergeable PR documentation in reviewer reports (#322)."""
+    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)
+        mergeable = entry.get("mergeable")
+        verified = entry.get("mergeability_verified")
+        classification = (entry.get("classification") or "").strip().upper()
+        head_sha = (entry.get("head_sha") or "").strip() or None
+
+        item_reasons: list[str] = []
+
+        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 mergeable is False or verified is False:
+            if not _MERGEABILITY_UNVERIFIED_RE.search(section) and verified is False:
+                if "MERGEABILITY_UNVERIFIED" not in classification:
+                    item_reasons.append(
+                        f"PR #{pr_number} conflict proof unavailable; report must classify "
+                        "MERGEABILITY_UNVERIFIED"
+                    )
+            elif verified is not False:
+                if not _MERGEABILITY_FALSE_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing mergeability:false proof"
+                    )
+                if not _has_head_sha(section, head_sha):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing current head SHA"
+                    )
+                if not _CONFLICT_PROOF_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing conflict proof command/tool"
+                    )
+                if entry.get("head_changed_since_prior_blocker") is not None:
+                    if not _HEAD_CHANGED_RE.search(section):
+                        item_reasons.append(
+                            f"PR #{pr_number} skip missing head-changed-since-blocker proof"
+                        )
+
+        if (
+            entry.get("conflicting_files")
+            and not _CONFLICTING_FILES_RE.search(section)
+            and not any(
+                path.lower() in section.lower()
+                for path in (entry.get("conflicting_files") or [])
+            )
+        ):
+            item_reasons.append(
+                f"PR #{pr_number} has conflicting files in session proof but "
+                "report omits file-level conflict proof"
+            )
+
+        proven = not item_reasons
+        assessments.append({
+            "pr_number": pr_number,
+            "proven": proven,
+            "classification": classification or (
+                "MERGEABILITY_UNVERIFIED" if verified is False else "NON_MERGEABLE_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": (
+            "document each skipped non-mergeable PR with head SHA, mergeability result, "
+            "conflict proof command, conflicting files, and head-change status"
+            if reasons
+            else "proceed"
+        ),
+    }

skills/llm-project-workflow/schemas/review-merge-final-report.md

@@ -77,6 +77,18 @@ When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`:
 
 Identity format: `username / profile` (not personal email unless required — #305).
 
+### Queue-status-only runs (no selected PR)
+
+When the run inventories the queue but selects no PR for review:
+
+- Selected PR: `none`
+- Already-landed gate, Author-safety result, Merge preflight: `not applicable` or `not run` — never `passed`
+- Review worktree detail fields: `not applicable` or `none` when Review worktree used is `false`
+- Blockers must not be `none` if the narrative says all open PRs are conflicted, blocked, or unverified
+- Inventory pagination proof must cite final-page metadata, not default page-size assumptions
+
+Verifier: `review_proofs.assess_queue_status_report()`.
+
 Narrative final report and controller handoff must agree on eligibility class,
 candidate/reviewed head SHA, mutation state, worktree usage, review decision,
 terminal review mutation, merge result, and linked issue status.

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_llm_workflow_split.py

@@ -118,7 +118,19 @@ def test_start_issue_template_references_work_issue_workflow():
     assert "workflows/work-issue.md" in text
 
 
+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)
+    assert callable(assess_create_issue_final_report)
+
+
+def test_non_mergeable_skip_verifier_exported():
+    from review_proofs import assess_non_mergeable_skip_proof
+
+    assert callable(assess_non_mergeable_skip_proof)

tests/test_mcp_server.py

@@ -48,6 +48,21 @@ import mcp_server
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
+_NO_BLOCKER_FEEDBACK = {
+    "success": True,
+    "has_blocking_change_requests": False,
+    "review_feedback_stale": False,
+}
+
+
+def _mark_request_changes_ready(pr_number=8, **kwargs):
+    """Mark a request_changes decision ready with the #332 duplicate-
+    suppression feedback fetch stubbed to 'no existing blocker'."""
+    with patch("mcp_server.gitea_get_pr_review_feedback",
+               return_value=dict(_NO_BLOCKER_FEEDBACK)):
+        return gitea_mark_final_review_decision(
+            pr_number, "request_changes", **kwargs)
+
 
 def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
     return {
@@ -1839,7 +1854,7 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
-        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+        _mark_request_changes_ready(remote="prgs")
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"),
             {"id": 9, "state": "REQUEST_CHANGES"},
@@ -1862,7 +1877,7 @@ class TestSubmitPrReview(unittest.TestCase):
         self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
 
     def test_request_changes_blocked_without_eligibility(self):
-        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+        _mark_request_changes_ready(remote="prgs")
         with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
@@ -2144,7 +2159,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 reason="operator approved correcting mistaken approve",
                 operator_authorized=True,
             )
-            gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+            _mark_request_changes_ready(remote="prgs")
             second = gitea_submit_pr_review(
                 pr_number=8, action="request_changes", remote="prgs",
                 final_review_decision_ready=True,

tests/test_proof_wording.py

@@ -0,0 +1,93 @@
+"""Doc-contract checks for proof wording enforcement (#330)."""
+import unittest
+
+from review_proofs import assess_proof_wording, build_final_report
+
+
+class TestProofWording(unittest.TestCase):
+    def test_rejects_inventory_complete_without_pagination_proof(self):
+        result = assess_proof_wording(
+            "Open PR inventory is complete because only 3 PRs were returned."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_rejects_default_page_size_assumption(self):
+        result = assess_proof_wording(
+            "Inventory exhaustive: fewer than the default Gitea page size returned."
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("default_page_size_assumption", result["flagged_rules"])
+
+    def test_allows_inventory_complete_with_finality_metadata(self):
+        result = assess_proof_wording(
+            "Inventory complete. pagination_complete: true, is_final_page: true"
+        )
+        self.assertTrue(result["proven"])
+
+    def test_allows_inventory_complete_with_session_evidence(self):
+        result = assess_proof_wording(
+            "Queue inventory complete.",
+            session_evidence={"pagination_complete": True},
+        )
+        self.assertTrue(result["proven"])
+
+    def test_rejects_live_proof_without_session_evidence(self):
+        result = assess_proof_wording("Live proof confirms mergeable state.")
+        self.assertFalse(result["proven"])
+        self.assertIn("live_proof", result["flagged_rules"])
+
+    def test_allows_prior_blocker_reuse_wording(self):
+        result = assess_proof_wording(
+            "Prior blocker reused; head SHA unchanged since blocking review. "
+            "Live proof not claimed for this skip."
+        )
+        self.assertTrue(result["proven"])
+
+    def test_allows_live_proof_with_current_session_evidence(self):
+        result = assess_proof_wording(
+            "Live proof: gitea_view_pr revalidated in the current session."
+        )
+        self.assertTrue(result["proven"])
+
+    def test_rejects_same_as_master_without_baseline_proof(self):
+        result = assess_proof_wording("Diff is same as master.")
+        self.assertFalse(result["proven"])
+        self.assertIn("same_as_master", result["flagged_rules"])
+
+    def test_allows_same_as_master_with_baseline_wording(self):
+        result = assess_proof_wording(
+            "Clean baseline worktree proof: diff base matches master."
+        )
+        self.assertTrue(result["proven"])
+
+    def test_rejects_file_edits_none_without_ledger(self):
+        result = assess_proof_wording("Workspace mutations: file edits none.")
+        self.assertFalse(result["proven"])
+        self.assertIn("no_file_edits", result["flagged_rules"])
+
+    def test_allows_file_edits_none_with_mutation_ledger(self):
+        result = assess_proof_wording(
+            "Mutation ledger: tracked file edits: none",
+            session_evidence={"mutation_ledger_clean": True},
+        )
+        self.assertTrue(result["proven"])
+
+    def test_build_final_report_downgrades_on_proof_wording_violation(self):
+        report = build_final_report(
+            {"proven": True},
+            {"complete": True},
+            {"claimable": True, "verdict": "strong"},
+            {"status": "clean"},
+            True,
+            False,
+            True,
+            report_text="Inventory complete after listing 4 open PRs.",
+        )
+        self.assertEqual(report["grade"], "downgraded")
+        self.assertFalse(report["proof_wording_proven"])
+        self.assertTrue(report["proof_wording_violations"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_queue_status_report.py

@@ -0,0 +1,91 @@
+"""Doc-contract checks for queue-status report verifier (#339)."""
+import unittest
+
+from review_proofs import assess_queue_status_report, build_final_report
+
+BAD_QUEUE_STATUS_REPORT = """
+## PR Queue Status
+
+Loaded workflows/review-merge-pr.md. Inventory complete because gitea_list_prs
+returned 6 open PRs, fewer than the default Gitea page size.
+
+PR #286 skipped: prior diagnostic worktree simulation showed merge conflicts.
+Live blocker proof for all skipped PRs.
+
+All open PRs are conflicted or blocked. Blockers: none.
+
+## Controller Handoff
+
+- Task: review queue status
+- Repo: Scaled-Tech-Consulting/Gitea-Tools
+- Role: reviewer
+- Identity: reviewer1 / prgs-reviewer
+- Selected PR: none
+- Inventory pagination proof: assumed complete (6 < 50)
+- Already-landed gate: passed
+- Author-safety result: passed
+- Merge preflight: passed
+- Review worktree used: false
+- Review worktree dirty before validation: false
+- Blockers: none
+- Current status: queue inspected; no PR selected
+"""
+
+
+class TestQueueStatusReport(unittest.TestCase):
+    def test_bad_fixture_fails_closed(self):
+        result = assess_queue_status_report(BAD_QUEUE_STATUS_REPORT)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(result["queue_status_only"])
+        joined = " ".join(result["violations"]).lower()
+        self.assertIn("already-landed gate", joined)
+        self.assertIn("author-safety", joined)
+        self.assertIn("merge preflight", joined)
+        self.assertIn("prior diagnostic", joined)
+        self.assertIn("blockers", joined)
+
+    def test_passes_with_not_applicable_gates(self):
+        report = """
+## Controller Handoff
+- Selected PR: none
+- Inventory pagination proof: is_final_page: true, has_more: false
+- Already-landed gate: not applicable
+- Author-safety result: not run
+- Merge preflight: not applicable
+- Review worktree used: false
+- Review worktree dirty before validation: not applicable
+- Blockers: all PRs skipped pending reviewer
+"""
+        result = assess_queue_status_report(report)
+        self.assertTrue(result["proven"])
+
+    def test_rejects_passed_gates_without_selected_pr(self):
+        report = """
+## Controller Handoff
+- Selected PR: none
+- Already-landed gate: passed
+- Blockers: PR #1 conflicted
+"""
+        result = assess_queue_status_report(report)
+        self.assertFalse(result["proven"])
+        self.assertIn("already-landed gate", " ".join(result["violations"]).lower())
+
+    def test_build_final_report_downgrades_bad_queue_status(self):
+        report = build_final_report(
+            {"proven": True},
+            {"complete": True},
+            {"claimable": True, "verdict": "strong"},
+            {"status": "clean"},
+            True,
+            False,
+            True,
+            report_text=BAD_QUEUE_STATUS_REPORT,
+        )
+        self.assertEqual(report["grade"], "downgraded")
+        self.assertFalse(report["queue_status_report_proven"])
+        self.assertTrue(report["queue_status_violations"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_review_proofs.py

@@ -25,6 +25,8 @@ from review_proofs import (  # noqa: E402
     ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
     assess_author_pr_report,
     assess_validation_environment_proof,
+    assess_queue_ordering_proof,
+    assess_reviewer_baseline_validation_proof,
     assess_capability_evidence,
     assess_capability_proof,
     assess_contradictory_no_pr_claim,
@@ -940,13 +942,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",
@@ -958,6 +965,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")
@@ -1080,6 +1096,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."""
@@ -2263,5 +2417,229 @@ class TestValidationEnvironmentProof(unittest.TestCase):
         self.assertFalse(report["validation_environment_proven"])
 
 
+        result = assess_queue_ordering_proof(
+            report_text="Selected the next eligible PR: PR #286.",
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["claims_selection"])
+
+    def test_next_eligible_claim_with_policy_passes(self):
+        result = assess_queue_ordering_proof(
+            report_text=(
+                "Queue ordering policy: oldest PR number first\n"
+                "Selected the next eligible PR: PR #286."
+            ),
+        )
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["policy"], "oldest PR number first")
+
+    def test_newest_first_api_with_oldest_first_selection_passes(self):
+        result = assess_queue_ordering_proof(
+            report_text=(
+                "Queue ordering policy: oldest PR number first\n"
+                "API order was newest-first (PR #291 before PR #286).\n"
+                "Selected the next eligible PR: PR #286."
+            ),
+            queue_ordering={
+                "policy": "oldest PR number first",
+                "selected_pr_number": 286,
+                "api_pr_numbers": [291, 286],
+            },
+        )
+        self.assertTrue(result["proven"])
+
+    def test_earlier_actionable_pr_without_skip_reason_fails(self):
+        result = assess_queue_ordering_proof(
+            report_text=(
+                "Queue ordering policy: oldest PR number first\n"
+                "Selected the next eligible PR: PR #300."
+            ),
+            queue_ordering={
+                "policy": "oldest PR number first",
+                "selected_pr_number": 300,
+                "api_pr_numbers": [300, 286],
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_priority_override_requires_explanation(self):
+        result = assess_queue_ordering_proof(
+            report_text="Selected the next eligible PR: PR #300.",
+            queue_ordering={
+                "policy": "priority label",
+                "selected_pr_number": 300,
+                "api_pr_numbers": [286, 300],
+            },
+        )
+        self.assertFalse(result["proven"])
+
+    def test_priority_override_with_explanation_passes(self):
+        result = assess_queue_ordering_proof(
+            report_text=(
+                "Queue ordering policy: priority label\n"
+                "Priority override: security label on PR #300.\n"
+                "Selected the next eligible PR: PR #300."
+            ),
+            queue_ordering={
+                "policy": "priority label",
+                "selected_pr_number": 300,
+                "api_pr_numbers": [286, 300],
+                "priority_override": "security label on PR #300",
+            },
+        )
+        self.assertTrue(result["proven"])
+
+    def test_build_final_report_downgrades_missing_ordering_proof(self):
+        report = build_final_report(
+            checkout_proof=_good_checkout(),
+            inventory=_good_inventory(),
+            validation=_good_validation(),
+            contamination=_good_contamination(),
+            identity_eligible=True,
+            merge_performed=False,
+            issue_status_verified=True,
+            capability_evidence=_good_capability_evidence(),
+            sweep=_good_sweep(),
+            live_state=_good_live_state(),
+            role_boundary=_good_role_boundary(),
+            review_mutation=_good_review_mutation(),
+            controller_handoff=_good_handoff(),
+            capability_proof=_good_capability_proof(),
+            sweep_proof=_good_secret_sweep(),
+            worktree_proof={
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+            },
+            report_text="Selected the next eligible PR: PR #286.",
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["queue_ordering_proven"])
+
+class TestReviewerBaselineValidationProof(unittest.TestCase):
+    """Issue #325: baseline validation must use branches/ worktrees."""
+
+    ROOT = "/repo/Gitea-Tools"
+
+    def _good_baseline_proof(self, **overrides):
+        proof = {
+            "worktree_path": f"{self.ROOT}/branches/baseline-master-pr280",
+            "baseline_target_sha": PINNED,
+            "pr_head_sha": OTHER,
+            "baseline_failures": ["tests/test_foo.py::test_bar"],
+            "pr_failures": ["tests/test_foo.py::test_bar"],
+            "failure_signatures_match": True,
+            "clean_before": True,
+            "clean_after": True,
+        }
+        proof.update(overrides)
+        return proof
+
+    def test_main_checkout_test_run_blocks(self):
+        result = assess_reviewer_baseline_validation_proof(
+            validation_runs=[
+                {
+                    "command": "venv/bin/pytest tests/",
+                    "working_directory": self.ROOT,
+                    "project_root": self.ROOT,
+                }
+            ],
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(result["violations"])
+
+    def test_branches_worktree_test_run_passes(self):
+        result = assess_reviewer_baseline_validation_proof(
+            validation_runs=[
+                {
+                    "command": "venv/bin/pytest tests/",
+                    "working_directory": f"{self.ROOT}/branches/review-pr-280",
+                    "project_root": self.ROOT,
+                }
+            ],
+            project_root=self.ROOT,
+        )
+        self.assertTrue(result["proven"])
+
+    def test_preexisting_claim_without_baseline_proof_fails(self):
+        result = assess_reviewer_baseline_validation_proof(
+            report_text=(
+                "Validation: full-suite failures are pre-existing on master."
+            ),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["claims_preexisting"])
+
+    def test_preexisting_claim_with_complete_baseline_proof_passes(self):
+        result = assess_reviewer_baseline_validation_proof(
+            report_text="Failures are pre-existing on master.",
+            baseline_proof=self._good_baseline_proof(),
+            project_root=self.ROOT,
+        )
+        self.assertTrue(result["proven"])
+
+    def test_mismatched_failures_block_preexisting_claim(self):
+        result = assess_reviewer_baseline_validation_proof(
+            report_text="Same as master.",
+            baseline_proof=self._good_baseline_proof(
+                failure_signatures_match=False,
+                pr_failures=["tests/test_other.py::test_other"],
+            ),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_report_parsed_main_checkout_cwd_blocks(self):
+        result = assess_reviewer_baseline_validation_proof(
+            report_text=(
+                "- Validation command: venv/bin/pytest tests/\n"
+                f"- Working directory: {self.ROOT}\n"
+            ),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_build_final_report_downgrades_main_checkout_validation(self):
+        report = build_final_report(
+            checkout_proof=_good_checkout(),
+            inventory=_good_inventory(),
+            validation=_good_validation(),
+            contamination=_good_contamination(),
+            identity_eligible=True,
+            merge_performed=False,
+            issue_status_verified=True,
+            capability_evidence=_good_capability_evidence(),
+            sweep=_good_sweep(),
+            live_state=_good_live_state(),
+            role_boundary=_good_role_boundary(),
+            review_mutation=_good_review_mutation(),
+            controller_handoff=_good_handoff(),
+            capability_proof=_good_capability_proof(),
+            sweep_proof=_good_secret_sweep(),
+            worktree_proof={
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+            },
+            report_text=(
+                "- Validation command: venv/bin/pytest tests/\n"
+                f"- Working directory: {self.ROOT}\n"
+            ),
+            project_root=self.ROOT,
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["baseline_validation_proven"])
+
+
+if __name__ == "__main__":
+    unittest.main()
+
 if __name__ == "__main__":
     unittest.main()

tests/test_reviewer_mergeability_skip.py

@@ -0,0 +1,150 @@
+"""Tests for non-mergeable skip conflict-proof verifier (#322)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_mergeability_skip import assess_non_mergeable_skip_proof  # noqa: E402
+
+HEAD_A = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+HEAD_B = "1111111111111111111111111111111111111111"
+
+
+def _good_skip_report(pr_number: int, head: str, files: str = "review_proofs.py") -> str:
+    return "\n".join([
+        f"Skipped PR #{pr_number} (non-mergeable).",
+        f"- PR number: #{pr_number}",
+        f"- Current head SHA: {head}",
+        "- Mergeability result: false",
+        "- Conflict proof command: git merge-tree $(git merge-base prgs/master prgs/feat) prgs/master prgs/feat",
+        f"- Conflicting files: {files}",
+        "- Head unchanged since prior blocker",
+        "- Blocking category: merge conflict",
+    ])
+
+
+class TestNonMergeableSkipProof(unittest.TestCase):
+    def test_no_skips_passes(self):
+        report = "Selected PR #203 for review. No earlier PRs skipped."
+        result = assess_non_mergeable_skip_proof(report, skipped_prs=[])
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_documented_non_mergeable_skip_passes(self):
+        report = _good_skip_report(282, HEAD_A)
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 282,
+                "mergeable": False,
+                "head_sha": HEAD_A,
+                "mergeability_verified": True,
+                "conflicting_files": ["review_proofs.py"],
+                "head_changed_since_prior_blocker": False,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_skip_without_conflict_proof_blocks(self):
+        report = "\n".join([
+            "Skipped PR #284 because it was not mergeable.",
+            f"Head SHA: {HEAD_B}",
+        ])
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 284,
+                "mergeable": False,
+                "head_sha": HEAD_B,
+                "mergeability_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("conflict proof" in r for r in result["reasons"]))
+
+    def test_missing_head_sha_blocks(self):
+        report = _good_skip_report(282, HEAD_A).replace(HEAD_A, "")
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 282,
+                "mergeable": False,
+                "head_sha": HEAD_A,
+                "mergeability_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+
+    def test_mergeability_unverified_classification_passes(self):
+        report = "\n".join([
+            "Skipped PR #282.",
+            "Classification: MERGEABILITY_UNVERIFIED",
+            "Conflict proof could not be retrieved in this session.",
+        ])
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 282,
+                "mergeable": False,
+                "mergeability_verified": False,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unverified_without_classification_blocks(self):
+        report = "Skipped PR #282 due to conflicts."
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 282,
+                "mergeable": False,
+                "mergeability_verified": False,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("MERGEABILITY_UNVERIFIED" in r for r in result["reasons"]))
+
+    def test_non_mergeable_without_file_details_still_passes_with_command(self):
+        report = _good_skip_report(284, HEAD_B, files="not available")
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 284,
+                "mergeable": False,
+                "head_sha": HEAD_B,
+                "mergeability_verified": True,
+                "conflicting_files": [],
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_stale_head_requires_head_change_field(self):
+        report = _good_skip_report(282, HEAD_A).replace(
+            "- Head unchanged since prior blocker",
+            "",
+        )
+        result = assess_non_mergeable_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 282,
+                "mergeable": False,
+                "head_sha": HEAD_A,
+                "mergeability_verified": True,
+                "head_changed_since_prior_blocker": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head-changed" in r for r in result["reasons"]))
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_non_mergeable_skip_proof as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_terminal_review_hard_stop.py

@@ -0,0 +1,187 @@
+"""Tests for the session hard-stop after a terminal review mutation (#332).
+
+Extends the single-terminal-decision machinery (#211): after a live
+REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence
+for that same PR may continue; a different PR can never be reviewed or
+merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is
+suppressed.
+"""
+import os
+import unittest
+from unittest.mock import patch
+
+import mcp_server
+
+
+def _lock(mutations=None, correction=False):
+    return {
+        "task": "review_pr",
+        "remote": "prgs",
+        "session_pid": os.getpid(),
+        "session_profile": "prgs-reviewer",
+        "session_profile_lock": "",
+        "final_review_decision_ready": False,
+        "ready_pr_number": None,
+        "ready_action": None,
+        "ready_expected_head_sha": None,
+        "ready_remote": None,
+        "ready_org": None,
+        "ready_repo": None,
+        "live_mutations": list(mutations or []),
+        "correction_authorized": correction,
+        "correction_reason": None,
+    }
+
+
+def _seed(mutations=None, correction=False):
+    mcp_server._save_review_decision_lock(_lock(mutations, correction))
+
+
+APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
+              "review_state": "approve"}
+RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
+        "review_state": "request_changes"}
+
+
+class TestTerminalHardStopReasons(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_no_lock_no_reasons(self):
+        mcp_server._save_review_decision_lock(None)
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
+
+    def test_no_terminal_mutation_no_reasons(self):
+        _seed()
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
+
+    def test_request_changes_blocks_everything(self):
+        _seed([RC_A])
+        for op in ("merge", "mark_ready", "review"):
+            for pr in (5, 6):
+                reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op)
+                self.assertTrue(reasons, f"{op}/{pr}")
+                self.assertIn("request_changes on PR #5", reasons[0])
+                self.assertIn("#332", reasons[0])
+
+    def test_approve_allows_same_pr_merge_only(self):
+        _seed([APPROVED_A])
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "review"))
+
+    def test_correction_reopens_review_path_only(self):
+        _seed([RC_A], correction=True)
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
+        # correction never opens a cross-PR merge
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
+
+
+class TestMergeHardStopWiring(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_merge_blocked_after_request_changes(self):
+        _seed([RC_A])
+        result = mcp_server.gitea_merge_pr(
+            pr_number=6, confirmation="MERGE PR 6", remote="prgs")
+        self.assertFalse(result["performed"])
+        self.assertTrue(
+            any("#332" in r for r in result["reasons"]), result["reasons"])
+
+    def test_merge_of_other_pr_blocked_after_approve(self):
+        _seed([APPROVED_A])
+        result = mcp_server.gitea_merge_pr(
+            pr_number=6, confirmation="MERGE PR 6", remote="prgs")
+        self.assertFalse(result["performed"])
+        self.assertTrue(
+            any("#332" in r for r in result["reasons"]), result["reasons"])
+
+
+class TestMarkFinalHardStopWiring(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_mark_ready_blocked_after_terminal_mutation(self):
+        _seed([RC_A])
+        result = mcp_server.gitea_mark_final_review_decision(
+            pr_number=6, action="approve", remote="prgs")
+        self.assertFalse(result["marked_ready"])
+        self.assertTrue(
+            any("#332" in r for r in result["reasons"]), result["reasons"])
+
+
+def _feedback(blocking, stale=False, success=True):
+    return {
+        "success": success,
+        "has_blocking_change_requests": blocking,
+        "review_feedback_stale": stale,
+        "current_head_sha": "abc123",
+    }
+
+
+class TestDuplicateRequestChangesSuppression(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_duplicate_request_changes_blocked_at_same_head(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=True, stale=False)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertFalse(result["marked_ready"])
+        self.assertTrue(
+            any("duplicate" in r for r in result["reasons"]),
+            result["reasons"])
+
+    def test_request_changes_allowed_when_prior_blocker_stale(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=True, stale=True)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
+    def test_request_changes_allowed_when_no_blocker(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=False)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
+    def test_request_changes_fails_closed_when_feedback_unavailable(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=False,
+                                                 success=False)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertFalse(result["marked_ready"])
+        self.assertTrue(
+            any("could not verify" in r for r in result["reasons"]),
+            result["reasons"])
+
+    def test_approve_does_not_fetch_feedback(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          side_effect=AssertionError("must not be called")):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="approve", remote="prgs")
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_workspace_mutation_consistency.py

@@ -0,0 +1,165 @@
+"""Tests for workspace-vs-worktree mutation consistency verifier (#313).
+
+Final reports must not claim ``Workspace mutations: none`` when worktree
+mutations (reset --hard, checkout, clean, worktree add/remove, ...)
+occurred, and must report observed worktree / git ref mutations under the
+precise category fields.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_workspace_mutation_consistency  # noqa: E402
+
+
+class TestWorkspaceNoneContradiction(unittest.TestCase):
+    def test_reset_hard_with_workspace_none_fails(self):
+        report = (
+            "Controller Handoff\n"
+            "- Workspace mutations: none (no local file edits)\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(result["downgraded"])
+        self.assertTrue(
+            any("workspace mutations" in r.lower() for r in result["reasons"])
+        )
+
+    def test_checkout_with_workspace_none_fails(self):
+        report = "- Workspace mutations: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git checkout feat/some-branch"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_worktree_add_with_workspace_none_fails(self):
+        report = "- Workspace mutations: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git worktree add /tmp/review-pr1 abc123"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_clean_with_workspace_none_fails(self):
+        report = "- Workspace mutations: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git clean -fd"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_self_reported_worktree_mutation_contradicts_workspace_none(self):
+        # No observed command log; the report itself carries the
+        # contradiction (#313 observed behavior).
+        report = (
+            "- Worktree mutations: git reset --hard FETCH_HEAD\n"
+            "- Workspace mutations: none (no local file edits)\n"
+        )
+        result = assess_workspace_mutation_consistency(report)
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("workspace mutations" in r.lower() for r in result["reasons"])
+        )
+
+
+class TestPreciseCategoriesPass(unittest.TestCase):
+    def test_file_edits_none_with_worktree_mutation_listed_passes(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: git reset --hard FETCH_HEAD\n"
+            "- Git ref mutations: git fetch prgs\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=[
+                "git fetch prgs",
+                "git reset --hard FETCH_HEAD",
+            ],
+        )
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_noop_report_passes(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: none\n"
+            "- Git ref mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report, observed_commands=[]
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_fetch_only_with_ref_mutation_reported_passes(self):
+        # Fetch is a git ref mutation, not a worktree mutation; a
+        # workspace-none claim is not contradicted by fetch alone.
+        report = (
+            "- Workspace mutations: none\n"
+            "- Git ref mutations: git fetch prgs master\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertTrue(result["complete"])
+
+
+class TestObservedMutationsMustBeReported(unittest.TestCase):
+    def test_reset_without_worktree_mutation_field_fails(self):
+        report = "- File edits by reviewer: none\n"
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("worktree mutation" in r.lower() for r in result["reasons"])
+        )
+
+    def test_reset_with_worktree_mutations_none_fails(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["complete"])
+
+    def test_fetch_without_ref_mutation_field_fails(self):
+        report = (
+            "- File edits by reviewer: none\n"
+            "- Worktree mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("git ref mutation" in r.lower() for r in result["reasons"])
+        )
+
+    def test_fetch_with_ref_mutations_none_fails(self):
+        report = (
+            "- Worktree mutations: none\n"
+            "- Git ref mutations: none\n"
+        )
+        result = assess_workspace_mutation_consistency(
+            report,
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()