review_proofs.py

@@ -1351,6 +1351,7 @@ 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.
 
@@ -1378,6 +1379,22 @@ 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)
@@ -1529,6 +1546,11 @@ 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)"
@@ -1555,10 +1577,12 @@ 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(
@@ -1611,6 +1635,11 @@ 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")),
@@ -3189,6 +3218,162 @@ 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)
 # ---------------------------------------------------------------------------

tests/test_review_proofs.py

@@ -24,6 +24,7 @@ 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,
@@ -2161,6 +2162,111 @@ 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."""