From f4443dca4f7a20037aa0f9dbb0556c4efc04af51 Mon Sep 17 00:00:00 2001
From: openqj <openqj@gmail.com>
Date: Tue, 27 Oct 2026 03:17:51 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AE=A1=E5=88=92=E6=A8=A1=E5=BC=8F=20?=
 =?UTF-8?q?+=20=E5=90=8E=E7=BB=AD=E6=B6=88=E6=81=AF=20+=20=E4=BC=9A?=
 =?UTF-8?q?=E8=AF=9D=E5=88=86=E6=94=AF=E6=8E=A5=E4=BA=91=E7=AB=AF=20(#65)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 计划模式：tasks 加 plan 列；plan 与 approval=full / goal 冲突时
  400；执行器在 approval 参数后追加 --sandbox read-only（本地
  dispatch/run/goal/popen 与 Docker 内层命令全覆盖），host 任务
  存下 sandbox 后续轮次自动继承只读。
- 后续消息：新表 task_followups；POST/GET /tasks/{id}/followups
  （仅 queued/running 的非 goal 任务；direct 立即往链路追加
  phase=followup-direct 条目，本轮跑完后优先于排队执行；排队
  按 direct 优先、队内先来先服务逐条再跑一轮，同一 thread；
  一条失败即停、剩余 pending 留给重试入口；取消时回 pending）。
- 会话分支：POST /tasks/{id}/fork（仅终态任务；继承 executor/
  模型/强度/权限/计划，goal 字段置空）；worker 先 fork 再跑首轮，
  首轮失败不自动重试、不重复分叉，失败后走重试入口重跑首轮。
- 修任务级模型被覆盖：worker 此前无条件写成主模型，控制台
  选的模型实际不生效；现为任务级 > worker 主模型 > 配置。
- Docker 租户准备抽成 _docker_task_context，供普通尝试与 fork 共用。
- console：新建任务计划勾选（含冲突拦截）、详情计划徽标与
  分支来源、分支按钮与表单、后续消息区、列表 [计划]/[分支] 标记。
- 测试：cloud/tests/test_task_links.py 17 个（server 校验与继承、
  worker 模型优先、只读命令、排空顺序与停排、goal 不排空、
  fork 单次性）；全量 cloud 542、host 49、deploy 14。

Co-authored-by: host <host@test.local>
---
 README.md                      |   9 +
 cloud/console.html             |  72 ++++-
 cloud/db.py                    |  14 +
 cloud/sandbox.py               |  68 ++++-
 cloud/server.py                | 258 +++++++++++++++++-
 cloud/tests/test_task_links.py | 485 +++++++++++++++++++++++++++++++++
 cloud/worker.py                | 288 +++++++++++++++-----
 7 files changed, 1120 insertions(+), 74 deletions(-)
 create mode 100644 cloud/tests/test_task_links.py

diff --git a/README.md b/README.md
index 811840c..c5f2338 100644
--- a/README.md
+++ b/README.md
@@ -239,6 +239,15 @@ python3 host.py daemon-stop
   `GET /tasks/{id}/trace/stream`（运行中任务的 SSE：先 snapshot 再逐条
   entry，终态发 done 关流；命令台详情的链路区用 fetch 流式读它，
   EventSource 不能带 X-API-Key 头故不用）。
+  新业务链路：建任务可带 `plan`（计划模式：只读沙箱，执行器传
+  `--sandbox read-only`；不能与 `approval=full` 或 goal 同用）；
+  `POST /tasks/{id}/followups`（body/mode，mode 为 queued 或
+  direct；仅 queued/running 的非 goal 任务可接）与
+  `GET /tasks/{id}/followups`（按创建时间排序）——排队不打断当前
+  轮、跑完按序再跑一轮，直接发送立即进链路（phase=followup-direct）
+  且跑完当前轮后优先于排队执行；`POST /tasks/{id}/fork`
+  （message 必填；仅终态任务可分支，继承 executor/模型/强度/
+  权限/计划，goal 字段置空，首轮失败不自动重试、不重复分叉）。
 
 ## GitHub PR/CI 的外部条件（真实 PR 仍缺这两样）
 
diff --git a/cloud/console.html b/cloud/console.html
index 31f8ee4..9047f4a 100644
--- a/cloud/console.html
+++ b/cloud/console.html
@@ -128,6 +128,7 @@
     <option value="cloud">cloud</option>
   </select>
   <label><input id="goalMode" type="checkbox"> goal 模式</label>
+  <label><input id="planMode" type="checkbox"> 计划模式（只读）</label>
 </div>
 <textarea id="prompt" placeholder="任务 prompt（必填）"></textarea>
 <div class="bar">
@@ -183,13 +184,28 @@
   <button id="btnCancel" class="danger">取消</button>
   <button id="btnMerge">合并（completed → merged）</button>
   <button id="btnPr">提 PR（须 merged）</button>
+  <button id="btnFork">分支</button>
   <button id="btnCost">查成本</button>
   <span id="actionMsg" class="muted"></span>
 </div>
+<div id="forkForm" style="display:none">
+  <input id="forkMsg" type="text" placeholder="分支首轮消息（必填，带历史的新会话）" style="width:100%; box-sizing:border-box; margin:6px 0;">
+  <div class="bar"><button id="btnForkConfirm" class="primary">创建分支任务</button><span id="forkMsgSpan" class="muted"></span></div>
+</div>
 <pre id="detailBody" style="display:none"></pre>
 <h3>实时链路</h3>
 <div id="traceBox" class="muted">点任务行查看详情后，这里显示过程链路</div>
 <div id="costBox"></div>
+<h3>后续消息</h3>
+<div id="followupForm">
+  <textarea id="followupText" placeholder="后续消息（任务运行中/排队中可用；排队不打断当前轮，直接发送跑完当前轮后优先执行）"></textarea>
+  <div class="bar">
+    <button id="btnQueueFollowup">排队</button>
+    <button id="btnDirectFollowup" class="primary">直接发送</button>
+    <span id="followupMsg" class="muted"></span>
+  </div>
+</div>
+<div id="followupList" class="muted"></div>
 </section>
 
 <script>
@@ -300,7 +316,7 @@ async function loadTasks() {
       var pr = t.pr_url ? "<a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>#" + esc(t.pr_number) + "</a>" : "";
       return "<tr data-id='" + esc(t.id) + "' class='" + (currentTask && currentTask.id === t.id ? "selected" : "") + "'>" +
         "<td>" + esc(shortId(t.id)) + "</td>" +
-        "<td>" + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
+        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
         "<td class='st-" + esc(t.status) + "'>" + esc(t.status) + "</td>" +
         "<td>" + esc(t.executor || "") + "</td>" +
         "<td>" + esc(t.worktree_host || "") + "</td>" +
@@ -424,6 +440,8 @@ async function showDetail(taskId) {
       "executor " + esc(t.executor || "") + " ｜ 主机 " + esc(t.worktree_host || "") +
       " ｜ 模型 " + esc(t.model || "默认") + " ｜ 强度 " + esc(t.effort || "默认") +
       " ｜ 权限 " + esc(t.approval || "默认") +
+      (t.plan ? " ｜ <b>计划模式（只读）</b>" : "") +
+      (t.forked_from ? " ｜ 分支自 " + esc(shortId(t.forked_from)) : "") +
       (t.pr_url ? " ｜ <a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>PR #" + esc(t.pr_number) + "</a>" : "") +
       "<br>prompt：<br>" + esc(t.prompt || "");
     var body = "";
@@ -434,9 +452,15 @@ async function showDetail(taskId) {
     pre.textContent = body;
     pre.style.display = "block";
     $("#detailActions").style.display = "flex";
+    $("#btnFork").style.display = TRACE_TERMINAL.indexOf(t.status) >= 0 ? "" : "none";
     $("#actionMsg").textContent = "";
     $("#costBox").innerHTML = "";
+    $("#forkForm").style.display = "none";
+    $("#forkMsgSpan").textContent = "";
+    $("#followupMsg").textContent = "";
+    $("#followupForm").style.display = ["queued", "running"].indexOf(t.status) >= 0 ? "block" : "none";
     startTrace(taskId, t.status);
+    loadFollowups(taskId);
   } catch (e) { $("#detail").innerHTML = "<span class='err'>加载失败：" + esc(e.message) + "</span>"; }
 }
 
@@ -452,6 +476,29 @@ async function doAction(kind) {
   } catch (e) { $("#actionMsg").innerHTML = "<span class='err'>" + kind + " 失败：" + esc(e.message) + "</span>"; }
 }
 
+async function loadFollowups(taskId) {
+  var box = $("#followupList");
+  try {
+    var d = await api("/tasks/" + encodeURIComponent(taskId) + "/followups", "GET");
+    var list = d.followups || [];
+    if (!list.length) { box.innerHTML = "（暂无后续消息）"; return; }
+    box.innerHTML = list.map(function (f) {
+      return "[" + esc(f.mode) + "｜" + esc(f.status) + "] " + esc(String(f.body || "").slice(0, 120));
+    }).join("<br>");
+  } catch (e) { box.innerHTML = "<span class='err'>后续消息加载失败：" + esc(e.message) + "</span>"; }
+}
+async function postFollowup(mode) {
+  if (!currentTask) return;
+  var text = $("#followupText").value.trim();
+  if (!text) { $("#followupMsg").textContent = "后续消息不能为空"; return; }
+  try {
+    await api("/tasks/" + encodeURIComponent(currentTask.id) + "/followups", "POST", { body: text, mode: mode });
+    $("#followupMsg").innerHTML = "<span class='ok'>已" + (mode === "direct" ? "直接发送" : "排队") + "</span>";
+    $("#followupText").value = "";
+    loadFollowups(currentTask.id);
+  } catch (e) { $("#followupMsg").innerHTML = "<span class='err'>发送失败：" + esc(e.message) + "</span>"; }
+}
+
 $("#saveKey").onclick = function () {
   var v = $("#key").value.trim();
   if (!v) { $("#keyState").textContent = "key 为空，未保存"; return; }
@@ -503,6 +550,11 @@ $("#createTask").onclick = async function () {
     if (ap === "full" && !confirm("完全访问 = 自动批准 + 不受工作区限制，agent 可读写工作区外路径。确认提交吗？")) return;
     body.approval = ap;
   }
+  if ($("#planMode").checked) {
+    if (ap === "full") { $("#taskMsg").textContent = "计划模式不能与完全访问同用"; return; }
+    if ($("#goalMode").checked) { $("#taskMsg").textContent = "计划模式不能与 goal 模式同用"; return; }
+    body.plan = 1;
+  }
   if ($("#goalMode").checked) {
     body.goal = $("#goal").value.trim() || prompt;
     body.verify_command = $("#verifyCommand").value.trim();
@@ -519,6 +571,24 @@ $("#btnRetry").onclick = function () { doAction("retry"); };
 $("#btnCancel").onclick = function () { doAction("cancel"); };
 $("#btnMerge").onclick = function () { doAction("merge"); };
 $("#btnPr").onclick = function () { doAction("pr"); };
+$("#btnFork").onclick = function () {
+  if (!currentTask) return;
+  if (TRACE_TERMINAL.indexOf(currentTask.status) < 0) { $("#forkMsgSpan").textContent = "仅终态任务可分支"; return; }
+  $("#forkForm").style.display = "block";
+};
+$("#btnForkConfirm").onclick = async function () {
+  if (!currentTask) return;
+  var msg = $("#forkMsg").value.trim();
+  if (!msg) { $("#forkMsgSpan").textContent = "分支消息必填"; return; }
+  try {
+    var t = await api("/tasks/" + encodeURIComponent(currentTask.id) + "/fork", "POST", { message: msg });
+    $("#forkMsgSpan").innerHTML = "<span class='ok'>分支任务已建：" + esc(shortId(t.id)) + "</span>";
+    $("#forkMsg").value = "";
+    loadTasks();
+  } catch (e) { $("#forkMsgSpan").innerHTML = "<span class='err'>分支失败：" + esc(e.message) + "</span>"; }
+};
+$("#btnQueueFollowup").onclick = function () { postFollowup("queued"); };
+$("#btnDirectFollowup").onclick = function () { postFollowup("direct"); };
 $("#btnCost").onclick = async function () {
   if (!currentTask) return;
   try {
diff --git a/cloud/db.py b/cloud/db.py
index af20290..c2b57d6 100644
--- a/cloud/db.py
+++ b/cloud/db.py
@@ -834,8 +834,22 @@ POSTGRES_SCHEMA = (
     "model TEXT, "
     "effort TEXT, "
     "approval TEXT, "
+    "plan INTEGER, "
+    "forked_from TEXT, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
+    "CREATE TABLE IF NOT EXISTS task_followups("
+    "id TEXT PRIMARY KEY, "
+    "task_id TEXT, "
+    "project_id TEXT, "
+    "body TEXT, "
+    "mode TEXT, "
+    "status TEXT, "
+    "created_by TEXT, "
+    "created_at DOUBLE PRECISION, "
+    "updated_at DOUBLE PRECISION)",
+    "CREATE INDEX IF NOT EXISTS idx_followups_task"
+    " ON task_followups(task_id, status)",
     "CREATE TABLE IF NOT EXISTS summaries("
     "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, "
     "project_id TEXT, "
diff --git a/cloud/sandbox.py b/cloud/sandbox.py
index d76c448..ac2317d 100644
--- a/cloud/sandbox.py
+++ b/cloud/sandbox.py
@@ -187,7 +187,8 @@ class LocalExecutor(Executor):
         # 不改 host.py 的 dispatch/run 签名。
         self.model = model
 
-    def _base_command(self, model=None, effort=None, approval=None):
+    def _base_command(self, model=None, effort=None, approval=None,
+                      sandbox=None):
         cmd = [sys.executable, self.host_py_path]
         if self.config_path is not None:
             cmd += ["--config", self.config_path]
@@ -197,6 +198,11 @@ class LocalExecutor(Executor):
         if effort:
             cmd += ["--effort", effort]
         cmd += _approval_cli_args(approval)
+        # 显式沙箱放最后：argparse 里后出现的 --sandbox 覆盖前面
+        # 的（approval=full 自带 danger-full-access）；server 侧
+        # 已拒绝 plan+full 同存，这里是第二道防线。
+        if sandbox:
+            cmd += ["--sandbox", sandbox]
         return cmd
 
     def _command(self, task_id):
@@ -282,7 +288,9 @@ class LocalExecutor(Executor):
             raise ValueError("任务缺少 prompt")
         cmd = self._base_command(model=task.get("model"),
                                   effort=task.get("effort"),
-                                  approval=task.get("approval"))
+                                  approval=task.get("approval"),
+                                  sandbox=("read-only"
+                                           if task.get("plan") else None))
         cmd += ["dispatch", "--name", self._host_name(task),
                 "--task", task["prompt"]]
         repo = task.get("repo") or self.repo
@@ -291,6 +299,27 @@ class LocalExecutor(Executor):
         return self._result_from_runner(
             task["id"], self.runner(cmd))
 
+    def fork_task(self, task, source_id, timeout=None, extra_mounts=None,
+                  network=None, extra_env=None) -> dict:
+        """调 host.py fork：只分叉 thread 历史与工作区，不跑回合。
+
+        fork 成功后 host 任务已存在，worker 再走正常 dispatch→
+        已存在→run 回退跑首轮（见 worker._run_fork），不直接
+        在这里跑 turn。timeout/extra_* 是与 DockerExecutor 对齐
+        的签名，本地忽略。"""
+        if not isinstance(task, dict) or task.get("id") is None:
+            raise ValueError("任务缺少 id")
+        if not source_id:
+            raise ValueError("fork 缺少源任务")
+        cmd = self._base_command(model=task.get("model"),
+                                  effort=task.get("effort"),
+                                  approval=task.get("approval"))
+        cmd += ["fork", "--task",
+                self._host_name({"id": source_id}),
+                "--name", self._host_name(task)]
+        return self._result_from_runner(
+            task["id"], self.runner(cmd))
+
     def run_task(self, task) -> dict:
         """调 host.py run：在已有 host 任务上再跑一轮 turn。
 
@@ -305,7 +334,9 @@ class LocalExecutor(Executor):
             raise ValueError("任务缺少 prompt")
         cmd = self._base_command(model=task.get("model"),
                                   effort=task.get("effort"),
-                                  approval=task.get("approval"))
+                                  approval=task.get("approval"),
+                                  sandbox=("read-only"
+                                           if task.get("plan") else None))
         cmd += ["run", "--task", self._host_name(task),
                 "--prompt", task["prompt"]]
         return self._result_from_runner(
@@ -403,7 +434,9 @@ class LocalExecutor(Executor):
             raise ValueError("任务缺少 prompt")
         cmd = self._base_command(model=task.get("model"),
                                   effort=task.get("effort"),
-                                  approval=task.get("approval"))
+                                  approval=task.get("approval"),
+                                  sandbox=("read-only"
+                                           if task.get("plan") else None))
         if kind == "run":
             cmd += ["run", "--task", self._host_name(task),
                     "--prompt", task["prompt"]]
@@ -669,7 +702,7 @@ class DockerExecutor(Executor):
         return cmd
 
     def _inner_host_command(self, *args, model=None, effort=None,
-                            approval=None):
+                            approval=None, sandbox=None):
         """容器内 host.py 命令：python3 /app/host.py [args...]。
 
         config_path 若已给，容器内用 /app/<basename> 指代它——
@@ -691,6 +724,8 @@ class DockerExecutor(Executor):
         if effort:
             cmd += ["--effort", effort]
         cmd += _approval_cli_args(approval)
+        if sandbox:
+            cmd += ["--sandbox", sandbox]
         cmd += list(args)
         return cmd
 
@@ -828,7 +863,8 @@ class DockerExecutor(Executor):
         inner = self._inner_host_command(
             "dispatch", "--name", self._host_name(task),
             "--task", task["prompt"], model=task.get("model"),
-            effort=task.get("effort"), approval=task.get("approval"))
+            effort=task.get("effort"), approval=task.get("approval"),
+            sandbox=("read-only" if task.get("plan") else None))
         repo = task.get("repo")
         if repo:
             inner += ["--repo", repo]
@@ -837,6 +873,23 @@ class DockerExecutor(Executor):
                                       network=network,
                                       extra_env=extra_env)
 
+    def fork_task(self, task, source_id, timeout=None, extra_mounts=None,
+                  network=None, extra_env=None) -> dict:
+        """容器内调 host.py fork：只分叉，不跑回合（同本地版）。"""
+        if not isinstance(task, dict) or task.get("id") is None:
+            raise ValueError("任务缺少 id")
+        if not source_id:
+            raise ValueError("fork 缺少源任务")
+        inner = self._inner_host_command(
+            "fork", "--task", self._host_name({"id": source_id}),
+            "--name", self._host_name(task),
+            model=task.get("model"), effort=task.get("effort"),
+            approval=task.get("approval"))
+        return self._run_in_container(task, inner, timeout=timeout,
+                                      extra_mounts=extra_mounts,
+                                      network=network,
+                                      extra_env=extra_env)
+
     def run_task(self, task, timeout=None, extra_mounts=None,
                  network=None, extra_env=None) -> dict:
         """容器内调 host.py run：在已有 host 任务上再跑一轮 turn。"""
@@ -847,7 +900,8 @@ class DockerExecutor(Executor):
         inner = self._inner_host_command(
             "run", "--task", self._host_name(task),
             "--prompt", task["prompt"], model=task.get("model"),
-            effort=task.get("effort"), approval=task.get("approval"))
+            effort=task.get("effort"), approval=task.get("approval"),
+            sandbox=("read-only" if task.get("plan") else None))
         return self._run_in_container(task, inner, timeout=timeout,
                                       extra_mounts=extra_mounts,
                                       network=network,
diff --git a/cloud/server.py b/cloud/server.py
index 2ccd3f0..99e0c7d 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -445,6 +445,10 @@ def migrate_tasks_table(conn):
         "model": "TEXT",
         "effort": "TEXT",
         "approval": "TEXT",
+        # 计划模式（只读）与分支来源：plan=1 时 worker 用
+        # read-only 沙箱；forked_from 记源任务 id（非分支为 NULL）。
+        "plan": "INTEGER",
+        "forked_from": "TEXT",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -468,6 +472,41 @@ def migrate_projects_table(conn):
         conn.execute("ALTER TABLE projects ADD COLUMN auto_pr INTEGER")
 
 
+def _ensure_followups_table(conn):
+    """幂等建后续消息表：server init_db 与各 followup 接口都调它，
+    老库第一次跑到这里时自动补表，不丢数据。DOUBLE PRECISION 在
+    SQLite 与 Postgres 下都可用（SQLite 按 NUMERIC 亲和处理）。"""
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS task_followups("
+        "id TEXT PRIMARY KEY, "
+        "task_id TEXT, "
+        "project_id TEXT, "
+        "body TEXT, "
+        "mode TEXT, "
+        "status TEXT, "
+        "created_by TEXT, "
+        "created_at DOUBLE PRECISION, "
+        "updated_at DOUBLE PRECISION)"
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_followups_task"
+        " ON task_followups(task_id, status)")
+
+
+def _followup_to_dict(row):
+    return {
+        "id": row["id"],
+        "task_id": row["task_id"],
+        "project_id": row["project_id"],
+        "body": row["body"],
+        "mode": row["mode"],
+        "status": row["status"],
+        "created_by": row["created_by"],
+        "created_at": row["created_at"],
+        "updated_at": row["updated_at"],
+    }
+
+
 def _is_postgres_backend():
     backend = os.environ.get("CLOUD_DB_BACKEND") or "sqlite"
     return str(backend).strip().lower() in ("postgres", "postgresql")
@@ -487,6 +526,7 @@ def init_db(db_path):
             _init_postgres_schema(conn)
             migrate_tasks_table(conn)
             migrate_projects_table(conn)
+            _ensure_followups_table(conn)
             conn.commit()
             return
         conn.execute(
@@ -515,6 +555,7 @@ def init_db(db_path):
         )
         migrate_tasks_table(conn)
         migrate_projects_table(conn)
+        _ensure_followups_table(conn)
         # worker 每轮 poll 都按 status 取任务（见 load_tasks 的
         # WHERE status IN (...)），无索引时是全表扫描；任务表随
         # 时间变大后索引是必须的。IF NOT EXISTS 保证老库幂等。
@@ -583,6 +624,8 @@ def _task_to_dict(row):
         "model": _row_get(row, "model"),
         "effort": _row_get(row, "effort"),
         "approval": _row_get(row, "approval"),
+        "plan": _row_get(row, "plan") or 0,
+        "forked_from": _row_get(row, "forked_from"),
     }
 
 
@@ -911,6 +954,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._retry_task(user_id, rest[: -len("/retry")])
             elif rest.endswith("/heartbeat") and method == "POST":
                 self._heartbeat_task(user_id, rest[: -len("/heartbeat")])
+            elif rest.endswith("/followups") and method == "POST":
+                self._create_followup(user_id, rest[: -len("/followups")])
+            elif rest.endswith("/followups") and method == "GET":
+                self._list_followups(user_id, rest[: -len("/followups")])
+            elif rest.endswith("/fork") and method == "POST":
+                self._fork_task(user_id, rest[: -len("/fork")])
             else:
                 self._send_json(404, {"error": "not found"})
         else:
@@ -1484,6 +1533,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._send_json(400, {
                 "error": "approval must be ask, auto or full"})
             return
+        # 计划模式：只读沙箱，不改任何文件；与“完全访问”
+        # （danger-full-access）语义直接冲突，拒绝建任务。
+        plan = 1 if body.get("plan") in (True, 1) else 0
+        if plan and approval == "full":
+            self._send_json(400, {
+                "error": "plan 模式与 approval=full 冲突（只读 vs 完全访问）"})
+            return
         # goal 字段：verify_command 非空即 goal 模式（worker 走
         # host.py goal 循环）；goal 文本缺省用 prompt；轮次缺省 3
         # （与 host.py 默认一致），非法值 400 且不建任务。
@@ -1497,6 +1553,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._send_json(400, {
                 "error": "verify_command must be a string"})
             return
+        # 计划模式与 goal 模式也冲突：goal 循环就是要改文件
+        # 直到验测通过，只读沙箱下它永远做不成。
+        if plan and verify_command:
+            self._send_json(400, {
+                "error": "plan 模式与 goal 模式冲突（只读 vs 改文件）"})
+            return
         try:
             goal_max_rounds = int(body.get("goal_max_rounds", 3))
         except (TypeError, ValueError):
@@ -1534,13 +1596,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     " idempotency_key, depends_on, priority, created_by,"
                     " created_at, updated_at, last_heartbeat, executor,"
                     " goal, verify_command, goal_max_rounds,"
-                    " model, effort, approval)"
+                    " model, effort, approval, plan)"
                     " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?,"
-                    " ?, ?, ?, ?, ?, ?)",
+                    " ?, ?, ?, ?, ?, ?, ?)",
                     (task_id, project_id, prompt, idempotency_key,
                      json.dumps(depends_on), priority, user_id, now, now,
                      executor, goal, verify_command, goal_max_rounds,
-                     model, effort, approval),
+                     model, effort, approval, plan),
                 )
                 conn.commit()
             except (sqlite3.IntegrityError, _PgIntegrityError):
