From 29c2cf2ef6bc427417045a7474a67fcd40ea25bd Mon Sep 17 00:00:00 2001
From: Jason Walker <913443@dadeschools.net>
Date: Fri, 17 Jul 2026 13:25:34 -0400
Subject: [PATCH] fix(mcp): fail closed when merger lease cannot resolve live
 PR head

gitea_acquire_merger_pr_lease previously continued when GET /pulls returned
an empty head SHA. Fail closed so exact-head scoping never proceeds with an
unresolvable live head. Add regression coverage for empty head payloads.

Fixes #718
Refs #723
---
 gitea_mcp_server.py                | 15 +++++++++-
 tests/test_merger_lease_acquire.py | 46 ++++++++++++++++++++++++++++++
 2 files changed, 60 insertions(+), 1 deletion(-)

diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py
index 3ebf71a..e7551db 100644
--- a/gitea_mcp_server.py
+++ b/gitea_mcp_server.py
@@ -10821,7 +10821,20 @@ def gitea_acquire_merger_pr_lease(
         or ""
     )
     live_head = str(live_head).strip()
-    if live_head and live_head != head_pin:
+    # Fail closed when live PR head cannot be resolved — never proceed with
+    # an unpinned/unknown head even if candidate_head was supplied (#718).
+    if not live_head:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [
+                "live PR head could not be resolved from GET /pulls response "
+                "(exact-head scoping; fail closed)"
+            ],
+            "live_head": None,
+            "candidate_head": head_pin,
+        }
+    if live_head != head_pin:
         return {
             "success": False,
             "acquired": False,
diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py
index 3c9da2a..88fa94b 100644
--- a/tests/test_merger_lease_acquire.py
+++ b/tests/test_merger_lease_acquire.py
@@ -176,6 +176,52 @@ class TestMergerLeaseAcquireTool(unittest.TestCase):
         self.assertTrue(any("does not match live PR head" in r for r in res["reasons"]))
         self.assertIsNone(leases._SESSION_LEASE)
 
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_acquire_merger_fails_closed_when_live_head_unresolvable(
+        self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
+    ):
+        """Empty/missing live head must not silently proceed past exact-head gate."""
+        mock_profile.return_value = self._merger_profile()
+        # PR payload present but head.sha missing / empty / non-dict.
+        mock_api.return_value = {"state": "open", "head": {}}
+        res = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            session_id="merger-session",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res.get("acquired"), res)
+        self.assertTrue(
+            any("live PR head could not be resolved" in r for r in res["reasons"]),
+            res,
+        )
+        self.assertIsNone(res.get("live_head"))
+        self.assertEqual(res.get("candidate_head"), HEAD)
+        self.assertEqual(
+            [c for c in mock_api.call_args_list if c[0][0] == "POST"], []
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
+        # Completely empty GET /pulls body also fails closed.
+        mock_api.return_value = {}
+        res2 = mcp_server.gitea_acquire_merger_pr_lease(
+            pr_number=718,
+            worktree="branches/issue-718",
+            candidate_head=HEAD,
+        )
+        self.assertFalse(res2["success"], res2)
+        self.assertTrue(
+            any("live PR head could not be resolved" in r for r in res2["reasons"]),
+            res2,
+        )
+        self.assertIsNone(leases._SESSION_LEASE)
+
     @patch("gitea_mcp_server._profile_operation_gate")
     def test_author_denied_merge_permission(self, mock_gate):
         """Author profile lacks gitea.pr.merge → fail closed before mutation."""