diff --git a/review_proofs.py b/review_proofs.py index b8b41a5..73985b2 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -664,6 +664,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination, report_text, review_decision_lock ) + empty_queue_report = assess_empty_queue_report(report_text) + contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) validation_claimable = bool(validation.get("claimable")) @@ -787,6 +789,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "reviewer worktree safety proof missing or failed (#233)" ) downgrade_reasons.extend(worktree.get("reasons", [])) + if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"): + downgrade_reasons.append( + "empty-queue report missing or failed trust-gate proof (#198)" + ) + downgrade_reasons.extend(empty_queue_report.get("reasons", [])) merge_allowed = ( identity_eligible @@ -846,6 +853,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "unrelated_mutations_avoided": bool( worktree.get("unrelated_mutations_avoided") ), + "empty_queue_trust_gate_proven": ( + empty_queue_report.get("proven") + if empty_queue_report.get("claimed") + else True + ), + "empty_queue_trust_gate_status": empty_queue_report.get("status"), } @@ -1016,6 +1029,14 @@ HANDOFF_ROLE_FIELDS = { ("Repositories checked", ("repositories checked", "repos checked")), ("Open PR counts", ("open pr counts", "open pr count", "open prs per repo")), + ("PR inventory trust gate", ("pr inventory trust gate", + "pr_inventory_trust_gate.status", + "trust gate status")), + ("Trust gate reasons", ("trust gate reasons", + "pr_inventory_trust_gate.reason")), + ("Trust gate corroborated", ("trust gate corroborated", + "pr_inventory_trust_gate.corroborated")), + ("Inventory profile", ("inventory profile", "inventory mcp profile")), ("Selected PR or reason", ("selected pr", "none selected", "reason none selected")), ("Inventory completeness", ("inventory complete", "inventory scoped", @@ -1399,6 +1420,148 @@ def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]: return lines +_EMPTY_QUEUE_CLAIM = re.compile( + r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|" + r"nothing to review|queue cleared|inventory empty|" + r"open pr count:\s*0|workflow correctly stops with nothing", + re.I, +) + +_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile( + r"latest commit.*(?:merge|pr #)|merge of pr #|" + r"master latest commit|recent merge proves", + re.I, +) + +_TRUST_GATE_STATUS_LINE = re.compile( + r"pr_inventory_trust_gate\.status:\s*(\S+)", + re.I, +) + + +def parse_trust_gate_status_from_report(report_text: str | None) -> str | None: + """Extract ``pr_inventory_trust_gate.status`` from report text, if present.""" + match = _TRUST_GATE_STATUS_LINE.search(report_text or "") + return match.group(1).strip().lower() if match else None + + +def assess_empty_queue_report( + report_text: str | None, + *, + trust_gate: dict | None = None, + task_role: str | None = None, + inventory_profile: str | None = None, +) -> dict: + """Issue #198: empty-queue reports must cite the formal trust-gate result. + + Blocks reports that claim an empty queue without + ``pr_inventory_trust_gate.status == trusted_empty``, required inventory + metadata, or that rely on weak corroboration (e.g. a recent merge commit). + """ + text = report_text or "" + lower = text.lower() + reasons: list[str] = [] + missing: list[str] = [] + + if not _EMPTY_QUEUE_CLAIM.search(text): + return { + "claimed": False, + "proven": True, + "block": False, + "status": None, + "missing_fields": [], + "reasons": [], + } + + status = ( + (trust_gate or {}).get("status") + or parse_trust_gate_status_from_report(text) + ) + status_norm = (status or "").strip().lower() or None + + if not status_norm: + missing.append("pr_inventory_trust_gate.status") + reasons.append( + "empty-queue report missing pr_inventory_trust_gate.status; " + "fail closed" + ) + elif status_norm != "trusted_empty": + reasons.append( + f"empty-queue report has trust-gate status '{status_norm}', " + "not trusted_empty" + ) + + has_gate_reasons = ( + "pr_inventory_trust_gate.reason" in lower + or bool((trust_gate or {}).get("reasons")) + ) + if status_norm and status_norm != "trusted_empty" and not has_gate_reasons: + missing.append("pr_inventory_trust_gate.reasons") + + if status_norm == "trusted_empty": + if "pr_inventory_trust_gate.corroborated" not in lower and ( + trust_gate or {} + ).get("corroborated") is not True: + missing.append("pr_inventory_trust_gate.corroborated") + + inventory_markers = ( + "repository:", + "remote:", + "owner:", + "state filter:", + "state_filter:", + ) + if not any(marker in lower for marker in inventory_markers): + missing.append("inventory remote/owner/repo/state filter") + + profile_markers = ( + "mcp profile:", + "mcp-profile:", + "inventory profile:", + "active profile:", + ) + has_profile = ( + any(marker in lower for marker in profile_markers) + or bool((inventory_profile or "").strip()) + ) + if not has_profile: + missing.append("inventory MCP profile") + + if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text): + if "pr_inventory_trust_gate.status: trusted_empty" not in lower: + reasons.append( + "weak corroboration (recent merge commit) cannot substitute " + "for pr_inventory_trust_gate.status == trusted_empty" + ) + + role = (task_role or "").strip().lower() + if role == "author" and re.search( + r"reviewer queue|nothing to review|review backlog empty", + text, + re.I, + ): + reasons.append( + "author-bound session presented reviewer queue inventory as a " + "reviewer decision" + ) + + if missing: + reasons.extend( + f"empty-queue report missing required field: {field}" + for field in missing + ) + + proven = not reasons and not missing + return { + "claimed": True, + "proven": proven, + "block": not proven, + "status": status_norm, + "missing_fields": missing, + "reasons": reasons, + } + + def assess_duplicate_search_proof(report_text, matches): """#207: reject LLM duplicate summaries that omit known title matches.""" return issue_duplicate_gate.assess_duplicate_search_proof( diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 9f9e57c..e0596c0 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -25,6 +25,7 @@ from review_proofs import ( # noqa: E402 assess_capability_evidence, assess_capability_proof, assess_controller_handoff, + assess_empty_queue_report, assess_inventory_completeness, assess_reviewer_queue_inventory, assess_live_state_recheck, @@ -1020,6 +1021,10 @@ class TestControllerHandoff(unittest.TestCase): complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Repositories checked: Gitea-Tools, mcp-control-plane", "- Open PR counts: 2 / 0", + "- PR inventory trust gate: trusted_nonempty / trusted_empty", + "- Trust gate reasons: none", + "- Trust gate corroborated: true", + "- Inventory profile: prgs-reviewer", "- Selected PR or reason: none eligible (self-authored)", "- Inventory completeness: complete, no pagination needed", ]) @@ -1280,6 +1285,89 @@ class TestAssessReviewerQueueInventory(unittest.TestCase): self.assertEqual(result["trust_gates"], {}) +class TestAssessEmptyQueueReport(unittest.TestCase): + """Issue #198: empty-queue reports require formal trust-gate proof.""" + + def _trusted_report(self, **extra): + lines = [ + "Queue inventory complete.", + "Repository: Scaled-Tech-Consulting/Gitea-Tools", + "Open PR count: 0", + "pr_inventory_trust_gate.status: trusted_empty", + "pr_inventory_trust_gate.corroborated: true", + "Inventory profile: prgs-reviewer", + "Workflow correctly stops with nothing to review.", + ] + lines.extend(extra) + return "\n".join(lines) + + def test_non_empty_report_not_claimed(self): + result = assess_empty_queue_report("Reviewed PR #236 and merged.") + self.assertFalse(result["claimed"]) + self.assertTrue(result["proven"]) + + def test_empty_claim_without_trust_gate_blocked(self): + report = ( + "Open PR count: 0\n" + "Pagination complete: yes\n" + "Queue cleared." + ) + result = assess_empty_queue_report(report) + self.assertTrue(result["claimed"]) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_trusted_empty_report_with_required_fields_passes(self): + result = assess_empty_queue_report(self._trusted_report()) + self.assertTrue(result["proven"]) + + def test_weak_merge_commit_corroboration_blocked(self): + report = ( + "Open PR count: 0\n" + "Master latest commit is merge of PR #79 so queue is empty." + ) + result = assess_empty_queue_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("weak corroboration" in r for r in result["reasons"]) + ) + + def test_author_session_reviewer_queue_wording_blocked(self): + result = assess_empty_queue_report( + self._trusted_report(), + task_role="author", + ) + self.assertFalse(result["proven"]) + + def test_build_final_report_downgrades_weak_empty_queue(self): + final = 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-pr-1", + "porcelain_status": "", + "scratch_used": True, + "scratch_path": "/repo/branches/review-pr-1", + }, + report_text="Open PR count: 0. Queue cleared.", + ) + self.assertNotEqual(final["grade"], "A") + self.assertFalse(final["empty_queue_trust_gate_proven"]) + + class TestCapabilityEvidence(unittest.TestCase): """#179 gap 1: capability claims need exact evidence."""