feat: 主线 /goal 命令自动建 goal

发给主线的一句话（含字面 /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

@@ -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 式通用互发）
 
 任意两个同项目任务可以互发消息、对方做完可以回消息，不

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)

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)

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()

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:
             # 本轮已跑完（无论成败）：续跑消息已消费，清掉；进程
             # 中途被杀时清不到，下次认领会重跑同一条消息——可接受