"""新 goal 语义单测：prompt 即目标，无验收命令，不限轮次， 模型在回合末尾单独一行输出“目标已达成”即达标。 不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时 SQLite；worker 侧注入假 runner；旧语义（verify_command）任务 仍按原样建任务、走 goal 循环。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import time import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.sandbox import LocalExecutor from cloud.server import create_server, init_db 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.sandbox import LocalExecutor from cloud.server import create_server, init_db 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 ServerGoalModeTestCase(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.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) 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 test_create_goal_mode_task(self): status, task = self._create(goal_mode=1) self.assertEqual(status, 201) self.assertEqual(task["goal_mode"], 1) # 新语义：prompt 本身就是目标，不存验收命令，轮次存 0。 self.assertEqual(task["goal"], "build it") self.assertIsNone(task["verify_command"]) self.assertEqual(task["goal_max_rounds"], 0) def test_goal_mode_with_verify_command_400(self): status, body = self._create(goal_mode=1, verify_command="true") self.assertEqual(status, 400) def test_goal_mode_with_plan_400(self): status, body = self._create(goal_mode=1, plan=1) self.assertEqual(status, 400) def test_old_style_verify_task_still_works(self): status, task = self._create(verify_command="true", goal_max_rounds=2) self.assertEqual(status, 201) self.assertEqual(task["goal_mode"], 0) self.assertEqual(task["verify_command"], "true") self.assertEqual(task["goal_max_rounds"], 2) def test_console_goal_hint_and_no_old_inputs(self): # console 壳是静态页：无 key 也能取，_req 对非 JSON 直接回文本。 # React 版（2026-10-29 起）：旧单页的 goalHint id 与 # body.goal_mode = 1 字面量已不存在，goal 提示改为 # DispatchSection 的条件渲染文案，提交时仍设 goal_mode=1； # 断言意图不变，只换成构建产物里实际存在的标记。 _status, text = self._req("GET", "/console") self.assertIn("goal 模式：输入框内容即目标", text) self.assertIn("goal_mode", text) # 旧的专用 prompt / 验收命令 / 轮次输入已移除。 self.assertNotIn('id="verifyCommand"', text) self.assertNotIn('id="goalRounds"', text) self.assertNotIn('id="goalFields"', text) class ExecutorGoalModeTestCase(unittest.TestCase): def test_local_goal_task_no_verify_zero_rounds(self): seen = [] def fake(cmd): seen.append(list(cmd)) return _goal_met_runner(cmd) result = LocalExecutor("/fake/host.py", runner=fake).goal_task( {"id": "t1", "prompt": "build it", "goal": "build it", "goal_mode": 1, "goal_max_rounds": 0}) self.assertEqual(result["status"], "completed") self.assertTrue(result["goal_met"]) # 新形态：不传 --verify，--max-rounds 原样传 0（不能被 # or 兜底成 3，否则不限轮次悄悄变回 3 轮上限）。 self.assertNotIn("--verify", seen[0]) self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "0") class WorkerGoalModeTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.env = {"XCM_API_KEY": "test-placeholder"} def tearDown(self): self.tmp.cleanup() def _insert(self, task_id, **extra): now = time.time() cols = ["id", "project_id", "prompt", "status", "idempotency_key", "depends_on", "priority", "created_by", "created_at", "updated_at", "last_heartbeat"] vals = [task_id, "p1", "do work", "queued", None, "[]", 0, "tester", now, now, None] for k, v in extra.items(): cols.append(k) vals.append(v) conn = sqlite3.connect(self.db_path) try: conn.execute( "INSERT INTO tasks(%s) VALUES(%s)" % (", ".join(cols), ", ".join("?" * len(cols))), vals) conn.commit() finally: conn.close() def test_goal_mode_task_routes_to_goal_not_dispatch(self): self._insert("T", goal_mode=1, goal="build it", goal_max_rounds=0) seen = [] def fake(cmd): seen.append(list(cmd)) if "goal" in cmd: return _goal_met_runner(cmd) raise AssertionError("goal 任务不应走 dispatch: %s" % cmd) worker = Worker(self.db_path, "/fake/host.py", runner=fake, env=self.env) result = worker.run_one() self.assertEqual(result["status"], "completed") self.assertTrue(any("goal" in cmd for cmd in seen)) self.assertNotIn("--verify", seen[0]) self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "0") if __name__ == "__main__": unittest.main()