From 880fca81da97ca16f2c268a3cc28e1a071ce9fe5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 5 Jul 2026 16:45:28 -0400 Subject: [PATCH] Enforce single terminal review decision per PR review run (Issue #211) Reviewer agents could post probe APPROVE/REQUEST_CHANGES reviews while testing lock paths, polluting PR audit trails. Add a review decision lock seeded by gitea_resolve_task_capability(review_pr) that requires gitea_mark_final_review_decision and final_review_decision_ready=True before gitea_submit_pr_review performs a live mutation. Add gitea_dry_run_pr_review for read-only submission validation, gitea_authorize_review_correction for operator-approved fixes, and assess_review_mutation_final_report for final-report proof. One live review mutation per run unless correction is explicitly authorized. --- mcp_server.py | 341 ++++++++++++++++++++++----- review_proofs.py | 48 ++++ skills/llm-project-workflow/SKILL.md | 9 + tests/test_mcp_server.py | 179 +++++++++++++- tests/test_review_proofs.py | 50 ++++ 5 files changed, 556 insertions(+), 71 deletions(-) diff --git a/mcp_server.py b/mcp_server.py index 0218616..1fe3f2b 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -13,6 +13,7 @@ Configuration (mcp_config.json): "env": {} } """ +import json import os import re import sys @@ -921,6 +922,121 @@ _REVIEW_ACTIONS = { "request_changes": ("request_changes", "REQUEST_CHANGES"), } +_TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"}) + +REVIEW_DECISION_FILE = "/tmp/gitea_review_decision.lock" + + +def _load_review_decision_lock(): + if not os.path.exists(REVIEW_DECISION_FILE): + return None + try: + with open(REVIEW_DECISION_FILE, "r", encoding="utf-8") as f: + return json.load(f) + except Exception: + return None + + +def _save_review_decision_lock(data): + try: + with open(REVIEW_DECISION_FILE, "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception: + pass + + +def init_review_decision_lock(remote: str | None, task: str | None): + """Seed read-only-until-ready state for reviewer PR review tasks.""" + if task != "review_pr": + return + _save_review_decision_lock({ + "task": task, + "remote": remote, + "final_review_decision_ready": False, + "ready_pr_number": None, + "ready_action": None, + "ready_expected_head_sha": None, + "live_mutations": [], + "correction_authorized": False, + "correction_reason": None, + }) + + +def check_review_decision_gate( + pr_number: int, + action: str, + *, + final_review_decision_ready: bool, +) -> list[str]: + """Fail closed unless validation completed and the final decision is ready.""" + reasons = [] + lock = _load_review_decision_lock() + if lock is None: + reasons.append( + "review decision lock missing; call gitea_resolve_task_capability " + "for review_pr before live review mutations (fail closed)" + ) + return reasons + + if not final_review_decision_ready: + reasons.append( + "final_review_decision_ready must be true; validation-phase live " + "review mutations are forbidden — use gitea_dry_run_pr_review " + "instead (fail closed)" + ) + return reasons + + if not lock.get("final_review_decision_ready"): + reasons.append( + "final review decision not marked ready; call " + "gitea_mark_final_review_decision after validation completes " + "(fail closed)" + ) + return reasons + + if lock.get("ready_pr_number") != pr_number: + reasons.append( + f"ready PR #{lock.get('ready_pr_number')} does not match " + f"requested PR #{pr_number} (fail closed)" + ) + if lock.get("ready_action") != action: + reasons.append( + f"ready action '{lock.get('ready_action')}' does not match " + f"requested action '{action}' (fail closed)" + ) + + prior = list(lock.get("live_mutations") or []) + if prior and not lock.get("correction_authorized"): + reasons.append( + "live review mutation already recorded in this run; only one live " + "review mutation is allowed unless " + "gitea_authorize_review_correction was invoked (fail closed)" + ) + elif ( + action in _TERMINAL_REVIEW_ACTIONS + and any(m.get("action") in _TERMINAL_REVIEW_ACTIONS for m in prior) + and not lock.get("correction_authorized") + ): + reasons.append( + "terminal review decision already submitted on this PR in this " + "run; blocked unless an operator-approved correction was " + "authorized (fail closed)" + ) + + return reasons + + +def record_live_review_mutation(pr_number: int, action: str): + lock = _load_review_decision_lock() or {} + mutations = list(lock.get("live_mutations") or []) + mutations.append({"pr_number": pr_number, "action": action}) + lock["live_mutations"] = mutations + if lock.get("correction_authorized"): + lock["correction_authorized"] = False + lock["correction_reason"] = None + _save_review_decision_lock(lock) + + # Patterns scrubbed from any surfaced error text so a credential can never leak. _SECRET_PREFIXES = ("token ", "Basic ") @@ -1085,9 +1201,7 @@ def gitea_get_pr_review_feedback( } -@mcp.tool() -@_audit_pr_result("submit_pr_review") -def gitea_submit_pr_review( +def _evaluate_pr_review_submission( pr_number: int, action: str, body: str = "", @@ -1096,55 +1210,17 @@ def gitea_submit_pr_review( host: str | None = None, org: str | None = None, repo: str | None = None, + *, + live: bool, + final_review_decision_ready: bool = False, ) -> dict: - """Gated PR review mutation: comment findings, request changes, or approve. - - This is the only tool that submits a Gitea PR *review*. It performs a - mutation **only after every safety gate passes**; if any gate fails it - returns ``performed=False`` and never calls the mutating endpoint. - - Gate order (fail-closed at each step): - - 1. Validate ``action`` is one of 'comment', 'approve', 'request_changes'. - 2. Reuse ``gitea_check_pr_eligibility`` (#14), which runs the authenticated - -user lookup, active-profile lookup, PR-author lookup, self-approval - block, and profile-allowed-operation check. ``approve`` requires - eligibility for 'approve', ``request_changes`` requires - 'request_changes', and ``comment`` requires 'review'. - 3. Redundantly block self-approval (authenticated user == PR author). - 4. If ``expected_head_sha`` is supplied and the PR head has moved, abort. - 5. Only then POST the review. - - Endpoint: ``POST /repos/{owner}/{repo}/pulls/{n}/reviews``. This is the - *formal review* API (it records an APPROVE / COMMENT / REQUEST_CHANGES - review state tied to the head commit), chosen over the plain issue-comment - endpoint (``/issues/{n}/comments``) so that approvals and change requests - carry real review state — a plain comment cannot approve or block a PR. - - Merge is intentionally NOT implemented here. - - Never returns the token, Authorization header, or any credential material. - - Args: - pr_number: Target PR number. - action: 'comment', 'approve', or 'request_changes'. - body: Review body / finding text. - expected_head_sha: Optional. If given and the PR head SHA differs, the - review is refused (guards against reviewing a changed PR). - remote: Known instance — 'dadeschools' or 'prgs'. - host: Override the Gitea host. - org: Override the owner/organization. - repo: Override the repository name. - - Returns: - dict describing the attempt: action, whether it was performed, the - authenticated user, profile name, PR author, PR number, head SHA - checked, and the reasons/gates passed or blocked. Never secrets. - """ + """Shared gate chain for live submit and dry-run review tools.""" action = (action or "").strip().lower() result = { "requested_action": action, "performed": False, + "dry_run": not live, + "would_perform": False, "authenticated_user": None, "profile_name": get_profile()["profile_name"], "pr_author": None, @@ -1156,7 +1232,6 @@ def gitea_submit_pr_review( } reasons = result["reasons"] - # Gate 1 — valid review action (no mutation on unknown action). if action not in _REVIEW_ACTIONS: reasons.append( f"unknown review action '{action}'; expected one of " @@ -1165,8 +1240,15 @@ def gitea_submit_pr_review( return result eligibility_action, event = _REVIEW_ACTIONS[action] - # Gate 2 — reuse #14 eligibility (identity + profile + author + self-approve - # + profile-allowed). This performs only read-only GETs. + if live: + reasons.extend(check_review_decision_gate( + pr_number, + action, + final_review_decision_ready=final_review_decision_ready, + )) + if reasons: + return result + elig = gitea_check_pr_eligibility( pr_number=pr_number, action=eligibility_action, @@ -1188,30 +1270,36 @@ def gitea_submit_pr_review( result["permission_report"] = elig["permission_report"] return result - # Gate 3 — redundant self-approval block (belt-and-suspenders over #14). auth_user = result["authenticated_user"] pr_author = result["pr_author"] if action == "approve" and auth_user and pr_author and auth_user == pr_author: reasons.append("self-approval blocked (authenticated user is PR author)") return result - # Gate 4 — head SHA must match if the caller pinned one. actual_sha = result["head_sha"] - if expected_head_sha and actual_sha and expected_head_sha != actual_sha: + pinned_sha = expected_head_sha + lock = _load_review_decision_lock() or {} + if live and lock.get("ready_expected_head_sha"): + pinned_sha = lock.get("ready_expected_head_sha") + if pinned_sha and actual_sha and pinned_sha != actual_sha: reasons.append( "expected head SHA does not match current PR head (fail closed)" ) return result if not actual_sha: - # Should be unreachable — eligibility fails closed without a head SHA — - # but never submit a review without a commit to pin it to. reasons.append("PR head SHA unavailable (fail closed)") return result + result["would_perform"] = True + if not live: + reasons.append( + f"dry-run only: would submit '{event}' review on PR #{pr_number}; " + "no live mutation performed" + ) + return result + # Gate 5 — in-process mutation authority (#199): the last check before # the mutating POST, using the identity the eligibility gate proved. - # A profile/identity flip or side-channel override between preflight - # and mutation fails closed here. try: verify_mutation_authority(remote, host, required_role="reviewer", active_identity=auth_user) @@ -1219,22 +1307,151 @@ def gitea_submit_pr_review( reasons.append(str(e)) return result - # All gates passed — perform the single mutating call. h, o, r = _resolve(remote, host, org, repo) + review_id = None try: auth = _auth(h) review_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews" payload = {"body": body, "event": event, "commit_id": actual_sha} - api_request("POST", review_url, auth, payload) + resp = api_request("POST", review_url, auth, payload) + if isinstance(resp, dict): + review_id = resp.get("id") except Exception as exc: # noqa: BLE001 — redact before surfacing reasons.append(f"review submission failed: {_redact(str(exc))}") return result + record_live_review_mutation(pr_number, action, review_id) result["performed"] = True reasons.append(f"all gates passed; submitted '{event}' review on PR #{pr_number}") return result +@mcp.tool() +def gitea_mark_final_review_decision( + pr_number: int, + action: str, + expected_head_sha: str | None = None, + remote: str = "dadeschools", +) -> dict: + """Mark validation complete; the final review decision is ready to submit.""" + action = (action or "").strip().lower() + lock = _load_review_decision_lock() + if lock is None: + return { + "marked_ready": False, + "reasons": [ + "review decision lock missing; resolve review_pr capability first" + ], + } + if lock.get("live_mutations"): + return { + "marked_ready": False, + "reasons": [ + "cannot mark final decision after a live review mutation was " + "already recorded in this run" + ], + } + if action not in _REVIEW_ACTIONS: + return { + "marked_ready": False, + "reasons": [ + f"unknown review action '{action}'; expected one of " + f"{sorted(_REVIEW_ACTIONS)}" + ], + } + lock["final_review_decision_ready"] = True + lock["ready_pr_number"] = pr_number + lock["ready_action"] = action + lock["ready_expected_head_sha"] = expected_head_sha + _save_review_decision_lock(lock) + return { + "marked_ready": True, + "pr_number": pr_number, + "action": action, + "expected_head_sha": expected_head_sha, + "final_review_decision_ready": True, + "reasons": [], + } + + +@mcp.tool() +def gitea_authorize_review_correction(reason: str) -> dict: + """Authorize one operator-approved correction after a mistaken live review.""" + reason = (reason or "").strip() + lock = _load_review_decision_lock() + if lock is None: + return {"authorized": False, "reasons": ["review decision lock missing"]} + if not lock.get("live_mutations"): + return { + "authorized": False, + "reasons": ["no prior live review mutation to correct"], + } + if not reason: + return {"authorized": False, "reasons": ["correction reason is required"]} + lock["correction_authorized"] = True + lock["correction_reason"] = reason + _save_review_decision_lock(lock) + return {"authorized": True, "correction_reason": reason, "reasons": []} + + +@mcp.tool() +def gitea_dry_run_pr_review( + pr_number: int, + action: str, + body: str = "", + expected_head_sha: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Validate review submission mechanics without a live PR mutation.""" + return _evaluate_pr_review_submission( + pr_number=pr_number, + action=action, + body=body, + expected_head_sha=expected_head_sha, + remote=remote, + host=host, + org=org, + repo=repo, + live=False, + ) + + +@mcp.tool() +@_audit_pr_result("submit_pr_review") +def gitea_submit_pr_review( + pr_number: int, + action: str, + body: str = "", + expected_head_sha: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + final_review_decision_ready: bool = False, +) -> dict: + """Gated PR review mutation: comment findings, request changes, or approve. + + Live mutations require ``final_review_decision_ready=True`` and a prior + ``gitea_mark_final_review_decision`` call. Use ``gitea_dry_run_pr_review`` + during validation instead of probing with live submissions. + """ + return _evaluate_pr_review_submission( + pr_number=pr_number, + action=action, + body=body, + expected_head_sha=expected_head_sha, + remote=remote, + host=host, + org=org, + repo=repo, + live=True, + final_review_decision_ready=final_review_decision_ready, + ) + + @mcp.tool() def gitea_edit_pr( pr_number: int, @@ -3790,6 +4007,12 @@ def gitea_resolve_task_capability( "STOP: the active profile cannot perform the requested task; " "follow exact_safe_next_action instead of improvising.") + if task == "review_pr": + init_review_decision_lock( + remote if remote in REMOTES else None, + task, + ) + record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task) return { diff --git a/review_proofs.py b/review_proofs.py index bd4b817..354be7a 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -582,6 +582,54 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, } +def assess_review_mutation_final_report(report_text, review_decision_lock): + """Require final reports to list exactly one live review mutation. + + Two mutations are allowed only when an operator-approved correction flow + was invoked and explained in the report. + """ + lock = review_decision_lock or {} + text = report_text or "" + lower = text.lower() + mutations = list(lock.get("live_mutations") or []) + missing = [] + + count = len(mutations) + if count == 0: + if any(term in lower for term in ("submitted 'approve'", "submitted 'request_changes'", "live review mutation")): + missing.append("review mutation count") + elif count == 1: + m = mutations[0] + action = m.get("action") + pr_number = m.get("pr_number") + if action and action.lower() not in lower: + missing.append("review mutation action") + if pr_number is not None and f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower: + missing.append("review mutation PR number") + elif count > 1: + if not lock.get("correction_authorized") and "correction" not in lower: + missing.append("correction flow explanation") + if "review mutations:" not in lower and "review mutation" not in lower: + missing.append("review mutation listing") + + if missing: + return { + "complete": False, + "downgraded": True, + "missing_fields": missing, + "reasons": [ + f"final report missing review-mutation field: {field}" + for field in missing + ], + } + return { + "complete": True, + "downgraded": False, + "missing_fields": [], + "reasons": [], + } + + def build_final_report(checkout_proof, inventory, validation, contamination, identity_eligible, merge_performed, issue_status_verified, diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index fd94eab..c24670c 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -231,6 +231,15 @@ Worktree folder = branch with `/` replaced by `-` differs from the repository's canonical validation command. Only claim a validation result after the command has completed and its output has been read (`review_proofs.assess_validation_report`). + During validation, review work is **read-only**: use + `gitea_dry_run_pr_review` to prove submission mechanics — never post live + APPROVE, REQUEST_CHANGES, or review comments to probe tool paths. After + validation completes, call `gitea_mark_final_review_decision`, then submit + exactly one live review via + `gitea_submit_pr_review(..., final_review_decision_ready=True)`. + Final reports must list exactly one review mutation + (`review_proofs.assess_review_mutation_final_report`) unless an + operator-approved correction flow was invoked and explained. 10. **Do not merge if checks fail. Do not merge if the reviewer is the author.** 11. **#179 A-bar proofs** (all fail closed when missing — `review_proofs.assess_capability_evidence`, `assess_sweep_evidence`, diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 5c2806f..9bf72d3 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -31,6 +31,11 @@ from mcp_server import ( # noqa: E402 gitea_get_profile, gitea_check_pr_eligibility, gitea_submit_pr_review, + gitea_dry_run_pr_review, + gitea_mark_final_review_decision, + gitea_authorize_review_correction, + init_review_decision_lock, + REVIEW_DECISION_FILE, gitea_list_issue_comments, gitea_create_issue_comment, ) @@ -1395,9 +1400,125 @@ class TestPrEligibility(unittest.TestCase): self.assertNotIn(secret, blob) +class TestReviewDecisionValidationGate(unittest.TestCase): + """Block incidental live review mutations during validation.""" + + PR = 203 + SHA = "abc123" + + def _pr(self, author, sha=SHA): + return { + "user": {"login": author}, + "state": "open", + "head": {"sha": sha}, + "mergeable": True, + } + + def setUp(self): + init_review_decision_lock("prgs", "review_pr") + + def tearDown(self): + if os.path.exists(REVIEW_DECISION_FILE): + os.remove(REVIEW_DECISION_FILE) + + def _env(self): + return patch.dict(os.environ, { + "GITEA_PROFILE_NAME": "gitea-reviewer", + "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes", + }, clear=True) + + def _assert_no_mutation(self, mock_api): + for c in mock_api.call_args_list: + method, url = c.args[0], c.args[1] + self.assertFalse( + method == "POST" and url.endswith("/reviews"), + f"unexpected review mutation: {method} {url}", + ) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_approve_during_validation_blocked(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "reviewer-bot"}, self._pr("author-bot"), + ] + with self._env(): + r = gitea_submit_pr_review( + pr_number=self.PR, action="approve", remote="prgs", + final_review_decision_ready=False, + ) + self.assertFalse(r["performed"]) + self.assertTrue(any( + "final_review_decision_ready must be true" in x for x in r["reasons"] + )) + self._assert_no_mutation(mock_api) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_request_changes_during_validation_blocked(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "reviewer-bot"}, self._pr("author-bot"), + ] + with self._env(): + r = gitea_submit_pr_review( + pr_number=self.PR, action="request_changes", remote="prgs", + final_review_decision_ready=False, + ) + self.assertFalse(r["performed"]) + self.assertTrue(any( + "final_review_decision_ready must be true" in x for x in r["reasons"] + )) + self._assert_no_mutation(mock_api) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_duplicate_terminal_decision_blocked(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1}, + {"login": "reviewer-bot"}, self._pr("author-bot"), + ] + gitea_mark_final_review_decision( + self.PR, "approve", expected_head_sha=self.SHA, remote="prgs") + with self._env(): + first = gitea_submit_pr_review( + pr_number=self.PR, action="approve", remote="prgs", + final_review_decision_ready=True, + ) + second = gitea_submit_pr_review( + pr_number=self.PR, action="request_changes", remote="prgs", + final_review_decision_ready=True, + ) + self.assertTrue(first["performed"]) + self.assertFalse(second["performed"]) + self.assertTrue(any( + "live review mutation already recorded" in x for x in second["reasons"] + )) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_dry_run_proves_submission_without_live_mutation(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "reviewer-bot"}, self._pr("author-bot"), + ] + with self._env(): + r = gitea_dry_run_pr_review( + pr_number=self.PR, action="approve", remote="prgs") + self.assertFalse(r["performed"]) + self.assertTrue(r["would_perform"]) + self.assertTrue(r["dry_run"]) + self._assert_no_mutation(mock_api) + + class TestSubmitPrReview(unittest.TestCase): """Gated review-mutation tool (#15).""" + def setUp(self): + init_review_decision_lock("prgs", "review_pr") + gitea_mark_final_review_decision(8, "approve", remote="prgs") + + def tearDown(self): + if os.path.exists(REVIEW_DECISION_FILE): + os.remove(REVIEW_DECISION_FILE) + def _pr(self, author, state="open", sha="abc123", mergeable=True): return { "user": {"login": author}, @@ -1427,7 +1548,10 @@ class TestSubmitPrReview(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} with patch.dict(os.environ, env, clear=True): - r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs") + r = gitea_submit_pr_review( + pr_number=8, action="approve", remote="prgs", + final_review_decision_ready=True, + ) self.assertFalse(r["performed"]) self.assertIn("authenticated user is PR author", r["reasons"]) self._assert_no_mutation(mock_api) @@ -1442,7 +1566,9 @@ class TestSubmitPrReview(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( - pr_number=8, action="approve", body="LGTM", remote="prgs") + pr_number=8, action="approve", body="LGTM", remote="prgs", + final_review_decision_ready=True, + ) self.assertTrue(r["performed"]) self.assertEqual(r["authenticated_user"], "reviewer-bot") self.assertEqual(r["pr_author"], "author-bot") @@ -1459,6 +1585,7 @@ class TestSubmitPrReview(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_request_changes_succeeds_when_eligible(self, _auth, mock_api): + gitea_mark_final_review_decision(8, "request_changes", remote="prgs") mock_api.side_effect = [ {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9}, ] @@ -1467,11 +1594,14 @@ class TestSubmitPrReview(unittest.TestCase): with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( pr_number=8, action="request_changes", - body="needs work", remote="prgs") + body="needs work", remote="prgs", + final_review_decision_ready=True, + ) self.assertTrue(r["performed"]) self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES") def test_request_changes_blocked_without_eligibility(self): + gitea_mark_final_review_decision(8, "request_changes", remote="prgs") with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \ patch("mcp_server.api_request") as mock_api: mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")] @@ -1479,7 +1609,9 @@ class TestSubmitPrReview(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,review"} # no request_changes with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( - pr_number=8, action="request_changes", remote="prgs") + pr_number=8, action="request_changes", remote="prgs", + final_review_decision_ready=True, + ) self.assertFalse(r["performed"]) self.assertIn("profile is not allowed to request_changes", r["reasons"]) self._assert_no_mutation(mock_api) @@ -1489,6 +1621,7 @@ class TestSubmitPrReview(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_comment_succeeds_when_review_eligible(self, _auth, mock_api): + gitea_mark_final_review_decision(8, "comment", remote="prgs") mock_api.side_effect = [ {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3}, ] @@ -1496,7 +1629,9 @@ class TestSubmitPrReview(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,review"} with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( - pr_number=8, action="comment", body="finding", remote="prgs") + pr_number=8, action="comment", body="finding", remote="prgs", + final_review_decision_ready=True, + ) self.assertTrue(r["performed"]) self.assertEqual(mock_api.call_args.args[3]["event"], "COMMENT") @@ -1510,8 +1645,11 @@ class TestSubmitPrReview(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review"} with patch.dict(os.environ, env, clear=True): + gitea_mark_final_review_decision(8, "comment", remote="prgs") r = gitea_submit_pr_review( - pr_number=8, action="comment", body="note", remote="prgs") + pr_number=8, action="comment", body="note", remote="prgs", + final_review_decision_ready=True, + ) self.assertTrue(r["performed"]) # -- identity / profile fail-closed --------------------------------------- @@ -1521,7 +1659,10 @@ class TestSubmitPrReview(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} with patch.dict(os.environ, env, clear=True): - r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs") + r = gitea_submit_pr_review( + pr_number=8, action="approve", remote="prgs", + final_review_decision_ready=True, + ) self.assertFalse(r["performed"]) self.assertIsNone(r["authenticated_user"]) self.assertIn("authenticated identity could not be determined", r["reasons"]) @@ -1534,7 +1675,9 @@ class TestSubmitPrReview(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,pr.create"} with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( - pr_number=8, action="approve", remote="prgs") + pr_number=8, action="approve", remote="prgs", + final_review_decision_ready=True, + ) self.assertFalse(r["performed"]) self.assertIn("profile is not allowed to approve", r["reasons"]) self._assert_no_mutation(mock_api) @@ -1552,7 +1695,9 @@ class TestSubmitPrReview(unittest.TestCase): with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( pr_number=8, action="approve", - expected_head_sha="deadbeef", remote="prgs") + expected_head_sha="deadbeef", remote="prgs", + final_review_decision_ready=True, + ) self.assertFalse(r["performed"]) self.assertIn( "expected head SHA does not match current PR head (fail closed)", @@ -1571,7 +1716,9 @@ class TestSubmitPrReview(unittest.TestCase): with patch.dict(os.environ, env, clear=True): r = gitea_submit_pr_review( pr_number=8, action="approve", - expected_head_sha="abc123", remote="prgs") + expected_head_sha="abc123", remote="prgs", + final_review_decision_ready=True, + ) self.assertTrue(r["performed"]) # -- invalid action ------------------------------------------------------- @@ -1596,7 +1743,11 @@ class TestSubmitPrReview(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,review,approve", "GITEA_TOKEN": "super-secret-token"} with patch.dict(os.environ, env, clear=True): - r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs") + gitea_mark_final_review_decision(5, "approve", remote="prgs") + r = gitea_submit_pr_review( + pr_number=5, action="approve", remote="prgs", + final_review_decision_ready=True, + ) blob = repr(r).lower() for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()): self.assertNotIn(secret, blob) @@ -1612,7 +1763,11 @@ class TestSubmitPrReview(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} with patch.dict(os.environ, env, clear=True): - r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs") + gitea_mark_final_review_decision(5, "approve", remote="prgs") + r = gitea_submit_pr_review( + pr_number=5, action="approve", remote="prgs", + final_review_decision_ready=True, + ) self.assertFalse(r["performed"]) blob = repr(r) self.assertIn("[REDACTED]", blob) diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 8bd3276..da712bc 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -25,6 +25,7 @@ from review_proofs import ( # noqa: E402 assess_controller_handoff, assess_inventory_completeness, assess_live_state_recheck, + assess_review_mutation_final_report, assess_role_boundary, assess_self_review_contamination, assess_sweep_evidence, @@ -860,6 +861,55 @@ class TestControllerHandoff(unittest.TestCase): self.assertIn("issue #182", skill) +class TestReviewMutationFinalReport(unittest.TestCase): + """Final reports must list exactly one live review mutation.""" + + LOCK_ONE = { + "live_mutations": [{"pr_number": 203, "action": "request_changes"}], + "correction_authorized": False, + } + + def test_single_mutation_report_complete(self): + report = ( + "Review complete on PR #203.\n" + "Review mutations: one request_changes on #203.\n" + "Review decision: request_changes." + ) + result = assess_review_mutation_final_report(report, self.LOCK_ONE) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + + def test_missing_mutation_details_downgraded(self): + result = assess_review_mutation_final_report( + "Review complete. No details.", self.LOCK_ONE + ) + self.assertFalse(result["complete"]) + self.assertTrue(result["downgraded"]) + + def test_two_mutations_require_correction_explanation(self): + lock = { + "live_mutations": [ + {"pr_number": 203, "action": "approve"}, + {"pr_number": 203, "action": "request_changes"}, + ], + "correction_authorized": True, + "correction_reason": "operator approved correcting mistaken approve", + } + incomplete = assess_review_mutation_final_report( + "Submitted approve then request_changes on #203.", lock + ) + self.assertFalse(incomplete["complete"]) + + complete = assess_review_mutation_final_report( + "\n".join([ + "Review mutations: approve (mistake), then request_changes (final).", + "Correction flow: operator approved correcting mistaken approve on PR #203.", + ]), + lock, + ) + self.assertTrue(complete["complete"]) + + class TestPRInventoryTrustGate(unittest.TestCase): """Issue #194: unit tests for the PR inventory trust gate."""