review_proofs.py

@@ -1259,6 +1259,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         )
 
     empty_queue_report = assess_empty_queue_report(report_text)
+    queue_status_report = (
+        assess_queue_status_report(report_text)
+        if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
+        else {"proven": True, "block": False, "reasons": [], "violations": []}
+    )
+    proof_wording = (
+        assess_proof_wording(report_text)
+        if report_text
+        else {"proven": True, "block": False, "reasons": [], "violations": []}
+    )
     if baseline_validation is None and report_text:
         baseline_validation = assess_reviewer_baseline_validation_proof(
             report_text=report_text,
@@ -1400,6 +1410,16 @@ 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 proof_wording.get("proven"):
+        downgrade_reasons.append(
+            "unsupported proof wording in final report (#330)"
+        )
+        downgrade_reasons.extend(proof_wording.get("reasons", []))
+    if not queue_status_report.get("proven"):
+        downgrade_reasons.append(
+            "queue-status report violates loaded workflow proof gates (#339)"
+        )
+        downgrade_reasons.extend(queue_status_report.get("reasons", []))
     if not baseline_validation.get("proven"):
         downgrade_reasons.append(
             "reviewer baseline validation proof missing or failed (#325)"
@@ -1472,6 +1492,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             else True
         ),
         "empty_queue_trust_gate_status": empty_queue_report.get("status"),
+        "proof_wording_proven": bool(proof_wording.get("proven")),
+        "proof_wording_violations": list(proof_wording.get("violations") or []),
+        "queue_status_report_proven": bool(queue_status_report.get("proven")),
+        "queue_status_violations": list(
+            queue_status_report.get("violations") or []
+        ),
         "baseline_validation_proven": bool(baseline_validation.get("proven")),
         "baseline_validation_violations": list(
             baseline_validation.get("violations") or []
@@ -3299,6 +3325,257 @@ def assess_email_disclosure(
     }
 
 
+_QUEUE_STATUS_REPORT_HINT = re.compile(
+    r"queue[- ]status|selected pr:\s*none|no pr selected|"
+    r"queue[- ]status[- ]only",
+    re.I,
+)
+
+_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
+
+_NOT_APPLICABLE_VALUE = re.compile(
+    r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b",
+    re.I,
+)
+
+_PRIOR_PROOF_LABEL = re.compile(
+    r"prior (?:blocker|proof|request[- ]changes|feedback)|"
+    r"head sha unchanged since|prior blocker reused|labeled as prior",
+    re.I,
+)
+
+_PAGINATION_FINALITY_EVIDENCE = re.compile(
+    r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
+    r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
+    r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
+    r"total_count\s*:|pagination.*(?:final|complete)|"
+    r"inventory pagination proof:",
+    re.I,
+)
+
+_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
+    r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
+    r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|"
+    r"(?:complete|exhaustive).*(?:default )?page[- ]?size|"
+    r"page[- ]?size assumption",
+    re.I,
+)
+
+_PROOF_WORDING_RULES: tuple[dict, ...] = (
+    {
+        "id": "live_proof",
+        "pattern": re.compile(r"\blive (?:blocker )?proof\b", re.I),
+        "session_keys": ("live_session_proof", "session_proof_commands"),
+        "text_evidence": re.compile(
+            r"current[- ]session|ran in (?:the )?current session|"
+            r"proof (?:command|tool).*(?:current|this) session|"
+            r"revalidated in (?:the )?current session",
+            re.I,
+        ),
+        "allow_prior_label": True,
+    },
+    {
+        "id": "inventory_complete",
+        "pattern": re.compile(
+            r"\binventory (?:is )?complete\b|\binventory completeness\b",
+            re.I,
+        ),
+        "session_keys": ("pagination_complete", "inventory_complete"),
+        "text_evidence": _PAGINATION_FINALITY_EVIDENCE,
+        "allow_prior_label": False,
+    },
+    {
+        "id": "no_file_edits",
+        "pattern": re.compile(
+            r"\b(?:no file edits|file edits none|workspace mutations none)\b",
+            re.I,
+        ),
+        "session_keys": ("no_file_edits_proven", "mutation_ledger_clean"),
+        "text_evidence": re.compile(
+            r"mutation ledger|worktree mutations:\s*none|"
+            r"file edits:\s*none|tracked file edits:\s*none|"
+            r"no tracked (?:file )?edits",
+            re.I,
+        ),
+        "allow_prior_label": False,
+    },
+)
+
+_QUEUE_STATUS_GATES_NO_PR = (
+    "already-landed gate",
+    "author-safety result",
+    "merge preflight",
+)
+
+_REVIEW_WORKTREE_DETAIL_FIELDS = (
+    "review worktree dirty before validation",
+    "review worktree dirty after validation",
+    "review worktree head state",
+)
+
+
+def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]:
+    """Parse ``- Field: value`` lines from the Controller Handoff section."""
+    section = _handoff_section_lines(report_text)
+    if section is None:
+        return {}
+    fields: dict[str, str] = {}
+    for line in section:
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" not in stripped:
+            continue
+        key, value = stripped.split(":", 1)
+        fields[key.strip().lower()] = value.strip()
+    return fields
+
+
+def _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool:
+    selected = fields.get("selected pr", "")
+    if selected and _NOT_APPLICABLE_VALUE.search(selected):
+        return True
+    if re.search(r"\bnone\b", selected, re.I):
+        return True
+    if re.search(
+        r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none",
+        report_text or "",
+        re.I,
+    ):
+        return True
+    return not selected
+
+
+def assess_proof_wording(
+    report_text: str | None,
+    *,
+    session_evidence: dict | None = None,
+) -> dict:
+    """Issue #330: reject unsupported proof phrases in final reports."""
+    text = report_text or ""
+    evidence = dict(session_evidence or {})
+    violations: list[str] = []
+    flagged_rules: list[str] = []
+
+    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
+        if not _PAGINATION_FINALITY_EVIDENCE.search(text) and not (
+            evidence.get("pagination_complete") or evidence.get("inventory_complete")
+        ):
+            violations.append(
+                "inventory pagination assumed from default page size without "
+                "final-page/no-next-page/total-count/traversal proof"
+            )
+            flagged_rules.append("default_page_size_assumption")
+
+    for rule in _PROOF_WORDING_RULES:
+        if not rule["pattern"].search(text):
+            continue
+        if rule.get("allow_prior_label") and _PRIOR_PROOF_LABEL.search(text):
+            continue
+        if any(evidence.get(key) for key in rule.get("session_keys") or ()):
+            continue
+        if rule["text_evidence"].search(text):
+            continue
+        violations.append(
+            f"forbidden proof phrase ({rule['id']}) without matching evidence"
+        )
+        flagged_rules.append(rule["id"])
+
+    proven = not violations
+    return {
+        "proven": proven,
+        "block": not proven,
+        "violations": violations,
+        "flagged_rules": flagged_rules,
+        "reasons": violations,
+    }
+
+
+def assess_queue_status_report(
+    report_text: str | None,
+    *,
+    session_evidence: dict | None = None,
+) -> dict:
+    """Issue #339: reject contradictory reviewer queue-status-only reports."""
+    text = report_text or ""
+    fields = _controller_handoff_field_map(text)
+    violations: list[str] = []
+
+    proof = assess_proof_wording(text, session_evidence=session_evidence)
+    violations.extend(proof.get("violations", []))
+
+    if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search(
+        text
+    ):
+        violations.append(
+            "prior diagnostic worktree simulation used as current conflict proof"
+        )
+
+    queue_only = _queue_status_only_run(fields, text)
+    if queue_only:
+        for gate in _QUEUE_STATUS_GATES_NO_PR:
+            value = fields.get(gate, "")
+            if value and _GATE_PASSED_VALUE.search(value):
+                if not _NOT_APPLICABLE_VALUE.search(value):
+                    violations.append(
+                        f"{gate} cannot be 'passed' when no PR is selected (#339)"
+                    )
+
+        worktree_used = fields.get("review worktree used", "")
+        if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I):
+            for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS:
+                detail_value = fields.get(detail_field, "")
+                if (
+                    detail_value
+                    and not _NOT_APPLICABLE_VALUE.search(detail_value)
+                ):
+                    violations.append(
+                        f"{detail_field} must be 'not applicable' or 'none' "
+                        "when no review worktree was created"
+                    )
+
+    blockers = fields.get("blockers", "")
+    if blockers and re.search(r"\bnone\b", blockers, re.I):
+        if re.search(
+            r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|"
+            r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)",
+            text,
+            re.I,
+        ):
+            violations.append(
+                "Blockers: none contradicts report stating all PRs are "
+                "conflicted, blocked, or unverified"
+            )
+
+    if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I):
+        has_skip_proof = bool(
+            re.search(
+                r"current[- ]session|prior blocker reused|conflict proof:|"
+                r"gitea_view_pr|review[- ]feedback proof|unchanged head",
+                text,
+                re.I,
+            )
+            or (session_evidence or {}).get("skip_proof_per_pr")
+        )
+        if not has_skip_proof:
+            violations.append(
+                "skipped PRs listed without current-session or labeled prior proof"
+            )
+
+    if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations:
+        violations.append(
+            "canonical workflow cited but mandatory queue-status proof gates violated"
+        )
+
+    deduped = list(dict.fromkeys(violations))
+    proven = not deduped
+    return {
+        "proven": proven,
+        "block": not proven,
+        "queue_status_only": queue_only,
+        "violations": deduped,
+        "reasons": deduped,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Git ref mutations (#297)
 # ---------------------------------------------------------------------------

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

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

tests/test_queue_status_report.py

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