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 5e5d7e0..b8b41a5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1005,6 +1005,7 @@ HANDOFF_ROLE_FIELDS = { ), "author": ( ("Selected issue", ("selected issue",)), + ("Issue lock proof", ("issue lock proof", "lock before diff")), ("Claim/comment status", ("claim/comment status", "claim status", "claim")), ("PR number opened", ("pr number opened", "pr opened", "pr number")), @@ -1311,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_mcp_server.py b/tests/test_mcp_server.py index aa90d88..9a80d8d 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -65,6 +65,20 @@ CREATE_PR_ENV = { ), } +ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" + + +def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides): + record = { + "issue_number": issue_number, + "branch_name": branch_name, + "remote": "dadeschools", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + } + record.update(overrides) + return record + # --------------------------------------------------------------------------- # Create Issue @@ -127,15 +141,19 @@ class TestCreatePR(unittest.TestCase): @patch("os.path.exists", return_value=True) @patch("builtins.open") def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role): - mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' + lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) + mock_open.return_value.__enter__.return_value.read.return_value = lock_json mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} with patch.dict(os.environ, CREATE_PR_ENV, clear=True): result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertEqual(result["number"], 3) self.assertNotIn("url", result) + mock_exists.assert_called_with(ISSUE_LOCK_FILE) + mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") payload = mock_api.call_args[0][3] self.assertEqual(payload["head"], "feat/x") self.assertEqual(payload["base"], "main") + self.assertIn("Closes #123", payload["title"]) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @@ -144,13 +162,28 @@ class TestCreatePR(unittest.TestCase): @patch("os.path.exists", return_value=True) @patch("builtins.open") def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role): - mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' + lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) + mock_open.return_value.__enter__.return_value.read.return_value = lock_json mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} with patch.dict(os.environ, env, clear=True): result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertIn("pulls/3", result["url"]) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("os.path.exists", return_value=True) + @patch("builtins.open") + def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role): + lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) + mock_open.return_value.__enter__.return_value.read.return_value = lock_json + with patch.dict(os.environ, CREATE_PR_ENV, clear=True): + with self.assertRaises(ValueError) as ctx: + gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main") + self.assertIn("Closes #123", str(ctx.exception)) + mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") + # --------------------------------------------------------------------------- # Close Issue @@ -2770,8 +2803,8 @@ class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" def tearDown(self): - if os.path.exists("/tmp/gitea_issue_lock.json"): - os.remove("/tmp/gitea_issue_lock.json") + if os.path.exists(ISSUE_LOCK_FILE): + os.remove(ISSUE_LOCK_FILE) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2779,7 +2812,7 @@ class TestIssueLocking(unittest.TestCase): mock_api.return_value = [] # no open PRs res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertTrue(res["success"]) - self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json")) + self.assertTrue(os.path.exists(ISSUE_LOCK_FILE)) def test_lock_issue_mismatch_branch_fails(self): with self.assertRaises(ValueError) as ctx: @@ -2816,8 +2849,8 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_missing_lock_fails(self, _auth, _role): - if os.path.exists("/tmp/gitea_issue_lock.json"): - os.remove("/tmp/gitea_issue_lock.json") + if os.path.exists(ISSUE_LOCK_FILE): + os.remove(ISSUE_LOCK_FILE) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(RuntimeError) as ctx: gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs") @@ -2827,8 +2860,9 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_branch_mismatch_fails(self, _auth, _role): - with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=196, branch_name="feat/issue-196-mutations"), f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(ValueError) as ctx: gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs") @@ -2838,8 +2872,9 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_forbidden_terms_fails(self, _auth, _role): - with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=196, branch_name="feat/issue-196-mutations"), f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): for term in ("equivalent to #196", "related to #196", "same as #196"): with self.assertRaises(ValueError) as ctx: @@ -2850,8 +2885,9 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_missing_closes_ref_fails(self, _auth, _role): - with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=196, branch_name="feat/issue-196-mutations"), f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(ValueError) as ctx: gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs") 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 eb3a462..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, @@ -943,24 +944,46 @@ class TestControllerHandoff(unittest.TestCase): result = assess_controller_handoff(complete, role="review") self.assertEqual(result["verdict"], "complete") - def test_author_role_requires_author_fields(self): - complete = self.BASE_HANDOFF + "\n" + "\n".join([ - "- Selected issue: #182", + def _author_role_fields(self, issue_number=182, pr_number=999): + return [ + f"- Selected issue: #{issue_number}", + "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", - "- PR number opened: #999", + f"- PR number opened: #{pr_number}", "- No review/merge: confirmed", - ]) + ] + + def test_handoff_role_fields_author_includes_issue_lock_proof(self): + from review_proofs import HANDOFF_ROLE_FIELDS + names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]] + self.assertIn("Issue lock proof", names) + + def test_author_role_requires_author_fields(self): + complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields()) result = assess_controller_handoff(complete, role="author") self.assertEqual(result["verdict"], "complete") result = assess_controller_handoff(self.BASE_HANDOFF, role="author") self.assertEqual(result["verdict"], "incomplete") + self.assertIn("Issue lock proof", result["missing_fields"]) self.assertIn("No review/merge confirmation", result["missing_fields"]) + def test_author_role_requires_issue_lock_proof(self): + without_lock = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: #182", + "- Claim/comment status: comment-claimed", + "- PR number opened: #999", + "- No review/merge: confirmed", + ]) + result = assess_controller_handoff(without_lock, role="author") + self.assertEqual(result["verdict"], "incomplete") + self.assertIn("Issue lock proof", result["missing_fields"]) + def test_author_role_rejects_equivalent_or_multiple_issues(self): # 1. equivalent reference blocked incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: Issue #194 / #196 equivalent", + "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", @@ -972,6 +995,7 @@ class TestControllerHandoff(unittest.TestCase): # 2. multiple issues blocked incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #194, #196", + "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", @@ -983,6 +1007,7 @@ class TestControllerHandoff(unittest.TestCase): def test_author_role_rejects_fuzzy_pr_number(self): incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #196", + "- Issue lock proof: lock before diff on feat/issue-196 @ master", "- Claim/comment status: comment-claimed", "- PR number opened: PR #203 / #204 equivalent", "- No review/merge: confirmed", @@ -1014,23 +1039,17 @@ class TestControllerHandoff(unittest.TestCase): def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self): # 1. Workspace mutations: none is rejected when local_edits is True - incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ - "- Selected issue: #196", - "- Claim/comment status: comment-claimed", - "- PR number opened: #203", - "- No review/merge: confirmed", - ]) + incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join( + self._author_role_fields(issue_number=196, pr_number=203)) res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True) self.assertEqual(res["verdict"], "incomplete") self.assertIn("Workspace mutations", res["missing_fields"]) # 2. Workspace mutations: edited files is allowed when local_edits is True - complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([ - "- Selected issue: #196", - "- Claim/comment status: comment-claimed", - "- PR number opened: #203", - "- No review/merge: confirmed", - ]) + complete_eq = self.BASE_HANDOFF.replace( + "- Workspace mutations: none", + "- Workspace mutations: edited review_proofs.py", + ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203)) res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True) self.assertEqual(res2["verdict"], "complete") @@ -1193,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."""