fix: resolve conflicts for issue 324 (master refresh)

Add assess_reviewer_fallback_report to fail closed when reviewer final
reports or action logs show profiles.json access or local helper scripts
while MCP tools are available. Recovery mode requires explicit fallback
classification and identity/profile/repo/capability proof.

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

final_report_validator.py

@@ -1,878 +0,0 @@
-"""Composable final-report validator for reviewer and reconciliation handoffs (#327).
-
-Single entry point ``assess_final_report_validator`` runs task-specific rule
-modules and returns structured findings suitable for controller handoff
-``Blocker`` / ``Safe next action`` fields.
-"""
-
-from __future__ import annotations
-
-import inspect
-import re
-from typing import Any, Callable
-
-from review_proofs import (
-    HANDOFF_HEADING,
-    assess_controller_handoff,
-    assess_email_disclosure,
-    assess_issue_filing_final_report,
-    assess_mutation_ledger_report,
-    assess_review_mutation_final_report,
-    assess_validation_report,
-)
-
-FINAL_REPORT_TASK_KINDS = frozenset({
-    "review_pr",
-    "reconcile_already_landed",
-    "author_issue",
-    "work_issue",
-    "issue_filing",
-    "issue_selection",
-    "inventory",
-})
-
-_TASK_KIND_ALIASES = {
-    "review": "review_pr",
-    "review-merge-pr": "review_pr",
-    "reconcile-landed-pr": "reconcile_already_landed",
-    "reconcile_already_landed": "reconcile_already_landed",
-    "work-issue": "work_issue",
-    "create_issue": "issue_filing",
-}
-
-_HANDOFF_ROLE_BY_TASK = {
-    "review_pr": "review",
-    "reconcile_already_landed": None,
-    "author_issue": "author",
-    "work_issue": "author",
-    "issue_filing": "issue_filing",
-    "issue_selection": None,
-    "inventory": "inventory",
-}
-
-_LEGACY_WORKSPACE_MUTATIONS_RE = re.compile(
-    r"^\s*[-*]?\s*workspace\s+mutations\s*:",
-    re.IGNORECASE | re.MULTILINE,
-)
-_VAGUE_MUTATIONS_NONE_RE = re.compile(
-    r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
-    re.IGNORECASE | re.MULTILINE,
-)
-_BARE_PYTEST_RE = re.compile(r"\b(?:pytest|python\s+-m\s+pytest)\b", re.IGNORECASE)
-_VALIDATION_ENV_RE = re.compile(
-    r"(?:working directory|cwd|executed (?:in|from)|command\s+path)\s*:|\bin\s+branches/",
-    re.IGNORECASE,
-)
-_MAIN_CHECKOUT_RE = re.compile(
-    r"(?:/Gitea-Tools(?:/|$)(?!branches/)|main checkout|shared checkout|control checkout)",
-    re.IGNORECASE,
-)
-_SAME_AS_MASTER_RE = re.compile(
-    r"same as master|same failures as master|matches master baseline",
-    re.IGNORECASE,
-)
-_BASELINE_WORKTREE_RE = re.compile(
-    r"baseline worktree",
-    re.IGNORECASE,
-)
-_APPROVAL_CLAIM_RE = re.compile(
-    r"(?:submitted\s+['\"]approve['\"]|review decision\s*:\s*approve|claims?\s+approve)",
-    re.IGNORECASE,
-)
-_FULL_SUITE_FAIL_RE = re.compile(
-    r"(?:full suite failed|full test suite failed|\d+\s+failed(?:,|\s))",
-    re.IGNORECASE,
-)
-_ALREADY_LANDED_RE = re.compile(
-    r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed",
-    re.IGNORECASE,
-)
-_ELIGIBLE_REVIEW_RE = re.compile(
-    r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
-    re.IGNORECASE,
-)
-_REVIEWED_LANDED_RE = re.compile(
-    r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
-    re.IGNORECASE | re.DOTALL,
-)
-_RECONCILE_STALE_FIELDS = (
-    "pr number opened",
-    "pinned reviewed head",
-    "scratch worktree used",
-    "review decision",
-    "merge result",
-)
-_RECONCILE_REQUIRED_FIELDS = (
-    "Selected PR",
-    "Eligibility class",
-    "Linked issue live status",
-    "Safe next action",
-    "No review/merge confirmation",
-)
-_LINKED_ISSUE_STATUS_RE = re.compile(
-    r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$",
-    re.IGNORECASE | re.MULTILINE,
-)
-_LINKED_ISSUE_NUMBER_RE = re.compile(
-    r"linked issue\s*:\s*#?(\d+)",
-    re.IGNORECASE,
-)
-_INVENTORY_COMPLETE_RE = re.compile(
-    r"inventory\s+(?:complete|completeness\s*:\s*complete)",
-    re.IGNORECASE,
-)
-_PAGINATION_PROOF_RE = re.compile(
-    r"(?:pagination\s+complete|final page|no next page|total\s+(?:open\s+)?pr\s+count|traversed\s+all\s+pages)",
-    re.IGNORECASE,
-)
-_GIT_FETCH_RE = re.compile(r"\bgit\s+fetch\b", re.IGNORECASE)
-_READONLY_DIAG_RE = re.compile(
-    r"read[- ]only diagnostics\s*:\s*(.+)$",
-    re.IGNORECASE | re.MULTILINE,
-)
-_MUTATION_CATEGORY_FIELDS = (
-    "file edits by reviewer",
-    "worktree/index mutations",
-    "git ref mutations",
-    "mcp/gitea mutations",
-    "reconciliation mutations",
-    "file edits by reconciler",
-)
-
-
-def _normalize_task_kind(task_kind: str | None) -> str:
-    raw = (task_kind or "").strip().lower().replace(" ", "_")
-    return _TASK_KIND_ALIASES.get(raw, raw)
-
-
-def validator_finding(
-    rule_id: str,
-    severity: str,
-    field: str,
-    reason: str,
-    safe_next_action: str,
-) -> dict[str, str]:
-    """Structured finding for controller handoff consumption."""
-    return {
-        "rule_id": rule_id,
-        "severity": severity,
-        "field": field,
-        "reason": reason,
-        "safe_next_action": safe_next_action,
-    }
-
-
-def _findings_from_reasons(
-    rule_id: str,
-    reasons: list[str],
-    *,
-    field: str = "",
-    severity: str = "downgrade",
-    safe_next_action: str = "downgrade final report; repair missing proof fields",
-) -> list[dict[str, str]]:
-    return [
-        validator_finding(rule_id, severity, field, reason, safe_next_action)
-        for reason in reasons
-    ]
-
-
-def _handoff_fields(report_text: str) -> dict[str, str]:
-    from review_proofs import _handoff_section_lines  # local import avoids cycle at load
-
-    section = _handoff_section_lines(report_text) or []
-    fields: dict[str, str] = {}
-    for line in section:
-        stripped = line.strip().lstrip("-*").strip()
-        if ":" not in stripped:
-            continue
-        key, value = stripped.split(":", 1)
-        fields[key.strip().lower()] = value.strip()
-    return fields
-
-
-def _rule_shared_controller_handoff(
-    report_text: str,
-    task_kind: str,
-    *,
-    local_edits: bool = False,
-) -> list[dict[str, str]]:
-    role = _HANDOFF_ROLE_BY_TASK.get(task_kind)
-    result = assess_controller_handoff(
-        report_text,
-        role=role,
-        local_edits=local_edits,
-    )
-    verdict = result.get("verdict")
-    if verdict == "complete":
-        return []
-    severity = "block" if verdict == "missing" else "downgrade"
-    return _findings_from_reasons(
-        "shared.controller_handoff",
-        result.get("reasons") or [],
-        field="Controller Handoff",
-        severity=severity,
-        safe_next_action=(
-            "add a complete Controller Handoff section with task-appropriate fields"
-            if verdict == "missing"
-            else "fill missing controller handoff fields before submission"
-        ),
-    )
-
-
-def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
-    result = assess_email_disclosure(report_text)
-    if result.get("proven"):
-        return []
-    return _findings_from_reasons(
-        "shared.email_disclosure",
-        result.get("reasons") or [],
-        field="Identity",
-        severity="downgrade",
-        safe_next_action="use username / profile identity; remove personal email",
-    )
-
-
-def _rule_reviewer_legacy_workspace_mutations(
-    report_text: str,
-    *,
-    mutations_observed: bool = False,
-) -> list[dict[str, str]]:
-    if not mutations_observed:
-        return []
-    if not _LEGACY_WORKSPACE_MUTATIONS_RE.search(report_text or ""):
-        return []
-    return [
-        validator_finding(
-            "reviewer.legacy_workspace_mutations",
-            "block",
-            "Workspace mutations",
-            "legacy 'Workspace mutations' field used after observed mutations; "
-            "use precise mutation categories instead",
-            "replace Workspace mutations with file/worktree/git/MCP category fields",
-        )
-    ]
-
-
-def _rule_reviewer_vague_mutations_none(
-    report_text: str,
-    *,
-    action_log: list[dict] | None = None,
-    mutations_observed: bool = False,
-) -> list[dict[str, str]]:
-    performed = any(
-        e.get("performed") is not False and not e.get("gated_rejected")
-        for e in (action_log or [])
-    )
-    if not (mutations_observed or performed):
-        return []
-    if not _VAGUE_MUTATIONS_NONE_RE.search(report_text or ""):
-        return []
-    return [
-        validator_finding(
-            "reviewer.vague_mutations_none",
-            "block",
-            "Mutations",
-            "report claims 'Mutations: none' while mutations were observed",
-            "list performed mutations under precise category fields",
-        )
-    ]
-
-
-def _rule_reviewer_mutation_categories(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    lower = text.lower()
-    if "mutations:" not in lower and "mutation categories" not in lower:
-        return [
-            validator_finding(
-                "reviewer.mutation_categories",
-                "downgrade",
-                "Mutations",
-                "reviewer handoff lacks precise mutation category fields",
-                "report file edits, worktree/index, git ref, MCP/Gitea, and review mutations separately",
-            )
-        ]
-    if any(term in lower for term in ("workspace mutations", "mutations: none")):
-        return []
-    if any(field in lower for field in _MUTATION_CATEGORY_FIELDS):
-        return []
-    vague_only = "mutations:" in lower and not any(
-        marker in lower
-        for marker in ("file edits", "worktree", "git ref", "mcp/gitea", "review mutation")
-    )
-    if vague_only:
-        return [
-            validator_finding(
-                "reviewer.mutation_categories",
-                "downgrade",
-                "Mutations",
-                "mutation section is vague; precise categories are required",
-                "split mutations into file edits, worktree/index, git ref, MCP/Gitea, review",
-            )
-        ]
-    return []
-
-
-def _rule_reviewer_git_fetch_readonly(
-    report_text: str,
-    *,
-    action_log: list[dict] | None = None,
-) -> list[dict[str, str]]:
-    text = report_text or ""
-    fetch_observed = any(
-        _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
-        for e in (action_log or [])
-    ) or _GIT_FETCH_RE.search(text)
-    if not fetch_observed:
-        return []
-
-    readonly_match = _READONLY_DIAG_RE.search(text)
-    readonly_value = (readonly_match.group(1) if readonly_match else "").strip().lower()
-    fields = _handoff_fields(text)
-    git_ref_value = fields.get("git ref mutations", "").strip().lower()
-    git_ref_reported = bool(git_ref_value) and git_ref_value not in {
-        "none",
-        "n/a",
-        "no",
-    }
-    if git_ref_reported:
-        return []
-    if readonly_match and _GIT_FETCH_RE.search(readonly_value):
-        return [
-            validator_finding(
-                "reviewer.git_fetch_readonly",
-                "block",
-                "Git ref mutations",
-                "git fetch occurred but was classified only as read-only diagnostics",
-                "classify git fetch under Git ref mutations, not read-only diagnostics",
-            )
-        ]
-    if not git_ref_reported and _GIT_FETCH_RE.search(text):
-        return [
-            validator_finding(
-                "reviewer.git_fetch_readonly",
-                "downgrade",
-                "Git ref mutations",
-                "git fetch mentioned without Git ref mutation classification",
-                "record git fetch under Git ref mutations",
-            )
-        ]
-    return []
-
-
-def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if not _BARE_PYTEST_RE.search(text):
-        return []
-    if _VALIDATION_ENV_RE.search(text):
-        return []
-    return [
-        validator_finding(
-            "reviewer.validation_command_proof",
-            "downgrade",
-            "Validation",
-            "bare pytest command without working-directory or executable path proof",
-            "report exact validation command, cwd, and path proof for bare commands",
-        )
-    ]
-
-
-def _rule_reviewer_linked_issue(
-    report_text: str,
-    *,
-    linked_issue_lock: dict | None = None,
-) -> list[dict[str, str]]:
-    lock = linked_issue_lock or {}
-    expected_issue = lock.get("issue_number")
-    expected_pr = lock.get("pr_number")
-    if expected_issue is None and expected_pr is None:
-        return []
-
-    text = report_text or ""
-    findings: list[dict[str, str]] = []
-    reported_issue = None
-    issue_match = _LINKED_ISSUE_NUMBER_RE.search(text)
-    if issue_match:
-        reported_issue = int(issue_match.group(1))
-    elif expected_issue is not None:
-        if f"#{expected_issue}" not in text and f"issue {expected_issue}" not in text.lower():
-            findings.append(
-                validator_finding(
-                    "reviewer.linked_issue_missing",
-                    "downgrade",
-                    "Linked issue",
-                    f"linked issue #{expected_issue} not documented in final report",
-                    "document the linked issue number matching the selected PR",
-                )
-            )
-
-    if expected_issue is not None and reported_issue is not None and reported_issue != expected_issue:
-        findings.append(
-            validator_finding(
-                "reviewer.linked_issue_mismatch",
-                "block",
-                "Linked issue",
-                f"reported linked issue #{reported_issue} does not match "
-                f"selected PR linked issue #{expected_issue}",
-                "reconcile linked issue proof with the selected PR before reporting status",
-            )
-        )
-
-    if expected_pr is not None:
-        pr_tokens = (f"pr #{expected_pr}", f"pr{expected_pr}", f"#pr{expected_pr}")
-        if not any(token in text.lower().replace(" ", "") for token in pr_tokens):
-            if f"#{expected_pr}" not in text:
-                findings.append(
-                    validator_finding(
-                        "reviewer.linked_pr_missing",
-                        "downgrade",
-                        "Selected PR",
-                        f"selected PR #{expected_pr} not consistently referenced",
-                        "reference the selected PR number in handoff and diagnostics",
-                    )
-                )
-    return findings
-
-
-def _rule_reviewer_baseline_on_failure(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if not (_APPROVAL_CLAIM_RE.search(text) and _FULL_SUITE_FAIL_RE.search(text)):
-        return []
-    if _BASELINE_WORKTREE_RE.search(text) and "baseline worktree path" in text.lower():
-        return []
-    return [
-        validator_finding(
-            "reviewer.baseline_on_full_suite_failure",
-            "block",
-            "Baseline worktree",
-            "approval claimed after full-suite failure without baseline worktree proof",
-            "create a clean branches/ baseline worktree and report path plus results",
-        )
-    ]
-
-
-def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if not _SAME_AS_MASTER_RE.search(text):
-        return []
-    if _BASELINE_WORKTREE_RE.search(text):
-        return []
-    return [
-        validator_finding(
-            "reviewer.main_checkout_baseline",
-            "block",
-            "Baseline worktree",
-            "vague 'same as master' claim without clean baseline worktree proof",
-            "validate against a clean branches/ baseline worktree; do not use main checkout",
-        )
-    ]
-
-
-def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if "baseline worktree path" not in text.lower():
-        return []
-    fields = _handoff_fields(text)
-    baseline_path = ""
-    for key, value in fields.items():
-        if key.startswith("baseline worktree path"):
-            baseline_path = value
-            break
-    if baseline_path and _MAIN_CHECKOUT_RE.search(baseline_path):
-        return [
-            validator_finding(
-                "reviewer.main_checkout_path",
-                "block",
-                "Baseline worktree path",
-                "baseline validation used main/shared checkout instead of branches/ worktree",
-                "recreate baseline proof under branches/ and report that path",
-            )
-        ]
-    return []
-
-
-def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if not _ALREADY_LANDED_RE.search(text):
-        return []
-    if not _ELIGIBLE_REVIEW_RE.search(text):
-        return []
-    return [
-        validator_finding(
-            "reviewer.already_landed_eligible_wording",
-            "block",
-            "Eligibility class",
-            "already-landed PR described as eligible for review or merge",
-            "classify as ALREADY_LANDED_RECONCILE_REQUIRED and stop normal review",
-        )
-    ]
-
-
-def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]:
-    from review_proofs import _handoff_section_lines
-
-    if _handoff_section_lines(report_text) is None:
-        return [
-            validator_finding(
-                "reconcile.controller_handoff",
-                "block",
-                "Controller Handoff",
-                f"final report has no section titled exactly '{HANDOFF_HEADING}'",
-                "add a reconciliation Controller Handoff section",
-            )
-        ]
-
-    fields = _handoff_fields(report_text)
-    missing = [
-        name
-        for name in _RECONCILE_REQUIRED_FIELDS
-        if not (fields.get(name.lower()) or "").strip()
-    ]
-    if missing:
-        return [
-            validator_finding(
-                "reconcile.controller_handoff",
-                "downgrade",
-                "Controller Handoff",
-                f"reconciliation handoff missing required field: {field}",
-                "complete the reconciliation handoff schema before submission",
-            )
-            for field in missing
-        ]
-    return []
-
-
-def _rule_reconcile_stale_author_fields(report_text: str) -> list[dict[str, str]]:
-    text = (report_text or "").lower()
-    findings = []
-    for field in _RECONCILE_STALE_FIELDS:
-        if f"{field}:" in text:
-            findings.append(
-                validator_finding(
-                    "reconcile.stale_author_reviewer_fields",
-                    "block",
-                    field,
-                    f"reconciliation handoff includes stale author/reviewer field '{field}'",
-                    "use reconciliation schema only; remove author/reviewer review fields",
-                )
-            )
-    return findings
-
-
-def _rule_reconcile_eligible_reviewed(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if not _ALREADY_LANDED_RE.search(text):
-        return []
-    findings = []
-    if _ELIGIBLE_REVIEW_RE.search(text):
-        findings.append(
-            validator_finding(
-                "reconcile.eligible_reviewed_wording",
-                "block",
-                "Eligibility class",
-                "already-landed PR described as eligible for review or merge",
-                "report ALREADY_LANDED_RECONCILE_REQUIRED and reconciliation-only next steps",
-            )
-        )
-    if _REVIEWED_LANDED_RE.search(text):
-        findings.append(
-            validator_finding(
-                "reconcile.reviewed_landed_wording",
-                "block",
-                "Current status",
-                "already-landed PR described as reviewed or approval-eligible",
-                "use reconciliation status wording only",
-            )
-        )
-    return findings
-
-
-def _rule_reconcile_linked_issue_live(
-    report_text: str,
-    *,
-    linked_issue_lock: dict | None = None,
-) -> list[dict[str, str]]:
-    lock = linked_issue_lock or {}
-    text = report_text or ""
-    status_match = _LINKED_ISSUE_STATUS_RE.search(text)
-    if not status_match:
-        return []
-
-    status_value = status_match.group(1).strip().lower()
-    if "not verified" in status_value:
-        return []
-
-    live_proof = lock.get("live_fetched") is True or lock.get("verified_live") is True
-    if live_proof:
-        return []
-
-    if status_value not in {"", "none", "n/a", "unknown"}:
-        return [
-            validator_finding(
-                "reconcile.linked_issue_live_status",
-                "downgrade",
-                "Linked issue live status",
-                "linked issue status reported without live verification proof in this session",
-                "fetch linked issue live or report 'not verified in this session'",
-            )
-        ]
-    return []
-
-
-def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]:
-    text = report_text or ""
-    if not _INVENTORY_COMPLETE_RE.search(text):
-        return []
-    if _PAGINATION_PROOF_RE.search(text):
-        return []
-    return [
-        validator_finding(
-            "reconcile.inventory_pagination_proof",
-            "downgrade",
-            "PR inventory",
-            "inventory completeness claimed without pagination or final-page proof",
-            "include pagination complete, final page, or total open PR count evidence",
-        )
-    ]
-
-
-def _rule_reviewer_validation_structured(
-    validation_report: dict | None,
-) -> list[dict[str, str]]:
-    if not validation_report:
-        return []
-    result = assess_validation_report(validation_report)
-    if result.get("verdict") == "strong":
-        return []
-    severity = "block" if result.get("verdict") == "invalid" else "downgrade"
-    return _findings_from_reasons(
-        "reviewer.validation_structured",
-        result.get("reasons") or [],
-        field="Validation",
-        severity=severity,
-        safe_next_action="provide exact validation command, output read proof, and counts",
-    )
-
-
-def _rule_reviewer_mutation_ledger(
-    report_text: str,
-    *,
-    action_log: list[dict] | None = None,
-) -> list[dict[str, str]]:
-    if not action_log:
-        return []
-    result = assess_mutation_ledger_report(report_text, action_log=action_log)
-    if result.get("proven"):
-        return []
-    return _findings_from_reasons(
-        "reviewer.mutation_ledger",
-        result.get("reasons") or [],
-        field="File edits by reviewer",
-        severity="block",
-        safe_next_action="align mutation ledger with observed file edits",
-    )
-
-
-def _rule_reviewer_review_mutation(
-    report_text: str,
-    *,
-    review_decision_lock: dict | None = None,
-) -> list[dict[str, str]]:
-    if not review_decision_lock:
-        return []
-    result = assess_review_mutation_final_report(report_text, review_decision_lock)
-    if result.get("complete"):
-        return []
-    return _findings_from_reasons(
-        "reviewer.review_mutation_proof",
-        result.get("reasons") or [],
-        field="Review mutations",
-        severity="downgrade",
-        safe_next_action="document exactly one live review mutation or authorized correction flow",
-    )
-
-
-_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
-    "review_pr": [
-        _rule_shared_controller_handoff,
-        _rule_shared_email_disclosure,
-        _rule_reviewer_legacy_workspace_mutations,
-        _rule_reviewer_vague_mutations_none,
-        _rule_reviewer_mutation_categories,
-        _rule_reviewer_git_fetch_readonly,
-        _rule_reviewer_validation_command,
-        _rule_reviewer_validation_structured,
-        _rule_reviewer_linked_issue,
-        _rule_reviewer_baseline_on_failure,
-        _rule_reviewer_main_checkout_baseline,
-        _rule_reviewer_main_checkout_path,
-        _rule_reviewer_already_landed_eligible,
-        _rule_reviewer_mutation_ledger,
-        _rule_reviewer_review_mutation,
-    ],
-    "reconcile_already_landed": [
-        _rule_reconcile_controller_handoff,
-        _rule_shared_email_disclosure,
-        _rule_reconcile_stale_author_fields,
-        _rule_reconcile_eligible_reviewed,
-        _rule_reconcile_linked_issue_live,
-        _rule_reconcile_pagination_proof,
-        _rule_reviewer_git_fetch_readonly,
-        _rule_reviewer_legacy_workspace_mutations,
-        _rule_reviewer_vague_mutations_none,
-    ],
-    "author_issue": [
-        _rule_shared_controller_handoff,
-        _rule_shared_email_disclosure,
-        _rule_reviewer_vague_mutations_none,
-    ],
-    "work_issue": [
-        _rule_shared_controller_handoff,
-        _rule_shared_email_disclosure,
-        _rule_reviewer_vague_mutations_none,
-    ],
-    "issue_filing": [
-        _rule_shared_controller_handoff,
-        _rule_shared_email_disclosure,
-    ],
-    "inventory": [
-        _rule_shared_controller_handoff,
-        _rule_shared_email_disclosure,
-        _rule_reconcile_pagination_proof,
-    ],
-    "issue_selection": [
-        _rule_shared_controller_handoff,
-        _rule_shared_email_disclosure,
-    ],
-}
-
-
-def _aggregate_grade(findings: list[dict[str, str]]) -> tuple[str, bool, bool]:
-    blocked = any(f["severity"] == "block" for f in findings)
-    downgraded = any(f["severity"] == "downgrade" for f in findings)
-    if blocked:
-        return "blocked", True, downgraded
-    if downgraded:
-        return "downgraded", False, True
-    return "A", False, False
-
-
-def _primary_safe_next_action(findings: list[dict[str, str]]) -> str:
-    for severity in ("block", "downgrade", "warning"):
-        for finding in findings:
-            if finding["severity"] == severity:
-                return finding["safe_next_action"]
-    return "proceed"
-
-
-def _call_rule(
-    rule: Callable[..., list[dict[str, str]]],
-    report_text: str,
-    task_kind: str,
-    rule_kwargs: dict[str, Any],
-) -> list[dict[str, str]]:
-    if rule is _rule_shared_controller_handoff:
-        return rule(
-            report_text,
-            task_kind,
-            local_edits=bool(rule_kwargs.get("local_edits")),
-        )
-    if rule is _rule_reviewer_validation_structured:
-        return rule(rule_kwargs.get("validation_report"))
-    if rule is _rule_reconcile_controller_handoff:
-        return rule(report_text)
-
-    bound: dict[str, Any] = {}
-    for name, param in inspect.signature(rule).parameters.items():
-        if name == "report_text":
-            bound[name] = report_text
-        elif name in rule_kwargs:
-            bound[name] = rule_kwargs[name]
-        elif param.default is inspect.Parameter.empty and param.kind != inspect.Parameter.VAR_KEYWORD:
-            continue
-    return rule(**bound)
-
-
-def assess_final_report_validator(
-    report_text: str,
-    task_kind: str,
-    *,
-    review_decision_lock: dict | None = None,
-    linked_issue_lock: dict | None = None,
-    validation_report: dict | None = None,
-    action_log: list[dict] | None = None,
-    mutations_observed: bool = False,
-    local_edits: bool = False,
-    issue_filing_lock: dict | None = None,
-) -> dict[str, Any]:
-    """Validate final-report text against task-specific proof rules (#327).
-
-    Returns structured findings plus a composite grade suitable for workflow
-    controllers. Optional *proof_locks* carry live session facts used to fail
-    closed on stale or contradictory report fields.
-    """
-    normalized_kind = _normalize_task_kind(task_kind)
-    if normalized_kind not in FINAL_REPORT_TASK_KINDS:
-        return {
-            "task_kind": normalized_kind,
-            "grade": "blocked",
-            "blocked": True,
-            "downgraded": False,
-            "findings": [
-                validator_finding(
-                    "shared.unknown_task_kind",
-                    "block",
-                    "Task",
-                    f"unknown task kind '{task_kind}'",
-                    "use a supported task kind from FINAL_REPORT_TASK_KINDS",
-                )
-            ],
-            "checks": {},
-            "reasons": [f"unknown task kind '{task_kind}'"],
-            "safe_next_action": "set task_kind to a supported workflow mode",
-            "complete": False,
-        }
-
-    checks: dict[str, Any] = {}
-    findings: list[dict[str, str]] = []
-
-    if normalized_kind == "issue_filing" and issue_filing_lock is not None:
-        checks["issue_filing"] = assess_issue_filing_final_report(
-            report_text,
-            **issue_filing_lock,
-        )
-        if checks["issue_filing"].get("downgraded"):
-            findings.extend(
-                _findings_from_reasons(
-                    "issue_filing.composite",
-                    checks["issue_filing"].get("reasons") or [],
-                    field="Issue filing",
-                    severity="downgrade",
-                )
-            )
-
-    rule_kwargs = {
-        "review_decision_lock": review_decision_lock,
-        "linked_issue_lock": linked_issue_lock,
-        "validation_report": validation_report,
-        "action_log": action_log,
-        "mutations_observed": mutations_observed,
-        "local_edits": local_edits,
-    }
-
-    for rule in _RULES_BY_TASK.get(normalized_kind, ()):
-        findings.extend(_call_rule(rule, report_text, normalized_kind, rule_kwargs))
-
-    grade, blocked, downgraded = _aggregate_grade(findings)
-    reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
-    return {
-        "task_kind": normalized_kind,
-        "grade": grade,
-        "blocked": blocked,
-        "downgraded": downgraded,
-        "findings": findings,
-        "checks": checks,
-        "reasons": reasons,
-        "safe_next_action": _primary_safe_next_action(findings),
-        "complete": grade == "A",
-        "handoff_heading": HANDOFF_HEADING,
-    }

