host
3a4997d346
server：POST /tasks/{id}/continue（终端非 merged 任务重排队，
resume_message 列存本轮消息，不传选项沿用原值，plan/goal 冲突
与新建一致，goal_mode=1 时转新 goal 语义；CAS 防并发改状态，
审计 task_continued）；tasks.resume_message 三处迁移 + PG 建表。
worker：认领后本轮 prompt 换成续跑消息（只换内存副本），
goal 续跑走 goal-resume 而不是 goal，跑完清 resume_message
（进程被杀时不清，下次认领重跑同一条消息）。
host：goal-resume --goal-text（旧任务接续 goal 模式：跳过状态
与记录守卫，旧线程上从第 1 轮开新目标循环，新语义不限轮次；
无线程时拒绝）。
sandbox：Local/Docker goal_resume_task + popen goal-resume kind；
附带 popen goal 命令改成新语义（无验收不传 --verify，轮次 0
原样传，此前 goal 任务到本地 worker 会直接报错）。
console：详情页终端非 merged 任务出继续会话表单（消息 +
模型/强度/权限/计划/goal，模型下拉复用中转列表）。
测试：新增 test_task_continue.py 11 个 + test_host_continue.py 2 个。
316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""旧任务继续会话单测：终端任务带新消息 + 全部选项重排队，
|
||
worker 在同一会话线程上再跑一轮（不是新任务、不是分支）。
|
||
|
||
不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
|
||
SQLite；worker/sandbox 侧注入假 runner 录命令。
|
||
"""
|
||
|
||
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 DockerExecutor, 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 DockerExecutor, LocalExecutor
|
||
from cloud.server import create_server, init_db
|
||
from cloud.worker import Worker
|
||
|
||
|
||
class ServerContinueTestCase(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)
|
||
|
||
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, UnicodeDecodeError):
|
||
return resp.status, {}
|
||
|
||
def _task_in_status(self, status, **extra):
|
||
_s, task = self._req(
|
||
"POST", "/tasks",
|
||
{"project_id": self.project["id"], "prompt": "original"},
|
||
self.owner_key)
|
||
self.assertEqual(_s, 201, task)
|
||
conn = sqlite3.connect(self.db_path)
|
||
sets = ["status = ?"]
|
||
vals = [status]
|
||
for k, v in extra.items():
|
||
sets.append("%s = ?" % k)
|
||
vals.append(v)
|
||
vals.append(task["id"])
|
||
conn.execute("UPDATE tasks SET %s WHERE id = ?"
|
||
% ", ".join(sets), vals)
|
||
conn.commit()
|
||
conn.close()
|
||
return task["id"]
|
||
|
||
def _continue(self, task_id, body, key=None):
|
||
return self._req(
|
||
"POST", "/tasks/%s/continue" % task_id, body,
|
||
key or self.owner_key)
|
||
|
||
def test_continue_completed_with_options(self):
|
||
tid = self._task_in_status("completed")
|
||
status, task = self._continue(tid, {
|
||
"message": "follow up please", "model": "gpt-6-sol",
|
||
"effort": "high", "approval": "auto", "plan": 1})
|
||
self.assertEqual(status, 200, task)
|
||
self.assertEqual(task["status"], "queued")
|
||
self.assertEqual(task["resume_message"], "follow up please")
|
||
self.assertEqual(task["model"], "gpt-6-sol")
|
||
self.assertEqual(task["effort"], "high")
|
||
self.assertEqual(task["approval"], "auto")
|
||
self.assertEqual(task["plan"], 1)
|
||
self.assertEqual(task["prompt"], "original")
|
||
|
||
def test_continue_keeps_options_when_absent(self):
|
||
tid = self._task_in_status("failed", model="gpt-5.6-terra")
|
||
status, task = self._continue(tid, {"message": "again"})
|
||
self.assertEqual(status, 200, task)
|
||
self.assertEqual(task["model"], "gpt-5.6-terra")
|
||
self.assertEqual(task["resume_message"], "again")
|
||
|
||
def test_continue_goal_mode_converts_to_new_semantics(self):
|
||
tid = self._task_in_status(
|
||
"completed", goal="old goal", verify_command="true",
|
||
goal_max_rounds=3)
|
||
status, task = self._continue(
|
||
tid, {"message": "new goal", "goal_mode": 1})
|
||
self.assertEqual(status, 200, task)
|
||
self.assertEqual(task["goal_mode"], 1)
|
||
self.assertEqual(task["goal"], "new goal")
|
||
self.assertIsNone(task["verify_command"])
|
||
self.assertEqual(task["goal_max_rounds"], 0)
|
||
|
||
def test_continue_rejects_non_terminal_and_merged(self):
|
||
for status in ("queued", "running"):
|
||
tid = self._task_in_status(status)
|
||
code, _body = self._continue(tid, {"message": "hi"})
|
||
self.assertEqual(code, 409, status)
|
||
tid = self._task_in_status("merged")
|
||
code, _body = self._continue(tid, {"message": "hi"})
|
||
self.assertEqual(code, 409)
|
||
code, _body = self._continue("no-such-task", {"message": "hi"})
|
||
self.assertEqual(code, 404)
|
||
|
||
def test_continue_rejects_bad_input_and_conflicts(self):
|
||
tid = self._task_in_status("completed")
|
||
for body, code in [
|
||
({"message": ""}, 400),
|
||
({"message": "x" * 20001}, 400),
|
||
({"message": "hi", "effort": "ultra"}, 400),
|
||
({"message": "hi", "approval": "never"}, 400),
|
||
({"message": "hi", "plan": 1, "approval": "full"}, 400),
|
||
({"message": "hi", "plan": 1, "goal_mode": 1}, 400)]:
|
||
got, _body = self._continue(tid, body)
|
||
self.assertEqual(got, code, body)
|
||
|
||
def test_continue_forbidden_for_non_member(self):
|
||
tid = self._task_in_status("completed")
|
||
code, _body = self._continue(tid, {"message": "hi"},
|
||
key=self.other_key)
|
||
self.assertEqual(code, 403)
|
||
|
||
|
||
class WorkerContinueTestCase(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", "original", "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 _row(self, task_id):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT status, resume_message FROM tasks WHERE id = ?",
|
||
(task_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
return row
|
||
|
||
def test_resume_normal_runs_new_message_on_same_thread(self):
|
||
self._insert("t1", resume_message="follow up please")
|
||
seen = []
|
||
|
||
def fake(cmd):
|
||
seen.append(list(cmd))
|
||
if "dispatch" in cmd:
|
||
# host 任务已存在：dispatch 失败，worker 回落到 run。
|
||
return SimpleNamespace(returncode=1, stdout="任务已存在",
|
||
stderr="")
|
||
if "run" in cmd:
|
||
return SimpleNamespace(returncode=0, stdout="ok",
|
||
stderr="")
|
||
raise AssertionError("不应有第三种命令: %s" % cmd)
|
||
|
||
worker = Worker(self.db_path, "/fake/host.py", runner=fake,
|
||
env=self.env)
|
||
result = worker.run_one()
|
||
self.assertEqual(result["status"], "completed")
|
||
run_cmds = [c for c in seen if "run" in c]
|
||
self.assertTrue(run_cmds)
|
||
self.assertEqual(run_cmds[0][run_cmds[0].index("--prompt") + 1],
|
||
"follow up please")
|
||
status, resume = self._row("t1")
|
||
self.assertEqual(status, "completed")
|
||
self.assertIsNone(resume)
|
||
|
||
def test_resume_goal_routes_to_goal_resume(self):
|
||
self._insert("t2", goal_mode=1, goal="old goal",
|
||
goal_max_rounds=0,
|
||
resume_message="new goal")
|
||
seen = []
|
||
|
||
def fake(cmd):
|
||
seen.append(list(cmd))
|
||
if "goal-resume" in cmd:
|
||
return SimpleNamespace(
|
||
returncode=0,
|
||
stdout="round 1: turn=completed verify_rc=- "
|
||
"verify_out=\ngoal met after rounds\n",
|
||
stderr="")
|
||
raise AssertionError("goal 续跑不应走 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-resume" in c for c in seen))
|
||
cmd = [c for c in seen if "goal-resume" in c][0]
|
||
self.assertEqual(cmd[cmd.index("--goal-text") + 1], "new goal")
|
||
self.assertEqual(cmd[cmd.index("--max-rounds") + 1], "0")
|
||
status, resume = self._row("t2")
|
||
self.assertEqual(status, "completed")
|
||
self.assertIsNone(resume)
|
||
|
||
|
||
class SandboxContinueTestCase(unittest.TestCase):
|
||
def _recording(self, calls):
|
||
def fake(cmd):
|
||
calls.append(list(cmd))
|
||
return SimpleNamespace(returncode=0, stdout="ok", stderr="")
|
||
return fake
|
||
|
||
def test_local_goal_resume_command(self):
|
||
calls = []
|
||
ex = LocalExecutor("/fake/host.py",
|
||
runner=self._recording(calls))
|
||
ex.goal_resume_task({"id": "t1", "prompt": "new goal"},
|
||
"new goal")
|
||
self.assertIn("goal-resume", calls[0])
|
||
self.assertEqual(calls[0][calls[0].index("--goal-text") + 1],
|
||
"new goal")
|
||
self.assertEqual(calls[0][calls[0].index("--max-rounds") + 1],
|
||
"0")
|
||
|
||
def test_docker_goal_resume_command(self):
|
||
from unittest import mock
|
||
calls = []
|
||
ex = DockerExecutor(
|
||
"/fake/host.py", runner=self._recording(calls),
|
||
env={"XCM_API_KEY": "placeholder-not-a-real-key"})
|
||
with mock.patch("cloud.sandbox.shutil.which",
|
||
return_value="/usr/bin/docker"):
|
||
ex.goal_resume_task({"id": "t1", "prompt": "new goal"},
|
||
"new goal")
|
||
self.assertIn("goal-resume", calls[0])
|
||
self.assertIn("new goal", calls[0])
|
||
|
||
def test_popen_goal_resume_command(self):
|
||
import cloud.sandbox as sandbox_mod
|
||
calls = []
|
||
|
||
class FakePopen(object):
|
||
def __init__(self, cmd, **kwargs):
|
||
calls.append(list(cmd))
|
||
|
||
orig = sandbox_mod.subprocess.Popen
|
||
sandbox_mod.subprocess.Popen = FakePopen
|
||
try:
|
||
ex = LocalExecutor("/fake/host.py")
|
||
ex.popen_task({"id": "t1", "prompt": "new goal"},
|
||
kind="goal-resume")
|
||
finally:
|
||
sandbox_mod.subprocess.Popen = orig
|
||
self.assertIn("goal-resume", calls[0])
|
||
self.assertEqual(calls[0][calls[0].index("--goal-text") + 1],
|
||
"new goal")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|