feat: enforce session hard-stop after terminal review mutation (#332)

Extends the single-terminal-decision machinery (#211) to session-level hard-stop semantics: - terminal_review_hard_stop_reasons: after a live terminal verdict, only the merge sequence for the same approved PR may continue; REQUEST_CHANGES stops the run; operator-approved corrections re-open the review path only, never a cross-PR merge. - gitea_merge_pr: new Gate 2b consults the hard-stop before any API call, so a run can never verdict one PR and merge another. - gitea_mark_final_review_decision: hard-stop check with explicit stop guidance naming the consumed terminal mutation; plus duplicate REQUEST_CHANGES suppression — an unresolved request-changes at the current head SHA refuses a second request-changes (fail closed when feedback cannot be verified). - tests/test_terminal_review_hard_stop.py: 13 tests covering the acceptance criteria; existing TestSubmitPrReview setups stub the new feedback fetch. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 04:32:15 -04:00
parent fd396df334
commit 456e46d3fc
3 changed files with 288 additions and 3 deletions
@@ -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,
@@ -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()