Implement fail-closed PR inventory trust gate for Issue #194

2026-07-05 16:06:17 -04:00
parent 4dcf8fdfe4
commit b354c93710
2 changed files with 177 additions and 0 deletions
@@ -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,
+    }
@@ -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()