review_proofs.py

@@ -1052,303 +1052,6 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
     return None
 
 
-# ── Reconciliation inventory pagination proof (Issue #308) ───────────────────
-#
-# Already-landed reconciliation may not claim a complete queue scan (or
-# that every already-landed PR was found) unless the report carries the
-# pagination proof fields and the per-page gitea_list_prs metadata
-# proves the final page was reached or the page limit was not hit.
-
-_RECONCILE_COMPLETE_CLAIM_RE = re.compile(
-    r"complete queue scan|all already-landed|inventory complete|"
-    r"all open prs (?:were )?(?:inspected|scanned|listed)|found all",
-    re.IGNORECASE,
-)
-
-RECONCILIATION_INVENTORY_PROOF_FIELDS = (
-    "requested page size",
-    "pages fetched",
-    "returned pr count per page",
-    "final page proof",
-)
-
-
-def assess_reconciliation_inventory_proof(report_text, *,
-                                          pagination_evidence=None):
-    """#308: reconciliation completeness claims need pagination proof.
-
-    *pagination_evidence* is the list of per-page ``pagination`` dicts
-    from the session's ``gitea_list_prs`` calls, in fetch order. Reports
-    that do not claim inventory completeness pass without proof; any
-    completeness claim requires the report proof fields and evidence
-    that the final page was reached (``inventory_complete``,
-    ``is_final_page`` with ``has_more=false``, or a returned count below
-    the requested page size).
-
-    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
-    """
-    reasons = []
-    if not _RECONCILE_COMPLETE_CLAIM_RE.search(report_text or ""):
-        return {"complete": True, "downgraded": False, "reasons": []}
-
-    fields = _report_labeled_fields(report_text)
-    for required in RECONCILIATION_INVENTORY_PROOF_FIELDS:
-        if required not in fields:
-            reasons.append(
-                "reconciliation completeness claimed but report missing "
-                f"pagination proof field '{required}'"
-            )
-
-    if not pagination_evidence:
-        reasons.append(
-            "reconciliation completeness claimed without pagination "
-            "evidence from gitea_list_prs"
-        )
-        return {"complete": False, "downgraded": True, "reasons": reasons}
-
-    evidence_valid = True
-    for page in pagination_evidence:
-        if not isinstance(page, dict) or not all(
-            key in page for key in ("per_page", "returned_count")
-        ):
-            reasons.append(
-                "pagination evidence page missing 'per_page'/"
-                "'returned_count' metadata"
-            )
-            evidence_valid = False
-
-    if evidence_valid:
-        last = pagination_evidence[-1]
-        final_proven = (
-            last.get("inventory_complete") is True
-            or (last.get("is_final_page") is True
-                and last.get("has_more") is False)
-            or last["returned_count"] < last["per_page"]
-        )
-        if not final_proven:
-            reasons.append(
-                "pagination evidence does not prove final page (need "
-                "inventory_complete, is_final_page with has_more=false, "
-                "or a returned count below the requested page size)"
-            )
-
-        pages_fetched = fields.get("pages fetched")
-        if pages_fetched:
-            declared = re.findall(r"\d+", pages_fetched)
-            if declared and int(declared[0]) != len(pagination_evidence):
-                reasons.append(
-                    f"report 'Pages fetched' ({declared[0]}) does not match "
-                    f"pagination evidence ({len(pagination_evidence)} pages)"
-                )
-
-        counts_field = fields.get("returned pr count per page")
-        if counts_field:
-            declared_counts = [int(n) for n in re.findall(r"\d+", counts_field)]
-            actual_counts = [p["returned_count"] for p in pagination_evidence]
-            if declared_counts != actual_counts:
-                reasons.append(
-                    "report 'Returned PR count per page' "
-                    f"({declared_counts}) does not match pagination "
-                    f"evidence {actual_counts}"
-                )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-# ── Linked-issue consistency (Issue #314) ────────────────────────────────────
-#
-# Stale linked-issue text from a previous PR must not leak into a final
-# report: the "Linked issue status" field must match the issue(s) the
-# selected PR actually links, verified live this session, and no other
-# "Issue #N" mention may appear unless it is a verified linked issue.
-
-_ISSUE_MENTION_RE = re.compile(r"\bissue\s+#(\d+)", re.IGNORECASE)
-
-# ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
-#
-# A worktree reset/checkout/clean is a workspace mutation even when the
-# 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 _issue_numbers_mentioned(text):
-    """Return the set of ints referenced as 'Issue #N' in *text*."""
-    return {int(n) for n in _ISSUE_MENTION_RE.findall(text or "")}
-
-
-def assess_linked_issue_consistency(report_text, *, selected_pr=None,
-                                    linked_issues=None):
-    """#314: linked-issue status must match the selected PR, live-verified.
-
-    *linked_issues* is the list of issue numbers the selected PR was
-    live-verified to link this session, or None when no live verification
-    happened. Without live proof the report may only say the status was
-    not verified; with proof, the ``Linked issue status`` field must name
-    every linked issue and no stale issue number may appear anywhere in
-    the report.
-
-    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
-    """
-    fields = _report_labeled_fields(report_text)
-    status_value = fields.get("linked issue status")
-
-    reasons = []
-
-    if status_value is None:
-        reasons.append(
-            "final report missing 'Linked issue status' field for the "
-            "selected PR"
-        )
-    elif linked_issues is None:
-        if not status_value.strip().lower().startswith("not verified"):
-            reasons.append(
-                "linked issue status claimed without live proof; report "
-                "'Linked issue status: not verified in this session' or "
-                "verify the linked issue live"
-            )
-    else:
-        allowed = set(linked_issues)
-        status_mentions = _issue_numbers_mentioned(status_value)
-        for missing in sorted(allowed - status_mentions):
-            reasons.append(
-                f"linked issue #{missing} of selected PR "
-                f"#{selected_pr} not reported in 'Linked issue status'"
-            )
-        for stale in sorted(status_mentions - allowed):
-            reasons.append(
-                f"'Linked issue status' names issue #{stale}, which is "
-                f"not a live-verified linked issue of selected PR "
-                f"#{selected_pr}"
-            )
-        for stale in sorted(_issue_numbers_mentioned(report_text) - allowed):
-            reasons.append(
-                f"stale issue #{stale} mentioned in final report but not "
-                f"live-verified as linked to selected PR #{selected_pr}"
-            )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def _field_claims_none(fields, label):
-    """True when *label* is present and its value is empty or 'none...'."""
-    value = fields.get(label)
-    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.
@@ -1529,7 +1232,6 @@ 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,
-                       queue_ordering=None,
                        baseline_validation=None, project_root=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