@@ -1705,6 +1767,196 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         _emit({"type": "done", "status": task["status"],
                "reason": "timeout"})
 
+    def _create_followup(self, user_id, task_id):
+        """给进行中任务发后续消息：排队 / 直接发送。
+
+        语义（用户定稿）：排队进队尾，不打断本轮，本轮结束
+        后按序每条再跑一轮；直接发送也不中断本轮——消息立即
+        送达（链路里可见），本轮跑完后立刻以它开下一轮，且
+        优先于排队中的消息。v1 只做运行时（含排队中）的任务：
+        已终态的任务 409；goal 任务 409（goal 循环自有轮次）。
+        权限与取消同款：项目 create_task（写操作，不能只读）。
+        """
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "create_task"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        text = body.get("body")
+        if not text or not isinstance(text, str) or not text.strip():
+            self._send_json(400, {"error": "body required"})
+            return
+        text = text.strip()
+        if len(text) > 20000:
+            self._send_json(400, {"error": "body too long"})
+            return
+        mode = body.get("mode", "queued")
+        if mode not in ("queued", "direct"):
+            self._send_json(400, {
+                "error": "mode must be queued or direct"})
+            return
+        if task["status"] not in ("queued", "running"):
+            self._send_json(409, {
+                "error": "后续消息仅支持进行中任务",
+                "status": task["status"],
+            })
+            return
+        if task.get("verify_command"):
+            self._send_json(409, {"error": "goal 任务暂不支持后续消息"})
+            return
+        now = time.time()
+        conn = self._db()
+        try:
+            _ensure_followups_table(conn)
+            fu_id = uuid.uuid4().hex
+            conn.execute(
+                "INSERT INTO task_followups(id, task_id, project_id,"
+                " body, mode, status, created_by, created_at,"
+                " updated_at) VALUES(?, ?, ?, ?, ?, 'pending', ?, ?, ?)",
+                (fu_id, task_id, task["project_id"], text, mode,
+                 user_id, now, now),
+            )
+            conn.commit()
+            fu = _followup_to_dict(
+                conn.execute(
+                    "SELECT * FROM task_followups WHERE id = ?",
+                    (fu_id,),
+                ).fetchone()
+            )
+        finally:
+            conn.close()
+        if mode == "direct":
+            self._append_direct_trace(task, fu_id, text)
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"],
+                        "followup_created", "%s:%s" % (task_id, fu_id))
+        self._send_json(201, fu)
+
+    def _append_direct_trace(self, task, fu_id, text):
+        """直接发送的消息立即进链路：往 host 的 jsonl 里追加一条
+        message 条目（phase=followup-direct），运行中的链路流
+        下一轮 poll 就能看到它。只尽力而为：worktree 在别机时
+        本机文件不是 worker 写的那份，跳过（与 _trace_snapshot
+        的多机降级同规则）；写失败也绝不让 POST 失败。"""
+        try:
+            worktree_host = task.get("worktree_host")
+            if worktree_host and worktree_host != socket.gethostname():
+                return
+            d = _trace_dir_for(self.server)
+            os.makedirs(d, exist_ok=True)
+            entry = {"kind": "message", "id": "fu-" + fu_id,
+                     "ts": int(time.time() * 1000),
+                     "phase": "followup-direct", "text": text[:300]}
+            with open(os.path.join(
+                    d, _trace_file_name(task["id"]) + ".jsonl"),
+                    "a", encoding="utf-8") as f:
+                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+        except Exception:
+            pass
+
+    def _list_followups(self, user_id, task_id):
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "read"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        conn = self._db()
+        try:
+            _ensure_followups_table(conn)
+            rows = conn.execute(
+                "SELECT * FROM task_followups WHERE task_id = ?"
+                " ORDER BY created_at, id",
+                (task_id,),
+            ).fetchall()
+        finally:
+            conn.close()
+        self._send_json(200, {
+            "followups": [_followup_to_dict(r) for r in rows]})
+
+    def _fork_task(self, user_id, task_id):
+        """会话分支：从已结束任务的 thread 分叉出新任务。
+
+        新任务继承源任务的 executor/模型/强度/权限/计划模式，
+        不继承 goal（分支是普通任务，首轮 prompt 就是这里传的
+        message）。worker 认领后先调 host.py fork（分叉 thread
+        历史 + 拷 worktree 文件），再以 message 跑首轮。
+        只能从终态任务分支（进行中的 thread 正在变，分叉快照
+        不可信）；message 必填——分支没有首条消息就没有方向。
+        """
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "create_task"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        message = body.get("message")
+        if not message or not isinstance(message, str) \
+                or not message.strip():
+            self._send_json(400, {"error": "message required"})
+            return
+        message = message.strip()
+        if len(message) > 20000:
+            self._send_json(400, {"error": "message too long"})
+            return
+        if task["status"] not in TERMINAL_STATUSES:
+            self._send_json(409, {
+                "error": "只能从已结束的任务分支",
+                "status": task["status"],
+            })
+            return
+        now = time.time()
+        new_id = uuid.uuid4().hex
+        conn = self._db()
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " idempotency_key, depends_on, priority, created_by,"
+                " created_at, updated_at, last_heartbeat, executor,"
+                " goal, verify_command, goal_max_rounds,"
+                " model, effort, approval, plan, forked_from)"
+                " VALUES(?, ?, ?, 'queued', NULL, '[]', ?, ?, ?, ?,"
+                " NULL, ?, NULL, NULL, ?, ?, ?, ?, ?, ?)",
+                (new_id, task["project_id"], message, task["priority"],
+                 user_id, now, now, task["executor"],
+                 task["goal_max_rounds"] or 3, task["model"],
+                 task["effort"], task["approval"], task["plan"] or 0,
+                 task_id),
+            )
+            conn.commit()
+            new_task = _task_to_dict(
+                conn.execute(
+                    "SELECT * FROM tasks WHERE id = ?", (new_id,)
+                ).fetchone()
+            )
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"], "task_forked",
+                        "%s -> %s" % (task_id, new_id))
+        self._send_json(201, new_task)
+
     def _cancel_task(self, user_id, task_id):
         row = self._fetch_task(task_id)
         if row is None:
diff --git a/cloud/tests/test_task_links.py b/cloud/tests/test_task_links.py
new file mode 100644
index 0000000..7653c61
--- /dev/null
+++ b/cloud/tests/test_task_links.py
@@ -0,0 +1,485 @@
+"""新业务链路单测：计划模式、后续消息（排队/直接发送）、会话分支。
+
+不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
+SQLite；worker 侧注入假 runner（假 CLI）与假执行器；链路目录
+经 server.trace_dir 注入临时目录。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+
+
+def _completed_runner(cmd):
+    """假 CLI：dispatch/run 一律成功，打印 host.py 风格的 JSON。"""
+    if "dispatch" in cmd:
+        name = cmd[cmd.index("--name") + 1]
+    elif "run" in cmd:
+        name = cmd[cmd.index("--task") + 1]
+    else:
+        name = "x"
+    return SimpleNamespace(
+        returncode=0,
+        stdout=json.dumps({"task": name, "status": "completed",
+                           "text": "done"}),
+        stderr="")
+
+
+class ServerLinksTestCase(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()
+        self.trace_dir = os.path.join(self.tmp.name, "host-traces")
+        os.makedirs(self.trace_dir, exist_ok=True)
+        self.server.trace_dir = self.trace_dir
+        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 body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _create_task(self, **extra):
+        body = {"project_id": self.project["id"], "prompt": "做一个加法函数"}
+        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 test_plan_full_conflict_400(self):
+        status, body = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "plan": 1, "approval": "full"},
+            self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_plan_goal_conflict_400(self):
+        status, body = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "p",
+             "plan": 1, "verify_command": "true"},
+            self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_plan_create_roundtrip(self):
+        task = self._create_task(plan=1, approval="ask")
+        self.assertEqual(task.get("plan"), 1)
+        status, got = self._req(
+            "GET", "/tasks/" + task["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(got.get("plan"), 1)
+
+    # -- 后续消息 -------------------------------------------------------
+    def test_followup_validation(self):
+        status, _ = self._req(
+            "POST", "/tasks/nope/followups",
+            {"body": "hi"}, self.owner_key)
+        self.assertEqual(status, 404)
+        task = self._create_task()
+        status, _ = self._req(
+            "POST", "/tasks/%s/followups" % task["id"],
+            {"body": "   "}, self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "POST", "/tasks/%s/followups" % task["id"],
+            {"body": "hi", "mode": "bogus"}, self.owner_key)
+        self.assertEqual(status, 400)
+        # 终态任务拒绝后续消息。
+        self._set_status(task["id"], "completed")
+        status, _ = self._req(
+            "POST", "/tasks/%s/followups" % task["id"],
+            {"body": "hi"}, self.owner_key)
+        self.assertEqual(status, 409)
+        # goal 任务拒绝后续消息。
+        goal = self._create_task(verify_command="true")
+        status, _ = self._req(
+            "POST", "/tasks/%s/followups" % goal["id"],
+            {"body": "hi"}, self.owner_key)
+        self.assertEqual(status, 409)
+
+    def test_followup_create_and_list(self):
+        task = self._create_task()
+        status, fu1 = self._req(
+            "POST", "/tasks/%s/followups" % task["id"],
+            {"body": "先排队"}, self.owner_key)
+        self.assertEqual(status, 201, fu1)
+        self.assertEqual(fu1["status"], "pending")
+        status, fu2 = self._req(
+            "POST", "/tasks/%s/followups" % task["id"],
+            {"body": "直接发", "mode": "direct"}, self.owner_key)
+        self.assertEqual(status, 201, fu2)
+        status, listing = self._req(
+            "GET", "/tasks/%s/followups" % task["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual([f["body"] for f in listing["followups"]],
+                         ["先排队", "直接发"])
+
+    def test_followup_direct_appends_trace(self):
+        task = self._create_task()
+        status, _ = self._req(
+            "POST", "/tasks/%s/followups" % task["id"],
+            {"body": "立刻看到我", "mode": "direct"}, self.owner_key)
+        self.assertEqual(status, 201)
+        path = os.path.join(self.trace_dir, task["id"] + ".jsonl")
+        with open(path, encoding="utf-8") as f:
+            entries = [json.loads(line) for line in f if line.strip()]
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["phase"], "followup-direct")
+        self.assertIn("立刻看到我", entries[0]["text"])
+
+    # -- 会话分支 -------------------------------------------------------
+    def test_fork_terminal_inherits_fields(self):
+        src = self._create_task(model="gpt-5.6-sol", effort="high",
+                                approval="ask", plan=1)
+        self._set_status(src["id"], "completed")
+        status, forked = self._req(
+            "POST", "/tasks/%s/fork" % src["id"],
+            {"message": "换个方向继续"}, self.owner_key)
+        self.assertEqual(status, 201, forked)
+        self.assertEqual(forked["forked_from"], src["id"])
+        self.assertEqual(forked["status"], "queued")
+        self.assertEqual(forked["prompt"], "换个方向继续")
+        self.assertEqual(forked["model"], "gpt-5.6-sol")
+        self.assertEqual(forked["effort"], "high")
+        self.assertEqual(forked["approval"], "ask")
+        self.assertEqual(forked["plan"], 1)
+        self.assertIsNone(forked.get("verify_command"))
+
+    def test_fork_validation(self):
+        task = self._create_task()
+        # 运行中（queued）不能分支。
+        status, _ = self._req(
+            "POST", "/tasks/%s/fork" % task["id"],
+            {"message": "x"}, self.owner_key)
+        self.assertEqual(status, 409)
+        self._set_status(task["id"], "completed")
+        status, _ = self._req(
+            "POST", "/tasks/%s/fork" % task["id"],
+            {"message": "   "}, self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "POST", "/tasks/nope/fork",
+            {"message": "x"}, self.owner_key)
+        self.assertEqual(status, 404)
+
+
+def _insert_task(db_path, task_id, prompt="do work", status="queued",
+                 project_id="p1", **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, project_id, prompt, status, None, "[]", 0, "tester",
+            now, now, None]
+    for k, v in extra.items():
+        cols.append(k)
+        vals.append(v)
+    conn = sqlite3.connect(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(db_path, task_id, body, mode, created_at,
+                     project_id="p1", status="pending"):
+    # id 是 TEXT 主键（服务端生成 uuid）：不传 id 会插成 NULL，
+    # CAS 认领 WHERE id = NULL 永远匹配不到，排空会静默跳过。
+    import uuid
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO task_followups(id, task_id, project_id, body,"
+            " mode, status, created_by, created_at, updated_at)"
+            " VALUES(?, ?, ?, ?, ?, ?, 'tester', ?, ?)",
+            (uuid.uuid4().hex, task_id, project_id, body, mode, status,
+             created_at, created_at))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, table, task_id=None, fu_id=None):
+    conn = sqlite3.connect(db_path)
+    try:
+        if fu_id is not None:
+            row = conn.execute(
+                "SELECT * FROM %s WHERE id = ?" % table,
+                (fu_id,)).fetchone()
+        else:
+            row = conn.execute(
+                "SELECT * FROM %s WHERE id = ?" % table,
+                (task_id,)).fetchone()
+        if row is None:
+            return None
+        cols = [d[0] for d in conn.execute(
+            "SELECT * FROM %s LIMIT 0" % table).description]
+        return dict(zip(cols, row))
+    finally:
+        conn.close()
+
+
+def _followup_statuses(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT body, status FROM task_followups"
+            " WHERE task_id = ? ORDER BY id", (task_id,)).fetchall()
+    finally:
+        conn.close()
+    return dict(rows)
+
+
+class WorkerLinksTestCase(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"}
+        self.calls = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self, runner=None, **kwargs):
+        kwargs.setdefault("env", self.env)
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=runner or _completed_runner, **kwargs)
+
+    def _recording(self, inner):
+        def fake(cmd):
+            self.calls.append(list(cmd))
+            return inner(cmd)
+        return fake
+
+    # -- 任务级模型优先 -------------------------------------------------
+    def test_task_model_beats_worker_model(self):
+        _insert_task(self.db_path, "T", model="gpt-5.6-sol")
+        worker = self._worker(runner=self._recording(_completed_runner),
+                              model="gpt-5.6-terra")
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        dispatch = [c for c in self.calls if "dispatch" in c][0]
+        self.assertEqual(dispatch[dispatch.index("--model") + 1],
+                         "gpt-5.6-sol")
+        self.assertEqual(_row(self.db_path, "tasks", "T")["actual_model"],
+                         "gpt-5.6-sol")
+
+    # -- 计划模式命令 ---------------------------------------------------
+    def test_plan_local_command_readonly(self):
+        _insert_task(self.db_path, "T", plan=1)
+        worker = self._worker(runner=self._recording(_completed_runner))
+        worker.run_one()
+        dispatch = [c for c in self.calls if "dispatch" in c][0]
+        self.assertIn("--sandbox", dispatch)
+        self.assertEqual(dispatch[dispatch.index("--sandbox") + 1],
+                         "read-only")
+
+    def test_plan_local_command_absent_without_plan(self):
+        _insert_task(self.db_path, "T")
+        worker = self._worker(runner=self._recording(_completed_runner))
+        worker.run_one()
+        dispatch = [c for c in self.calls if "dispatch" in c][0]
+        self.assertNotIn("--sandbox", dispatch)
+
+    def test_plan_docker_inner_command_readonly(self):
+        ex = DockerExecutor(config_path="/fake/cfg.json")
+        cmd = ex._inner_host_command(
+            "dispatch", "--name", "x", "--task", "hi", sandbox="read-only")
+        self.assertIn("--sandbox", cmd)
+        self.assertEqual(cmd[cmd.index("--sandbox") + 1], "read-only")
+        cmd2 = ex._inner_host_command(
+            "dispatch", "--name", "x", "--task", "hi")
+        self.assertNotIn("--sandbox", cmd2)
+
+    # -- 后续消息排空 ---------------------------------------------------
+    def test_followup_drain_order_direct_first(self):
+        _insert_task(self.db_path, "T", prompt="orig")
+        now = time.time()
+        _insert_followup(self.db_path, "T", "q1", "queued", now)
+        _insert_followup(self.db_path, "T", "d1", "direct", now + 1)
+        _insert_followup(self.db_path, "T", "q2", "queued", now + 2)
+
+        def fake(cmd):
+            if "dispatch" in cmd:
+                return SimpleNamespace(
+                    returncode=1, stdout="",
+                    stderr="Traceback: task 已存在：T，先换名或 discard 它")
+            return _completed_runner(cmd)
+
+        worker = self._worker(runner=self._recording(fake))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        prompts = []
+        for c in self.calls:
+            if "run" in c:
+                prompts.append(c[c.index("--prompt") + 1])
+        self.assertEqual(prompts, ["orig", "d1", "q1", "q2"])
+        self.assertEqual(_followup_statuses(self.db_path, "T"),
+                         {"q1": "done", "d1": "done", "q2": "done"})
+
+    def test_followup_failure_stops_drain(self):
+        _insert_task(self.db_path, "T", prompt="orig")
+        now = time.time()
+        _insert_followup(self.db_path, "T", "bad", "queued", now)
+        _insert_followup(self.db_path, "T", "later", "queued", now + 1)
+
+        def fake(cmd):
+            if "dispatch" in cmd and "bad" in cmd:
+                return SimpleNamespace(returncode=1, stdout="",
+                                       stderr="boom")
+            return _completed_runner(cmd)
+
+        worker = self._worker(runner=self._recording(fake))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(_followup_statuses(self.db_path, "T"),
+                         {"bad": "failed", "later": "pending"})
+
+    def test_goal_task_does_not_drain(self):
+        _insert_task(self.db_path, "T", prompt="orig",
+                     verify_command="true")
+
+        def fake(cmd):
+            if "goal" in cmd:
+                # goal 结论行：host.py goal 两种结局都 exit 0，
+                # 结论靠这行解析，没有它会被判无结论。
+                return SimpleNamespace(returncode=0,
+                                       stdout="goal met after rounds\n",
+                                       stderr="")
+            return _completed_runner(cmd)
+
+        _insert_followup(self.db_path, "T", "later", "queued", time.time())
+        worker = self._worker(runner=self._recording(fake))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(_followup_statuses(self.db_path, "T"),
+                         {"later": "pending"})
+
+    # -- 会话分支 -------------------------------------------------------
+    def test_fork_runs_first_round_once(self):
+        _insert_task(self.db_path, "T", prompt="branch msg",
+                     forked_from="SRC")
+        forks = []
+
+        def fake(cmd):
+            if "fork" in cmd:
+                forks.append(list(cmd))
+                return SimpleNamespace(returncode=0,
+                                       stdout="forked SRC -> T\n",
+                                       stderr="")
+            if "dispatch" in cmd:
+                return SimpleNamespace(
+                    returncode=1, stdout="",
+                    stderr="Traceback: task 已存在：T，先换名或 discard 它")
+            return _completed_runner(cmd)
+
+        worker = self._worker(runner=self._recording(fake))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(len(forks), 1)
+        fork = forks[0]
+        self.assertIn("SRC", fork)
+        self.assertIn("T", fork)
+        runs = [c for c in self.calls if "run" in c]
+        self.assertEqual(len(runs), 1)
+        self.assertEqual(runs[0][runs[0].index("--prompt") + 1],
+                         "branch msg")
+
+    def test_fork_failure_not_retried(self):
+        _insert_task(self.db_path, "T", prompt="branch msg",
+                     forked_from="SRC")
+        forks = []
+
+        def fake(cmd):
+            if "fork" in cmd:
+                forks.append(list(cmd))
+                return SimpleNamespace(returncode=0,
+                                       stdout="forked SRC -> T\n",
+                                       stderr="")
+            if "dispatch" in cmd:
+                return SimpleNamespace(
+                    returncode=1, stdout="",
+                    stderr="Traceback: task 已存在：T，先换名或 discard 它")
+            return SimpleNamespace(returncode=1, stdout="",
+                                   stderr="model boom")
+
+        worker = self._worker(runner=self._recording(fake))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        # range(0) 跳过重试循环：fork 只调一次，绝不重复分叉。
+        self.assertEqual(len(forks), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/worker.py b/cloud/worker.py
index c1d9503..ded13e9 100644
--- a/cloud/worker.py
+++ b/cloud/worker.py
@@ -238,6 +238,8 @@ def _task_from_row(row):
         "model": _col("model"),
         "effort": _col("effort"),
         "approval": _col("approval"),
+        "plan": _col("plan"),
+        "forked_from": _col("forked_from"),
     }
 
 
@@ -563,7 +565,9 @@ class Worker:
                 "worktree_host": "TEXT",
                 "model": "TEXT",
                 "effort": "TEXT",
-                "approval": "TEXT"}.items():
+                "approval": "TEXT",
+                "plan": "INTEGER",
+                "forked_from": "TEXT"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -1155,6 +1159,158 @@ class Worker:
         return None
 
     # -- 执行 ----------------------------------------------------------
+    def _docker_task_context(self, task):
+        """Docker 任务的租户挂载/网络/代理准备（_run_attempt 与
+        _run_fork 共用）。成功返回 (True, mounts, network,
+        proxy_env, None)；失败返回 (False, None, None, None,
+        error_text)。网络建失败绝不降级默认网络跑容器（降级会
+        让隔离悄悄失效）。"""
+        tenant_mounts = None
+        tenant_network = None
+        proxy_env = None
+        if self.tenant_isolation:
+            try:
+                tenant_mounts = self._tenant_mounts(
+                    task.get("project_id"))
+                tenant_network = self._tenant_network(
+                    task.get("project_id"))
+            except (ValueError, RuntimeError) as exc:
+                return (False, None, None, None,
+                        "租户隔离准备失败: %s" % exc)
+            if not self.executor.ensure_network(tenant_network):
+                return (False, None, None, None,
+                        "租户 Docker 网络创建失败: %s" % tenant_network)
+        if self.egress_proxy_port:
+            # 代理在宿主上，容器经所在网络的网关访问；租户模式
+            # 用项目网络，非租户用默认 bridge。
+            gw = self.executor.network_gateway(tenant_network)
+            if not gw:
+                return (False, None, None, None,
+                        "egress 代理网关解析失败，拒绝无代理"
+                        "运行（避免绕过出口白名单）")
+            proxy_url = "http://%s:%s" % (gw, self.egress_proxy_port)
+            proxy_env = {
+                "HTTP_PROXY": proxy_url,
+                "HTTPS_PROXY": proxy_url,
+                "http_proxy": proxy_url,
+                "https_proxy": proxy_url,
+                "NO_PROXY": "localhost,127.0.0.1",
+                "no_proxy": "localhost,127.0.0.1",
+            }
+        return (True, tenant_mounts, tenant_network, proxy_env, None)
+
+    def _run_fork(self, task):
+        """分支任务执行：host.py fork 建分支 thread，再跑首轮。
+        返回 (result, error)，语义同 _run_attempt。fork 本身不
+        跑模型；首轮走 _run_attempt（dispatch 撞已存在 → run
+        续跑，三路取消/超时处理与普通任务一致）。首轮失败不
+        自动重试/fallback：失败后走重试入口重跑首轮（已存在
+        续跑），不重复 fork。"""
+        task_id = task["id"]
+        source_id = task.get("forked_from")
+        if not hasattr(self.executor, "fork_task"):
+            return ({"executor": self._executor_name(),
+                     "task_id": task_id,
+                     "status": "failed", "stdout": "",
+                     "returncode": None, "host_status": None,
+                     "stderr": ""},
+                    "executor 不支持 fork")
+        is_docker = isinstance(self.executor, DockerExecutor)
+        if is_docker:
+            (ok, tenant_mounts, tenant_network, proxy_env,
+             prep_err) = self._docker_task_context(task)
+            if not ok:
+                return ({"executor": "docker",
+                         "task_id": task_id,
+                         "status": "failed", "stdout": "",
+                         "returncode": None, "host_status": None,
+                         "stderr": ""},
+                        prep_err)
+        else:
+            tenant_mounts = tenant_network = proxy_env = None
+        try:
+            fork_res = self.executor.fork_task(
+                task, source_id,
+                timeout=(self.task_timeout if is_docker else None),
+                extra_mounts=tenant_mounts,
+                network=tenant_network,
+                extra_env=proxy_env)
+        except Exception as exc:
+            return ({"executor": self._executor_name(),
+                     "task_id": task_id,
+                     "status": "failed", "stdout": "",
+                     "returncode": None, "host_status": None,
+                     "stderr": ""},
+                    str(exc))
+        if (fork_res or {}).get("status") != "completed":
+            err = ((fork_res or {}).get("stderr", "") or
+                   (fork_res or {}).get("stdout", "") or "fork failed")
+            err = err.strip()
+            return (fork_res,
+                    err[-1000:] if err else "fork failed")
+        return self._run_attempt(task)
+
+    def _claim_next_followup(self, task_id):
+        """取下一条后续消息：direct 优先、队内按创建时间先来
+        先服务。CAS 认领（pending → running），抢不到或表不
+        在返回 None（表由 server init_db 建；worker 直连老库
+        时缺表不炸 run_one，只当没有后续消息）。"""
+        try:
+            conn = _db_connect(self.db_path)
+        except Exception:
+            return None
+        try:
+            row = conn.execute(
+                "SELECT * FROM task_followups"
+                " WHERE task_id = ? AND status = 'pending'"
+                " ORDER BY CASE WHEN mode = 'direct' THEN 0 ELSE 1 END,"
+                " created_at, id LIMIT 1",
+                (task_id,)).fetchone()
+            if row is None:
+                return None
+            cur = conn.execute(
+                "UPDATE task_followups SET status = 'running',"
+                " updated_at = ? WHERE id = ? AND status = 'pending'",
+                (time.time(), row["id"]))
+            conn.commit()
+            if cur.rowcount != 1:
+                return None
+            return {"id": row["id"], "body": row["body"],
+                    "mode": row["mode"]}
+        except Exception:
+            return None
+        finally:
+            conn.close()
+
+    def _finish_followup(self, fu_id, status):
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                conn.execute(
+                    "UPDATE task_followups SET status = ?,"
+                    " updated_at = ? WHERE id = ?",
+                    (status, time.time(), fu_id))
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
+    def _restore_followup(self, fu_id):
+        """取消时把跑中的后续消息放回 pending，留给重试入口。"""
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                conn.execute(
+                    "UPDATE task_followups SET status = 'pending',"
+                    " updated_at = ? WHERE id = ? AND status = 'running'",
+                    (time.time(), fu_id))
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
     def _run_attempt(self, task):
         """单次执行尝试（不含认领、key 门、重试、成本与最终审计）。
 
