Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge pull request 'feat(reviewer): wire PR inventory trust gate into live queue path (closes #196)' (#236) from feat/issue-196-pr-inventory-trust-gate-live-path into master

Browse Source
This commit was merged in pull request #236.
This commit is contained in:
sysadmin
2026-07-06 10:14:22 -05:00
parent 1071619532 1033a22407
commit be6feabf70
6 changed files with 357 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+42 -1
View File
@@ -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)
review_proofs.py
+88
View File
@@ -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(
skills/llm-project-workflow/templates/review-pr.md
+6
View File
@@ -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.
tests/test_mcp_server.py
+49 -13
View File
@@ -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")
tests/test_pr_queue_inventory.py
+68 -1
View File
@@ -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",
tests/test_review_proofs.py
+104 -17
View File
@@ -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."""