gitea_mcp_server.py

@@ -2852,6 +2852,85 @@ def gitea_get_pr_review_feedback(
     }
 
 
+def _fetch_pr_lease_comments_safe(
+    pr_number: int,
+    *,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    limit: int = 100,
+    require_open: bool = False,
+) -> dict:
+    """Fetch PR thread comments with structured fail-closed errors (#519)."""
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    resolved_repo = f"{o}/{r}"
+    pr_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
+    try:
+        pr = api_request("GET", pr_url, auth)
+    except RuntimeError as exc:
+        return {
+            "success": False,
+            "comments": [],
+            "reasons": [
+                "PR lookup failed before conflict-fix push assessment "
+                f"(pr_number={pr_number}, repo={resolved_repo}, remote={remote}): "
+                f"{_redact(str(exc))}"
+            ],
+            "pr_lookup": "failed",
+            "resolved_repo": resolved_repo,
+            "remote": remote,
+            "pr_number": pr_number,
+        }
+    pr_state = (pr.get("state") or "").strip().lower()
+    if require_open and pr_state != "open":
+        return {
+            "success": False,
+            "comments": [],
+            "reasons": [
+                f"PR #{pr_number} on {resolved_repo} is not open "
+                f"(state={pr_state or 'unknown'})"
+            ],
+            "pr_lookup": "not_open",
+            "resolved_repo": resolved_repo,
+            "remote": remote,
+            "pr_number": pr_number,
+            "head_sha": (pr.get("head") or {}).get("sha"),
+        }
+    api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    try:
+        comments = api_request("GET", api, auth)
+    except RuntimeError as exc:
+        return {
+            "success": False,
+            "comments": [],
+            "reasons": [
+                "PR comment fetch failed during conflict-fix push assessment "
+                f"(pr_number={pr_number}, issue_index={pr_number}, "
+                f"repo={resolved_repo}, remote={remote}): "
+                f"{_redact(str(exc))}"
+            ],
+            "pr_lookup": "ok",
+            "resolved_repo": resolved_repo,
+            "remote": remote,
+            "pr_number": pr_number,
+            "head_sha": (pr.get("head") or {}).get("sha"),
+        }
+    if not isinstance(comments, list):
+        comments = []
+    return {
+        "success": True,
+        "comments": list(comments[:limit]),
+        "reasons": [],
+        "pr_lookup": "ok",
+        "resolved_repo": resolved_repo,
+        "remote": remote,
+        "pr_number": pr_number,
+        "head_sha": (pr.get("head") or {}).get("sha"),
+    }
+
+
 def _list_pr_lease_comments(
     pr_number: int,
     *,
@@ -2861,7 +2940,15 @@ def _list_pr_lease_comments(
     repo: str | None,
     limit: int = 100,
 ) -> list[dict]:
-    """Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
+    """Fetch PR/issue thread comments used for reviewer/conflict-fix leases.
+
+    Intentionally does **not** pre-fetch the PR via GET /pulls/{n}. Shared
+    callers (merge approval feedback, reviewer lease gates) depend on a single
+    comments GET so mock sequences and fail-open #485 non-list handling stay
+    stable. Structured PR-lookup failures belong only to
+    :func:`_fetch_pr_lease_comments_safe` used by conflict-fix push assessment
+    (#519).
+    """
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
@@ -7996,22 +8083,39 @@ def gitea_assess_conflict_fix_push(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
-    comments = _list_pr_lease_comments(
+    fetched = _fetch_pr_lease_comments_safe(
         pr_number,
         remote=remote,
         host=host,
         org=org,
         repo=repo,
+        require_open=True,
     )
-    return pr_work_lease.assess_conflict_fix_push(
+    if not fetched["success"]:
+        return {
+            "push_allowed": False,
+            "block": True,
+            "reasons": fetched["reasons"],
+            "pr_lookup": fetched.get("pr_lookup"),
+            "resolved_repo": fetched.get("resolved_repo"),
+            "remote": fetched.get("remote"),
+            "pr_number": pr_number,
+            "assessment_failed": True,
+        }
+    assessment = pr_work_lease.assess_conflict_fix_push(
         pr_number=pr_number,
-        comments=comments,
+        comments=fetched["comments"],
         branch_head_before=branch_head_before,
         branch_head_after=branch_head_after,
         worktree_path=worktree_path,
         push_cwd=push_cwd,
         is_fast_forward=is_fast_forward,
     )
+    assessment["pr_lookup"] = fetched.get("pr_lookup")
+    assessment["resolved_repo"] = fetched.get("resolved_repo")
+    assessment["remote"] = fetched.get("remote")
+    assessment["live_pr_head_sha"] = fetched.get("head_sha")
+    return assessment
 
 
 @mcp.tool()

skills/llm-project-workflow/workflows/work-issue.md

@@ -626,9 +626,14 @@ When pushing to an existing PR branch to resolve merge conflicts:
    * session worktree path
    * push cwd
    * whether the push is fast-forward
-3. Do not push when a reviewer holds an active lease on the same PR.
-4. Do not force-push.
-5. Do not push from the main checkout or wrong cwd.
+   * explicit `remote`, `org`, and `repo` when not using defaults
+3. If assessment returns `assessment_failed: true` or `pr_lookup: failed`, stop
+   and produce a recovery handoff with the structured `reasons` and
+   `resolved_repo` fields — do not treat an MCP HTTP 500 as proof the push was
+   unsafe or safe (#519).
+4. Do not push when a reviewer holds an active lease on the same PR.
+5. Do not force-push.
+6. Do not push from the main checkout or wrong cwd.
 
 Conflict-fix final reports must state:
 

tests/test_assess_conflict_fix_push_tool.py

@@ -0,0 +1,139 @@
+"""Regression tests for gitea_assess_conflict_fix_push structured failures (#519)."""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from mcp_server import (  # noqa: E402
+    _fetch_pr_lease_comments_safe,
+    gitea_assess_conflict_fix_push,
+)
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+AUTHOR_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-author",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.branch.push,gitea.pr.create",
+}
+
+HEAD_BEFORE = "dad1dc8d5108ab01ed83065334116d7425a4471c"
+HEAD_AFTER = "3f3d6cb35d0fe225dcb236247f2a2d0ec193fa35"
+OPEN_PR = {
+    "number": 508,
+    "state": "open",
+    "head": {"sha": HEAD_AFTER},
+}
+
+
+class TestFetchPrLeaseCommentsSafe(unittest.TestCase):
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_pr_lookup_404_returns_structured_failure(self, _auth, mock_api):
+        mock_api.side_effect = RuntimeError(
+            'HTTP 404: {"message":"issue does not exist","index":508}'
+        )
+        result = _fetch_pr_lease_comments_safe(
+            508, remote="prgs", host=None, org=None, repo=None)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["comments"], [])
+        self.assertTrue(result["reasons"])
+        self.assertIn("PR lookup failed", result["reasons"][0])
+        self.assertEqual(result["pr_lookup"], "failed")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_comment_fetch_404_after_pr_lookup(self, _auth, mock_api):
+        def _api(method, url, auth, *args, **kwargs):
+            if "/pulls/" in url:
+                return OPEN_PR
+            raise RuntimeError(
+                'HTTP 404: {"message":"issue does not exist","index":508}'
+            )
+
+        mock_api.side_effect = _api
+        result = _fetch_pr_lease_comments_safe(
+            508, remote="prgs", host=None, org=None, repo=None)
+        self.assertFalse(result["success"])
+        self.assertIn("comment fetch failed", result["reasons"][0].lower())
+        self.assertEqual(result["pr_lookup"], "ok")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_success_returns_comments_and_head_sha(self, _auth, mock_api):
+        comment = {"id": 1, "body": "<!-- mcp-conflict-fix-lease:v1 -->"}
+
+        def _api(method, url, auth, *args, **kwargs):
+            if "/pulls/" in url:
+                return OPEN_PR
+            return [comment]
+
+        mock_api.side_effect = _api
+        result = _fetch_pr_lease_comments_safe(
+            508, remote="prgs", host=None, org=None, repo=None)
+        self.assertTrue(result["success"])
+        self.assertEqual(result["comments"], [comment])
+        self.assertEqual(result["head_sha"], OPEN_PR["head"]["sha"])
+
+
+class TestAssessConflictFixPushTool(unittest.TestCase):
+    @patch("mcp_server._fetch_pr_lease_comments_safe")
+    @patch("mcp_server._profile_operation_gate", return_value=None)
+    def test_returns_structured_block_when_pr_lookup_fails(
+        self, _gate, mock_fetch,
+    ):
+        mock_fetch.return_value = {
+            "success": False,
+            "comments": [],
+            "reasons": ["PR lookup failed"],
+            "pr_lookup": "failed",
+            "resolved_repo": "Scaled-Tech-Consulting/Gitea-Tools",
+            "remote": "prgs",
+            "pr_number": 508,
+        }
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_assess_conflict_fix_push(
+                pr_number=508,
+                branch_head_before=HEAD_BEFORE,
+                branch_head_after=HEAD_AFTER,
+                worktree_path="/proj/branches/fix-pr508",
+                push_cwd="/proj/branches/fix-pr508",
+                is_fast_forward=True,
+                remote="prgs",
+            )
+        self.assertFalse(result["push_allowed"])
+        self.assertTrue(result.get("assessment_failed"))
+        self.assertEqual(result["pr_lookup"], "failed")
+
+    @patch("mcp_server._fetch_pr_lease_comments_safe")
+    @patch("mcp_server._profile_operation_gate", return_value=None)
+    def test_valid_push_assessment_when_pr_and_comments_resolve(
+        self, _gate, mock_fetch,
+    ):
+        mock_fetch.return_value = {
+            "success": True,
+            "comments": [],
+            "reasons": [],
+            "pr_lookup": "ok",
+            "resolved_repo": "Scaled-Tech-Consulting/Gitea-Tools",
+            "remote": "prgs",
+            "pr_number": 508,
+            "head_sha": HEAD_AFTER,
+        }
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_assess_conflict_fix_push(
+                pr_number=508,
+                branch_head_before=HEAD_BEFORE,
+                branch_head_after=HEAD_AFTER,
+                worktree_path="/proj/branches/fix-pr508",
+                push_cwd="/proj/branches/fix-pr508",
+                is_fast_forward=True,
+                remote="prgs",
+            )
+        self.assertTrue(result["push_allowed"])
+        self.assertEqual(result.get("live_pr_head_sha"), HEAD_AFTER)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_lease_comments_non_list_guard.py

@@ -26,6 +26,11 @@ class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None)
         self.assertEqual(result, [])
+        # Single comments GET only — no pre-flight PR lookup (#519 isolation).
+        mock_api.assert_called_once()
+        _method, url, _auth_arg = mock_api.call_args[0][:3]
+        self.assertIn("/issues/12/comments", url)
+        self.assertNotIn("/pulls/", url)
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -73,4 +78,4 @@ class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()