@@ -1214,62 +1370,20 @@ class Worker:
                         deadline_seconds=self.task_timeout)
             else:
                 if is_docker:
-                    tenant_mounts = None
-                    tenant_network = None
-                    proxy_env = None
-                    if self.tenant_isolation:
-                        try:
-                            tenant_mounts = self._tenant_mounts(
-                                task.get("project_id"))
-                            tenant_network = self._tenant_network(
-                                task.get("project_id"))
-                        except (ValueError, RuntimeError) as exc:
-                            # 提前返回会跳过下面的 finally（清容器
-                            # 标记），这里先手动清掉，否则 _running
-                            # 里永远残留该任务，心跳与退出清理
-                            # 都会把它当在跑任务。
-                            self._running_clear(task_id)
-                            return ({"executor": "docker",
-                                     "task_id": task_id,
-                                     "status": "failed", "stdout": "",
-                                     "returncode": None, "host_status": None,
-                                     "stderr": ""},
-                                    "租户隔离准备失败: %s" % exc)
-                        # 网络建失败绝不降级默认网络跑容器（降级会
-                        # 让隔离悄悄失效）；直接让本次尝试 failed，
-                        # 调用方按普通失败处理（重试/fallback 照常）。
-                        if not self.executor.ensure_network(tenant_network):
-                            self._running_clear(task_id)
-                            return ({"executor": "docker",
-                                     "task_id": task_id,
-                                     "status": "failed", "stdout": "",
-                                     "returncode": None, "host_status": None,
-                                     "stderr": ""},
-                                    "租户 Docker 网络创建失败: %s"
-                                    % tenant_network)
-                    if self.egress_proxy_port:
-                        # 代理在宿主上，容器经所在网络的网关访问；
-                        # 租户模式用项目网络，非租户用默认 bridge。
-                        gw = self.executor.network_gateway(tenant_network)
-                        if not gw:
-                            self._running_clear(task_id)
-                            return ({"executor": "docker",
-                                     "task_id": task_id,
-                                     "status": "failed", "stdout": "",
-                                     "returncode": None, "host_status": None,
-                                     "stderr": ""},
-                                    "egress 代理网关解析失败，拒绝无代理"
-                                    "运行（避免绕过出口白名单）")
-                        proxy_url = "http://%s:%s" % (
-                            gw, self.egress_proxy_port)
-                        proxy_env = {
-                            "HTTP_PROXY": proxy_url,
-                            "HTTPS_PROXY": proxy_url,
-                            "http_proxy": proxy_url,
-                            "https_proxy": proxy_url,
-                            "NO_PROXY": "localhost,127.0.0.1",
-                            "no_proxy": "localhost,127.0.0.1",
-                        }
+                    (ok, tenant_mounts, tenant_network, proxy_env,
+                     prep_err) = self._docker_task_context(task)
+                    if not ok:
+                        # 提前返回会跳过下面的 finally（清容器
+                        # 标记），这里先手动清掉，否则 _running
+                        # 里永远残留该任务，心跳与退出清理
+                        # 都会把它当在跑任务。
+                        self._running_clear(task_id)
+                        return ({"executor": "docker",
+                                 "task_id": task_id,
+                                 "status": "failed", "stdout": "",
+                                 "returncode": None, "host_status": None,
+                                 "stderr": ""},
+                                prep_err)
                     if goal_mode:
                         result = self.executor.goal_task(
                             task, timeout=effective_docker_timeout,
@@ -1484,12 +1598,14 @@ class Worker:
         actual_model = self.model
         fallback_model = None
         first_error = None
-        # 模型经任务的 model 字段传给执行器（执行器再经 host.py 的
-        # 顶层 --model 参数传给 dispatch/run）；不设时 host.py 用
-        # 配置文件里的 model。这里显式设成 worker 的主模型，让
-        # --model 覆盖配置真正生效，fallback 时再换成新模型。
+        # 模型优先级：任务级 > worker 主模型 > 配置文件（host.py
+        # 再兜底）。此前这里无条件写成 worker 主模型，导致控制
+        # 台选的模型被悄悄覆盖（任务级模型实际不生效）；修掉。
+        # fallback 时下面再把 task["model"] 换成 fallback 模型。
         task = dict(task)
-        task["model"] = self.model
+        if not task.get("model"):
+            task["model"] = self.model
+        actual_model = task["model"]
         # 有限重试与真实 fallback 的优先级：取消不重试；一次尝试失败
         # 且可 fallback、且显式配了 fallback_models 时，优先真换模型
         # 再跑一次（最多 1 次，同一任务、同一 worktree/分支，dispatch
@@ -1501,7 +1617,16 @@ class Worker:
         # 各次尝试的真实用量都要算进成本（失败尝试花的钱也是钱）：
         # 主尝试与 fallback 尝试的结果都先累加，最后只记一条汇总。
         usage_total = None
-        for attempt in range(self.max_retries + 1):
+        # 分支任务：先 fork 再跑首轮（见 _run_fork）；首轮失败
+        # 不自动重试/fallback，失败后走重试入口重跑首轮（已存在
+        # 续跑），不重复 fork。range(0) 让下面的尝试循环跳过，
+        # 直接进统一的收尾（超时/取消/成本/审计/通知）逻辑。
+        if task.get("forked_from"):
+            result, error = self._run_fork(task)
+            usage_total = sum_usage(usage_total,
+                                    (result or {}).get("usage"))
+        for attempt in range(0 if task.get("forked_from")
+                             else self.max_retries + 1):
             result, error = self._run_attempt(task)
             usage_total = sum_usage(usage_total, (result or {}).get("usage"))
             if self._was_cancelled(task_id) or \
@@ -1566,6 +1691,43 @@ class Worker:
             error = "task timeout after %s seconds" % self.task_timeout
             self._audit(task["project_id"], "worker_timeout", task_id)
 
+        # 后续消息（排队/直接发送）：首轮 completed 且非 goal 时，
+        # 按 direct 优先、队内先来先服务逐条再跑一轮（同一 thread，
+        # 历史与 worktree 改动都在）。直接发送在 POST 时已进链路，
+        # 这里只负责把轮次跑出来。一条失败即停：任务记 failed，
+        # 剩余 pending 留给重试入口（重试走已存在续跑，跑完继续
+        # 排空剩余消息）；取消时该条回 pending，任务走中断收尾。
+        if (result is not None and result.get("status") == "completed"
+                and not task.get("verify_command")):
+            while True:
+                fu = self._claim_next_followup(task_id)
+                if fu is None:
+                    break
+                fu_task = dict(task)
+                fu_task["prompt"] = fu["body"]
+                fu_result, fu_error = self._run_attempt(fu_task)
+                usage_total = sum_usage(
+                    usage_total, (fu_result or {}).get("usage"))
+                if self._was_cancelled(task_id) or \
+                        (fu_result or {}).get("status") == "cancelled":
+                    self._restore_followup(fu["id"])
+                    result = fu_result
+                    break
+                if (fu_result or {}).get("status") == "timeout":
+                    fu_result = dict(fu_result, status="failed")
+                    fu_error = ("task timeout after %s seconds"
+                                % self.task_timeout)
+                if (fu_result or {}).get("status") == "completed":
+                    self._finish_followup(fu["id"], "done")
+                    result = fu_result
+                    error = None
+                    continue
+                self._finish_followup(fu["id"], "failed")
+                result = fu_result
+                error = fu_error or self._attempt_error(
+                    fu_result, None, fu_result.get("host_status"))
+                break
+
         # 取消优先：执行期间被 cancel 的任务保持 interrupted，
         # 绝不用执行结果覆盖它（见模块 docstring 的取消说明）。
         if self._was_cancelled(task_id) or result.get("status") == "cancelled":