diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 37a8e9b..8e8683d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -535,6 +535,7 @@ import issue_lock_worktree # noqa: E402 import issue_lock_provenance # noqa: E402 import issue_lock_store # noqa: E402 import issue_lock_adoption # noqa: E402 +import merge_approval_gate # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 @@ -2472,6 +2473,10 @@ def gitea_get_pr_review_feedback( e for e in latest_by_reviewer.values() if e["verdict"] == "APPROVED" and not e["dismissed"] ] + approval_head = merge_approval_gate.assess_merge_approval_head( + current_head_sha=current_head, + latest_by_reviewer=latest_by_reviewer, + ) return { "success": True, "pr_number": pr_number, @@ -2482,6 +2487,9 @@ def gitea_get_pr_review_feedback( login: e["verdict"] for login, e in latest_by_reviewer.items()}, "has_blocking_change_requests": bool(blocking), "approval_visible": bool(approvals), + "approval_at_current_head": approval_head["approval_at_current_head"], + "latest_approved_head_sha": approval_head["latest_approved_head_sha"], + "stale_approval_block_reason": approval_head["stale_approval_block_reason"], "latest_reviewed_head_sha": latest_reviewed_head, "review_feedback_stale": bool( latest_reviewed_head and current_head @@ -3442,6 +3450,9 @@ def gitea_merge_pr( result["permission_report"] = feedback["permission_report"] return result result["approval_visible"] = feedback.get("approval_visible") + result["approval_at_current_head"] = feedback.get("approval_at_current_head") + result["latest_approved_head_sha"] = feedback.get("latest_approved_head_sha") + result["review_feedback_stale"] = feedback.get("review_feedback_stale") result["has_blocking_change_requests"] = feedback.get( "has_blocking_change_requests") if feedback.get("has_blocking_change_requests"): @@ -3455,6 +3466,16 @@ def gitea_merge_pr( "completed before merge (fail closed)" ) return result + if not feedback.get("approval_at_current_head"): + reasons.append( + feedback.get("stale_approval_block_reason") + or ( + "approval does not apply to current PR head SHA " + "(fail closed); required next action: re-review PR at " + "current head before merge" + ) + ) + return result # Gate 8 — in-process mutation authority (#199): the last check before # the merge mutation, using the identity the eligibility gate proved. diff --git a/merge_approval_gate.py b/merge_approval_gate.py new file mode 100644 index 0000000..08fc8ad --- /dev/null +++ b/merge_approval_gate.py @@ -0,0 +1,61 @@ +"""Merge approval must pin the current PR head SHA (#471). + +Formal APPROVED reviews that predate the live PR head must not satisfy +``gitea_merge_pr`` eligibility. Pure assessment helpers are isolated here +for hermetic unit tests apart from MCP HTTP calls. +""" + +from __future__ import annotations + + +def assess_merge_approval_head( + *, + current_head_sha: str | None, + latest_by_reviewer: dict, +) -> dict: + """Return whether a visible approval applies to the live PR head. + + Args: + current_head_sha: Current PR head commit SHA. + latest_by_reviewer: Map of reviewer login → review entry dicts with + ``verdict``, ``dismissed``, and ``reviewed_head_sha`` keys. + + Returns: + dict with ``approval_at_current_head``, ``latest_approved_head_sha``, + and ``stale_approval_block_reason`` (set when merge must fail closed). + """ + current = (current_head_sha or "").strip() + approved_entries = [ + entry + for entry in (latest_by_reviewer or {}).values() + if (entry.get("verdict") or "").upper() == "APPROVED" + and not entry.get("dismissed") + ] + at_current = any( + (entry.get("reviewed_head_sha") or "").strip() == current + for entry in approved_entries + if current + ) + latest_approved = None + if approved_entries: + latest_entry = sorted( + approved_entries, + key=lambda entry: ( + entry.get("submitted_at") or "", + entry.get("reviewed_head_sha") or "", + ), + )[-1] + latest_approved = (latest_entry.get("reviewed_head_sha") or "").strip() or None + reason = None + if approved_entries and not at_current: + reason = ( + f"stale approval: approved SHA '{latest_approved}' does not match " + f"current live PR head SHA '{current or '(unknown)'}' (fail closed); " + "required next action: re-review PR at current head before merge" + ) + + return { + "approval_at_current_head": at_current, + "latest_approved_head_sha": latest_approved, + "stale_approval_block_reason": reason, + } \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 4f9f6eb..80e9218 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -762,6 +762,7 @@ Before merge, rerun fresh live checks: * author safety * PR re-fetch * reviewed head SHA unchanged +* visible APPROVED review applies to the **current live PR head SHA** (`approval_at_current_head`); if the head moved after approval, re-review at the new head before merge (#471) * target branch freshly fetched * PR still open * PR still mergeable @@ -778,6 +779,7 @@ Do not merge if: * capability state is stale * worktree is dirty * PR head changed +* approval is stale (approved SHA ≠ current live head SHA) * validation failed * inventory was incomplete * PR is already landed diff --git a/tests/test_audit.py b/tests/test_audit.py index 11cecd3..83fd9b9 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -313,7 +313,8 @@ class TestGatedToolAudit(_AuditWiringBase): {"login": "merger-bot"}, self._pr("author-bot"), self._pr("author-bot"), [{"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED", - "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}], + "commit_id": "abc123", "submitted_at": "2026-07-06T10:00:00Z", + "dismissed": False}], {}, {"merged_commit_sha": "c1"}, ] env = self._env(GITEA_PROFILE_NAME="gitea-merger", diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index ff2f4a4..397ac81 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -972,6 +972,29 @@ class TestMergePR(unittest.TestCase): self.assertIn("[REDACTED]", blob) self.assertNotIn("abc-secret-xyz", blob) + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_merge_blocked_on_stale_approval_head(self, _auth, mock_api): + old_sha = "8b61c4b41f1b49b271ed3b99657431cf06eeda3e" + new_sha = "3e4b721d60e97147ba0704773cf57cd0d42cbe31" + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot", sha=new_sha), + self._pr("author-bot", sha=new_sha), + [_formal_review("reviewer-bot", "APPROVED", sha=old_sha)], + ] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertTrue(r.get("approval_visible")) + self.assertFalse(r.get("approval_at_current_head")) + self.assertTrue(any("stale approval" in x for x in r["reasons"])) + self.assertTrue(any(old_sha in x for x in r["reasons"])) + self.assertTrue(any(new_sha in x for x in r["reasons"])) + self._assert_no_merge_call(mock_api) + @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_merge_blocked_without_visible_approval(self, _auth, mock_api): diff --git a/tests/test_merge_approval_gate.py b/tests/test_merge_approval_gate.py new file mode 100644 index 0000000..94f11ac --- /dev/null +++ b/tests/test_merge_approval_gate.py @@ -0,0 +1,59 @@ +"""Hermetic tests for merge approval head pinning (#471).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from merge_approval_gate import assess_merge_approval_head # noqa: E402 + +HEAD_OLD = "8b61c4b41f1b49b271ed3b99657431cf06eeda3e" +HEAD_NEW = "3e4b721d60e97147ba0704773cf57cd0d42cbe31" + + +class TestMergeApprovalGate(unittest.TestCase): + def test_fresh_approval_at_current_head(self): + result = assess_merge_approval_head( + current_head_sha=HEAD_NEW, + latest_by_reviewer={ + "reviewer1": { + "verdict": "APPROVED", + "dismissed": False, + "reviewed_head_sha": HEAD_NEW, + "submitted_at": "2026-07-06T12:00:00Z", + } + }, + ) + self.assertTrue(result["approval_at_current_head"]) + self.assertIsNone(result["stale_approval_block_reason"]) + + def test_stale_approval_after_rebase(self): + result = assess_merge_approval_head( + current_head_sha=HEAD_NEW, + latest_by_reviewer={ + "reviewer1": { + "verdict": "APPROVED", + "dismissed": False, + "reviewed_head_sha": HEAD_OLD, + "submitted_at": "2026-07-06T10:00:00Z", + } + }, + ) + self.assertFalse(result["approval_at_current_head"]) + self.assertEqual(result["latest_approved_head_sha"], HEAD_OLD) + self.assertIn("stale approval", result["stale_approval_block_reason"]) + self.assertIn(HEAD_OLD, result["stale_approval_block_reason"]) + self.assertIn(HEAD_NEW, result["stale_approval_block_reason"]) + self.assertIn("re-review PR at current head", result["stale_approval_block_reason"]) + + def test_no_approval_entries(self): + result = assess_merge_approval_head( + current_head_sha=HEAD_NEW, + latest_by_reviewer={}, + ) + self.assertFalse(result["approval_at_current_head"]) + self.assertIsNone(result["latest_approved_head_sha"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_review_feedback.py b/tests/test_review_feedback.py index 01fff0c..72cbe61 100644 --- a/tests/test_review_feedback.py +++ b/tests/test_review_feedback.py @@ -115,6 +115,22 @@ class TestPRReviewFeedbackDiscovery(unittest.TestCase): result["latest_review_state_by_reviewer"], {"reviewer1": "APPROVED"}) self.assertFalse(result["has_blocking_change_requests"]) self.assertTrue(result["approval_visible"]) + self.assertTrue(result["approval_at_current_head"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_stale_approval_not_at_current_head(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(head_sha="newhead3"), + [_review("reviewer1", "APPROVED", commit_id="oldhead1")], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertTrue(result["approval_visible"]) + self.assertFalse(result["approval_at_current_head"]) + self.assertEqual(result["latest_approved_head_sha"], "oldhead1") + self.assertIn("stale approval", result["stale_approval_block_reason"]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)