From 1033a22407a96c5813861457ce99f4220e01a962 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 02:08:55 -0400 Subject: [PATCH] feat(reviewer): wire PR inventory trust gate into live queue path (#196) Run pr_inventory_trust_gate on empty gitea_review_pr inventory results, add assess_reviewer_queue_inventory for multi-repo completeness checks, and require trusted_empty before empty-queue claims in reports/templates. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 43 ++++++++- review_proofs.py | 87 +++++++++++++++++++ .../templates/review-pr.md | 6 ++ tests/test_pr_queue_inventory.py | 69 ++++++++++++++- tests/test_review_proofs.py | 69 +++++++++++++++ 5 files changed, 272 insertions(+), 2 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 9b915da..20bd535 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -259,6 +259,7 @@ import issue_duplicate_gate # noqa: E402 import role_session_router # noqa: E402 import role_namespace_gate # noqa: E402 import task_capability_map # noqa: E402 +import review_proofs # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -2274,6 +2275,23 @@ def gitea_merge_pr( return result +def _local_git_remote_url(remote_name: str) -> str | None: + """Best-effort local ``git remote get-url`` for trust-gate corroboration.""" + try: + proc = subprocess.run( + ["git", "remote", "get-url", remote_name], + capture_output=True, + text=True, + cwd=PROJECT_ROOT, + ) + if proc.returncode != 0: + return None + url = (proc.stdout or "").strip() + return url or None + except Exception: + return None + + @mcp.tool() def gitea_review_pr( pr_number: int, @@ -2341,6 +2359,8 @@ def gitea_review_pr( prs_found_count = 0 pr_details_list = [] inventory_msg = "" + inventory_trust_gate = None + prs: list = [] h, o, r = _resolve(remote, host, org, repo) auth = None @@ -2425,7 +2445,28 @@ def gitea_review_pr( if inventory_msg: report_lines.append(inventory_msg) else: - report_lines.append("Open PRs found: 0") + inventory_trust_gate = review_proofs.pr_inventory_trust_gate( + prs if inventory_attempted else None, + remote=remote, + org=o, + repo=r, + state="open", + authenticated_profile=profile, + local_remote_url=_local_git_remote_url(remote), + has_finality_metadata=True, + ) + report_lines.extend( + review_proofs.format_pr_inventory_trust_gate_report( + inventory_trust_gate + ) + ) + if inventory_trust_gate.get("status") == "trusted_empty": + report_lines.append("Open PRs found: 0 (trusted_empty)") + else: + report_lines.append( + "Empty-queue claim blocked: " + "pr_inventory_trust_gate did not return trusted_empty" + ) else: report_lines.append(inventory_msg) diff --git a/review_proofs.py b/review_proofs.py index c9fa6f5..b8b41a5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1312,6 +1312,93 @@ def pr_inventory_trust_gate( } +def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]: + parts = (full_repo or "").split("/", 1) + if len(parts) == 2: + return parts[0].strip() or None, parts[1].strip() or None + return None, None + + +def assess_reviewer_queue_inventory( + repo_reports: list[dict] | None, + required_repos: list[str] | None = None, + *, + user_context: str | None = None, +) -> dict: + """Canonical reviewer queue path: completeness plus per-repo trust gates (#196). + + Any repository reporting ``open_pr_count == 0`` must pass + ``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue + claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient. + """ + required = list(required_repos or [ + "Scaled-Tech-Consulting/Gitea-Tools", + "Scaled-Tech-Consulting/mcp-control-plane", + ]) + completeness = assess_inventory_completeness(repo_reports, required) + + trust_gates: dict[str, dict] = {} + blockers: list[str] = [] + can_claim_empty = bool(completeness.get("complete")) + + for report in repo_reports or []: + repo = (report.get("repo") or "").strip() + count = report.get("open_pr_count") + if not isinstance(count, int) or count != 0: + continue + + org, repo_name = _split_repo_slug(repo) + list_response = report.get("list_prs_response") + if list_response is None: + list_response = [] + + gate = pr_inventory_trust_gate( + list_response, + remote=report.get("remote"), + org=org, + repo=repo_name, + state=report.get("state_filter"), + authenticated_profile=report.get("authenticated_profile"), + local_remote_url=report.get("local_remote_url"), + user_context=user_context or report.get("user_context"), + corroboration_open_pr_counter=report.get( + "corroboration_open_pr_counter" + ), + has_finality_metadata=report.get("pagination_complete") is True, + ) + trust_gates[repo] = gate + status = gate.get("status") + if status != "trusted_empty": + can_claim_empty = False + blockers.append( + f"repository '{repo}': empty PR list trust gate is " + f"'{status}'; cannot claim 'no open PRs'" + ) + blockers.extend(gate.get("reasons") or []) + + return { + "complete": bool(completeness.get("complete")), + "can_claim_empty_queue": can_claim_empty, + "can_claim_exhaustive": ( + bool(completeness.get("can_claim_exhaustive")) and can_claim_empty + ), + "inventory_reasons": list(completeness.get("reasons") or []), + "trust_gates": trust_gates, + "blockers": blockers, + "reasons": list(completeness.get("reasons") or []) + blockers, + } + + +def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]: + """Render trust-gate lines for MCP inventory output.""" + lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"] + if gate.get("corroborated"): + lines.append("pr_inventory_trust_gate.corroborated: true") + for reason in gate.get("reasons") or []: + lines.append(f"pr_inventory_trust_gate.reason: {reason}") + return lines + + 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/skills/llm-project-workflow/templates/review-pr.md b/skills/llm-project-workflow/templates/review-pr.md index 15f348a..7116774 100644 --- a/skills/llm-project-workflow/templates/review-pr.md +++ b/skills/llm-project-workflow/templates/review-pr.md @@ -17,6 +17,12 @@ Repo name disambiguation (Gitea-Tools blind review hardening): configured repos were not checked. This is not a complete queue inventory." - A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs" if the other configured repo was not inventoried. +- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty", + the workflow must run `pr_inventory_trust_gate` (via the live inventory path or + `review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a + clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and + corroboration in the final report. A bare `[]` from `gitea_list_prs` is never + sufficient proof. Rules (llm-project-workflow): - Review in a SEPARATE detached review worktree, never the author's folder. diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py index 4866f99..8919cfb 100644 --- a/tests/test_pr_queue_inventory.py +++ b/tests/test_pr_queue_inventory.py @@ -261,12 +261,79 @@ class TestPRQueueInventory(unittest.TestCase): for call in mock_api.call_args_list: self.assertEqual(call.args[0], "GET") + @patch("mcp_server._local_git_remote_url") @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty( + self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url + ): + mock_get_profile.return_value = { + "profile_name": "gitea-author", + "allowed_operations": ["read"], + "forbidden_operations": [], + "base_url": None, + } + mock_get_all.return_value = [] + mock_api.return_value = {"login": "jcwalker3"} + mock_local_url.return_value = None + + result = gitea_review_pr( + pr_number=1, + event="APPROVE", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertFalse(result["success"]) + msg = result["message"] + self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg) + self.assertIn("Empty-queue claim blocked", msg) + self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg) + + @patch("mcp_server._local_git_remote_url") + @patch("mcp_server.api_get_all") + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_empty_inventory_trusted_empty_when_gate_passes( + self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url + ): + mock_get_profile.return_value = { + "profile_name": "gitea-author", + "allowed_operations": ["read"], + "forbidden_operations": [], + "base_url": None, + } + mock_get_all.return_value = [] + mock_api.return_value = {"login": "jcwalker3"} + mock_local_url.return_value = ( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + ) + + result = gitea_review_pr( + pr_number=1, + event="APPROVE", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertFalse(result["success"]) + msg = result["message"] + self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg) + self.assertIn("Open PRs found: 0 (trusted_empty)", msg) + + @patch("mcp_server._local_git_remote_url") + @patch("mcp_server.api_get_all") + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url): """Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory.""" + mock_local_url.return_value = ( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + ) for event in ["APPROVE", "REQUEST_CHANGES"]: mock_get_profile.return_value = { "profile_name": "gitea-author", diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 20ea1e9..9f9e57c 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -26,6 +26,7 @@ from review_proofs import ( # noqa: E402 assess_capability_proof, assess_controller_handoff, assess_inventory_completeness, + assess_reviewer_queue_inventory, assess_live_state_recheck, assess_review_mutation_final_report, assess_role_boundary, @@ -1211,6 +1212,74 @@ class TestPRInventoryTrustGate(unittest.TestCase): self.assertTrue(res["corroborated"]) +class TestAssessReviewerQueueInventory(unittest.TestCase): + """Issue #196: trust gate wired into canonical queue inventory.""" + + def _repo_report(self, **overrides): + report = { + "repo": "Scaled-Tech-Consulting/Gitea-Tools", + "state_filter": "open", + "pagination_complete": True, + "open_pr_count": 0, + "list_prs_response": [], + "remote": "prgs", + "authenticated_profile": { + "profile_name": "prgs-reviewer", + "allowed_operations": ["read", "gitea.read"], + }, + "local_remote_url": ( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + ), + } + report.update(overrides) + return report + + def test_empty_without_corroboration_blocks_empty_queue_claim(self): + result = assess_reviewer_queue_inventory([ + self._repo_report(pagination_complete=False), + self._repo_report( + repo="Scaled-Tech-Consulting/mcp-control-plane", + open_pr_count=0, + pagination_complete=False, + ), + ]) + self.assertFalse(result["can_claim_empty_queue"]) + self.assertIn("untrusted_empty", str(result["trust_gates"])) + + def test_trusted_empty_with_finality_allows_empty_queue_claim(self): + result = assess_reviewer_queue_inventory([ + self._repo_report(), + self._repo_report( + repo="Scaled-Tech-Consulting/mcp-control-plane", + local_remote_url=( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/" + "mcp-control-plane.git" + ), + ), + ]) + self.assertTrue(result["can_claim_empty_queue"]) + self.assertTrue(result["can_claim_exhaustive"]) + + def test_user_context_indicating_open_prs_blocks_empty_claim(self): + result = assess_reviewer_queue_inventory( + [self._repo_report()], + user_context="please review open PR #195 in the queue", + ) + self.assertFalse(result["can_claim_empty_queue"]) + self.assertTrue(result["blockers"]) + + def test_nonempty_inventory_skips_empty_trust_gate_block(self): + result = assess_reviewer_queue_inventory([ + self._repo_report(open_pr_count=2), + self._repo_report( + repo="Scaled-Tech-Consulting/mcp-control-plane", + open_pr_count=1, + ), + ]) + self.assertTrue(result["complete"]) + self.assertEqual(result["trust_gates"], {}) + + class TestCapabilityEvidence(unittest.TestCase): """#179 gap 1: capability claims need exact evidence."""