gitea_mcp_server.py

@@ -1677,6 +1677,51 @@ def record_live_review_mutation(pr_number: int, action: str, review_id: int | No
     _save_review_decision_lock(lock)
 
 
+def terminal_review_hard_stop_reasons(pr_number: int, operation: str) -> list[str]:
+    """Session hard-stop after a terminal live review mutation (#332).
+
+    After a terminal verdict is consumed in this run, the only permitted
+    continuation is the merge sequence for the same PR that was approved.
+    A REQUEST_CHANGES (or an approval of a different PR) blocks every
+    further review/mark-ready/merge mutation. An operator-approved
+    correction (#211) re-opens the review path only — never a cross-PR
+    merge.
+
+    *operation* is one of 'merge', 'mark_ready', or 'review'.
+    Returns [] when the operation may proceed.
+    """
+    lock = _load_review_decision_lock()
+    if lock is None:
+        return []
+    terminals = [
+        m for m in (lock.get("live_mutations") or [])
+        if m.get("action") in _TERMINAL_REVIEW_ACTIONS
+    ]
+    if not terminals:
+        return []
+    last = terminals[-1]
+    if (
+        operation == "merge"
+        and last.get("action") == "approve"
+        and last.get("pr_number") == pr_number
+    ):
+        return []
+    if operation in ("mark_ready", "review") and lock.get("correction_authorized"):
+        return []
+    if last.get("action") == "approve":
+        guidance = (
+            f"only the merge sequence for approved PR "
+            f"#{last.get('pr_number')} may continue"
+        )
+    else:
+        guidance = "the session must stop and produce a final report"
+    return [
+        "terminal review mutation already consumed in this run "
+        f"({last.get('action')} on PR #{last.get('pr_number')}); {guidance} "
+        "(fail closed, #332)"
+    ]
+
+
 # Patterns scrubbed from any surfaced error text so a credential can never leak.
 _SECRET_PREFIXES = ("token ", "Basic ")
 
@@ -2092,6 +2137,9 @@ def gitea_mark_final_review_decision(
         }
     org = resolved_org
     repo = resolved_repo
+    hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
+    if hard_stop:
+        return {"marked_ready": False, "reasons": hard_stop}
     if lock.get("live_mutations") and not lock.get("correction_authorized"):
         return {
             "marked_ready": False,
@@ -2109,6 +2157,33 @@ def gitea_mark_final_review_decision(
                 f"{sorted(_REVIEW_ACTIONS)}"
             ],
         }
+    if action == "request_changes":
+        # Duplicate request-changes suppression (#332): an unresolved
+        # REQUEST_CHANGES at the current head must not be duplicated.
+        feedback = gitea_get_pr_review_feedback(
+            pr_number, remote=remote, org=org, repo=repo)
+        if not feedback.get("success"):
+            return {
+                "marked_ready": False,
+                "reasons": [
+                    "could not verify existing request-changes state for "
+                    f"PR #{pr_number}; refusing to submit a possibly "
+                    "duplicate REQUEST_CHANGES (fail closed, #332)"
+                ],
+            }
+        if (
+            feedback.get("has_blocking_change_requests")
+            and not feedback.get("review_feedback_stale")
+        ):
+            return {
+                "marked_ready": False,
+                "reasons": [
+                    f"PR #{pr_number} already has an unresolved "
+                    "REQUEST_CHANGES at the current head SHA; do not "
+                    "duplicate the request-changes review (fail closed, "
+                    "#332)"
+                ],
+            }
     lock["final_review_decision_ready"] = True
     lock["ready_pr_number"] = pr_number
     lock["ready_action"] = action
@@ -2678,6 +2753,14 @@ def gitea_merge_pr(
         )
         return result
 
+    # Gate 2b — session hard-stop after a terminal review mutation (#332):
+    # merge may only continue for the same PR that was just approved.
+    # Local in-process check; zero API calls.
+    hard_stop = terminal_review_hard_stop_reasons(pr_number, "merge")
+    if hard_stop:
+        reasons.extend(hard_stop)
+        return result
+
     # Gate 3 — reuse #14 eligibility (identity + profile + merge-allowed +
     # author + self-merge block + open + mergeable/unknown fail-closed).
     # Read-only GETs only.

tests/test_mcp_server.py

@@ -47,6 +47,21 @@ import mcp_server
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
+_NO_BLOCKER_FEEDBACK = {
+    "success": True,
+    "has_blocking_change_requests": False,
+    "review_feedback_stale": False,
+}
+
+
+def _mark_request_changes_ready(pr_number=8, **kwargs):
+    """Mark a request_changes decision ready with the #332 duplicate-
+    suppression feedback fetch stubbed to 'no existing blocker'."""
+    with patch("mcp_server.gitea_get_pr_review_feedback",
+               return_value=dict(_NO_BLOCKER_FEEDBACK)):
+        return gitea_mark_final_review_decision(
+            pr_number, "request_changes", **kwargs)
+
 
 def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
     return {
@@ -1775,7 +1790,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")
+        _mark_request_changes_ready(remote="prgs")
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"),
             {"id": 9, "state": "REQUEST_CHANGES"},
@@ -1798,7 +1813,7 @@ class TestSubmitPrReview(unittest.TestCase):
         self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
 
     def test_request_changes_blocked_without_eligibility(self):
-        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+        _mark_request_changes_ready(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")]
@@ -2080,7 +2095,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 reason="operator approved correcting mistaken approve",
                 operator_authorized=True,
             )
-            gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+            _mark_request_changes_ready(remote="prgs")
             second = gitea_submit_pr_review(
                 pr_number=8, action="request_changes", remote="prgs",
                 final_review_decision_ready=True,

tests/test_terminal_review_hard_stop.py

@@ -0,0 +1,187 @@
+"""Tests for the session hard-stop after a terminal review mutation (#332).
+
+Extends the single-terminal-decision machinery (#211): after a live
+REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence
+for that same PR may continue; a different PR can never be reviewed or
+merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is
+suppressed.
+"""
+import os
+import unittest
+from unittest.mock import patch
+
+import mcp_server
+
+
+def _lock(mutations=None, correction=False):
+    return {
+        "task": "review_pr",
+        "remote": "prgs",
+        "session_pid": os.getpid(),
+        "session_profile": "prgs-reviewer",
+        "session_profile_lock": "",
+        "final_review_decision_ready": False,
+        "ready_pr_number": None,
+        "ready_action": None,
+        "ready_expected_head_sha": None,
+        "ready_remote": None,
+        "ready_org": None,
+        "ready_repo": None,
+        "live_mutations": list(mutations or []),
+        "correction_authorized": correction,
+        "correction_reason": None,
+    }
+
+
+def _seed(mutations=None, correction=False):
+    mcp_server._save_review_decision_lock(_lock(mutations, correction))
+
+
+APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
+              "review_state": "approve"}
+RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
+        "review_state": "request_changes"}
+
+
+class TestTerminalHardStopReasons(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_no_lock_no_reasons(self):
+        mcp_server._save_review_decision_lock(None)
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
+
+    def test_no_terminal_mutation_no_reasons(self):
+        _seed()
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
+
+    def test_request_changes_blocks_everything(self):
+        _seed([RC_A])
+        for op in ("merge", "mark_ready", "review"):
+            for pr in (5, 6):
+                reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op)
+                self.assertTrue(reasons, f"{op}/{pr}")
+                self.assertIn("request_changes on PR #5", reasons[0])
+                self.assertIn("#332", reasons[0])
+
+    def test_approve_allows_same_pr_merge_only(self):
+        _seed([APPROVED_A])
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "review"))
+
+    def test_correction_reopens_review_path_only(self):
+        _seed([RC_A], correction=True)
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
+        # correction never opens a cross-PR merge
+        self.assertTrue(
+            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
+
+
+class TestMergeHardStopWiring(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_merge_blocked_after_request_changes(self):
+        _seed([RC_A])
+        result = mcp_server.gitea_merge_pr(
+            pr_number=6, confirmation="MERGE PR 6", remote="prgs")
+        self.assertFalse(result["performed"])
+        self.assertTrue(
+            any("#332" in r for r in result["reasons"]), result["reasons"])
+
+    def test_merge_of_other_pr_blocked_after_approve(self):
+        _seed([APPROVED_A])
+        result = mcp_server.gitea_merge_pr(
+            pr_number=6, confirmation="MERGE PR 6", remote="prgs")
+        self.assertFalse(result["performed"])
+        self.assertTrue(
+            any("#332" in r for r in result["reasons"]), result["reasons"])
+
+
+class TestMarkFinalHardStopWiring(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_mark_ready_blocked_after_terminal_mutation(self):
+        _seed([RC_A])
+        result = mcp_server.gitea_mark_final_review_decision(
+            pr_number=6, action="approve", remote="prgs")
+        self.assertFalse(result["marked_ready"])
+        self.assertTrue(
+            any("#332" in r for r in result["reasons"]), result["reasons"])
+
+
+def _feedback(blocking, stale=False, success=True):
+    return {
+        "success": success,
+        "has_blocking_change_requests": blocking,
+        "review_feedback_stale": stale,
+        "current_head_sha": "abc123",
+    }
+
+
+class TestDuplicateRequestChangesSuppression(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+
+    def test_duplicate_request_changes_blocked_at_same_head(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=True, stale=False)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertFalse(result["marked_ready"])
+        self.assertTrue(
+            any("duplicate" in r for r in result["reasons"]),
+            result["reasons"])
+
+    def test_request_changes_allowed_when_prior_blocker_stale(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=True, stale=True)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
+    def test_request_changes_allowed_when_no_blocker(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=False)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
+    def test_request_changes_fails_closed_when_feedback_unavailable(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          return_value=_feedback(blocking=False,
+                                                 success=False)):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="request_changes", remote="prgs")
+        self.assertFalse(result["marked_ready"])
+        self.assertTrue(
+            any("could not verify" in r for r in result["reasons"]),
+            result["reasons"])
+
+    def test_approve_does_not_fetch_feedback(self):
+        _seed()
+        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
+                          side_effect=AssertionError("must not be called")):
+            result = mcp_server.gitea_mark_final_review_decision(
+                pr_number=6, action="approve", remote="prgs")
+        self.assertTrue(result["marked_ready"], result.get("reasons"))
+
+
+if __name__ == "__main__":
+    unittest.main()