Harden controller review and runtime gates

orchestration_tests/test_nq_approve_merge.py

@@ -2,6 +2,7 @@ from pathlib import Path
 import runpy
 import tempfile
 import unittest
+from unittest.mock import patch
 
 
 APPROVE = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-approve-merge"))
@@ -22,6 +23,19 @@ class MergeGateTests(unittest.TestCase):
             with self.assertRaisesRegex(RuntimeError, "exact head"):
                 APPROVE["validate_review"](review, HEAD)
 
+    def test_requires_distinct_gitea_reviewer_on_exact_commit(self):
+        records = [{"state": "APPROVED", "commit_id": HEAD, "user": {"login": "review_agent"}}]
+        with patch.dict(APPROVE["has_platform_review"].__globals__, {"request": lambda *_: records}):
+            self.assertTrue(APPROVE["has_platform_review"](149, HEAD, "unused"))
+            self.assertFalse(APPROVE["has_platform_review"](149, "b" * 40, "unused"))
+            records[0]["user"]["login"] = "orchestrator-agent"
+            self.assertFalse(APPROVE["has_platform_review"](149, HEAD, "unused"))
+            records[:] = [
+                {"id": 1, "state": "APPROVED", "commit_id": HEAD, "user": {"login": "review_agent"}},
+                {"id": 2, "state": "REQUEST_CHANGES", "commit_id": HEAD, "user": {"login": "review_agent"}},
+            ]
+            self.assertFalse(APPROVE["has_platform_review"](149, HEAD, "unused"))
+
 
 if __name__ == "__main__":
     unittest.main()

orchestration_tests/test_nq_controller.py

@@ -19,6 +19,8 @@ class ControllerProtocolTests(unittest.TestCase):
     def test_error_event_does_not_publish_provider_message(self):
         event = {"type": "error", "error": {"name": "APIError", "data": {"message": "secret provider response"}}}
         self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (None, "APIError"))
+        self.assertEqual(CONTROL["parse_event"]("null"), (None, None))
+        self.assertEqual(CONTROL["parse_event"]("[]"), (None, None))
 
     def test_tick_uses_final_text(self):
         event = json.dumps({"type": "text", "part": {"type": "text", "text": "ORCHESTRATOR_STATE: QUIESCENT\n"}})
@@ -53,6 +55,11 @@ class ControllerProtocolTests(unittest.TestCase):
             self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
             (entry / "cmdline").write_bytes(b"opencode\0--dir\0" + str(root).encode() + b"\0")
             self.assertFalse(CONTROL["interactive_nestquest_session"](project, proc, database))
+            with sqlite3.connect(database) as db:
+                db.execute("DROP TABLE session")
+            (entry / "cmdline").write_bytes(b"opencode\0-s\0ses_test\0")
+            with self.assertRaisesRegex(RuntimeError, "lookup failed"):
+                CONTROL["interactive_nestquest_session"](project, proc, database)
 
 
 if __name__ == "__main__":

orchestration_tests/test_nq_review.py

@@ -1,43 +1,17 @@
 from pathlib import Path
-import os
-import subprocess
-import tempfile
+import runpy
 import unittest
 
-
-SCRIPT = Path(__file__).resolve().parents[1] / "scripts/nq-review"
-
+REVIEW = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-review"))
+HEAD = "a" * 40
 
 class ReviewLauncherTests(unittest.TestCase):
-    def test_pins_head_and_model_in_worktree(self):
-        with tempfile.TemporaryDirectory() as directory:
-            root = Path(directory)
-            repo = root / "repo"
-            repo.mkdir()
-            subprocess.run(["git", "init", "-q", str(repo)], check=True)
-            subprocess.run(["git", "-C", str(repo), "config", "user.email", "test@example.invalid"], check=True)
-            subprocess.run(["git", "-C", str(repo), "config", "user.name", "Test"], check=True)
-            (repo / "README").write_text("initial\n")
-            subprocess.run(["git", "-C", str(repo), "add", "README"], check=True)
-            subprocess.run(["git", "-C", str(repo), "commit", "-qm", "initial"], check=True)
-            head = subprocess.check_output(["git", "-C", str(repo), "rev-parse", "HEAD"], text=True).strip()
-            prompt = root / "review.txt"
-            prompt.write_text("Review this change.")
-            stub = root / "claude"
-            stub.write_text("#!/bin/sh\nprintf '%s\\n' \"$PWD\" \"$@\"\n")
-            stub.chmod(0o700)
-            env = os.environ.copy()
-            env["CLAUDE_BIN"] = str(stub)
-            result = subprocess.run([str(SCRIPT), str(repo), head, head, str(prompt)], env=env, text=True, capture_output=True)
-            self.assertEqual(result.returncode, 0, result.stderr)
-            self.assertIn(str(repo), result.stdout)
-            self.assertIn("--model\nclaude-sonnet-5", result.stdout)
-            self.assertIn("--effort\nhigh", result.stdout)
-            self.assertIn(head, result.stdout)
-
-            result = subprocess.run([str(SCRIPT), str(repo), head, "HEAD~1", str(prompt)], env=env, text=True, capture_output=True)
-            self.assertNotEqual(result.returncode, 0)
-
+    def test_exact_head_verdict(self):
+        self.assertEqual(REVIEW["verdict"](f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n", HEAD), "APPROVED")
+        with self.assertRaisesRegex(RuntimeError, "one verdict"):
+            REVIEW["verdict"](f"REVIEWED_HEAD: {'b' * 40}\nREVIEW_VERDICT: APPROVED\n", HEAD)
+        with self.assertRaisesRegex(RuntimeError, "one verdict"):
+            REVIEW["verdict"](f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\nREVIEW_VERDICT: CHANGES_REQUIRED\n", HEAD)
 
 if __name__ == "__main__":
     unittest.main()