Test merge decision gates and CTXD preflight

deploy/nestquest-controller.service

@@ -8,7 +8,7 @@ ExecStartPre=/home/overseer/.local/share/nestquest-controller/scripts/nq-control
 ExecStart=/home/overseer/.local/share/nestquest-controller/scripts/nq-controller
 Restart=on-failure
 RestartSec=60s
-RestartPreventExitStatus=75 77
+RestartPreventExitStatus=77
 TimeoutStartSec=0
 KillMode=control-group
 Environment=PYTHONUNBUFFERED=1

orchestration_tests/test_nq_approve_merge.py

@@ -1,5 +1,8 @@
 from pathlib import Path
+import contextlib
+import io
 import runpy
+import sys
 import tempfile
 import unittest
 from unittest.mock import patch
@@ -10,6 +13,46 @@ HEAD = "a" * 40
 
 
 class MergeGateTests(unittest.TestCase):
+    def test_main_rejects_moved_head_unmergeable_and_pending_checks(self):
+        root = Path("/tmp/opencode")
+        root.mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=root) as directory:
+            review_file = Path(directory) / "review.txt"
+            review_file.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n")
+            for problem in ("moved", "unmergeable", "pending", "none"):
+                with self.subTest(problem=problem):
+                    merged = False
+                    calls = []
+                    def fake_request(path, token, payload=None):
+                        nonlocal merged
+                        calls.append((path, payload))
+                        if path.endswith("/reviews"):
+                            return [{"id": 1, "state": "APPROVED", "commit_id": HEAD,
+                                     "user": {"login": "review_agent"}}]
+                        if path.endswith("/status"):
+                            return {"total_count": 1, "state": "pending" if problem == "pending" else "success"}
+                        if path.endswith("/merge"):
+                            merged = True
+                            return {}
+                        if path.endswith("/pulls/149"):
+                            return {"number": 149, "base": {"ref": "dev"},
+                                    "head": {"sha": "b" * 40 if problem == "moved" else HEAD},
+                                    "state": "open", "merged": merged,
+                                    "mergeable": problem != "unmergeable"}
+                        raise AssertionError(path)
+                    globals_ = APPROVE["main"].__globals__
+                    with patch.dict(globals_, {"request": fake_request,
+                                               "token_from_file": lambda: "test"}), \
+                         patch.object(sys, "argv", ["nq-approve-merge", "merge", "149", HEAD,
+                                                    str(review_file)]), \
+                         contextlib.redirect_stdout(io.StringIO()):
+                        if problem == "none":
+                            self.assertEqual(APPROVE["main"](), 0)
+                        else:
+                            with self.assertRaises(RuntimeError):
+                                APPROVE["main"]()
+                    self.assertEqual(merged, problem == "none")
+
     def test_review_evidence_must_match_exact_head_and_approval(self):
         root = Path("/tmp/opencode")
         root.mkdir(exist_ok=True)

orchestration_tests/test_nq_controller.py

@@ -1,16 +1,38 @@
 import json
+import io
 from pathlib import Path
 import runpy
 import sqlite3
 import sys
 import tempfile
 import unittest
+from unittest.mock import patch
 
 
 CONTROL = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-controller"))
 
 
 class ControllerProtocolTests(unittest.TestCase):
+    def test_ctxd_preflight_reads_configured_endpoint_and_checks_identity(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = root / ".config/opencode/opencode.json"
+            config.parent.mkdir(parents=True)
+            config.write_text(json.dumps({"mcp": {"ctxd": {
+                "url": "http://localhost:9091/mcp",
+                "headers": {"Authorization": "Bearer test"},
+            }}}))
+            record = {"project_id": CONTROL["CTXD_PROJECT"],
+                      "file_path": CONTROL["CTXD_FILE"],
+                      "content": "# CONTROLLER_TICK", "version": 6}
+            with patch.object(Path, "home", return_value=root), \
+                 patch.dict(CONTROL["check_ctxd"].__globals__["urllib"].request.__dict__,
+                            {"urlopen": lambda request, timeout: io.BytesIO(json.dumps(record).encode())}):
+                self.assertEqual(CONTROL["check_ctxd"](), 6)
+                record["project_id"] = "wrong"
+                with self.assertRaisesRegex(RuntimeError, "wrong NestQuest prompt"):
+                    CONTROL["check_ctxd"]()
+
     def test_installed_opencode_text_event(self):
         event = {"type": "text", "sessionID": "ses_test", "part": {"type": "text", "text": "Next task queued.\nORCHESTRATOR_STATE: CONTINUE\n"}}
         self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (event["part"]["text"], None))