orchestrator-agent
4ac4d3f1ec
Co-authored-by: orchestrator-agent <orchestrator@cubecraftlabs.com> Co-committed-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
109 lines
5.7 KiB
Python
109 lines
5.7 KiB
Python
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"]()
|
|
record["project_id"] = CONTROL["CTXD_PROJECT"]
|
|
config.write_text(json.dumps({"mcp": {"ctxd": {
|
|
"url": "http://localhost:9091/mcp/",
|
|
"headers": {"Authorization": "Bearer test"},
|
|
}}}))
|
|
self.assertEqual(CONTROL["check_ctxd"](), 6)
|
|
config.write_text(json.dumps({"mcp": {"ctxd": {
|
|
"url": "http://localhost:9091/invalid",
|
|
"headers": {"Authorization": "Bearer test"},
|
|
}}}))
|
|
with self.assertRaisesRegex(RuntimeError, "must end in /mcp"):
|
|
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))
|
|
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"))
|
|
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"}})
|
|
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\0web\0--dir\0" + str(project).encode() + b"\0")
|
|
self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
|
|
(entry / "cmdline").write_bytes(b"opencode\0serve\0")
|
|
(entry / "cwd").unlink()
|
|
(entry / "cwd").symlink_to(project)
|
|
self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
|
|
(entry / "cwd").unlink()
|
|
(entry / "cwd").symlink_to(root)
|
|
(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))
|
|
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__":
|
|
unittest.main()
|