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.

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,
+    }

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

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",

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):

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()

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)