"""G10 云端 goal 断点续跑单测：端点口径 + worker 断点路由。 不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时 SQLite；worker 侧注入假 runner，只断言命令形状与最终状态。 本文件不含任何真实密钥、token、证书或私钥。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.server import create_server from cloud.worker import Worker except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.server import create_server from cloud.worker import Worker def _goal_met_runner(cmd): return SimpleNamespace( returncode=0, stdout="round 1: turn=completed verify_rc=- verify_out=\n" "goal met after rounds\n", stderr="") class GoalResumeTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.other_id, self.other_key = self.auth.create_user("other") self.server = create_server(self.db_path, "127.0.0.1", 0) self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201, self.project) self.env = {"XCM_API_KEY": "test-placeholder"} def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection( "127.0.0.1", self.port, timeout=10) hdrs = {} payload = None if key: hdrs["X-API-Key"] = key if body is not None: payload = json.dumps(body) hdrs["Content-Type"] = "application/json" conn.request(method, path, body=payload, headers=hdrs) resp = conn.getresponse() raw = resp.read() conn.close() try: return resp.status, json.loads(raw.decode("utf-8")) except ValueError: return resp.status, raw.decode("utf-8", errors="replace") def _create(self, **extra): body = {"project_id": self.project["id"], "prompt": "build it"} body.update(extra) return self._req("POST", "/tasks", body, self.owner_key) def _set_status(self, task_id, status): conn = sqlite3.connect(self.db_path) try: conn.execute("UPDATE tasks SET status = ? WHERE id = ?", (status, task_id)) conn.commit() finally: conn.close() def _row(self, task_id): conn = sqlite3.connect(self.db_path) try: row = conn.execute( "SELECT status, goal_resume_pending FROM tasks" " WHERE id = ?", (task_id,)).fetchone() return row finally: conn.close() def test_interrupted_goal_task_resumes_to_queued_with_flag(self): status, task = self._create(goal_mode=1) self.assertEqual(status, 201, task) self._set_status(task["id"], "interrupted") status, body = self._req( "POST", "/tasks/%s/goal-resume" % task["id"], {}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["status"], "queued") self.assertEqual(body["goal_resume_pending"], 1) row = self._row(task["id"]) self.assertEqual(row[0], "queued") self.assertEqual(row[1], 1) conn = sqlite3.connect(self.db_path) try: audit = conn.execute( "SELECT action FROM audit_log WHERE project_id = ?" " AND action = 'task_goal_resumed'", (self.project["id"],)).fetchall() finally: conn.close() self.assertEqual(len(audit), 1) def test_terminal_and_failed_and_non_goal_all_409(self): status, goal_task = self._create(goal_mode=1) self.assertEqual(status, 201, goal_task) status, plain_task = self._create() self.assertEqual(status, 201, plain_task) for task_id, status_value in ( (goal_task["id"], "completed"), (goal_task["id"], "failed"), (plain_task["id"], "interrupted")): self._set_status(task_id, status_value) status, body = self._req( "POST", "/tasks/%s/goal-resume" % task_id, {}, self.owner_key) self.assertEqual(status, 409, (task_id, status_value, body)) row = self._row(task_id) self.assertEqual(row[0], status_value) self.assertEqual(row[1], 0) def test_non_maintainer_forbidden(self): status, task = self._create(goal_mode=1) self.assertEqual(status, 201, task) self._set_status(task["id"], "interrupted") status, body = self._req( "POST", "/tasks/%s/goal-resume" % task["id"], {}, self.other_key) self.assertEqual(status, 403, body) row = self._row(task["id"]) self.assertEqual(row[0], "interrupted") def test_worker_breakpoint_uses_goal_resume_without_goal_text(self): # 端点置标记后，worker 认领应走 goal-resume（无 --goal-text）， # 而不是 goal（撞已存在）或 goal-resume --goal-text（开新循环）。 # main（#86）起未指定 executor 的任务默认成 docker，本地 # worker 按认领过滤跳过它是正确行为；本测试要的是本地断点 # 路由，所以显式指定 executor="local"。 status, task = self._create(goal_mode=1, executor="local") self.assertEqual(status, 201, task) self._set_status(task["id"], "interrupted") status, body = self._req( "POST", "/tasks/%s/goal-resume" % task["id"], {}, self.owner_key) self.assertEqual(status, 200, body) # 本测试测的是“有断点的任务走 goal-resume”：先给任务补 # 一行库断点（模拟跑过一轮），否则按从未启动回落规则， # 它会被当成从未启动而走全新 goal（见 # test_neverstart_fresh.py）。 conn = sqlite3.connect(self.db_path) try: conn.execute( "INSERT INTO task_checkpoints(task_id, round_no," " thread_id, goal_status, state_sha, verify_out," " created_at) VALUES(?, ?, ?, ?, ?, ?, ?)", (task["id"], 1, "th-test", "goal-running", "sha", "", 0)) conn.commit() finally: conn.close() seen = [] def fake(cmd): seen.append(list(cmd)) return _goal_met_runner(cmd) worker = Worker(self.db_path, "/fake/host.py", runner=fake, env=self.env) result = worker.run_one() self.assertEqual(result["status"], "completed", result) self.assertEqual(len(seen), 1) cmd = seen[0] self.assertIn("goal-resume", cmd) self.assertNotIn("--goal-text", cmd) self.assertNotIn("--max-rounds", cmd) row = self._row(task["id"]) self.assertEqual(row[0], "completed") self.assertEqual(row[1], 0) if __name__ == "__main__": unittest.main()