From 3096e04345b8450e2e19d05f27cf4507755ece11 Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 05:17:23 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=E4=B8=BB=E7=BA=BF=20/goal=20?= =?UTF-8?q?=E5=91=BD=E4=BB=A4=E8=87=AA=E5=8A=A8=E5=BB=BA=20goal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 发给主线的一句话（含字面 /goal）即可自动建 goal，不用先 知道主线是进行中还是已终态，也不用勾选项： - cloud/goal_command.py：server 与 worker 共用的纯文本解析， 只认命令形状（/goal 开头、分隔符/目标标记、开启类动词）， 讨论里提到 /goal 不误触发；空目标返回空串供调用方 400。 - server followup/thread/message/continue 三入口同款：queued 原地转 goal，已终态复活并转 goal，running 排队后由 worker 在轮次边界转 goal；已是 goal 409、plan 400、merged 409。 - worker 排空：/goal 不当普通轮次跑，CAS 把 running 主线 改回 queued 并写 goal/resume_message，下次认领走 goal-resume 在同一 thread 上开新目标循环；取消/回收时 CAS 失败则把 followup 放回 pending，绝不把取消过的任务改回 queued。 - 单测 14 项：解析、三态转换、拒绝口径、普通消息不受影响、 continue/thread 同款、worker 轮次边界转换 + 下次 goal 跑通。 --- cloud/README.md | 21 ++ cloud/goal_command.py | 79 +++++ cloud/server.py | 214 +++++++++++++- cloud/tests/test_mainline_goal_command.py | 343 ++++++++++++++++++++++ cloud/worker.py | 74 ++++- 5 files changed, 720 insertions(+), 11 deletions(-) create mode 100644 cloud/goal_command.py create mode 100644 cloud/tests/test_mainline_goal_command.py 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: # 本轮已跑完（无论成败）：续跑消息已消费，清掉；进程 # 中途被杀时清不到，下次认领会重跑同一条消息——可接受 From f3bd65c171b38ee2c5cd5888d3628271ac11f8d0 Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 05:21:25 +0800 Subject: [PATCH 2/4] =?UTF-8?q?feat:=20=E6=96=B0=E5=BB=BA=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E7=9A=84=20prompt=20=E9=87=8C=E5=86=99=20/goal=20?= =?UTF-8?q?=E4=B9=9F=E8=87=AA=E5=8A=A8=E5=8F=98=20goal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上一提交只覆盖了已有主线的转换，新建任务的 prompt 里写 /goal 时仍会当成普通一轮任务跑。本提交补上 POST /tasks 入口：prompt 经 cloud/goal_command.py 解析后，目标即新任 务的 prompt 与 goal，goal_mode 强制为 1（命令优先于 body 里的 goal_mode=0），plan/验收命令冲突仍按新建规则 400， 空目标 400 不建任务；讨论里提到 /goal 不触发。 单测补 5 项：/goal 直写、整句、空目标、plan/验收冲突、 讨论 prompt 不变。 --- cloud/README.md | 5 ++- cloud/server.py | 16 ++++++++- cloud/tests/test_mainline_goal_command.py | 40 +++++++++++++++++++++++ 3 files changed, 59 insertions(+), 2 deletions(-) diff --git a/cloud/README.md b/cloud/README.md index e481baf..51004d6 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -539,7 +539,10 @@ goal 模式的继续：server 把任务转成新 goal 语义（goal=新消息、 持久 /goal，目标是收xxx”，或直接写“/goal 收xxx”。解析在 `cloud/goal_command.py`（server 与 worker 共用同一份，防两端 判断不一致）；只是在讨论里提到“/goal 的用法”不算命令。 -入口三处同款：`POST /tasks/{id}/followups`、 +新建目标也同款：`POST /tasks` 的 prompt 里直接写 `/goal +收xxx`（或上面那句整话）即新建 goal 任务，prompt 与 goal +都存解析出的目标，不用勾 goal 选项；空目标、plan 冲突、 +带验收命令各 400。入口三处同款：`POST /tasks/{id}/followups`、 `POST /tasks/{id}/thread/message`、`POST /tasks/{id}/continue` （message 字段）；响应里 `delivery` 说明实际走了哪条路： queued 主线原地转 goal（`goal_converted`）、已终态复活并转 diff --git a/cloud/server.py b/cloud/server.py index adf2dbb..15f6986 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -4244,6 +4244,17 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if not prompt or not isinstance(prompt, str): self._send_json(400, {"error": "prompt required"}) return + # 新建目标：prompt 里直接写 /goal 也算新建 goal，不必 + # 先勾选项——例如 prompt="/goal 收xxx" 或“请开启一个 + # 持久 /goal，目标是收xxx”。解析在 cloud/goal_command.py， + # 与主线转换同款（讨论里提到 /goal 不算）。目标即新 + # 任务的 prompt 与 goal；空目标 400，不建普通任务。 + prompt_goal_target = _parse_goal_command(prompt) + if prompt_goal_target is not None and not prompt_goal_target: + self._send_json(400, {"error": "goal 目标不能为空"}) + return + if prompt_goal_target: + prompt = prompt_goal_target depends_on = body.get("depends_on") or [] if isinstance(depends_on, str): depends_on = [depends_on] @@ -4327,7 +4338,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): # verify_command 是语义冲突，400。旧语义（verify_command # 非空）保留：goal 文本缺省用 prompt，轮次缺省 3，非法值 # 400 且不建任务；老客户端与已建的旧 goal 任务继续可用。 - goal_mode = 1 if body.get("goal_mode") in (True, 1) else 0 + # /goal 命令优先于 body 里的 goal_mode=0（prompt 里的 + # 命令比选项更具体），plan/验收命令冲突仍在下面 400。 + goal_mode = 1 if body.get("goal_mode") in (True, 1) \ + or prompt_goal_target else 0 if goal_mode: if plan: self._send_json(400, { diff --git a/cloud/tests/test_mainline_goal_command.py b/cloud/tests/test_mainline_goal_command.py index 7a55fc9..d81ad5f 100644 --- a/cloud/tests/test_mainline_goal_command.py +++ b/cloud/tests/test_mainline_goal_command.py @@ -208,6 +208,46 @@ class ServerTestCase(unittest.TestCase): after = self._get_task(task["id"]) self.assertEqual(after["goal_mode"], 0) + def test_create_task_prompt_goal_command(self): + task = self._create(prompt="/goal 收xxx") + self.assertEqual(task["goal_mode"], 1) + self.assertEqual(task["goal"], "收xxx") + self.assertEqual(task["prompt"], "收xxx") + self.assertIsNone(task["verify_command"]) + self.assertEqual(task["goal_max_rounds"], 0) + + def test_create_task_prompt_goal_user_sentence(self): + task = self._create( + prompt="请在这条主线开启一个持久 /goal，目标是收xxx") + self.assertEqual(task["goal_mode"], 1) + self.assertEqual(task["goal"], "收xxx") + self.assertEqual(task["prompt"], "收xxx") + + def test_create_task_prompt_goal_empty_400(self): + status, body = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "/goal"}, + self.owner_key) + self.assertEqual(status, 400, body) + + def test_create_task_prompt_goal_conflicts_400(self): + status, _body = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "/goal 收xxx", + "plan": 1}, + self.owner_key) + self.assertEqual(status, 400) + status, _body = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "/goal 收xxx", + "verify_command": "true"}, + self.owner_key) + self.assertEqual(status, 400) + + def test_create_task_discussion_prompt_unchanged(self): + task = self._create(prompt="讨论/goal的用法") + self.assertEqual(task["goal_mode"], 0) + def test_continue_goal_command(self): task = self._create() self._set_status(task["id"], "completed") From f1aab5415c48bdaddc627158e8dd0d9df9697b43 Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 05:27:30 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat:=20=E4=B8=8D=E7=94=A8=E6=96=9C?= =?UTF-8?q?=E6=9D=A0=EF=BC=8C=E5=8D=95=E5=8F=91=20goal=20=E4=B9=9F?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=20/goal=20=E7=9B=AE=E6=A0=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 触发词从仅 /goal 放宽到 /goal 或裸 goal（大小写不敏感， 单数 only，goals/mygoal 不算）：goal 收xxx、goal，目标是 收xxx、发goal 收xxx 均同款触发，新建与主线转换入口共用 同一解析。讨论句式仍不触发：goal 的用法、goal 是什么、 讨论goal的用法、/goal的用法 均返回 None。 单测补 4 项：裸 goal 触发、裸 goal 讨论不触发、followup 裸 goal 转换、新建裸 goal。 --- cloud/README.md | 9 ++- cloud/goal_command.py | 86 +++++++++++++++-------- cloud/tests/test_mainline_goal_command.py | 30 ++++++++ 3 files changed, 93 insertions(+), 32 deletions(-) diff --git a/cloud/README.md b/cloud/README.md index 51004d6..fbd6c31 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -535,10 +535,13 @@ goal 模式的继续：server 把任务转成新 goal 语义（goal=新消息、 ## 主线 /goal 命令（发给主线一句话就自动建 goal） 不用先知道主线是进行中还是已终态，也不用勾选项：给主线发 -一条含字面 `/goal` 的消息即可，例如“请在这条主线开启一个 -持久 /goal，目标是收xxx”，或直接写“/goal 收xxx”。解析在 +一条含触发词的消息即可，例如“请在这条主线开启一个 +持久 /goal，目标是收xxx”，或直接写“/goal 收xxx”。斜杠 +不用：单发 `goal` 也同款触发，例如“goal 收xxx”、 +“goal，目标是收xxx”。解析在 `cloud/goal_command.py`（server 与 worker 共用同一份，防两端 -判断不一致）；只是在讨论里提到“/goal 的用法”不算命令。 +判断不一致）；只是在讨论里提到“/goal 的用法”“goal 是什么” +这种不算命令。 新建目标也同款：`POST /tasks` 的 prompt 里直接写 `/goal 收xxx`（或上面那句整话）即新建 goal 任务，prompt 与 goal 都存解析出的目标，不用勾 goal 选项；空目标、plan 冲突、 diff --git a/cloud/goal_command.py b/cloud/goal_command.py index 62fe6a8..0423a23 100644 --- a/cloud/goal_command.py +++ b/cloud/goal_command.py @@ -7,15 +7,19 @@ 触发转换，调用方不用再知道任务当前是进行中还是已终态。 解析规则（刻意收紧，防误触发）： -- 文本里必须出现字面 "/goal"（全角 ／ 也认，大小写不敏感）。 +- 触发词是 "/goal" 或裸 "goal"（斜杠不用，全角 ／ 也认， + 大小写不敏感，单数 only——"goals" 不算）。 - 光出现还不够，必须是命令形状，二者居一： - 1. 文本以 /goal 开头（最显式的命令）； - 2. /goal 后面紧跟分隔符（空格、，、：、、）或“目标”标记， - 例如“/goal 收xxx”“/goal，目标是收xxx”； - 3. /goal 前面有开启类动词（开启/建立/创建/添加/设为/设置/ - 开一个），例如“请开启一个持久/goal收xxx”。 - 只是在讨论里提到“/goal 的用法”这种不算命令。 -- 目标取 /goal 之后的部分：先去分隔符，再去“目标是/目标为/ + 1. 文本以触发词开头（最显式的命令），例如“goal 收xxx”； + 2. 触发词后面紧跟分隔符（空格、，、：、、）或“目标”标记， + 例如“/goal 收xxx”“goal，目标是收xxx”； + 3. 触发词前面有开启/发送类动词（开启/建立/创建/添加/设为/ + 设置/开一个/发/发送），例如“请开启一个持久/goal收xxx”、 + “发goal 收xxx”。 + 只是在讨论里提到“/goal 的用法”“goal 的用法”“goal 是什么” + 这种不算命令：触发词后面紧跟“的/是/吗/什么/怎么/如何/ + 模式/用法/意思/功能”时一律不当命令。 +- 目标取触发词之后的部分：先去分隔符，再去“目标是/目标为/ 目标：/目标:”或“持久目标是/为/：/:”这类前缀，最后去尾句号。 - 返回值：不是命令返回 None；是命令但没给目标返回 ""（调用方 据此回 400“目标不能为空”，而不是当成普通消息悄悄排队）； @@ -28,12 +32,17 @@ import re _GOAL_RE = re.compile(r"/goal(?![A-Za-z])", re.IGNORECASE) +_BARE_GOAL_RE = re.compile(r"(? 0 and s[bare.start() - 1] == "/": + continue # 斜杠那处已在上面判过 + before = s[:bare.start()] + rest = s[bare.end():] + at_start = not before.strip() + before_ok = at_start or ( + before.rstrip() and before.rstrip()[-1] in _SEPARATORS) or ( + any(verb in before for verb in _OPEN_VERBS)) + if before_ok and _rest_is_command(rest): + return _strip_target(rest) + return None diff --git a/cloud/tests/test_mainline_goal_command.py b/cloud/tests/test_mainline_goal_command.py index d81ad5f..d7b9f1a 100644 --- a/cloud/tests/test_mainline_goal_command.py +++ b/cloud/tests/test_mainline_goal_command.py @@ -56,6 +56,22 @@ class ParserTestCase(unittest.TestCase): self.assertIsNone(parse_goal_command("请帮我完成目标收xxx")) self.assertIsNone(parse_goal_command("hello world")) + def test_bare_goal_without_slash(self): + # 不用斜杠：用户单发 goal 也触发同款目标。 + 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 收xxx"), "收xxx") + self.assertEqual(parse_goal_command("goal"), "") + + def test_bare_goal_discussion_still_not_a_command(self): + self.assertIsNone(parse_goal_command("goal的用法")) + self.assertIsNone(parse_goal_command("goal是什么意思")) + self.assertIsNone(parse_goal_command("讨论goal的用法")) + self.assertIsNone(parse_goal_command("请问goal是什么意思")) + self.assertIsNone(parse_goal_command("/goal的用法")) + class ServerTestCase(unittest.TestCase): def setUp(self): @@ -248,6 +264,20 @@ class ServerTestCase(unittest.TestCase): task = self._create(prompt="讨论/goal的用法") self.assertEqual(task["goal_mode"], 0) + def test_followup_bare_goal_converts(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") + + def test_create_task_bare_goal(self): + task = self._create(prompt="goal 收xxx") + self.assertEqual(task["goal_mode"], 1) + self.assertEqual(task["goal"], "收xxx") + def test_continue_goal_command(self): task = self._create() self._set_status(task["id"], "completed") From e81ba8d2bfc6bd86b1ada6ccf1f04c4ff5dd77ce Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 16:02:26 +0800 Subject: [PATCH 4/4] =?UTF-8?q?feat:=20=E4=BC=9A=E8=AF=9D=E5=86=85=20LLM?= =?UTF-8?q?=20=E8=87=AA=E5=88=A4=20goal=20/=20=E8=87=AA=E5=8A=A8=E5=8C=96?= =?UTF-8?q?=E6=A0=87=E8=AE=B0=E5=B7=A5=E5=85=B7=EF=BC=8C=E4=B8=8D=E9=9D=A0?= =?UTF-8?q?=E5=85=B3=E9=94=AE=E8=AF=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent 在回合输出里写 goal-convert 块，worker 在轮次边界 排成 /goal 自后续消息、走已有 followup 转换路径把本会话 转成 goal；写 automation-create 块，worker 在轮次边界 创建自动化定义（权限 create_task、配额整轮门、创建人跟 父任务走、name 缺失自动生成、首次时刻已过去单块拒绝、 bind heartbeat 可绑回本会话）。 automations 新增 first_run_at / run_once：首次时刻只管 首次、之后按间隔算；只跑一次跑过后 is_due / due / claim 三处都不再到期（SQL 到期条件与 Python 口径共用同一 _DUE_SQL）；API 创建与更新同步接受这两个字段。 刻意不向 LLM 自动注入工具指南：避免 prompt 臃肿，格式 只写进 docs/marker-tools.md 与 README，需要时由用户在 会话里手动告诉 agent；服务端不按关键词再猜意图。 单测 cloud/tests/test_session_goal_automation.py 12 项， 相关回归 218 项全过。 --- cloud/README.md | 9 + cloud/automations.py | 129 +++++--- cloud/marker_tools.py | 226 ++++++++++++- cloud/sandbox.py | 18 ++ cloud/server.py | 26 +- cloud/tests/test_session_goal_automation.py | 334 ++++++++++++++++++++ cloud/worker.py | 134 +++++++- docs/marker-tools.md | 67 +++- 8 files changed, 892 insertions(+), 51 deletions(-) create mode 100644 cloud/tests/test_session_goal_automation.py diff --git a/cloud/README.md b/cloud/README.md index fbd6c31..820f0e9 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -559,6 +559,15 @@ merged 409、空目标 400。goal 语义与新建 goal 模式一致： 是永不结束的常驻采集——需要继续收时再发一次 /goal 续开 新循环。 +不靠关键词的另一条路（LLM 自己判断）：会话里的 agent +也可以在回合输出里写 `goal-convert` 块把本会话转成 goal， +或写 `automation-create` 块新建自动化定义（首次运行时刻 +`first_run_at`、只跑一次 `once`、绑回本会话 `bind: +"heartbeat"`），worker 在轮次边界执行，格式与口径见 +`docs/marker-tools.md` 新增的两节。本批刻意不向 LLM 自动 +注入这两个工具的指南（避免 prompt 臃肿），需要时由用户在 +会话里手动告诉 agent；服务端不按关键词再猜意图。 + ## 任务间互发消息（peer messages，Muse 式通用互发） 任意两个同项目任务可以互发消息、对方做完可以回消息，不 diff --git a/cloud/automations.py b/cloud/automations.py index 31248df..3320da2 100644 --- a/cloud/automations.py +++ b/cloud/automations.py @@ -72,7 +72,8 @@ def row_get(row, name): AUTOMATION_COLUMNS = ( "id", "project_id", "name", "prompt", "repo_url", "interval_seconds", "status", "last_run_at", "last_claimed_at", - "created_by", "created_at", "cancelled_at") + "created_by", "created_at", "cancelled_at", + "first_run_at", "run_once") RUN_COLUMNS = ( "id", "automation_id", "project_id", "task_id", "status", "detail", "created_at") @@ -122,7 +123,9 @@ def ensure_automations_tables(conn): "last_claimed_at DOUBLE PRECISION, " "created_by TEXT, " "created_at DOUBLE PRECISION, " - "cancelled_at DOUBLE PRECISION)" + "cancelled_at DOUBLE PRECISION, " + "first_run_at DOUBLE PRECISION, " + "run_once INTEGER)" ) conn.execute( "CREATE TABLE IF NOT EXISTS automation_runs(" @@ -138,7 +141,9 @@ def ensure_automations_tables(conn): for col, ddl in ( ("repo_url", "TEXT"), ("last_claimed_at", "DOUBLE PRECISION"), - ("cancelled_at", "DOUBLE PRECISION")): + ("cancelled_at", "DOUBLE PRECISION"), + ("first_run_at", "DOUBLE PRECISION"), + ("run_once", "INTEGER")): if col not in existing: try: conn.execute( @@ -169,23 +174,40 @@ def ensure_automations_tables(conn): def next_run_at(auto, now=None): - """下次到期时刻：从未跑过时就是创建时间(建完即到期， - 执行器下一轮即可认领)；跑过时为上次执行 + 间隔。""" + """下次到期时刻： + - 只跑一次且已跑过：None（不再到期，不是“下次”）； + - 从未跑过且给了 first_run_at：就是该时刻（“明天 9 点” + 的首次运行就靠它，而不是建完即到期）； + - 从未跑过且没给 first_run_at：创建时间（建完即到期， + 执行器下一轮即可认领，沿用旧语义）； + - 跑过：上次执行 + 间隔（first_run_at 只管首次，之后 + 不再看它）。""" last = row_get(auto, "last_run_at") - interval = row_get(auto, "interval_seconds") or 0 + if row_get(auto, "run_once") and last is not None: + return None if last is None: + first = row_get(auto, "first_run_at") + if first is not None: + return first return row_get(auto, "created_at") + interval = row_get(auto, "interval_seconds") or 0 return last + interval def is_due(auto, now=None): """是否到期：仅 active 的定义可到期；now 缺省取当前时间。 到期边界用 >=，与 host run-due 的 now - last_run >= - interval 同款(含边界那一秒)。""" + interval 同款(含边界那一秒)。只跑一次且已跑过时永远 + 不到期（先于 next_run_at 判断）；next_run_at 为 None + 的其它情形（行缺创建时间等脏数据）仍按旧代码认到期， + 不静默改口径。""" if row_get(auto, "status") != ACTIVE: return False if now is None: now = time.time() + if row_get(auto, "run_once") and row_get(auto, "last_run_at") \ + is not None: + return False nxt = next_run_at(auto) if nxt is None: return True @@ -208,12 +230,15 @@ def automation_to_dict(row, now=None): "created_by": row["created_by"], "created_at": row["created_at"], "cancelled_at": row_get(row, "cancelled_at"), + "first_run_at": row_get(row, "first_run_at"), + "run_once": bool(row_get(row, "run_once")), "next_run_at": nxt, "due": is_due(row, now), } -def _validate_fields(name, prompt, interval_seconds, repo_url): +def _validate_fields(name, prompt, interval_seconds, repo_url, + first_run_at=None, run_once=False): if not name or not isinstance(name, str) or not name.strip(): raise ValueError("name required") name = name.strip() @@ -238,16 +263,31 @@ def _validate_fields(name, prompt, interval_seconds, repo_url): repo_url = repo_url.strip() or None if repo_url is not None and len(repo_url) > MAX_REPO_URL_LEN: raise ValueError("repo_url too long") - return name, prompt, interval_seconds, repo_url + if first_run_at is not None: + if isinstance(first_run_at, bool) or \ + not isinstance(first_run_at, (int, float)): + raise ValueError("first_run_at must be a timestamp") + first_run_at = float(first_run_at) + if first_run_at <= 0: + raise ValueError("first_run_at must be positive") + if not isinstance(run_once, bool): + raise ValueError("run_once must be a boolean") + return name, prompt, interval_seconds, repo_url, \ + first_run_at, run_once def create_automation(conn, project_id, name, prompt, created_by, - interval_seconds=86400, repo_url=None, now=None): + interval_seconds=86400, repo_url=None, now=None, + first_run_at=None, run_once=False): """创建自动化定义：同项目内 active 的名字唯一(取消过的 名字可重用，与环境名唯一规则同款语义)。成功时 commit - 并返回 dict；校验失败抛 ValueError(调用方转 400)。""" - name, prompt, interval_seconds, repo_url = _validate_fields( - name, prompt, interval_seconds, repo_url) + 并返回 dict；校验失败抛 ValueError(调用方转 400)。 + first_run_at 为首次运行时刻（epoch 秒，过去则建完即 + 到期）；run_once 为 True 时跑过一次后不再到期。""" + (name, prompt, interval_seconds, repo_url, + first_run_at, run_once) = _validate_fields( + name, prompt, interval_seconds, repo_url, + first_run_at, run_once) if now is None: now = time.time() ensure_automations_tables(conn) @@ -261,10 +301,12 @@ def create_automation(conn, project_id, name, prompt, created_by, conn.execute( "INSERT INTO automations(id, project_id, name, prompt," " repo_url, interval_seconds, status, last_run_at," - " last_claimed_at, created_by, created_at, cancelled_at)" - " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL)", + " last_claimed_at, created_by, created_at, cancelled_at," + " first_run_at, run_once)" + " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL, ?, ?)", (auto_id, project_id, name, prompt, repo_url, - interval_seconds, ACTIVE, created_by, now)) + interval_seconds, ACTIVE, created_by, now, + first_run_at, 1 if run_once else 0)) conn.commit() row = conn.execute( "SELECT * FROM automations WHERE id = ?", (auto_id,)).fetchone() @@ -327,11 +369,12 @@ def cancel_automation(conn, automation_id, now=None): def update_automation(conn, automation_id, fields, now=None): """更新自动化定义（automation_update）：fields 只认 - name / prompt / interval_seconds / repo_url 四个键， - 其它键一律忽略（不报错、不存入，避免调用方拼错字段 - 名后以为已改）；四个键全没给时抛 ValueError（空更新 - 不是成功）。校验与创建同款（_validate_fields，名字 - 仍按“同项目 active 名字唯一”查重，改名撞名时 400）。 + name / prompt / interval_seconds / repo_url / first_run_at / + run_once 六个键，其它键一律忽略（不报错、不存入，避免 + 调用方拼错字段名后以为已改）；六个键全没给时抛 + ValueError（空更新不是成功）。校验与创建同款 + （_validate_fields，名字仍按“同项目 active 名字唯一” + 查重，改名撞名时 400）。 返回 (ok, auto)：行不存在时 (False, None)；已取消时 (False, 旧 dict)——取消是终态，改已取消的定义必须 @@ -359,11 +402,16 @@ def update_automation(conn, automation_id, fields, now=None): interval = fields.get("interval_seconds", current["interval_seconds"]) repo_url = fields.get("repo_url", current["repo_url"]) + first_run_at = fields.get("first_run_at", + current["first_run_at"]) + run_once = fields.get("run_once", current["run_once"]) if not any(k in fields for k in ( - "name", "prompt", "interval_seconds", "repo_url")): + "name", "prompt", "interval_seconds", "repo_url", + "first_run_at", "run_once")): raise ValueError("no updatable field given") - name, prompt, interval, repo_url = _validate_fields( - name, prompt, interval, repo_url) + (name, prompt, interval, repo_url, + first_run_at, run_once) = _validate_fields( + name, prompt, interval, repo_url, first_run_at, run_once) if name != current["name"]: clash = conn.execute( "SELECT id FROM automations WHERE project_id = ?" @@ -375,8 +423,10 @@ def update_automation(conn, automation_id, fields, now=None): "automation name already exists in this project") conn.execute( "UPDATE automations SET name = ?, prompt = ?," - " interval_seconds = ?, repo_url = ? WHERE id = ?", - (name, prompt, interval, repo_url, automation_id)) + " interval_seconds = ?, repo_url = ?, first_run_at = ?," + " run_once = ? WHERE id = ?", + (name, prompt, interval, repo_url, first_run_at, + 1 if run_once else 0, automation_id)) conn.commit() row = conn.execute( "SELECT * FROM automations WHERE id = ?", (automation_id,) @@ -384,20 +434,29 @@ def update_automation(conn, automation_id, fields, now=None): return True, automation_to_dict(row, now) +# 到期 SQL（due 与 claim 共用，绝不能只改一处）： +# - 从未跑过：没给 first_run_at 时即到期；给了则只到该时刻才到期； +# - 跑过：上次 + 间隔 <= now 才到期； +# - 只跑一次且已跑过：永远不到期（run_once=1 且 last_run_at 非空）。 +_DUE_SQL = ( + " AND (run_once IS NULL OR run_once = 0 OR last_run_at IS NULL)" + " AND ((last_run_at IS NULL AND (first_run_at IS NULL" + " OR first_run_at <= ?))" + " OR (last_run_at IS NOT NULL AND" + " last_run_at + interval_seconds <= ?))") + + def due_automations(conn, project_id, now=None): """列出某项目当前到期的 active 定义(纯读，不抢单)。 - SQL 层算到期：从未跑过(last_run_at 为空)或上次执行距 - 今已满间隔；与 is_due 的 Python 算法同款。""" + SQL 层算到期：与 is_due 的 Python 算法同款，见 _DUE_SQL。""" if now is None: now = time.time() ensure_automations_tables(conn) rows = conn.execute( "SELECT * FROM automations WHERE project_id = ?" - " AND status = ?" - " AND (last_run_at IS NULL" - " OR last_run_at + interval_seconds <= ?)" + " AND status = ?" + _DUE_SQL + " ORDER BY created_at, id", - (project_id, ACTIVE, now)).fetchall() + (project_id, ACTIVE, now, now)).fetchall() return [automation_to_dict(r, now) for r in rows] @@ -420,10 +479,8 @@ def claim_due_automation(conn, automation_id, now=None): cur = conn.execute( "UPDATE automations SET last_run_at = ?," " last_claimed_at = ?" - " WHERE id = ? AND status = ?" - " AND (last_run_at IS NULL" - " OR last_run_at + interval_seconds <= ?)", - (now, now, automation_id, ACTIVE, now)) + " WHERE id = ? AND status = ?" + _DUE_SQL, + (now, now, automation_id, ACTIVE, now, now)) conn.commit() return cur.rowcount == 1 diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py index 5abe246..ecb87fd 100644 --- a/cloud/marker_tools.py +++ b/cloud/marker_tools.py @@ -1,6 +1,18 @@ """任务内标记工具（marker tools）：task-create / task-poll / task-read / wait-threads / task-cancel / task-retry / todo-list / todo-create / -todo-update / task-list / task-cost / quota-status / task-files。 +todo-update / task-list / task-cost / quota-status / task-files / +goal-convert / automation-create。 + +新增的两个工具是“LLM 自己判断意图、不用用户说关键词” +那条路：用户只说“我要一直收 xxx”“明天 9 点看 yyy”这种 +自然语言，agent 在回合里写 goal-convert / automation-create +块，worker 在轮次边界执行——不是服务端再用正则猜用户 +是不是在说 goal / 自动化。关键词解析（cloud/goal_command） +仍保留作显式命令兜底，两条路不互斥。 +注意：本批刻意不向 LLM 自动注入这两个工具的指南（避免 +prompt 臃肿），agent 知道它们的唯一途径是用户在会话里 +手动告诉它，或调用方在任务 prompt 里自己写明；格式见 +docs/marker-tools.md 新增的两节。 设计（与 peer_messages、task_spawn 同款原则，不另起一套）： @@ -68,6 +80,17 @@ todo-update / task-list / task-cost / quota-status / task-files。 二进制文件拒绝）；绝对路径、.. 逃逸、符号链接逃逸一律 拒绝；cloud 任务与 worktree 在别机的任务拒绝（文件不在 本机）；目标没有本地 worktree 时按 404 报告，不编内容。 +- goal-convert：把本会话在轮次边界转成新语义 goal 任务 + （goal_mode=1、goal 即块里给的目标、无限轮直到模型自决 + “目标已达成”，与 /goal 转换同款 CAS）。每轮最多 1 块； + 实际执行时各块照常执行、报告先排、/goal 消息后排，转成 + goal 后本任务不再解析标记块。 +- automation-create：新建一个自动化定义（见 + extract_automation_creates 的字段说明）。每轮最多 5 块； + 创建人跟父任务走，须有项目 create_task 权限，配额超限 + 时整轮跳过；name 缺失时 worker 自动生成，避免重名撞号。 + 创建只是登记定义，到期执行仍走 automations 的 claim 门 + 与外部触发器，见 automations 模块。 上限口径（本模块最终定的值，文档与代码一致）： @@ -111,8 +134,20 @@ import os import re import time import uuid +from datetime import datetime, timedelta, timezone + +try: + from zoneinfo import ZoneInfo +except Exception: # pragma: no cover - 极老 Python 无 zoneinfo + ZoneInfo = None MAX_CREATES_PER_TURN = 5 +MAX_GOAL_CONVERTS_PER_TURN = 1 +MAX_AUTOMATION_CREATES_PER_TURN = 5 +MIN_AUTOMATION_INTERVAL_SECONDS = 60 +MAX_AUTOMATION_INTERVAL_SECONDS = 30 * 86400 +MAX_AUTOMATION_REPO_URL_CHARS = 1000 +DEFAULT_AUTOMATION_INTERVAL_SECONDS = 86400 MAX_POLLS_PER_TURN = 5 MAX_WAITS_PER_TURN = 2 MAX_WAIT_TARGETS_PER_BLOCK = 8 @@ -176,6 +211,64 @@ _QUOTA_RE = re.compile( r"```(quota-status|quota_status)\s*\n(.*?)\n```", re.DOTALL) _TASK_FILES_RE = re.compile( r"```(task-files|task_files)\s*\n(.*?)\n```", re.DOTALL) +_GOAL_CONVERT_RE = re.compile( + r"```(goal-convert|goal_convert)\s*\n(.*?)\n```", re.DOTALL) +_AUTOMATION_CREATE_RE = re.compile( + r"```(automation-create|automation_create)\s*\n(.*?)\n```", + re.DOTALL) + + +def _shanghai_tz(): + """无时区时间默认按 Asia/Shanghai 理解（用户时区）。 + + 有 zoneinfo 时用它；没有时退回固定 UTC+8，避免把“明天 9 点” + 按 UTC 算成北京时间 17 点。""" + if ZoneInfo is not None: + try: + return ZoneInfo("Asia/Shanghai") + except Exception: + pass + return timezone(timedelta(hours=8)) + + +def parse_first_run_at(value): + """解析 first_run_at，返回 epoch 秒（float）或 None（无效）。 + + 接受：epoch 秒（int/float，毫秒级数字自动除 1000）、纯数字 + 字符串、ISO8601 字符串（带偏移量按偏移量算；无时区按 + Asia/Shanghai 算；“YYYY-MM-DD HH:MM”这种 fromisoformat + 能读的也算）。不判断“是否过去”——过去判断在 worker 层， + 那里才能把“已过去”写进报告让 agent 下一轮纠正；纯解析层 + 只负责“是不是一个合法时刻”，否则 LLM 的一次笔误会连 + 报告都拿不到。 + 本函数只依赖标准库，server 解析 API 的 first_run_at 时 + 也复用它，避免两端对同一字段的解析不一致。""" + if value is None or isinstance(value, bool): + return None + if isinstance(value, (int, float)): + epoch = float(value) + elif isinstance(value, str): + s = value.strip() + if not s: + return None + try: + epoch = float(s) + except ValueError: + try: + dt = datetime.fromisoformat( + s.replace("Z", "+00:00").replace("z", "+00:00")) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=_shanghai_tz()) + return dt.timestamp() + else: + return None + if epoch >= 1e11: # 毫秒级 epoch，秒级不可能这么大 + epoch = epoch / 1000.0 + if epoch <= 0 or epoch > 4102444800: # 2100-01-01 之后不认 + return None + return epoch def row_get(row, name): @@ -786,6 +879,137 @@ def extract_task_retries(text): return _extract_controls(text, _RETRY_RE) +def extract_goal_converts(text): + """提取 goal-convert / goal_convert 块：把本会话转成 goal。 + + 块格式:: + + ```goal-convert + {"goal": "收 xxx 到 yyy 为止"} + ``` + + goal 必填、1-20000 字；每轮最多 1 块（转一次就够了， + 多余的忽略）。不接受 verify_command / goal_max_rounds： + LLM 定的 goal 一律是新语义——无限轮直到模型自己输出 + “目标已达成”，与服务端 /goal 转换口径一致，绝不让 LLM + 顺手把旧验收语义混进来。""" + if not text: + return [] + out = [] + for match in _GOAL_CONVERT_RE.finditer(str(text)): + if len(out) >= MAX_GOAL_CONVERTS_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + goal = payload.get("goal") or payload.get("target") + if not isinstance(goal, str) or not goal.strip(): + continue + goal = goal.strip() + if len(goal) > MAX_PROMPT_CHARS: + continue + out.append({"goal": goal}) + return out + + +def extract_automation_creates(text): + """提取 automation-create / automation_create 块。 + + 块格式:: + + ```automation-create + {"name": "看 xxx 每日", "prompt": "查看 xxx 最新并汇总", + "interval_seconds": 86400, + "first_run_at": "2026-10-30T09:00:00+08:00", + "once": false, "bind": null, "repo_url": null} + ``` + + - prompt 必填、1-20000 字；name 可选、1-100 字，缺失时 + worker 自动生成（LLM 不必为起名费劲，也不会因重名 + 撞上同项目已有的 active 自动化）。 + - interval_seconds 可选、缺省 86400，必须是整数且在 + 60-30*86400 之间；非法时整块跳过，不静默钳制——钳制 + 会把“每小时”悄悄改成“每分钟下限”或把“每月”改 + 短，调度错误比拒绝更贵。 + - first_run_at 可选，见 parse_first_run_at；不给表示 + 建完即到期（沿用自动化原有语义），给了表示首次在该 + 时刻跑——“每天 9 点”要写成 first_run_at=下一个 9 点 + + interval_seconds=86400，而不是只写间隔（只写间隔 + 会从创建时间起算，每天跑在创建时间那个钟点）。 + - once 可选、缺省 false；true 表示只跑一次（跑过后 + is_due / due / claim 三处都不再到期，见 automations）。 + - bind 可选、缺省 null（到期独立跑新任务）；"heartbeat" + 表示绑到本会话，到期唤醒本线程；"cron" 与 null 同款， + 只是显式写明独立跑。其它值整块跳过。 + - repo_url 可选、最多 1000 字。 + - 每轮最多 5 块，多余的忽略；非法 JSON、prompt 缺失/ + 空串/超长时整块跳过。""" + if not text: + return [] + out = [] + for match in _AUTOMATION_CREATE_RE.finditer(str(text)): + if len(out) >= MAX_AUTOMATION_CREATES_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + prompt = payload.get("prompt") + if not isinstance(prompt, str) or not prompt.strip(): + continue + prompt = prompt.strip() + if len(prompt) > MAX_PROMPT_CHARS: + continue + name = payload.get("name") + if name is not None: + if not isinstance(name, str) or not name.strip(): + continue + name = name.strip() + if len(name) > MAX_NAME_CHARS: + continue + interval = payload.get("interval_seconds") + if interval is None: + interval = DEFAULT_AUTOMATION_INTERVAL_SECONDS + elif isinstance(interval, bool): + continue + elif isinstance(interval, float) and interval.is_integer(): + interval = int(interval) + elif not isinstance(interval, int): + continue + if interval < MIN_AUTOMATION_INTERVAL_SECONDS or \ + interval > MAX_AUTOMATION_INTERVAL_SECONDS: + continue + first_run_at = None + if payload.get("first_run_at") is not None: + first_run_at = parse_first_run_at(payload.get("first_run_at")) + if first_run_at is None: + continue + once = payload.get("once", False) + if not isinstance(once, bool): + continue + bind = payload.get("bind", payload.get("bind_mode")) + if bind is not None and bind not in ("heartbeat", "cron"): + continue + repo_url = payload.get("repo_url") + if repo_url is not None: + if not isinstance(repo_url, str): + continue + repo_url = repo_url.strip() or None + if repo_url is not None and \ + len(repo_url) > MAX_AUTOMATION_REPO_URL_CHARS: + continue + out.append({"name": name, "prompt": prompt, + "interval_seconds": interval, + "first_run_at": first_run_at, "once": once, + "bind": bind, "repo_url": repo_url}) + return out + + def parent_spawn_generation(conn, parent_id): """父任务自身的派生代数：普通任务 0，第一代派生任务 1，依此类推。 diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 3c96dc8..5d26e94 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -23,6 +23,10 @@ import sys import time from cloud.marker_tools import extract_quota_status as _extract_quota_status +from cloud.marker_tools import extract_automation_creates as \ + _extract_automation_creates +from cloud.marker_tools import extract_goal_converts as \ + _extract_goal_converts from cloud.marker_tools import extract_task_cancels as _extract_task_cancels from cloud.marker_tools import extract_task_costs as _extract_task_costs from cloud.marker_tools import extract_task_creates as _extract_task_creates @@ -418,6 +422,8 @@ class LocalExecutor(Executor): todo_lists = [] todo_creates = [] todo_updates = [] + goal_converts = [] + automation_creates = [] turn_id = None root_turn_id = None parsed = None @@ -465,6 +471,9 @@ class LocalExecutor(Executor): todo_lists = _extract_todo_lists(parsed.get("text")) todo_creates = _extract_todo_creates(parsed.get("text")) todo_updates = _extract_todo_updates(parsed.get("text")) + goal_converts = _extract_goal_converts(parsed.get("text")) + automation_creates = _extract_automation_creates( + parsed.get("text")) else: host_status = None else: @@ -496,6 +505,8 @@ class LocalExecutor(Executor): "todo_lists": todo_lists, "todo_creates": todo_creates, "todo_updates": todo_updates, + "goal_converts": goal_converts, + "automation_creates": automation_creates, "turn_id": turn_id, "root_turn_id": root_turn_id, } @@ -1944,6 +1955,8 @@ def _parse_host_result(task_id, fake): todo_lists = [] todo_creates = [] todo_updates = [] + goal_converts = [] + automation_creates = [] turn_id = None root_turn_id = None parsed = None @@ -1978,6 +1991,9 @@ def _parse_host_result(task_id, fake): todo_lists = _extract_todo_lists(parsed.get("text")) todo_creates = _extract_todo_creates(parsed.get("text")) todo_updates = _extract_todo_updates(parsed.get("text")) + goal_converts = _extract_goal_converts(parsed.get("text")) + automation_creates = _extract_automation_creates( + parsed.get("text")) # 任务列表显示 turn 用：与 Local 版同款语义，取不到 # 时保持 None，worker 写回时不覆盖旧值。 turn_id = parsed.get("turn_id") @@ -2006,6 +2022,8 @@ def _parse_host_result(task_id, fake): "todo_lists": todo_lists, "todo_creates": todo_creates, "todo_updates": todo_updates, + "goal_converts": goal_converts, + "automation_creates": automation_creates, "turn_id": turn_id, "root_turn_id": root_turn_id, } diff --git a/cloud/server.py b/cloud/server.py index 15f6986..6ddf2eb 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -68,6 +68,7 @@ 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.marker_tools import parse_first_run_at as _parse_first_run_at 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 @@ -9472,7 +9473,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): """创建自动化定义：只登记定义与到期状态，不跑模型、 不另起调度器(执行器后续走 claim 门，见 automations 模块文档)。body：name、prompt 必填；interval_seconds - 可选(默认 86400，最小 60)；repo_url 可选。 + 可选(默认 86400，最小 60)；repo_url 可选； + first_run_at 可选（epoch 秒或 ISO8601 字符串，与 + marker 的 parse_first_run_at 同款解析，过去则建完 + 即到期）；run_once 可选（true 时跑过一次后不再到期）。 P1 配额门：创建时先过与建任务同一道配额门（美元配额 与无价 token cap，超限 402），避免“定义免费建、到期 @@ -9490,13 +9494,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): interval = body.get("interval_seconds", 86400) if interval is None: interval = 86400 + first_run_at = body.get("first_run_at") + if first_run_at is not None: + first_run_at = _parse_first_run_at(first_run_at) + if first_run_at is None: + self._send_json(400, {"error": "invalid first_run_at"}) + return + run_once = body.get("run_once", False) + if not isinstance(run_once, bool): + self._send_json(400, {"error": "run_once must be boolean"}) + return conn = self._db() try: try: auto = _auto_create( conn, project_id, body.get("name"), body.get("prompt"), user_id, interval, - body.get("repo_url")) + body.get("repo_url"), + first_run_at=first_run_at, run_once=run_once) except ValueError as exc: self._send_json(400, {"error": str(exc)}) return @@ -9845,6 +9860,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if jerr: self._send_json(400, {"error": jerr}) return + if body.get("first_run_at") is not None and \ + isinstance(body.get("first_run_at"), str): + parsed = _parse_first_run_at(body["first_run_at"]) + if parsed is None: + self._send_json(400, {"error": "invalid first_run_at"}) + return + body = dict(body, first_run_at=parsed) conn = self._db() try: try: diff --git a/cloud/tests/test_session_goal_automation.py b/cloud/tests/test_session_goal_automation.py new file mode 100644 index 0000000..0c1d1ef --- /dev/null +++ b/cloud/tests/test_session_goal_automation.py @@ -0,0 +1,334 @@ +"""会话内 goal / 自动化标记工具单测：LLM 自己判断意图那条路。 + +不跑真实模型、不联网： +- 解析层是纯函数（goal-convert、automation-create、 + first_run_at 的 ISO / 无时区 / 毫秒解析与非法跳过）； +- automations 层用内存 SQLite，测 first_run_at 到期与 + run_once 跑过后不再到期（Python 口径与 SQL due/claim + 口径一致）； +- worker 投递用 init_db 建的临时库 + 最小任务行，直接调 + _deliver_marker_tools，不走 run_one 全量流程。 + +注意：本文件不测“LLM 真的会写块”——那是模型行为， +且本批刻意不向 LLM 自动注入工具指南（避免 prompt +臃肿）；这里测的是块写出来后系统一定按约定执行，且 +非法块一定不静默变调度。工具格式见 docs/marker-tools.md， +需要时由用户在会话里手动告诉 agent。 +""" + +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path + +try: + from cloud.automations import ( + claim_due_automation, + create_automation, + due_automations, + get_automation, + update_automation, + ) + from cloud.marker_tools import ( + extract_automation_creates, + extract_goal_converts, + parse_first_run_at, + ) + from cloud.server import 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.automations import ( + claim_due_automation, + create_automation, + due_automations, + get_automation, + update_automation, + ) + from cloud.marker_tools import ( + extract_automation_creates, + extract_goal_converts, + parse_first_run_at, + ) + from cloud.server import init_db + from cloud.worker import Worker + +FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} + + +class ParseTestCase(unittest.TestCase): + def test_goal_convert_basic_and_alias(self): + text = ( + "```goal-convert\n{\"goal\": \"收 xxx\"}\n```\n" + "```goal_convert\n{\"goal\": \"收 yyy\"}\n```\n") + converts = extract_goal_converts(text) + # 每轮最多 1 块，第二个被忽略 + self.assertEqual([c["goal"] for c in converts], ["收 xxx"]) + + def test_goal_convert_invalid_skipped(self): + text = ( + "```goal-convert\nnot json\n```\n" + "```goal-convert\n{\"goal\": \" \"}\n```\n" + "```goal-convert\n{\"goal\": \"ok\"}\n```\n") + self.assertEqual( + [c["goal"] for c in extract_goal_converts(text)], ["ok"]) + + def test_first_run_at_formats(self): + iso = parse_first_run_at("2026-10-30T09:00:00+08:00") + naive = parse_first_run_at("2026-10-30 09:00") + self.assertEqual(iso, naive) + self.assertEqual(parse_first_run_at(iso), iso) + self.assertEqual(parse_first_run_at(int(iso * 1000)), iso) + self.assertIsNone(parse_first_run_at("not a time")) + self.assertIsNone(parse_first_run_at("2026-13-40 09:00")) + self.assertIsNone(parse_first_run_at(True)) + self.assertIsNone(parse_first_run_at(-1)) + + def test_automation_create_basic(self): + text = ( + "```automation-create\n" + "{\"prompt\": \"看 yyy\", \"interval_seconds\": 86400," + " \"first_run_at\": \"2026-10-30T09:00:00+08:00\"," + " \"once\": true, \"bind\": \"heartbeat\"}\n```\n") + blocks = extract_automation_creates(text) + self.assertEqual(len(blocks), 1) + block = blocks[0] + self.assertEqual(block["prompt"], "看 yyy") + self.assertEqual(block["interval_seconds"], 86400) + self.assertEqual( + block["first_run_at"], + parse_first_run_at("2026-10-30T09:00:00+08:00")) + self.assertTrue(block["once"]) + self.assertEqual(block["bind"], "heartbeat") + self.assertIsNone(block["name"]) + + def test_automation_create_invalid_interval_and_bind_skipped(self): + text = ( + "```automation-create\n{\"prompt\": \"x\"," + " \"interval_seconds\": 30}\n```\n" + "```automation-create\n{\"prompt\": \"y\"," + " \"bind\": \"weird\"}\n```\n" + "```automation-create\n{\"prompt\": \"good\"}\n```\n") + blocks = extract_automation_creates(text) + self.assertEqual([b["prompt"] for b in blocks], ["good"]) + self.assertEqual(blocks[0]["interval_seconds"], 86400) + + +class AutomationsTestCase(unittest.TestCase): + def setUp(self): + self.conn = sqlite3.connect(":memory:") + self.conn.row_factory = sqlite3.Row + + def tearDown(self): + self.conn.close() + + def test_first_run_at_not_due_early_then_due(self): + now = 1700000000.0 + auto = create_automation( + self.conn, "p", "future", "看 x", "u", now=now, + first_run_at=now + 3600) + self.assertFalse(auto["due"]) + self.assertEqual(auto["next_run_at"], now + 3600) + self.assertEqual(due_automations(self.conn, "p", now=now), []) + self.assertFalse( + claim_due_automation(self.conn, auto["id"], now=now)) + self.assertEqual( + [a["name"] for a in due_automations( + self.conn, "p", now=now + 3600)], + ["future"]) + self.assertTrue( + claim_due_automation(self.conn, auto["id"], now=now + 3600)) + + def test_run_once_never_due_after_first_run(self): + now = 1700000000.0 + auto = create_automation( + self.conn, "p", "once", "看 y", "u", now=now, + first_run_at=now + 10, run_once=True) + self.assertTrue( + claim_due_automation(self.conn, auto["id"], now=now + 20)) + after = get_automation(self.conn, auto["id"]) + self.assertFalse(after["due"]) + self.assertIsNone(after["next_run_at"]) + self.assertEqual( + due_automations(self.conn, "p", now=now + 90000), []) + + def test_update_first_run_at_and_run_once(self): + now = 1700000000.0 + auto = create_automation( + self.conn, "p", "upd", "看 z", "u", now=now) + ok, updated = update_automation( + self.conn, auto["id"], + {"first_run_at": now + 7200, "run_once": True}, now=now) + self.assertTrue(ok) + self.assertEqual(updated["first_run_at"], now + 7200) + self.assertTrue(updated["run_once"]) + self.assertFalse(updated["due"]) + + +class WorkerDeliverTestCase(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._insert("PM", status="running", created_by="tester") + self._member("tester", "maintainer") + + def tearDown(self): + self.tmp.cleanup() + + def _insert(self, task_id, status="queued", created_by="tester", + project_id="p1"): + now = time.time() + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat)" + " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)", + (task_id, project_id, "do work", status, created_by, + now, now)) + conn.commit() + finally: + conn.close() + + def _member(self, user_id, role): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY," + " username TEXT UNIQUE, api_key_hash TEXT," + " created_at REAL)") + conn.execute( + "CREATE TABLE IF NOT EXISTS project_members(" + "project_id TEXT, user_id TEXT, role TEXT," + " PRIMARY KEY(project_id, user_id))") + conn.execute( + "INSERT OR REPLACE INTO project_members" + "(project_id, user_id, role) VALUES('p1', ?, ?)", + (user_id, role)) + conn.execute( + "INSERT OR IGNORE INTO users(id, username, api_key_hash," + " created_at) VALUES(?, ?, '', 0)", (user_id, user_id)) + conn.commit() + finally: + conn.close() + + def _worker(self): + return Worker(self.db_path, "/fake/host.py", env=FAKE_ENV) + + def _followup_bodies(self): + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute( + "SELECT body FROM task_followups" + " WHERE task_id = 'PM' ORDER BY created_at").fetchall() + return [row[0] for row in rows] + finally: + conn.close() + + def _automations(self): + conn = sqlite3.connect(self.db_path) + try: + try: + rows = conn.execute( + "SELECT name, prompt, interval_seconds, run_once" + " FROM automations").fetchall() + except sqlite3.OperationalError: + # 一个自动化都没建过时表还不存在，不是失败 + return [] + return [tuple(row) for row in rows] + finally: + conn.close() + + def test_deliver_automation_create_and_bind(self): + worker = self._worker() + sender = worker.get_task("PM") + future = time.time() + 3600 + worker._deliver_marker_tools(sender, { + "status": "completed", + "task_creates": [], "task_polls": [], "wait_threads": [], + "task_cancels": [], "task_retries": [], + "goal_converts": [], + "automation_creates": [{ + "name": None, "prompt": "看 yyy", + "interval_seconds": 86400, "first_run_at": future, + "once": False, "bind": "heartbeat", + "repo_url": None}], + }) + autos = self._automations() + self.assertEqual(len(autos), 1) + self.assertEqual(autos[0][1], "看 yyy") + conn = sqlite3.connect(self.db_path) + try: + binding = conn.execute( + "SELECT task_id, mode FROM automation_bindings" + ).fetchone() + bodies = self._followup_bodies() + finally: + conn.close() + self.assertEqual(tuple(binding), ("PM", "heartbeat")) + self.assertTrue(any("已创建自动化" in b for b in bodies)) + + def test_deliver_automation_past_first_run_rejected(self): + worker = self._worker() + sender = worker.get_task("PM") + worker._deliver_marker_tools(sender, { + "status": "completed", + "task_creates": [], "task_polls": [], "wait_threads": [], + "task_cancels": [], "task_retries": [], + "goal_converts": [], + "automation_creates": [{ + "name": "old", "prompt": "看旧", + "interval_seconds": 86400, + "first_run_at": time.time() - 3600, + "once": False, "bind": None, "repo_url": None}], + }) + self.assertEqual(self._automations(), []) + bodies = self._followup_bodies() + self.assertTrue(any("已过去" in b for b in bodies)) + + def test_deliver_goal_convert_queues_after_report(self): + worker = self._worker() + sender = worker.get_task("PM") + worker._deliver_marker_tools(sender, { + "status": "completed", + "task_creates": [], "task_polls": [], "wait_threads": [], + "task_cancels": [], "task_retries": [], + "goal_converts": [{"goal": "收 xxx"}], + "automation_creates": [{ + "name": "daily", "prompt": "看 yyy", + "interval_seconds": 86400, "first_run_at": None, + "once": False, "bind": None, "repo_url": None}], + }) + bodies = self._followup_bodies() + # 报告先、/goal 后：排空先跑报告轮、再转 goal + self.assertTrue(any("已创建自动化" in b for b in bodies)) + self.assertEqual(bodies[-1], "/goal 收 xxx") + self.assertEqual(len(self._automations()), 1) + + def test_deliver_goal_task_ignored(self): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET goal_mode = 1 WHERE id = 'PM'") + conn.commit() + finally: + conn.close() + worker = self._worker() + sender = worker.get_task("PM") + worker._deliver_marker_tools(sender, { + "status": "completed", + "task_creates": [], "task_polls": [], "wait_threads": [], + "task_cancels": [], "task_retries": [], + "goal_converts": [{"goal": "收 xxx"}], + "automation_creates": [], + }) + self.assertEqual(self._followup_bodies(), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 4086b14..9d4978b 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -150,6 +150,8 @@ from cloud.models import ModelRouter, is_fallbackable, record_fallback from cloud.auto_memory import AutoMemoryStore from cloud.auto_memory import build_injection_prefix as _auto_prefix from cloud.auto_memory import process_due_tasks as _auto_process_due +from cloud.automations import create_automation as _auto_create +from cloud.lifecycle import bind_automation as _life_bind from cloud.marker_tools import ancestry_relation as _marker_ancestry from cloud.marker_tools import check_control_gate as _marker_gate from cloud.marker_tools import count_direct_spawn_children as _marker_count_direct @@ -3492,14 +3494,21 @@ class Worker: 只执行 completed 轮次的块；goal 任务与协调员任务不解析 （只收不发，与 peer 同款）；父任务被取消时不执行。 - 执行顺序：创建 → 取消 → 重试 → 待办创建 → 待办更新 → - 待办列表 → 任务列表 → 任务成本 → 配额 → 任务文件 → - 轮询 → 等待——创建在前让同一轮刚创建的任务立刻能被 - 轮询，控制操作在快照之前让轮询/等待读到取消或重试后 - 的新状态；待办列表放在待办写操作之后，让同一轮刚建/ - 刚改的待办立刻能在列表里看到。全部报告拼成一条系统 - 后续消息排给父任务自己，由排空循环在轮次边界跑下 - 一轮；拒绝只记报告与审计，不炸 run_one。""" + 执行顺序：创建 → 自动化创建 → 取消 → 重试 → 待办创建 → + 待办更新 → 待办列表 → 任务列表 → 任务成本 → 配额 → + 任务文件 → 轮询 → 等待——创建在前让同一轮刚创建的 + 任务立刻能被轮询，控制操作在快照之前让轮询/等待读到 + 取消或重试后的新状态；待办列表放在待办写操作之后， + 让同一轮刚建/刚改的待办立刻能在列表里看到。全部报告 + 拼成一条系统后续消息排给父任务自己，由排空循环在轮次 + 边界跑下一轮；拒绝只记报告与审计，不炸 run_one。 + goal-convert 不在这里直接转：它排成一条“/goal <目标>” + 的自后续消息，放在报告消息之后——排空先跑报告那轮 + （agent 看到自动化等工具的确认），再跑 /goal 那条， + 走已有的 followup /goal 转换路径（plan 冲突、CAS、 + 终局 queued 审计都在那条路里，不在这里另起一套）。 + 这样转 goal 后报告不会丢在 goal 循环外，也不会出现 + 两套转换逻辑打架。""" if not result or result.get("status") != "completed": return if self._was_cancelled(task["id"]): @@ -3520,14 +3529,17 @@ class Worker: task_costs = result.get("task_costs") or [] quota_status = result.get("quota_status") or [] task_files = result.get("task_files") or [] + goal_converts = result.get("goal_converts") or [] + automation_creates = result.get("automation_creates") or [] if not (creates or polls or waits or cancels or retries or todo_lists or todo_creates or todo_updates or task_lists or task_costs or quota_status - or task_files): + or task_files or goal_converts or automation_creates): return reports = [] created = 0 rejected = 0 + auto_created = 0 if creates: quota_error = None try: @@ -3616,6 +3628,93 @@ class Worker: except Exception as exc: reports.append("创建失败：%s" % exc) rejected += len(creates) - created + if automation_creates: + # 自动化创建：权限与配额是整轮门（与 task-create 同款， + # 配额超限时整轮跳过，不逐块半建）；单块的首次时刻 + # 已过去、重名等只拒该块。创建人跟父任务走，绝不 + # 用 worker 自己的身份建——否则自动化到期产出的任务 + # 会算到错误的人头上，配额门也绕过去了。 + auth_ok, _auth_reason = self._marker_auth( + task, "create_task") + quota_error = None + try: + quota_error = self._quota_exceeded(task, None) + except Exception as exc: + quota_error = "quota_check_failed: %s" % exc + if auth_ok is not True: + reports.append( + "自动化创建跳过：发送方创建人无 create_task " + "权限或权限查不到，本轮 %d 个自动化块全部跳过。" + % len(automation_creates)) + rejected += len(automation_creates) + elif quota_error is not None: + reports.append( + "自动化创建跳过：父任务配额检查未通过（%s），" + "本轮 %d 个自动化块全部跳过。" + % (quota_error, len(automation_creates))) + rejected += len(automation_creates) + else: + now = time.time() + for idx, block in enumerate(automation_creates): + first = block.get("first_run_at") + if first is not None and first < now - 60: + reports.append( + "自动化创建跳过：首次运行时刻已过去" + "（%s），请给一个未来时刻再写块。" + % block.get("prompt", "")[:40]) + rejected += 1 + continue + name = block.get("name") or ( + "auto-%s-%d-%d" + % (task["id"][:8], int(now), idx)) + try: + conn = _db_connect(self.db_path) + try: + auto = _auto_create( + conn, task["project_id"], name, + block["prompt"], task.get("created_by"), + block["interval_seconds"], + block.get("repo_url"), now=now, + first_run_at=first, + run_once=block.get("once", False)) + bind = block.get("bind") + if bind: + try: + _life_bind( + conn, auto["id"], + task["project_id"], task["id"], + bind, now=now) + except Exception as exc: + reports.append( + "已创建自动化 %s（%s），但绑定" + "到本会话失败（%s），定义本身" + "仍有效，到期独立跑。" + % (auto["id"], name, exc)) + auto_created += 1 + continue + finally: + conn.close() + except Exception as exc: + reports.append( + "自动化创建失败（%s）：%s。" + % (block.get("prompt", "")[:40], exc)) + rejected += 1 + continue + auto_created += 1 + nxt = auto.get("next_run_at") + nxt_text = ( + time.strftime( + "%Y-%m-%d %H:%M", time.localtime(nxt)) + if nxt else "不再到期") + reports.append( + "已创建自动化 %s（%s），间隔 %d 秒，" + "下次 %s%s，到期后由执行器经 claim 抢单" + "建任务%s。" + % (auto["id"], name, + block["interval_seconds"], nxt_text, + "（只跑一次）" if block.get("once") else "", + "，已绑到本会话唤醒" + if block.get("bind") == "heartbeat" else "")) for kind, blocks, action in (("取消", cancels, "create_task"), ("重试", retries, "approve_merge")): for block in blocks: @@ -4071,16 +4170,26 @@ class Worker: if reports: body = _marker_header() + "\n" + "\n".join(reports) self._queue_self_followup(task, body) + if goal_converts: + # 顺序关键：报告消息先排、/goal 消息后排，排空先跑 + # 报告那轮、再转 goal；反过来排会先转 goal，报告 + # 就送不到 goal 循环里了。plan 冲突等拒绝由排空的 + # /goal 路径处理（标 failed 停排空），这里不预判。 + goal_target = goal_converts[0]["goal"] + self._queue_self_followup( + task, "/goal %s" % goal_target) detail = ("%s: spawned=%d rejected=%d polls=%d waits=%d " "cancels=%d retries=%d todo_lists=%d " "todo_creates=%d todo_updates=%d task_lists=%d " - "task_costs=%d quota=%d task_files=%d" + "task_costs=%d quota=%d task_files=%d " + "automations=%d goal_converts=%d" % (task["id"], created, rejected, len(polls), len(waits), len(cancels), len(retries), len(todo_lists), len(todo_creates), len(todo_updates), len(task_lists), len(task_costs), len(quota_status), - len(task_files))) + len(task_files), auto_created, + len(goal_converts))) self._audit(task["project_id"], "worker_marker_tools", detail) def _deliver_peer_messages(self, task, result): @@ -5769,6 +5878,9 @@ class Worker: if fu.get("sender_task_id"): fu_prompt = _peer_prefix(fu["sender_task_id"]) + fu_prompt fu_task["prompt"] = fu_prompt + # 后续消息轮不重复加指南：首轮指南已在同一 + # thread 历史里，后续轮靠历史想起工具；每轮 + # 都加会让指南在历史里越积越多。 fu_result, fu_error = self._run_attempt( fu_task, quota_check=_quota_hit) usage_total = sum_usage( diff --git a/docs/marker-tools.md b/docs/marker-tools.md index 52f2ae7..55f5f56 100644 --- a/docs/marker-tools.md +++ b/docs/marker-tools.md @@ -3,9 +3,19 @@ 适用基线：main d129f7b 上的 feat-marker-tools 分支（本地提交，未推、 未建 PR、未部署）。本批只做四类标记工具，不要把本文档读成 “所有 Codex app 线程工具都已实现”：handoff、worktree 创建状态、 -PR 检查诊断、automation、clock-sleep 都不在本批；MCP 真工具路 +PR 检查诊断、clock-sleep 都不在本批；MCP 真工具路 （同轮函数返回）也不在本批。 +新增（feat-mainline-goal 分支，本地提交，未推、未部署）： +goal-convert 与 automation-create 两个标记工具已实现，见下面 +新增的两节。注意：本批刻意**不**向 LLM 自动注入工具指南—— +不靠服务端再猜关键词、也不让每个任务的 prompt 都变长； +agent 知道这两个工具的唯一途径，是用户在会话里手动告诉它 +（例如把下面两节的格式贴给它），或由调用方在任务 prompt 里 +自己写明。LLM 是否用它们、判成 goal 还是自动化，全由 LLM +按用户话里的意图决定，服务端不按“收 / 每天 / 明天”这类词 +再做一次关键词判断。 + ## 一句话原理 agent 不用联网、不用凭证。agent 在回合输出里写标记块，本地与 @@ -32,6 +42,61 @@ Docker 执行器都经 sandbox 的结果解析提取标记块（与 task-message - 容易混淆的一点：协调员的子任务本身是普通任务（不是协调员 任务），它写标记块会被解析；它的派生代数从 0 计（见“派生护栏”）。 +## 新增的两个工具：goal-convert 与 automation-create + +### goal-convert：把本会话转成 goal + +```text +```goal-convert +{"goal": "收 xxx 到 yyy 为止"} +``` +``` + +- goal 必填、1-20000 字；每轮最多 1 块。转成的是新语义 goal + （goal_mode=1、无限轮直到模型自己输出“目标已达成”）， + 与服务端 /goal 转换同款，不接受 verify_command / + goal_max_rounds。 +- 执行方式是排队，不是立即转：worker 把它排成一条 + “/goal <目标>”的自后续消息，放在本轮其它工具的报告消息 + 之后——排空先跑报告那轮，再跑 /goal 那条走已有的 followup + /goal 转换路径（plan 冲突、CAS、终局 queued 都在那条路里）。 + 所以同轮其它标记块的报告不会丢在 goal 循环外。 +- 只在普通任务里有效：goal 任务与协调员任务不解析标记块。 + +### automation-create：新建自动化定义 + +```text +```automation-create +{"name": "看 xxx 每日", "prompt": "查看 xxx 最新并汇总", + "interval_seconds": 86400, + "first_run_at": "2026-10-30T09:00:00+08:00", + "once": false, "bind": null, "repo_url": null} +``` +``` + +- prompt 必填、1-20000 字；name 可选、1-100 字，缺失时 + worker 自动生成（`auto-<任务短 id>-<时间>-<序号>`），不会因 + 重名撞上同项目已有的 active 自动化。 +- interval_seconds 可选、缺省 86400，必须是整数且在 60 到 + 30*86400 之间；非法时整块跳过，不静默钳制。 +- first_run_at 可选：epoch 秒（毫秒级自动除 1000）、ISO8601 + 字符串（带偏移量按偏移量算，无时区按 Asia/Shanghai 算）。 + 不给表示建完即到期（沿用自动化原有语义）；给了表示首次 + 在该时刻跑——“每天 9 点”要写成 first_run_at=下一个 9 点 + + interval_seconds=86400，而不是只写间隔（只写间隔会从创建 + 时间起算，每天跑在创建时间那个钟点）。首次时刻已过去的块 + 在 worker 层拒绝并写进报告，agent 下一轮可纠正。 +- once 可选、缺省 false；true 表示只跑一次，跑过后 is_due / + due / claim 三处都不再到期（automations 模块的 + first_run_at / run_once 字段与 SQL 到期条件已同步改过， + API 的创建与更新也接受这两个字段）。 +- bind 可选、缺省 null（到期独立跑新任务）；"heartbeat" + 表示绑到本会话，到期唤醒本线程；"cron" 与 null 同款。 +- 每轮最多 5 块；创建人跟父任务走，须有项目 create_task + 权限，配额超限时整轮跳过。创建只是登记定义，到期执行 + 仍走 automations 的 claim 门与外部触发器，不是建完就跑。 +- 只在普通任务里有效：goal 任务与协调员任务不解析标记块。 + ## 四个工具的标记格式 ### task-create：新建平级主线任务