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 7412cc0..b8b41a5 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")), @@ -978,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")), @@ -1284,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/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..7116774 100644 --- a/skills/llm-project-workflow/templates/review-pr.md +++ b/skills/llm-project-workflow/templates/review-pr.md @@ -17,9 +17,24 @@ 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. +- 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_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 47ee28d..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, @@ -183,6 +184,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 +614,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 +925,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", @@ -913,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", @@ -942,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", @@ -953,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", @@ -984,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") @@ -1163,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.""" @@ -1306,6 +1423,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