Add durable NestQuest orchestration roles and controller

orchestration_tests/test_nq_approve_merge.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+import runpy
+import tempfile
+import unittest
+
+
+APPROVE = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-approve-merge"))
+HEAD = "a" * 40
+
+
+class MergeGateTests(unittest.TestCase):
+    def test_review_evidence_must_match_exact_head_and_approval(self):
+        root = Path("/tmp/opencode")
+        root.mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=root) as directory:
+            review = Path(directory) / "review.txt"
+            review.write_text(f"Findings: none\nREVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n")
+            APPROVE["validate_review"](review, HEAD)
+            with self.assertRaisesRegex(RuntimeError, "exact head"):
+                APPROVE["validate_review"](review, "b" * 40)
+            review.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: CHANGES_REQUIRED\n")
+            with self.assertRaisesRegex(RuntimeError, "exact head"):
+                APPROVE["validate_review"](review, HEAD)
+
+
+if __name__ == "__main__":
+    unittest.main()

orchestration_tests/test_nq_controller.py

@@ -0,0 +1,59 @@
+import json
+from pathlib import Path
+import runpy
+import sqlite3
+import sys
+import tempfile
+import unittest
+
+
+CONTROL = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-controller"))
+
+
+class ControllerProtocolTests(unittest.TestCase):
+    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))
+        self.assertEqual(CONTROL["STATE_RE"].findall(event["part"]["text"]), ["CONTINUE"])
+
+    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"))
+
+    def test_tick_uses_final_text(self):
+        event = json.dumps({"type": "text", "part": {"type": "text", "text": "ORCHESTRATOR_STATE: QUIESCENT\n"}})
+        command = [sys.executable, "-c", f"import json,sys; sys.stdin.read(); print({event!r})"]
+        status, answer, error = CONTROL["run_tick"](command, {}, "tick", 10)
+        self.assertEqual((status, error), (0, None))
+        self.assertEqual(CONTROL["STATE_RE"].findall(answer), ["QUIESCENT"])
+
+    def test_timeout_kills_child_group_and_marks_tick_failed(self):
+        command = [sys.executable, "-c", "import sys,time; sys.stdin.read(); time.sleep(10)"]
+        status, answer, error = CONTROL["run_tick"](command, {}, "tick", 1)
+        self.assertEqual(status, 124)
+        self.assertEqual(answer, "")
+        self.assertIn("timeout", error)
+
+    def test_interactive_session_detects_dir_and_resumed_session(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            project = root / "project"
+            project.mkdir()
+            proc = root / "proc"
+            entry = proc / "123"
+            entry.mkdir(parents=True)
+            (entry / "cwd").symlink_to(root)
+            database = root / "opencode.db"
+            with sqlite3.connect(database) as db:
+                db.execute("CREATE TABLE session (id TEXT, directory TEXT)")
+                db.execute("INSERT INTO session VALUES (?, ?)", ("ses_test", str(project)))
+            (entry / "cmdline").write_bytes(b"opencode\0--dir\0" + str(project).encode() + b"\0")
+            self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
+            (entry / "cmdline").write_bytes(b"opencode\0-s\0ses_test\0")
+            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))
+
+
+if __name__ == "__main__":
+    unittest.main()

orchestration_tests/test_nq_review.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts/nq-review"
+
+
+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)
+
+
+if __name__ == "__main__":
+    unittest.main()