@@ -1557,22 +1259,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         )
 
     empty_queue_report = assess_empty_queue_report(report_text)
-    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)
@@ -1724,11 +1410,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "empty-queue report missing or failed trust-gate proof (#198)"
         )
         downgrade_reasons.extend(empty_queue_report.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)"
@@ -1755,12 +1436,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         # #179: no merge without a proven final live-state recheck.
         and live_state_proven
         and worktree_proven
-        and queue_ordering_proof.get("proven")
         and baseline_validation.get("proven")
     )
 
     violations = []
-    violations.extend(queue_ordering_proof.get("violations", []))
     violations.extend(baseline_validation.get("violations", []))
     if merge_performed and not merge_allowed:
         violations.append(
@@ -1813,11 +1492,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             else True
         ),
         "empty_queue_trust_gate_status": empty_queue_report.get("status"),
-        "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")),
@@ -1969,20 +1643,6 @@ 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",)),
@@ -1998,7 +1658,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")),
@@ -2149,25 +1809,6 @@ 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 = []
@@ -3429,162 +3070,6 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
     }
 
 
-# ---------------------------------------------------------------------------
-# 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)
 # ---------------------------------------------------------------------------
@@ -3847,13 +3332,6 @@ def assess_reviewer_fallback_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
-def assess_final_report_validator(report_text, task_kind, **kwargs):
-    """#327: single entry point for task-specific final-report validation."""
-    from final_report_validator import assess_final_report_validator as _validate
-
-    return _validate(report_text, task_kind, **kwargs)
-
-
 _QUEUE_STATUS_REPORT_HINT = re.compile(
     r"queue[- ]status|selected pr:\s*none|no pr selected|"
     r"queue[- ]status[- ]only",
@@ -3913,17 +3391,6 @@ _PROOF_WORDING_RULES: tuple[dict, ...] = (
         "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(
@@ -3989,12 +3456,7 @@ def assess_proof_wording(
     *,
     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*.
-    """
+    """Issue #330: reject unsupported proof phrases in final reports."""
     text = report_text or ""
     evidence = dict(session_evidence or {})
     violations: list[str] = []
@@ -4237,13 +3699,6 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
     }
 
 
-def assess_prior_blocker_skip_proof(report_text, **kwargs):
-    """#318: require live blocker proof before skipping earlier open PRs."""
-    from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess
-
-    return _assess(report_text, **kwargs)
-
-
 def assess_non_mergeable_skip_proof(report_text, **kwargs):
     """#322: require conflict proof when skipping non-mergeable PRs."""
     from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess

reviewer_blocker_skip.py

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

tests/test_final_report_validator.py

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

tests/test_linked_issue_consistency.py

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

tests/test_llm_workflow_split.py

@@ -124,25 +124,13 @@ def test_reviewer_fallback_verifier_exported():
     assert callable(assess_reviewer_fallback_report)
 
 
-def test_final_report_validator_exported():
-    from review_proofs import assess_final_report_validator
-
-    assert callable(assess_final_report_validator)
-
-
 def test_create_issue_final_report_verifier_exported():
     from review_proofs import assess_create_issue_final_report
 
     assert callable(assess_create_issue_final_report)
 
 
-def test_prior_blocker_skip_verifier_exported():
-    from review_proofs import assess_prior_blocker_skip_proof
-
-    assert callable(assess_prior_blocker_skip_proof)
-
-
 def test_non_mergeable_skip_verifier_exported():
     from review_proofs import assess_non_mergeable_skip_proof
 
-    assert callable(assess_non_mergeable_skip_proof)
+    assert callable(assess_non_mergeable_skip_proof)

tests/test_proof_wording.py

@@ -1,93 +0,0 @@
-"""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_reconciliation_inventory_proof.py

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

tests/test_review_proofs.py

@@ -24,7 +24,6 @@ from review_proofs import (  # noqa: E402
     ISSUE_SELECTION_CONTINUATION_EXPLICIT,
     ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
     assess_author_pr_report,
-    assess_queue_ordering_proof,
     assess_reviewer_baseline_validation_proof,
     assess_capability_evidence,
     assess_capability_proof,
@@ -941,18 +940,13 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("Safety", result["missing_fields"])
 
     def test_review_role_requires_review_fields(self):
-        # 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")
+        result = assess_controller_handoff(self.BASE_HANDOFF, 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 = review_base + "\n" + "\n".join([
+        complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected PR: #999",
             "- Reviewer eligibility: passed",
             "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
@@ -964,15 +958,6 @@ 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")
@@ -1095,144 +1080,6 @@ 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."""
@@ -2314,111 +2161,6 @@ class TestCreateIssueFinalReport(unittest.TestCase):
         self.assertTrue(result["downgraded"])
 
 
-class TestQueueOrderingProof(unittest.TestCase):
-    """Issue #321: reviewer queue selection must prove ordering."""
-
-    def test_next_eligible_claim_without_policy_fails(self):
-        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."""
 

tests/test_reviewer_blocker_skip.py

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

tests/test_workspace_mutation_consistency.py

@@ -1,165 +0,0 @@
-"""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()