diff --git a/review_proofs.py b/review_proofs.py index a0e4fed..2ec3239 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -533,3 +533,97 @@ def assess_controller_handoff(report_text, role=None): "missing_fields": [], "reasons": [], } + + +# ── PR Inventory Trust Gate (Issue #194) ────────────────────────────────────── +# +# A reviewer agent may not convert an empty PR list response into a definitive +# "no open PRs" conclusion unless the inventory result is independently proven +# trustworthy. + +def pr_inventory_trust_gate( + list_prs_response: list | None, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + state: str | None = None, + authenticated_profile: dict | None = None, + local_remote_url: str | None = None, + user_context: str | None = None, + corroboration_open_pr_counter: int | None = None, + has_finality_metadata: bool = False, +) -> dict: + """Evaluate whether an empty PR list is trusted or untrusted. + + Returns a dict with 'status', 'reasons', and 'corroborated'. + """ + if list_prs_response is None or not isinstance(list_prs_response, list): + return { + "status": "inventory_error", + "reasons": ["PR list response is invalid (not a list or None)"], + "corroborated": False, + } + + if len(list_prs_response) > 0: + return { + "status": "trusted_nonempty", + "reasons": [], + "corroborated": False, + } + + reasons = [] + + # 1. Exact remote, owner, repo, and state filter resolved correctly + if not remote or remote not in ("dadeschools", "prgs"): + reasons.append("remote instance is invalid or unresolved") + if not org or not org.strip(): + reasons.append("owner/org is invalid or unresolved") + if not repo or not repo.strip(): + reasons.append("repository name is invalid or unresolved") + if state != "open": + reasons.append("state filter is not 'open'") + + # 2. Authenticated profile permission check + if not authenticated_profile or not isinstance(authenticated_profile, dict): + reasons.append("authenticated profile is missing or invalid") + else: + allowed = authenticated_profile.get("allowed_operations") or [] + if "gitea.read" not in allowed and "read" not in allowed: + reasons.append("authenticated profile lacks read permissions") + + # 3. Pagination/finality metadata or independent read path corroboration + corroborated = False + if has_finality_metadata: + corroborated = True + elif corroboration_open_pr_counter == 0: + corroborated = True + else: + reasons.append("pagination finality not proven and open_pr_counter corroboration is missing or non-zero") + + # 4. Local checkout remote URL matching the target repo + if not local_remote_url or not isinstance(local_remote_url, str): + reasons.append("local checkout remote URL is missing or invalid") + else: + expected = f"{org}/{repo}".lower() + if expected not in local_remote_url.lower(): + reasons.append(f"local remote URL does not match target repository '{org}/{repo}'") + + # 5. User context check (indicators that PRs should exist) + if user_context and isinstance(user_context, str): + indicators = ["pr #", "pull request #", "open pr", "pr queue"] + found = [ind for ind in indicators if ind in user_context.lower()] + if found: + reasons.append(f"user context indicates open PRs should exist (matched: {', '.join(found)})") + + if reasons: + return { + "status": "untrusted_empty", + "reasons": reasons, + "corroborated": corroborated, + } + + return { + "status": "trusted_empty", + "reasons": [], + "corroborated": corroborated, + } diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 07dfc40..115d09e 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -26,6 +26,7 @@ from review_proofs import ( # noqa: E402 assess_self_review_contamination, assess_validation_report, build_final_report, + pr_inventory_trust_gate, resolve_repos_from_user_reference, verify_pinned_head_checkout, ) @@ -679,5 +680,87 @@ class TestControllerHandoff(unittest.TestCase): self.assertIn("issue #182", skill) +class TestPRInventoryTrustGate(unittest.TestCase): + """Issue #194: unit tests for the PR inventory trust gate.""" + + def setUp(self): + self.profile = { + "profile_name": "prgs-reviewer", + "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"], + } + self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + + def test_trusted_nonempty(self): + res = pr_inventory_trust_gate([{"number": 1}]) + self.assertEqual(res["status"], "trusted_nonempty") + self.assertFalse(res["corroborated"]) + + def test_inventory_error_none_or_not_list(self): + self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error") + self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error") + + def test_untrusted_empty_no_pagination_or_corroboration(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=None, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"]) + + def test_untrusted_empty_profile_permission_mismatch(self): + bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]} + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=bad_profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertIn("authenticated profile lacks read permissions", res["reasons"]) + + def test_untrusted_empty_remote_url_mismatch(self): + bad_url = "https://gitea.prgs.cc/other-org/other-repo.git" + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=bad_url, user_context=None, + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"]) + + def test_untrusted_empty_user_context_indicates_prs(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context="please check open PR #181", + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"])) + + def test_trusted_empty_with_corroboration(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "trusted_empty") + self.assertTrue(res["corroborated"]) + + def test_trusted_empty_with_finality_metadata(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=None, has_finality_metadata=True + ) + self.assertEqual(res["status"], "trusted_empty") + self.assertTrue(res["corroborated"]) + + if __name__ == "__main__": unittest.main()