From 6ac6b9528c80c5b4d1c16def41337adae7da73ac Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 02:02:26 -0400 Subject: [PATCH 1/3] test: require issue-lock proof in author handoffs (#208) Add Issue lock proof to author HANDOFF_ROLE_FIELDS and update controller handoff tests. Strengthen create_pr tests to read and validate the issue lock file (branch match, Closes #N, reveal opt-in unchanged). Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 1 + tests/test_mcp_server.py | 62 +++++++++++++++++++++++++++++-------- tests/test_review_proofs.py | 52 +++++++++++++++++++++---------- 3 files changed, 85 insertions(+), 30 deletions(-) diff --git a/review_proofs.py b/review_proofs.py index 7412cc0..eb5f61e 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -978,6 +978,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")), 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_review_proofs.py b/tests/test_review_proofs.py index 47ee28d..f3bfb5f 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -913,24 +913,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", @@ -942,6 +964,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", @@ -953,6 +976,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", @@ -984,23 +1008,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") From 4f466550ca6040e3d3b647855a0c79ab7edf18aa Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 02:04:34 -0400 Subject: [PATCH 2/3] feat(reviewer): fail closed on dirty worktrees and forbidden git cleanup (#233) Add reviewer_worktree proofs that block stash/reset/checkout -- cleanup of unrelated local files, require scratch worktree reporting when the main tree is dirty outside PR scope, and integrate the gate into final report grading and Controller Handoff review fields. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 29 ++- reviewer_worktree.py | 219 ++++++++++++++++++ .../templates/review-pr.md | 9 + tests/test_review_proofs.py | 37 +++ tests/test_reviewer_worktree.py | 166 +++++++++++++ 5 files changed, 459 insertions(+), 1 deletion(-) create mode 100644 reviewer_worktree.py create mode 100644 tests/test_reviewer_worktree.py diff --git a/review_proofs.py b/review_proofs.py index eb5f61e..c9fa6f5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -17,6 +17,7 @@ here weakens or replaces them. import re import issue_duplicate_gate +from reviewer_worktree import assess_reviewer_worktree_proof _FULL_SHA = re.compile(r"^[0-9a-f]{40}$") @@ -637,7 +638,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, role_boundary=None, review_mutation=None, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, - sweep_proof=None): + sweep_proof=None, worktree_proof=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -707,6 +708,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "downgraded": True, "reasons": ["review mutation proof not provided (#211)"], } + if worktree_proof is not None: + worktree = assess_reviewer_worktree_proof(worktree_proof) + else: + worktree = { + "proven": False, + "block": True, + "reasons": ["reviewer worktree proof not provided (#233)"], + } capability_proven = bool(capability_evidence.get("proven")) sweep_proven = bool(sweep.get("proven")) @@ -719,6 +728,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "reasons": ["review mutation proof missing"], } review_mutation_complete = bool(review_mutation.get("complete")) + worktree_proven = bool(worktree.get("proven")) downgrade_reasons = [] if not identity_eligible: @@ -772,6 +782,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, if not review_mutation_complete: downgrade_reasons.append("review mutation proof missing or incomplete (#211)") downgrade_reasons.extend(review_mutation.get("reasons", [])) + if not worktree_proven: + downgrade_reasons.append( + "reviewer worktree safety proof missing or failed (#233)" + ) + downgrade_reasons.extend(worktree.get("reasons", [])) merge_allowed = ( identity_eligible @@ -782,6 +797,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, and validation.get("verdict") != "invalid" # #179: no merge without a proven final live-state recheck. and live_state_proven + and worktree_proven ) violations = [] @@ -825,6 +841,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "live_state_recheck_proven": live_state_proven, "role_boundary_clean": role_boundary_clean, "review_mutation_complete": review_mutation_complete, + "worktree_proof_proven": worktree_proven, + "worktree_scratch_used": bool(worktree.get("scratch_used")), + "unrelated_mutations_avoided": bool( + worktree.get("unrelated_mutations_avoided") + ), } @@ -971,6 +992,12 @@ HANDOFF_ROLE_FIELDS = { ("Selected PR", ("selected pr",)), ("Reviewer eligibility", ("reviewer eligibility", "eligibility")), ("Pinned reviewed head", ("pinned reviewed head", "pinned head")), + ("Worktree path", ("worktree path", "starting worktree path")), + ("Worktree dirty", ("worktree dirty", "whether worktree was dirty")), + ("Scratch worktree used", ("scratch worktree used", "scratch clone used", + "scratch worktree")), + ("Unrelated local mutations", ("unrelated local mutations", + "unrelated files modified")), ("Review decision", ("review decision", "decision")), ("Merge result", ("merge result",)), ("Linked issue status", ("linked issue status", "linked issue")), diff --git a/reviewer_worktree.py b/reviewer_worktree.py new file mode 100644 index 0000000..c35155e --- /dev/null +++ b/reviewer_worktree.py @@ -0,0 +1,219 @@ +"""Fail-closed reviewer worktree and local-git safety proofs (#233). + +Reviewer sessions must never stash, reset, or otherwise manipulate unrelated +local changes from another session. When the active worktree has dirty tracked +files outside the PR scope, the workflow must stop or switch to a disposable +scratch worktree (``scripts/worktree-review``). +""" + +from __future__ import annotations + +import re +import shlex + +# Subcommands that mutate unrelated local state — forbidden for reviewers. +_FORBIDDEN_REVIEWER_GIT = re.compile( + r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?" + r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|" + r"checkout\s+--|" + r"restore\s+|" + r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|" + r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|" + r"cherry-pick|" + r"rebase|" + r"merge|" + r"commit(?:\s+(?:--amend|-a|-am))?|" + r"push(?:\s+(?:--force|--force-with-lease))?|" + r"branch\s+-[dD]|" + r"worktree\s+remove)", + re.IGNORECASE, +) + +# Read-only git operations reviewers may use for validation. +_READONLY_REVIEWER_GIT = re.compile( + r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?" + r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|" + r"worktree\s+list|worktree\s+add)", + re.IGNORECASE, +) + + +def parse_dirty_tracked_files(porcelain: str) -> list[str]: + """Return tracked paths with local modifications from ``git status --porcelain``. + + Untracked entries (``??``) are ignored — they do not block reviewer work + when a scratch worktree is used, and authors may have unrelated untracked + files without implying reviewer interference. + """ + paths: list[str] = [] + for line in (porcelain or "").splitlines(): + if not line or len(line) < 4: + continue + if line.startswith("??"): + continue + path = line[3:].strip() + if " -> " in path: + path = path.split(" -> ", 1)[1].strip() + if path: + paths.append(path) + return paths + + +def files_outside_pr_scope( + dirty_files: list[str] | None, + pr_scope_files: list[str] | None, +) -> list[str]: + """Dirty tracked files not explained by the PR diff file set.""" + dirty = [p for p in (dirty_files or []) if p] + scope = {p for p in (pr_scope_files or []) if p} + if not dirty: + return [] + if not scope: + return list(dirty) + return [path for path in dirty if path not in scope] + + +def is_forbidden_reviewer_git_command(command: str) -> bool: + """True when a shell command would mutate unrelated local/remote git state.""" + text = (command or "").strip() + if not text: + return False + return bool(_FORBIDDEN_REVIEWER_GIT.search(text)) + + +def is_readonly_reviewer_git_command(command: str) -> bool: + """True when the command is an explicitly allowed read-only git operation.""" + text = (command or "").strip() + if not text: + return False + if is_forbidden_reviewer_git_command(text): + return False + return bool(_READONLY_REVIEWER_GIT.search(text)) + + +def assess_reviewer_git_command_log(commands: list[str] | None) -> dict: + """Fail closed when reviewer shell history includes forbidden git mutations.""" + forbidden = [ + cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd) + ] + if forbidden: + return { + "proven": False, + "block": True, + "forbidden_commands": forbidden, + "reasons": [ + "reviewer workflow executed forbidden local git mutation: " + f"{cmd!r}" + for cmd in forbidden + ], + "safe_next_action": ( + "stop; report worktree interference; do not stash/reset/checkout " + "unrelated files — use scripts/worktree-review instead" + ), + } + return { + "proven": True, + "block": False, + "forbidden_commands": [], + "reasons": [], + "safe_next_action": "proceed", + } + + +def assess_reviewer_worktree_proof(proof: dict | None) -> dict: + """Evaluate reviewer worktree safety before checkout/diff/validation/review. + + *proof* keys: + - ``worktree_path`` (required) + - ``porcelain_status`` or ``dirty_files`` + - ``pr_scope_files`` (paths in the PR diff) + - ``scratch_used`` (bool) + - ``scratch_path`` (when scratch_used) + - ``git_commands`` (shell commands executed this session) + - ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported + """ + proof = dict(proof or {}) + reasons: list[str] = [] + worktree_path = (proof.get("worktree_path") or "").strip() + if not worktree_path: + reasons.append("reviewer worktree path not reported; fail closed") + + if proof.get("dirty_files") is not None: + dirty_files = list(proof.get("dirty_files") or []) + else: + dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "") + + pr_scope = list(proof.get("pr_scope_files") or []) + unrelated = files_outside_pr_scope(dirty_files, pr_scope) + scratch_used = bool(proof.get("scratch_used")) + scratch_path = (proof.get("scratch_path") or "").strip() + + is_dirty = bool(dirty_files) + unrelated_dirty = bool(unrelated) + + if unrelated_dirty and not scratch_used: + reasons.append( + "worktree has dirty tracked files outside PR scope " + f"({', '.join(unrelated)}); stop or use a scratch worktree" + ) + if scratch_used and not scratch_path: + reasons.append( + "scratch worktree was used but scratch_path was not reported" + ) + if proof.get("unrelated_mutations_claimed"): + reasons.append( + "reviewer reported stash/reset/checkout cleanup of unrelated " + "local changes; this is forbidden" + ) + + command_assessment = assess_reviewer_git_command_log( + list(proof.get("git_commands") or []) + ) + if command_assessment["block"]: + reasons.extend(command_assessment["reasons"]) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "worktree_path": worktree_path or None, + "is_dirty": is_dirty, + "dirty_files": dirty_files, + "unrelated_dirty_files": unrelated, + "scratch_used": scratch_used, + "scratch_path": scratch_path or None, + "unrelated_mutations_avoided": not bool( + proof.get("unrelated_mutations_claimed") + or command_assessment.get("forbidden_commands") + ), + "safe_next_action": ( + "proceed" + if proven + else command_assessment.get("safe_next_action") + or "stop; use scripts/worktree-review or report dirty worktree" + ), + "forbidden_commands": command_assessment.get("forbidden_commands", []), + } + + +def assess_author_worktree_continuity(proof: dict | None) -> dict: + """Authors may keep dirty feature worktrees; reviewers may not manipulate them. + + This helper only proves the task role is author when dirty unrelated files + exist — it does not grant reviewers an exception. + """ + proof = dict(proof or {}) + role = (proof.get("task_role") or "").strip().lower() + dirty_files = list(proof.get("dirty_files") or []) + if role == "author" and dirty_files: + return { + "allowed": True, + "reasons": [ + "author task may continue with dirty tracked files in its own " + "worktree; reviewer interference rules do not apply" + ], + } + if role == "reviewer" and dirty_files: + return assess_reviewer_worktree_proof(proof) + return {"allowed": True, "reasons": []} \ No newline at end of file diff --git a/skills/llm-project-workflow/templates/review-pr.md b/skills/llm-project-workflow/templates/review-pr.md index 3a2dfae..15f348a 100644 --- a/skills/llm-project-workflow/templates/review-pr.md +++ b/skills/llm-project-workflow/templates/review-pr.md @@ -20,6 +20,15 @@ Repo name disambiguation (Gitea-Tools blind review hardening): Rules (llm-project-workflow): - Review in a SEPARATE detached review worktree, never the author's folder. +- Worktree safety (#233): before checkout, diff, validation, review, or merge, + report the starting worktree path and whether it was dirty. If unrelated + tracked files exist outside the PR scope, STOP or run + `scripts/worktree-review ` and validate in the scratch path. + NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`, + or `git clean` to manage another session's dirty files. +- Final report must state: Worktree path, Worktree dirty (yes/no), + Scratch worktree used (yes/no + path if yes), and confirm no unrelated local + files were modified, stashed, reset, or dropped. - You must NOT be the PR author. If the authenticated user == PR author, stop. A different LLM-Agent-SHA does NOT make you a different actor — only a different authenticated Gitea user does (docs/llm-agent-sha.md). diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index f3bfb5f..20ea1e9 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -183,6 +183,24 @@ def _good_review_mutation(): return assess_review_mutation_final_report(report, lock) +def _good_worktree(**overrides): + proof = { + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + "scratch_path": "/repo/branches/review-feat-issue-224", + "git_commands": [ + "git fetch prgs master feat/issue-224-wiki-proof-refresh", + "git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh", + ], + } + proof.update(overrides) + from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402 + + return assess_reviewer_worktree_proof(proof) + + def _good_role_boundary_179(**overrides): kwargs = { "task_role": "reviewer", @@ -595,6 +613,13 @@ class TestFinalReport(unittest.TestCase): "controller_handoff": _good_handoff(), "capability_proof": _good_capability_proof(), "sweep_proof": _good_secret_sweep(), + "worktree_proof": { + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + "scratch_path": "/repo/branches/review-feat-issue-224", + }, } kwargs.update(overrides) return build_final_report(**kwargs) @@ -899,12 +924,17 @@ class TestControllerHandoff(unittest.TestCase): result = assess_controller_handoff(self.BASE_HANDOFF, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertIn("Pinned reviewed head", result["missing_fields"]) + self.assertIn("Worktree path", result["missing_fields"]) self.assertIn("Merge result", result["missing_fields"]) complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected PR: #999", "- Reviewer eligibility: passed", "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Worktree path: /repo/branches/review-pr-999", + "- Worktree dirty: no", + "- Scratch worktree used: yes (/repo/branches/review-pr-999)", + "- Unrelated local mutations: none", "- Review decision: approve", "- Merge result: merged", "- Linked issue status: closed", @@ -1324,6 +1354,13 @@ class TestFinalReport179Bar(unittest.TestCase): "controller_handoff": _good_handoff(), "capability_proof": _good_capability_proof(), "sweep_proof": _good_secret_sweep(), + "worktree_proof": { + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + "scratch_path": "/repo/branches/review-feat-issue-224", + }, } kwargs.update(overrides) return build_final_report(**kwargs) diff --git a/tests/test_reviewer_worktree.py b/tests/test_reviewer_worktree.py new file mode 100644 index 0000000..11dfee0 --- /dev/null +++ b/tests/test_reviewer_worktree.py @@ -0,0 +1,166 @@ +"""Tests for reviewer worktree safety proofs (Issue #233).""" +import sys +import unittest + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +from reviewer_worktree import ( # noqa: E402 + assess_author_worktree_continuity, + assess_reviewer_git_command_log, + assess_reviewer_worktree_proof, + files_outside_pr_scope, + is_forbidden_reviewer_git_command, + is_readonly_reviewer_git_command, + parse_dirty_tracked_files, +) + + +class TestParseDirtyTrackedFiles(unittest.TestCase): + def test_ignores_untracked_files(self): + porcelain = "?? untracked.txt\n M tracked.py\n" + self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"]) + + def test_parses_renamed_paths(self): + porcelain = "R old.py -> new.py\n" + self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"]) + + +class TestFilesOutsidePrScope(unittest.TestCase): + def test_all_dirty_in_scope_is_clean(self): + self.assertEqual( + files_outside_pr_scope( + ["docs/wiki/Repositories.md"], + ["docs/wiki/Repositories.md"], + ), + [], + ) + + def test_unrelated_dirty_files_detected(self): + self.assertEqual( + files_outside_pr_scope( + ["review_proofs.py", "docs/wiki/Repositories.md"], + ["docs/wiki/Repositories.md"], + ), + ["review_proofs.py"], + ) + + +class TestForbiddenGitCommands(unittest.TestCase): + def test_blocks_stash_operations(self): + self.assertTrue(is_forbidden_reviewer_git_command("git stash")) + self.assertTrue( + is_forbidden_reviewer_git_command( + 'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py' + ) + ) + self.assertTrue(is_forbidden_reviewer_git_command("git stash pop")) + self.assertTrue(is_forbidden_reviewer_git_command("git stash drop")) + + def test_blocks_checkout_reset_and_clean(self): + self.assertTrue( + is_forbidden_reviewer_git_command( + "git checkout -- review_proofs.py tests/test_mcp_server.py" + ) + ) + self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard")) + self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd")) + + def test_allows_readonly_commands(self): + for cmd in ( + "git fetch prgs master", + "git status --porcelain", + "git diff prgs/master...HEAD", + "git rev-parse HEAD", + "git -C /repo log -1", + ): + with self.subTest(cmd=cmd): + self.assertFalse(is_forbidden_reviewer_git_command(cmd)) + self.assertTrue(is_readonly_reviewer_git_command(cmd)) + + +class TestAssessReviewerWorktreeProof(unittest.TestCase): + def test_clean_worktree_proceeds(self): + result = assess_reviewer_worktree_proof({ + "worktree_path": "/repo/branches/review-pr-231", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": False, + "git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"], + }) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_dirty_unrelated_without_scratch_blocks(self): + result = assess_reviewer_worktree_proof({ + "worktree_path": "/repo", + "dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"], + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": False, + }) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0]) + + def test_scratch_worktree_allows_dirty_main_repo(self): + result = assess_reviewer_worktree_proof({ + "worktree_path": "/repo", + "dirty_files": ["review_proofs.py"], + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + "scratch_path": "/repo/branches/review-feat-issue-224", + }) + self.assertTrue(result["proven"]) + + def test_forbidden_command_history_blocks(self): + result = assess_reviewer_worktree_proof({ + "worktree_path": "/repo/branches/review-pr-231", + "porcelain_status": "", + "git_commands": ["git stash push -m temp -- review_proofs.py"], + }) + self.assertFalse(result["proven"]) + self.assertTrue(result["forbidden_commands"]) + + def test_unrelated_mutations_claimed_blocks(self): + result = assess_reviewer_worktree_proof({ + "worktree_path": "/repo", + "porcelain_status": "", + "unrelated_mutations_claimed": True, + }) + self.assertFalse(result["proven"]) + + +class TestAuthorContinuity(unittest.TestCase): + def test_author_may_keep_dirty_worktree(self): + result = assess_author_worktree_continuity({ + "task_role": "author", + "dirty_files": ["feat.py"], + }) + self.assertTrue(result["allowed"]) + + def test_reviewer_dirty_worktree_uses_reviewer_gate(self): + result = assess_author_worktree_continuity({ + "task_role": "reviewer", + "worktree_path": "/repo", + "dirty_files": ["other.py"], + "pr_scope_files": ["docs/a.md"], + "scratch_used": False, + }) + self.assertFalse(result["proven"]) + + +class TestAssessReviewerGitCommandLog(unittest.TestCase): + def test_empty_log_is_clean(self): + result = assess_reviewer_git_command_log([]) + self.assertTrue(result["proven"]) + + def test_mixed_log_blocks_on_forbidden(self): + result = assess_reviewer_git_command_log([ + "git fetch prgs", + "git stash", + ]) + self.assertFalse(result["proven"]) + self.assertEqual(len(result["forbidden_commands"]), 1) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file 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 3/3] 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."""