diff --git a/cloud/README.md b/cloud/README.md index baed554..e481baf 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -532,6 +532,27 @@ goal 模式的继续：server 把任务转成新 goal 语义（goal=新消息、 合并任务 worktree 已删、请分支（merged 409）；冲突与新建一致 （plan + approval=full、plan + goal 各 400）。 +## 主线 /goal 命令（发给主线一句话就自动建 goal） + +不用先知道主线是进行中还是已终态，也不用勾选项：给主线发 +一条含字面 `/goal` 的消息即可，例如“请在这条主线开启一个 +持久 /goal，目标是收xxx”，或直接写“/goal 收xxx”。解析在 +`cloud/goal_command.py`（server 与 worker 共用同一份，防两端 +判断不一致）；只是在讨论里提到“/goal 的用法”不算命令。 +入口三处同款：`POST /tasks/{id}/followups`、 +`POST /tasks/{id}/thread/message`、`POST /tasks/{id}/continue` +（message 字段）；响应里 `delivery` 说明实际走了哪条路： +queued 主线原地转 goal（`goal_converted`）、已终态复活并转 +goal（`goal_revived`）、running 主线先排队（`goal_queued`， +本轮不打断，worker 在本轮完成、排空到这条时把主线转成 goal +并回 queued，下次认领走 goal-resume 在同一 thread 上开新 +目标循环；本轮若失败，这条留在 pending，等下次成功轮次再转）。 +拒绝口径：已是 goal 任务 409（不重复创建）、plan 任务 400、 +merged 409、空目标 400。goal 语义与新建 goal 模式一致： +不限轮次、模型宣告“目标已达成”才结束，宣告后即终态，不 +是永不结束的常驻采集——需要继续收时再发一次 /goal 续开 +新循环。 + ## 任务间互发消息（peer messages，Muse 式通用互发） 任意两个同项目任务可以互发消息、对方做完可以回消息，不 diff --git a/cloud/goal_command.py b/cloud/goal_command.py new file mode 100644 index 0000000..62fe6a8 --- /dev/null +++ b/cloud/goal_command.py @@ -0,0 +1,79 @@ +"""主线 /goal 命令解析：发给主线会话的一句话里自动建 goal。 + +背景：此前主线不会因为收到“请在这条主线开启一个持久 /goal” +这样的自然语言就自己变成 goal 任务——goal 只在建任务时由 +调用方用 goal_mode=1 指定，或在 /continue 时显式勾 goal 模式。 +本模块把“/goal”当成显式命令标记，让发给主线的消息本身就能 +触发转换，调用方不用再知道任务当前是进行中还是已终态。 + +解析规则（刻意收紧，防误触发）： +- 文本里必须出现字面 "/goal"（全角 ／ 也认，大小写不敏感）。 +- 光出现还不够，必须是命令形状，二者居一： + 1. 文本以 /goal 开头（最显式的命令）； + 2. /goal 后面紧跟分隔符（空格、，、：、、）或“目标”标记， + 例如“/goal 收xxx”“/goal，目标是收xxx”； + 3. /goal 前面有开启类动词（开启/建立/创建/添加/设为/设置/ + 开一个），例如“请开启一个持久/goal收xxx”。 + 只是在讨论里提到“/goal 的用法”这种不算命令。 +- 目标取 /goal 之后的部分：先去分隔符，再去“目标是/目标为/ + 目标：/目标:”或“持久目标是/为/：/:”这类前缀，最后去尾句号。 +- 返回值：不是命令返回 None；是命令但没给目标返回 ""（调用方 + 据此回 400“目标不能为空”，而不是当成普通消息悄悄排队）； + 否则返回目标文本。 + +本模块只做纯文本解析，不碰数据库、不发请求，server 与 worker +共用同一份解析，避免两端对“是不是 /goal 命令”判断不一致。 +""" + +import re + +_GOAL_RE = re.compile(r"/goal(?![A-Za-z])", re.IGNORECASE) +_SEPARATORS = " \t\r\n:：,，、" +_TARGET_MARKERS = ( + "持久目标是", "持久目标为", "持久目标：", "持久目标:", + "目标是", "目标为", "目标：", "目标:", +) +_OPEN_VERBS = ("开启", "建立", "创建", "添加", "设为", "设置", "开一个") + + +def _strip_target(rest): + """去掉目标前的分隔符与“目标是”类前缀、尾句号。""" + rest = rest.lstrip(_SEPARATORS) + for marker in _TARGET_MARKERS: + if rest.startswith(marker): + rest = rest[len(marker):].lstrip(_SEPARATORS) + break + rest = rest.strip() + while rest.endswith("。"): + rest = rest[:-1].rstrip() + return rest + + +def parse_goal_command(text): + """解析主线消息里的 /goal 命令，返回目标或 None/""（见模块说明）。""" + if not isinstance(text, str): + return None + s = text.strip().replace("／", "/") + if not s: + return None + match = _GOAL_RE.search(s) + if match is None: + return None + before = s[:match.start()] + rest = s[match.end():] + stripped = s.lstrip() + is_command = False + if stripped.lower().startswith("/goal"): + is_command = True + elif rest and rest[0] in _SEPARATORS: + is_command = True + elif rest.startswith("目标"): + is_command = True + elif any(verb in before for verb in _OPEN_VERBS): + is_command = True + elif not rest: + # 单独一句“/goal”：是命令，但没目标，调用方回 400。 + is_command = True + if not is_command: + return None + return _strip_target(rest) diff --git a/cloud/server.py b/cloud/server.py index 04a309f..adf2dbb 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -67,6 +67,7 @@ from cloud.db import create_database as _create_database from cloud.db import init_postgres_schema as _init_postgres_schema from cloud.db import PostgresIntegrityError as _PgIntegrityError from cloud.db import PostgresOperationalError as _PgOperationalError +from cloud.goal_command import parse_goal_command as _parse_goal_command from cloud.peer_messages import deliver_peer_message as _deliver_peer_message from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups from cloud.peer_messages import peer_message_prefix as _peer_prefix @@ -4705,6 +4706,58 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): finally: conn.close() + def _goal_convert_queued(self, task_id, target): + """把尚未认领的 queued 主线原地转成新语义 goal 任务。 + + CAS 只改仍是 queued 的行：worker 在本方法执行间认领 + 时 rowcount 为 0，调用方回 409 请重试，绝不把已在跑 + 的任务悄悄改模式（在跑任务的转换走 followup 排队 + + worker 轮次边界转换，见 worker._convert_running_task_to_goal）。 + 成功返回新任务 dict，失败返回 None。""" + conn = self._db() + try: + cur = conn.execute( + "UPDATE tasks SET goal_mode = 1, goal = ?," + " verify_command = NULL, goal_max_rounds = 0," + " updated_at = ? WHERE id = ? AND status = 'queued'", + (target, time.time(), task_id)) + conn.commit() + if cur.rowcount != 1: + return None + return _task_to_dict(conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,) + ).fetchone()) + finally: + conn.close() + + def _goal_revive_terminal(self, task_id, target): + """把已终态主线复活并转成新语义 goal 任务（/continue 同款）。 + + CAS 要求状态在终态四态且 resume_message 为空：已有待续跑 + 消息时不覆盖它（那条消息会先跑，goal 请求不应把它顶掉）。 + 成功返回新任务 dict，失败返回 None。merged 不在此列， + 调用方须先按 merged 409 拦掉。""" + conn = self._db() + try: + cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " resume_message = ?, goal_mode = 1, goal = ?," + " verify_command = NULL, goal_max_rounds = 0," + " result_text = NULL, error_text = NULL," + " actual_model = NULL, updated_at = ?" + " WHERE id = ? AND status IN ('completed', 'failed'," + " 'interrupted', 'merge-failed')" + " AND resume_message IS NULL", + (target, target, time.time(), task_id)) + conn.commit() + if cur.rowcount != 1: + return None + return _task_to_dict(conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,) + ).fetchone()) + finally: + conn.close() + def _get_task(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: @@ -4878,6 +4931,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 送达（链路里可见），本轮跑完后立刻以它开下一轮，且 优先于排队中的消息。v1 只做运行时（含排队中）的任务： 已终态的任务 409；goal 任务 409（goal 循环自有轮次）。 + 例外：消息是 /goal 命令时（见 cloud/goal_command.py）， + 不走普通后续消息——queued 原地转 goal、running 排队后 + 由 worker 在轮次边界转 goal、已终态复活并转 goal； + 已是 goal、plan 任务、merged、空目标各按 409/400 拒绝。 权限与取消同款：项目 create_task（写操作，不能只读）。 """ body, err = self._read_json() @@ -4909,13 +4966,79 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, { "error": "mode must be queued or direct"}) return - if task["status"] not in ("queued", "running"): + goal_target = _parse_goal_command(text) + if goal_target is not None: + if not goal_target: + self._send_json(400, {"error": "goal 目标不能为空"}) + return + if task.get("goal_mode") or task.get("verify_command"): + self._send_json(409, { + "error": "主线已有 goal，/goal 不会重复创建", + "status": task["status"], + "goal": task.get("goal") or task.get("prompt"), + }) + return + if task.get("plan"): + self._send_json(400, { + "error": "plan 模式与 goal 模式冲突"}) + return + if task["status"] == "merged": + self._send_json(409, { + "error": "已合并任务不能继续（worktree 已删），请分支", + "status": task["status"], + }) + return + if task["status"] == "queued": + new_task = self._goal_convert_queued(task_id, goal_target) + if new_task is None: + self._send_json(409, { + "error": "任务状态已变化，请刷新后重试", + "status": task["status"], + }) + return + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], + "mainline_goal_created", + "%s:goal=%s" % (task_id, goal_target)) + self._send_json(200, dict( + new_task, delivery="goal_converted", + goal=goal_target)) + return + if task["status"] in TERMINAL_STATUSES: + new_task = self._goal_revive_terminal(task_id, goal_target) + if new_task is None: + self._send_json(409, { + "error": "任务已有待续跑消息或状态已变化，" + "请刷新后重试", + "status": task["status"], + }) + return + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], + "mainline_goal_created", + "%s:goal_revived=%s" + % (task_id, goal_target)) + self._send_json(200, dict( + new_task, delivery="goal_revived", + goal=goal_target)) + return + # running：不打断本轮，只排队；worker 在本轮完成、 + # 排空到这条时把主线转成 goal（见 worker 排空段）， + # 本轮若失败则这条留在 pending，等下次成功轮次再转。 + # 下面落到普通 followup 插入路，但响应里标明是 goal。 + if task["status"] not in ("queued", "running"): + self._send_json(409, { + "error": "后续消息仅支持进行中任务", + "status": task["status"], + }) + return + elif task["status"] not in ("queued", "running"): self._send_json(409, { "error": "后续消息仅支持进行中任务", "status": task["status"], }) return - if task.get("verify_command"): + if task.get("verify_command") and goal_target is None: self._send_json(409, {"error": "goal 任务暂不支持后续消息"}) return now = time.time() @@ -4944,6 +5067,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], "followup_created", "%s:%s" % (task_id, fu_id)) + if goal_target is not None: + store.audit(user_id, task["project_id"], + "mainline_goal_created", + "%s:goal_queued=%s" % (task_id, goal_target)) + if goal_target is not None: + self._send_json(201, dict( + fu, delivery="goal_queued", goal=goal_target)) + return self._send_json(201, fu) def _append_direct_trace(self, task, fu_id, text): @@ -6194,9 +6325,60 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 自动化 heartbeat 共用同一路，不各写一套复活逻辑）： goal 任务 409、merged 409、终态任务 CAS 复活（清旧 结果/旧错误）、进行中任务写 followup（direct 再追加 - 直接 trace）。返回 (http_code, payload)，不直接 - _send_json；配额检查由调用方在调本方法前做。""" + 直接 trace）。例外：文本是 /goal 命令时（见 + cloud/goal_command.py），按 /followups 的 goal 分支 + 处理——queued 原地转 goal、running 排队后由 worker + 在轮次边界转 goal、已终态复活并转 goal。返回 + (http_code, payload)，不直接 _send_json；配额检查 + 由调用方在调本方法前做。""" task_id = task["id"] + goal_target = _parse_goal_command(text) + if goal_target is not None: + if not goal_target: + return 400, {"error": "goal 目标不能为空"} + if task.get("verify_command") or task.get("goal_mode"): + return 409, {"error": "主线已有 goal，/goal 不会重复创建", + "status": task["status"], + "goal": task.get("goal") or task.get("prompt")} + if task.get("plan"): + return 400, {"error": "plan 模式与 goal 模式冲突"} + if task["status"] == "merged": + return 409, { + "error": "已合并任务不能继续（worktree 已删），请分支", + "status": task["status"]} + if task["status"] in TERMINAL_STATUSES: + new_task = self._goal_revive_terminal(task_id, goal_target) + if new_task is None: + return 409, { + "error": "target has a pending resume message" + " or status changed", + "status": task["status"]} + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], audit_action, + "%s:goal_revived" % task_id) + return 201, { + "task_id": task_id, "delivery": "goal_revived", + "followup_id": None, "status": "queued", + "mode": mode, "goal": goal_target} + if task["status"] == "queued": + new_task = self._goal_convert_queued(task_id, goal_target) + if new_task is None: + return 409, { + "error": "任务状态已变化，请刷新后重试", + "status": task["status"]} + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], audit_action, + "%s:goal_converted" % task_id) + return 201, { + "task_id": task_id, "delivery": "goal_converted", + "followup_id": None, "status": "queued", + "mode": mode, "goal": goal_target} + if task["status"] != "running": + return 409, { + "error": "后续消息仅支持进行中任务", + "status": task["status"]} + # running：落到下面的普通 followup 插入路，响应里 + # 标 delivery=goal_queued（见本函数末尾）。 if task.get("verify_command") or task.get("goal_mode"): return 409, {"error": "goal 任务暂不支持后续消息"} if task["status"] == "merged": @@ -6255,6 +6437,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], audit_action, "%s:%s" % (task_id, fu_id)) + if goal_target is not None: + store.audit(user_id, task["project_id"], + "mainline_goal_created", + "%s:goal_queued=%s" % (task_id, goal_target)) + if goal_target is not None: + return 201, dict(fu, delivery="goal_queued", + goal=goal_target) return 201, dict(fu, delivery="queued" if mode == "queued" else "direct") @@ -7362,7 +7551,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): goal_mode=1 时同时把任务转成新 goal 语义（goal=新消息、 verify_command 清空、轮次 0），worker 走 goal-resume 在旧线程上开新目标循环；goal_mode 不传时 goal 任务沿用 - 原语义续跑。状态回 queued 时上一轮 result/error 清空 + 原语义续跑。例外：message 本身是 /goal 命令时强制 + goal_mode=1，goal 与 resume_message 都取解析出的目标 + 而非整句（见 cloud/goal_command.py）。状态回 queued 时上一轮 result/error 清空 （与重试一致，否则排队中的任务还挂着旧结果）。 配额：全部校验通过后、写库之前按任务创建者查一次 （与重试同口径，费用记创建者名下）；超限回 402，不 @@ -7415,6 +7606,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if len(message) > 20000: self._send_json(400, {"error": "message too long"}) return + goal_target = _parse_goal_command(message) + if goal_target is not None and not goal_target: + self._send_json(400, {"error": "goal 目标不能为空"}) + return + resume_text = goal_target if goal_target else message model = body.get("model") if model is not None: if not isinstance(model, str): @@ -7469,6 +7665,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): goal_mode = None if "goal_mode" in body: goal_mode = 1 if body["goal_mode"] in (True, 1) else 0 + if goal_target is not None: + # /goal 命令本身就是显式 goal 请求，优先于 body 里的 + # goal_mode=0（消息比选项更具体），plan 冲突仍在下面 400。 + goal_mode = 1 plan_eff = plan if plan is not None else (task["plan"] or 0) goal_eff = goal_mode if goal_mode is not None \ else (task["goal_mode"] or 0) @@ -7485,7 +7685,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): sets = ["status = 'queued'", "resume_message = ?", "result_text = NULL", "error_text = NULL", "actual_model = NULL", "updated_at = ?"] - vals = [message, time.time()] + vals = [resume_text, time.time()] if model is not None or "model" in body: sets.append("model = ?") vals.append(model) @@ -7509,7 +7709,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): vals.append(goal_mode) if goal_mode == 1: sets.append("goal = ?") - vals.append(message) + vals.append(resume_text) sets.append("verify_command = NULL") sets.append("goal_max_rounds = 0") vals.append(task_id) diff --git a/cloud/tests/test_mainline_goal_command.py b/cloud/tests/test_mainline_goal_command.py new file mode 100644 index 0000000..7a55fc9 --- /dev/null +++ b/cloud/tests/test_mainline_goal_command.py @@ -0,0 +1,343 @@ +"""主线 /goal 命令单测：发给主线的一句话自动建 goal。 + +不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时 +SQLite；worker 侧注入假 runner。覆盖：解析、followup 的 +queued/running/已终态三态、已是 goal、空目标、plan 冲突、 +普通消息不受影响、/continue 与 /thread/message 同款转换、 +worker 在轮次边界把 running 主线转成 goal 并在下次认领走 +goal-resume。 +""" + +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.goal_command import parse_goal_command + from cloud.peer_messages import ensure_followups_table + 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.goal_command import parse_goal_command + from cloud.peer_messages import ensure_followups_table + from cloud.server import create_server, init_db + from cloud.worker import Worker + + +class ParserTestCase(unittest.TestCase): + def test_user_sentence(self): + self.assertEqual( + parse_goal_command("请在这条主线开启一个持久 /goal，目标是收xxx"), + "收xxx") + self.assertEqual( + parse_goal_command("请在这条主线开启一个持久 /goal，目标是收xxx。"), + "收xxx") + + def test_slash_forms(self): + self.assertEqual(parse_goal_command("/goal 收xxx"), "收xxx") + self.assertEqual(parse_goal_command("/goal，目标是收xxx"), "收xxx") + self.assertEqual(parse_goal_command("/goal: 收xxx"), "收xxx") + self.assertEqual(parse_goal_command("请开启/goal收xxx"), "收xxx") + self.assertEqual(parse_goal_command("/goal"), "") + + def test_discussion_is_not_a_command(self): + self.assertIsNone(parse_goal_command("讨论/goal的用法")) + self.assertIsNone(parse_goal_command("请帮我完成目标收xxx")) + self.assertIsNone(parse_goal_command("hello world")) + + +class ServerTestCase(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": "do work"} + body.update(extra) + status, task = self._req("POST", "/tasks", body, self.owner_key) + self.assertEqual(status, 201, task) + return task + + 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 _get_task(self, task_id): + status, task = self._req("GET", "/tasks/" + task_id, None, + self.owner_key) + self.assertEqual(status, 200, task) + return task + + def test_followup_queued_converts_in_place(self): + task = self._create() + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "请在这条主线开启一个持久 /goal，目标是收xxx"}, + self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["delivery"], "goal_converted") + self.assertEqual(body["goal"], "收xxx") + after = self._get_task(task["id"]) + self.assertEqual(after["status"], "queued") + self.assertEqual(after["goal_mode"], 1) + self.assertEqual(after["goal"], "收xxx") + self.assertIsNone(after["verify_command"]) + self.assertEqual(after["goal_max_rounds"], 0) + + def test_followup_running_queues_for_worker(self): + task = self._create() + self._set_status(task["id"], "running") + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "/goal 收xxx"}, self.owner_key) + self.assertEqual(status, 201, body) + self.assertEqual(body["delivery"], "goal_queued") + self.assertEqual(body["goal"], "收xxx") + # running 主线本轮不被打断：DB 里仍是 running，未转 goal。 + after = self._get_task(task["id"]) + self.assertEqual(after["status"], "running") + self.assertEqual(after["goal_mode"], 0) + + def test_followup_terminal_revives_as_goal(self): + task = self._create() + self._set_status(task["id"], "completed") + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "/goal 收xxx"}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["delivery"], "goal_revived") + after = self._get_task(task["id"]) + self.assertEqual(after["status"], "queued") + self.assertEqual(after["goal_mode"], 1) + self.assertEqual(after["goal"], "收xxx") + self.assertEqual(after["resume_message"], "收xxx") + + def test_followup_already_goal_409(self): + task = self._create(goal_mode=1) + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "/goal 收yyy"}, self.owner_key) + self.assertEqual(status, 409, body) + after = self._get_task(task["id"]) + self.assertEqual(after["goal"], "do work") + + def test_followup_empty_goal_400(self): + task = self._create() + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "/goal"}, self.owner_key) + self.assertEqual(status, 400, body) + + def test_followup_plan_conflict_400(self): + task = self._create(plan=1) + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "/goal 收xxx"}, self.owner_key) + self.assertEqual(status, 400, body) + + def test_followup_merged_409(self): + task = self._create() + self._set_status(task["id"], "merged") + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "/goal 收xxx"}, self.owner_key) + self.assertEqual(status, 409, body) + + def test_ordinary_followup_unchanged(self): + task = self._create() + status, body = self._req( + "POST", "/tasks/%s/followups" % task["id"], + {"body": "继续做"}, self.owner_key) + self.assertEqual(status, 201, body) + self.assertNotIn("goal", body) + after = self._get_task(task["id"]) + self.assertEqual(after["goal_mode"], 0) + + def test_continue_goal_command(self): + task = self._create() + self._set_status(task["id"], "completed") + status, body = self._req( + "POST", "/tasks/%s/continue" % task["id"], + {"message": "请开启/goal，目标是收xxx"}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["goal_mode"], 1) + self.assertEqual(body["goal"], "收xxx") + self.assertEqual(body["resume_message"], "收xxx") + + def test_thread_message_terminal_goal(self): + task = self._create() + self._set_status(task["id"], "failed") + status, body = self._req( + "POST", "/tasks/%s/thread/message" % task["id"], + {"body": "/goal 收xxx"}, self.owner_key) + self.assertEqual(status, 201, body) + self.assertEqual(body["delivery"], "goal_revived") + after = self._get_task(task["id"]) + self.assertEqual(after["goal_mode"], 1) + self.assertEqual(after["goal"], "收xxx") + + +class WorkerTestCase(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_task(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 _insert_followup(self, task_id, body): + conn = sqlite3.connect(self.db_path) + try: + ensure_followups_table(conn) + conn.execute( + "INSERT INTO task_followups(id, task_id, project_id," + " body, mode, status, created_by, created_at," + " updated_at) VALUES(?, ?, ?, ?, 'queued', 'pending'," + " ?, ?, ?)", + ("fu1", task_id, "p1", body, "tester", + time.time(), time.time())) + conn.commit() + finally: + conn.close() + + def _row(self, task_id): + conn = sqlite3.connect(self.db_path) + try: + row = conn.execute( + "SELECT status, goal_mode, goal, resume_message" + " FROM tasks WHERE id = ?", (task_id,)).fetchone() + fu = conn.execute( + "SELECT status FROM task_followups WHERE id = 'fu1'" + ).fetchone() + return row, (fu[0] if fu else None) + finally: + conn.close() + + def test_running_task_converts_at_turn_boundary_then_goal_runs(self): + self._insert_task("T") + self._insert_followup("T", "/goal 收xxx") + seen = [] + + def fake_dispatch(cmd): + seen.append(list(cmd)) + name = cmd[cmd.index("--name") + 1] + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": name, "status": "completed", + "text": "done"}), + stderr="") + + worker = Worker(self.db_path, "/fake/host.py", + runner=fake_dispatch, env=self.env) + result = worker.run_one() + self.assertEqual(result["status"], "queued") + (status, goal_mode, goal, resume), fu_status = self._row("T") + self.assertEqual(status, "queued") + self.assertEqual(goal_mode, 1) + self.assertEqual(goal, "收xxx") + self.assertEqual(resume, "收xxx") + self.assertEqual(fu_status, "done") + self.assertTrue(any("dispatch" in c for c in seen)) + self.assertFalse(any("goal" in c for c in seen)) + + # 下次认领走 goal-resume：goal 循环宣告达成即完成。 + def fake_goal(cmd): + seen.append(list(cmd)) + # goal-resume 的命令元素是 "goal-resume"，不是 "goal"， + # 用子串判，别用元素相等判。 + if any("goal" in part for part in cmd): + return SimpleNamespace( + returncode=0, + stdout="round 1: turn=completed verify_rc=-" + " verify_out=\n" + "goal met after rounds\n", + stderr="") + raise AssertionError("第二轮不应走 dispatch: %s" % cmd) + + worker2 = Worker(self.db_path, "/fake/host.py", + runner=fake_goal, env=self.env) + result2 = worker2.run_one() + self.assertEqual(result2["status"], "completed") + self.assertTrue(any(any("goal" in part for part in c) + for c in seen)) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 77ca0bd..4086b14 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -144,6 +144,7 @@ from cloud.coordinator import nonterminal_children as _coord_nonterminal from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \ sum_usage, unpriced_token_cap from cloud.db import connect as _db_connect +from cloud.goal_command import parse_goal_command as _parse_goal_command from cloud.logging_config import get_logger, log_event, setup_logging from cloud.models import ModelRouter, is_fallbackable, record_fallback from cloud.auto_memory import AutoMemoryStore @@ -3212,6 +3213,36 @@ class Worker: except Exception: pass + def _convert_running_task_to_goal(self, task_id, target): + """把正在跑的主线在轮次边界转成新语义 goal 任务。 + + 只在排空循环里调用：本轮首轮已 completed、任务在库里 + 仍是 running，CAS 把它改回 queued 并写 goal 字段与 + resume_message——下次 worker 认领时走 /continue 同款 + goal-resume，在同一 thread 上开新目标循环（不是新任务、 + 不是分支）。CAS 失败（任务已被取消/回收、状态已不是 + running）返回 False，调用方把 followup 放回 pending， + 绝不把取消过的任务悄悄改回 queued。 + 成功返回 True。""" + try: + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + cur = conn.execute( + "UPDATE tasks SET goal_mode = 1, goal = ?," + " verify_command = NULL, goal_max_rounds = 0," + " status = 'queued', resume_message = ?," + " result_text = NULL, error_text = NULL," + " actual_model = NULL, updated_at = ?" + " WHERE id = ? AND status = 'running'", + (target, target, time.time(), task_id)) + conn.commit() + return cur.rowcount == 1 + finally: + conn.close() + except Exception: + return False + def _restore_followup(self, fu_id): """取消时把跑中的后续消息放回 pending，留给重试入口。""" try: @@ -5693,6 +5724,11 @@ class Worker: # 这里只负责把轮次跑出来。一条失败即停：任务记 failed， # 剩余 pending 留给重试入口（重试走已存在续跑，跑完继续 # 排空剩余消息）；取消时该条回 pending，任务走中断收尾。 + # /goal 命令不在这里当普通轮次跑：排空到它时把主线转 + # 成 goal 任务并回 queued，下次认领走 goal-resume 开 + # 新目标循环（服务端已在 followup/thread 入口把 running + # 的 /goal 排成 followup，见 server._create_followup）。 + goal_converted = False if (result is not None and result.get("status") == "completed" and not quota_stop and not task.get("verify_command") @@ -5701,6 +5737,23 @@ class Worker: fu = self._claim_next_followup(task_id) if fu is None: break + goal_target = _parse_goal_command(fu["body"]) + if goal_target is not None: + if not goal_target or task.get("plan"): + # 服务端已拦空目标与 plan 冲突，这里是 + # 兜底：不跑、不转，标 failed 停排空。 + self._finish_followup(fu["id"], "failed") + break + if self._convert_running_task_to_goal( + task_id, goal_target): + self._finish_followup(fu["id"], "done") + self._audit(task["project_id"], + "worker_goal_converted", + "%s:goal=%s" % (task_id, goal_target)) + goal_converted = True + else: + self._restore_followup(fu["id"]) + break # 后续消息执行时查配额：超限时该条回 pending（留 # 给重试入口），任务走配额中断收尾，不再排空剩余； # 建消息时不查（排队本身不花钱，顶上来后可跑）。 @@ -5909,10 +5962,23 @@ class Worker: self._notify_task_finished( task_id, task["project_id"], final, error=error) else: - # 已被取消/回收：DB 状态以 interrupted 为准，不再 - # 补一条 worker_completed/failed 审计误导排查。 - final = "interrupted" - self._audit(task["project_id"], "worker_interrupted", task_id) + if goal_converted: + # /goal 转换：DB 状态已是本轮转换写回的 queued， + # _set_status_if_running 因状态不是 running 而 + # 没改行是预期，不是取消/回收；绝不能把 local + # 的 final 改成 interrupted（那只会误导返回体， + # DB 本身已是 queued）。审计也按转换记，不按中断记。 + final = "queued" + self._audit(task["project_id"], "worker_goal_converted", + task_id) + log_event(self.logger, "worker_goal_converted", + task_id=task_id, project_id=task["project_id"], + status="queued") + else: + # 已被取消/回收：DB 状态以 interrupted 为准，不再 + # 补一条 worker_completed/failed 审计误导排查。 + final = "interrupted" + self._audit(task["project_id"], "worker_interrupted", task_id) if resume_message and not quota_stop: # 本轮已跑完（无论成败）：续跑消息已消费，清掉；进程 # 中途被杀时清不到，下次认领会重跑同一条消息——可接受