任务内标记工具：task-create/task-poll/task-read 移植、wait-threads 接通、新增 task-cancel/task-retry

- 新模块 cloud/marker_tools.py：纯逻辑解析、校验、快照格式化、
  控制门禁、派生代数与直接上限计数
- sandbox 两处解析提取五种标记块；worker 在 5 处投递点与
  _deliver_peer_messages 同款位置执行，报告排成后续消息下一轮回父任务
- wait 每块 cap 到 8 个目标（与服务端 /wait 对齐，spawn 原先的 10 个收紧），
  每轮最多 2 块；取消/重试门禁：同项目、非自己、非 goal、
  派生链或同创建人，另查 create_task / approve_merge 权限，
  查不到时收紧到仅派生链内
- goal 与协调员任务不解析；单测 30 个，cloud 全量 1043 个、4 跳过、0 失败

cloud/marker_tools.py

@@ -0,0 +1,534 @@
+"""任务内标记工具（marker tools）：task-create / task-poll / task-read /
+wait-threads / task-cancel / task-retry。
+
+设计（与 peer_messages、task_spawn 同款原则，不另起一套）：
+
+- agent 不用联网、不用凭证：agent 只在回合输出里写标记块，
+  可信的 worker 在 completed 轮次后解析并执行——本地与
+  Docker 执行器都通（两者都经 sandbox 的结果解析提取标记块，
+  与 task-message 同款）。
+- 异步语义：写块 → 回合结束 → worker 执行 → 报告拼成后续
+  消息、下一轮才回到父任务。不是同轮函数返回；轮询与等待
+  在 worker 执行期间循环查库、占 worker 槽位（与协调员等齐
+  同款代价），超时或父任务被取消时中止。
+- 只收不发：goal 任务与协调员任务不解析这些标记块。本批
+  工具是给“普通任务当项目经理、自己派生员工”那条路用的，
+  不是给协调员 PM 用的——协调员任务本身不跑模型回合，等齐
+  循环也不会在中途排空后续消息做判断。
+- 拒绝只记报告与审计、不炸任务、不改目标状态：非法 JSON、
+  超长、超上限、跨项目、自操作、已终态、goal、已合并、门禁
+  不过，全部按“跳过 + 报告里写清原因”处理。
+
+四个工具的分工，不要混：
+
+- task-create：新建一个平级主线任务（不是子代理，也不是
+  fork）。新任务 prompt 由标记块给，不继承父任务的历史；
+  环境基线继承父任务的 environment_id，但 worker 会给它独立
+  的 tenant/worktree，不动父任务的 cwd。父任务未提交的改动
+  不会自动带过去。
+- task-poll / task-read：按任务 id 轮询阅读。task-poll 按
+  interval 查库直到终态或超时；task-read 是同款别名，
+  不传 wait_seconds 时立即读一次快照，传了则等这么久再读
+  一次。read-thread / read_thread 是同款别名。
+- wait-threads：一次等多个目标任务，并行查库（每轮查一次
+  全部目标，不是逐个串行等超时相加），全部终态或超时后把
+  每目标快照一起排回父任务。
+- task-cancel / task-retry：对已存在的任务做取消 / 重试。
+  取消复用 worker 现有的取消路径语义（取消后状态为
+  interrupted）；重试与服务端 POST /tasks/{id}/retry 对齐
+  （仅 failed / interrupted / merge-failed 可重试，重试时
+  清空上一轮 result_text / error_text / actual_model、状态
+  回 queued、prompt 与模型等原样保留）。
+
+上限口径（本模块最终定的值，文档与代码一致）：
+
+- 每轮创建、轮询、取消、重试各最多 5 块；wait 每轮最多
+  2 块，每块最多 8 个目标。8 这个数是刻意与服务端
+  POST /tasks/{id}/wait 的“一次最多 8 个线程”对齐的；
+  spawn 分支原先的每块 10 个在此收紧到 8 个，不要再用 10。
+- 超时口径与服务端 /wait 不同，且是故意的：服务端 /wait
+  是给外部调用方持 HTTP 连接等的，上限 120 秒；标记 wait
+  是 worker 在回合后直连库循环查，不持 HTTP 连接，超时
+  钳制与 task-poll 同款（0-3600 秒，0 = 立即快照），占的
+  是 worker 槽位而不是 HTTP 连接。两者不要混用、不要互抄
+  上限。
+- 派生护栏：创建 prompt 1-20000 字，name 可选 1-100 字
+  （仅用于派生报告，tasks 表没有 name 列，不落库）；
+  model 不做白名单；effort 必须是 codex 的 effort 档位；
+  executor 必须是 local / docker / cloud，未指定时继承父任务；
+  派生深度上限 3 代、单个父任务直接派生上限 20 个。
+
+门禁（task-cancel / task-retry，纯函数层 + worker 权限层）：
+
+- 纯函数层（本模块 check_control_gate）：目标存在、同项目、
+  非自己、非 goal、状态符合 kind（取消要求非终态，重试
+  要求仅三态）、关系为派生链内或同创建人，二选一满足即可。
+  “同项目猜到 id 就能取消任何人的任务”在此被挡住。
+- worker 权限层（worker.py _deliver_marker_tools）：取消要求
+  发送方创建人有项目 create_task 权限（与服务端取消口径
+  同款），重试要求有 approve_merge 权限（与服务端重试口
+  径同款）。权限检查本身抛异常时收紧到仅派生链内可用，
+  同创建人-only 的目标一律拒绝，并把原因写进报告——绝不
+  在查不到权限时静默放宽。
+
+本模块只依赖标准库，不 import cloud 其它模块，避免循环
+import。DB 助手函数只用 ? 占位符，与 peer_messages 同款写法。
+"""
+
+import json
+import re
+import time
+import uuid
+
+MAX_CREATES_PER_TURN = 5
+MAX_POLLS_PER_TURN = 5
+MAX_WAITS_PER_TURN = 2
+MAX_WAIT_TARGETS_PER_BLOCK = 8
+MAX_CONTROLS_PER_TURN = 5
+MAX_PROMPT_CHARS = 20000
+MAX_NAME_CHARS = 100
+MAX_SPAWN_DEPTH = 3
+MAX_DIRECT_CHILDREN = 20
+DEFAULT_POLL_INTERVAL_SECONDS = 5
+DEFAULT_POLL_TIMEOUT_SECONDS = 300
+MAX_POLL_TIMEOUT_SECONDS = 3600
+DEFAULT_SNAPSHOT_CHARS = 2000
+MAX_SNAPSHOT_CHARS = 8000
+TERMINAL_STATUSES = ("completed", "failed", "interrupted", "merged",
+                     "merge-failed")
+RETRYABLE_STATUSES = ("failed", "interrupted", "merge-failed")
+EFFORTS = ("low", "medium", "high", "xhigh", "max")
+EXECUTORS = ("local", "docker", "cloud")
+
+_CREATE_RE = re.compile(r"```task-create\s*\n(.*?)\n```", re.DOTALL)
+_POLL_RE = re.compile(
+    r"```(task-poll|task_poll|task-read|task_read|read-thread|read_thread)"
+    r"\s*\n(.*?)\n```", re.DOTALL)
+_WAIT_RE = re.compile(r"```(wait-threads|wait_threads)\s*\n(.*?)\n```",
+                      re.DOTALL)
+_CANCEL_RE = re.compile(r"```(task-cancel|task_cancel)\s*\n(.*?)\n```",
+                        re.DOTALL)
+_RETRY_RE = re.compile(r"```(task-retry|task_retry)\s*\n(.*?)\n```",
+                       re.DOTALL)
+
+
+def row_get(row, name):
+    """与 peer_messages.row_get 同款容错取值。"""
+    try:
+        return row[name]
+    except (KeyError, IndexError, TypeError):
+        return None
+
+
+def _as_int(value, default):
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float) and value.is_integer():
+        return int(value)
+    return default
+
+
+def _clamp(value, low, high, default):
+    value = _as_int(value, default)
+    if value < low:
+        return low
+    if value > high:
+        return high
+    return value
+
+
+def extract_task_creates(text):
+    """提取 task-create 块；prompt 缺失/空串/超长、name 超长、
+    effort/executor 非法时整块跳过；每轮最多 5 块；正文里的围栏
+    会把块截断成非法 JSON 而被跳过（与 task-message 同款非贪婪
+    匹配）。"""
+    if not text:
+        return []
+    out = []
+    for match in _CREATE_RE.finditer(str(text)):
+        if len(out) >= MAX_CREATES_PER_TURN:
+            break
+        try:
+            payload = json.loads(match.group(1))
+        except (TypeError, ValueError):
+            continue
+        if not isinstance(payload, dict):
+            continue
+        prompt = payload.get("prompt")
+        if not isinstance(prompt, str) or not prompt.strip():
+            continue
+        prompt = prompt.strip()
+        if len(prompt) > MAX_PROMPT_CHARS:
+            continue
+        name = payload.get("name")
+        if name is not None:
+            if not isinstance(name, str) or not name.strip():
+                continue
+            name = name.strip()
+            if len(name) > MAX_NAME_CHARS:
+                continue
+        model = payload.get("model")
+        if model is not None:
+            if not isinstance(model, str) or not model.strip():
+                continue
+            model = model.strip()
+        effort = payload.get("effort")
+        if effort is not None and effort not in EFFORTS:
+            continue
+        executor = payload.get("executor")
+        if executor is not None and executor not in EXECUTORS:
+            continue
+        out.append({"name": name, "prompt": prompt, "model": model,
+                    "effort": effort, "executor": executor})
+    return out
+
+
+def extract_task_polls(text):
+    """提取 task-poll / task-read（含 read-thread 别名）块。
+
+    task-read 不传 wait_seconds 时立即读一次快照（timeout=0）；
+    传了则等这么久再读一次。interval 钳制 1-60 秒，timeout
+    钳制 0-3600 秒，chars 钳制 200-8000 字；各类块合计每轮
+    最多 5 块，多余的忽略。"""
+    if not text:
+        return []
+    out = []
+    for match in _POLL_RE.finditer(str(text)):
+        if len(out) >= MAX_POLLS_PER_TURN:
+            break
+        kind, raw = match.group(1), match.group(2)
+        try:
+            payload = json.loads(raw)
+        except (TypeError, ValueError):
+            continue
+        if not isinstance(payload, dict):
+            continue
+        task_id = payload.get("task_id") or payload.get("to") \
+            or payload.get("to_task_id")
+        if not isinstance(task_id, str) or not task_id.strip():
+            continue
+        chars = _clamp(payload.get("chars"), 200, MAX_SNAPSHOT_CHARS,
+                       DEFAULT_SNAPSHOT_CHARS)
+        if kind in ("task-read", "task_read", "read-thread",
+                    "read_thread"):
+            wait = payload.get("wait_seconds")
+            if wait is None:
+                interval = DEFAULT_POLL_INTERVAL_SECONDS
+                timeout = 0
+            else:
+                timeout = _clamp(wait, 0, MAX_POLL_TIMEOUT_SECONDS, 0)
+                interval = timeout if timeout > 0 \
+                    else DEFAULT_POLL_INTERVAL_SECONDS
+        else:
+            interval = _clamp(payload.get("interval_seconds"), 1, 60,
+                              DEFAULT_POLL_INTERVAL_SECONDS)
+            timeout = _clamp(payload.get("timeout_seconds"), 0,
+                             MAX_POLL_TIMEOUT_SECONDS,
+                             DEFAULT_POLL_TIMEOUT_SECONDS)
+        out.append({"task_id": task_id.strip(),
+                    "interval_seconds": interval,
+                    "timeout_seconds": timeout, "chars": chars})
+    return out
+
+
+def extract_wait_threads(text):
+    """提取 wait-threads / wait_threads 块：一次等多个目标。
+
+    块格式::
+
+        ```wait-threads
+        {"task_ids": ["B", "C"], "interval_seconds": 5,
+         "timeout_seconds": 300, "chars": 2000}
+        ```
+
+    task_ids 也接受 thread_ids / threads 两个别名。去重保序，
+    每块最多 8 个目标（与服务端 /wait 口径对齐，见模块头），
+    每轮最多 2 块；无有效 id 的块跳过。interval 钳制 1-60 秒，
+    timeout 钳制 0-3600 秒（worker 直连库口径，见模块头），
+    chars 钳制 200-8000 字。"""
+    if not text:
+        return []
+    out = []
+    for match in _WAIT_RE.finditer(str(text)):
+        if len(out) >= MAX_WAITS_PER_TURN:
+            break
+        try:
+            payload = json.loads(match.group(2))
+        except (TypeError, ValueError):
+            continue
+        if not isinstance(payload, dict):
+            continue
+        raw_ids = (payload.get("task_ids") or payload.get("thread_ids")
+                   or payload.get("threads"))
+        if not isinstance(raw_ids, list):
+            continue
+        ids = []
+        for tid in raw_ids:
+            if isinstance(tid, str) and tid.strip() \
+                    and tid.strip() not in ids:
+                ids.append(tid.strip())
+        if not ids:
+            continue
+        ids = ids[:MAX_WAIT_TARGETS_PER_BLOCK]
+        out.append({
+            "task_ids": ids,
+            "interval_seconds": _clamp(
+                payload.get("interval_seconds"), 1, 60,
+                DEFAULT_POLL_INTERVAL_SECONDS),
+            "timeout_seconds": _clamp(
+                payload.get("timeout_seconds"), 0,
+                MAX_POLL_TIMEOUT_SECONDS, DEFAULT_POLL_TIMEOUT_SECONDS),
+            "chars": _clamp(payload.get("chars"), 200,
+                            MAX_SNAPSHOT_CHARS, DEFAULT_SNAPSHOT_CHARS),
+        })
+    return out
+
+
+def _extract_controls(text, regex):
+    """提取 task-cancel / task-retry 块的公共逻辑。
+
+    task_id 缺失/空串时整块跳过；to / to_task_id / thread_id
+    是同款别名；每种每轮最多 5 块，多余的忽略。"""
+    if not text:
+        return []
+    out = []
+    for match in regex.finditer(str(text)):
+        if len(out) >= MAX_CONTROLS_PER_TURN:
+            break
+        try:
+            payload = json.loads(match.group(2))
+        except (TypeError, ValueError):
+            continue
+        if not isinstance(payload, dict):
+            continue
+        task_id = (payload.get("task_id") or payload.get("to")
+                   or payload.get("to_task_id")
+                   or payload.get("thread_id"))
+        if not isinstance(task_id, str) or not task_id.strip():
+            continue
+        out.append({"task_id": task_id.strip()})
+    return out
+
+
+def extract_task_cancels(text):
+    """提取 task-cancel 块：{"task_id": "..."}。"""
+    return _extract_controls(text, _CANCEL_RE)
+
+
+def extract_task_retries(text):
+    """提取 task-retry 块：{"task_id": "..."}。"""
+    return _extract_controls(text, _RETRY_RE)
+
+
+def parent_spawn_generation(conn, parent_id):
+    """父任务自身的派生代数：普通任务 0，第一代派生任务 1，依此类推。
+
+    只数 parent_task_id 非空且 coordinator_part 为空的祖先链（即
+    派生链）；协调员子任务（coordinator_part 非空）本身不是派生
+    任务，代数为 0，链到此为止。走 32 步仍未到头时按已数到的算，
+    绝不死循环。"""
+    generation = 0
+    cur = parent_id
+    for _ in range(32):
+        if not cur:
+            break
+        row = conn.execute(
+            "SELECT parent_task_id, coordinator_part FROM tasks"
+            " WHERE id = ?", (cur,)).fetchone()
+        if row is None:
+            break
+        parent = row_get(row, "parent_task_id")
+        part = row_get(row, "coordinator_part")
+        if parent is not None and part is None:
+            generation += 1
+            cur = parent
+            continue
+        break
+    return generation
+
+
+def count_direct_spawn_children(conn, parent_id):
+    """单个父任务的直接派生子任务数（不含协调员子任务）。"""
+    row = conn.execute(
+        "SELECT COUNT(*) FROM tasks WHERE parent_task_id = ?"
+        " AND coordinator_part IS NULL", (parent_id,)).fetchone()
+    try:
+        return int(row[0]) if row is not None else 0
+    except (TypeError, ValueError, IndexError):
+        return 0
+
+
+def fetch_task_snapshot(conn, task_id):
+    """取目标任务的快照行；不存在时回 None。
+
+    用 SELECT * 而不是列清单：老库/最小测试表缺列时 row_get
+    容错回 None，不炸；门禁要用的 created_by / parent_task_id /
+    coordinator_part / verify_command / goal_mode 一并带上。"""
+    row = conn.execute(
+        "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
+    if row is None:
+        return None
+    return {"id": row_get(row, "id"),
+            "project_id": row_get(row, "project_id"),
+            "status": row_get(row, "status"),
+            "result_text": row_get(row, "result_text"),
+            "error_text": row_get(row, "error_text"),
+            "executor": row_get(row, "executor"),
+            "model": row_get(row, "model"),
+            "actual_model": row_get(row, "actual_model"),
+            "turn_id": row_get(row, "turn_id"),
+            "last_heartbeat": row_get(row, "last_heartbeat"),
+            "worktree_host": row_get(row, "worktree_host"),
+            "pr_url": row_get(row, "pr_url"),
+            "pr_number": row_get(row, "pr_number"),
+            "environment_id": row_get(row, "environment_id"),
+            "created_by": row_get(row, "created_by"),
+            "parent_task_id": row_get(row, "parent_task_id"),
+            "coordinator_part": row_get(row, "coordinator_part"),
+            "verify_command": row_get(row, "verify_command"),
+            "goal_mode": row_get(row, "goal_mode")}
+
+
+def ancestry_relation(conn, sender, target):
+    """发送方与目标的关系：descendant / same_creator / other。
+
+    descendant：沿目标的 parent_task_id 链（最多 32 步）能追溯
+    到发送方（直接派生、隔代派生都算；协调员子任务的链到
+    协调员为止，不会穿过协调员再往上算——因为协调员子任务的
+    coordinator_part 非空，链在此处按派生链定义截断，见
+    parent_spawn_generation 的同款处理；但本函数走的是
+    parent_task_id 原始链，目标本身若是协调员子任务、其父
+    是协调员时，发送方若恰好是该协调员也会算 descendant——
+    协调员任务不解析标记块，所以发送方不可能是协调员，
+    这一支在实际调用里到不了，不影响门禁）。
+    same_creator：链上追不到，但目标 created_by 与发送方
+    created_by 相同且非空。
+    other：两者都不是。"""
+    sender_id = (sender or {}).get("id")
+    target_parent = (target or {}).get("parent_task_id")
+    cur = target_parent
+    for _ in range(32):
+        if not cur:
+            break
+        if cur == sender_id:
+            return "descendant"
+        row = conn.execute(
+            "SELECT parent_task_id FROM tasks WHERE id = ?",
+            (cur,)).fetchone()
+        if row is None:
+            break
+        cur = row_get(row, "parent_task_id")
+    sender_creator = (sender or {}).get("created_by")
+    target_creator = (target or {}).get("created_by")
+    if sender_creator and target_creator \
+            and sender_creator == target_creator:
+        return "same_creator"
+    return "other"
+
+
+def is_goal_task(task):
+    """goal 任务判定：verify_command 非空或 goal_mode 为真。
+
+    与 worker run_one 的 goal 分支同款（verify_command 或
+    goal_mode）；goal 字段本身是 goal_mode=1 时 prompt 的副本，
+    不单独当判定条件，避免把普通任务的 goal 文本误判。"""
+    if not task:
+        return False
+    return bool(task.get("verify_command") or task.get("goal_mode"))
+
+
+def check_control_gate(target, sender, relation, kind):
+    """task-cancel / task-retry 的纯函数门禁，返回 (ok, reason)。
+
+    kind 只能是 "cancel" 或 "retry"，其它值一律拒绝。权限层
+    不在这里：worker 侧另查 create_task / approve_merge，见
+    模块头；本函数只管目标状态与关系，单测不需要 AuthStore。"""
+    if kind not in ("cancel", "retry"):
+        return False, "未知的控制操作"
+    if target is None:
+        return False, "目标任务不存在"
+    if (target.get("project_id") is not None
+            and sender.get("project_id") is not None
+            and target.get("project_id") != sender.get("project_id")):
+        return False, "跨项目任务不可操作"
+    if target.get("id") == sender.get("id"):
+        return False, "不能操作自己"
+    if is_goal_task(target):
+        return False, "goal 任务不可取消或重试（goal 循环自有轮次）"
+    status = target.get("status")
+    if kind == "cancel":
+        if status in TERMINAL_STATUSES:
+            return False, "目标已终态（%s），不可取消" % status
+    else:
+        if status not in RETRYABLE_STATUSES:
+            return False, ("仅 failed / interrupted / merge-failed"
+                           " 可重试，当前状态为 %s" % status)
+    if relation not in ("descendant", "same_creator"):
+        return False, "目标不在发送方派生链上且创建人不同"
+    return True, ""
+
+
+def _heartbeat_age_text(snapshot, now=None):
+    """最后心跳距今秒数，无心跳时回空串（不编造）。"""
+    hb = snapshot.get("last_heartbeat")
+    if hb is None:
+        return ""
+    try:
+        age = int((now if now is None else now) - float(hb))
+    except (TypeError, ValueError):
+        return ""
+    if age < 0:
+        age = 0
+    return "%d 秒前" % age
+
+
+def _progress_lines(snapshot, now=None):
+    """快照的进度行：执行器、模型、工作树主机、心跳、turn。"""
+    lines = []
+    if snapshot.get("executor"):
+        lines.append("执行器：%s" % snapshot["executor"])
+    model = snapshot.get("actual_model") or snapshot.get("model")
+    if model:
+        lines.append("模型：%s" % model)
+    if snapshot.get("worktree_host"):
+        lines.append("工作树主机：%s" % snapshot["worktree_host"])
+    age = _heartbeat_age_text(snapshot, now=now)
+    if age:
+        lines.append("最后心跳：%s" % age)
+    if snapshot.get("turn_id"):
+        lines.append("回合：%s" % snapshot["turn_id"])
+    return lines
+
+
+def format_snapshot(snapshot, chars, timed_out=False, cancelled=False):
+    """把快照格式化成给父任务下一轮的报告正文（截断防上下文爆炸）。"""
+    task_id = snapshot["id"]
+    status = snapshot["status"] or "?"
+    lines = ["[任务 %s]" % task_id, "状态：%s" % status]
+    lines.extend(_progress_lines(snapshot))
+    if cancelled:
+        lines.append("说明：等待期间父任务被取消，轮询已中止。")
+    elif timed_out:
+        lines.append("说明：等待超时，目标任务仍未终态，可再轮询一次。")
+    elif status in TERMINAL_STATUSES:
+        result = (snapshot["result_text"] or "").strip()
+        error = (snapshot["error_text"] or "").strip()
+        if result:
+            lines.append("结果：%s" % result[:chars])
+        if error:
+            lines.append("错误：%s" % error[:chars])
+        if not result and not error:
+            lines.append("结果：（空）")
+    else:
+        lines.append("说明：目标任务仍在运行或排队，可再轮询一次。")
+    return "\n".join(lines)
+
+
+def new_child_id():
+    return uuid.uuid4().hex
+
+
+def marker_report_header():
+    return "[任务工具报告]"

cloud/sandbox.py

@@ -20,6 +20,11 @@ import shutil
 import subprocess
 import sys
 
+from cloud.marker_tools import extract_task_cancels as _extract_task_cancels
+from cloud.marker_tools import extract_task_creates as _extract_task_creates
+from cloud.marker_tools import extract_task_polls as _extract_task_polls
+from cloud.marker_tools import extract_task_retries as _extract_task_retries
+from cloud.marker_tools import extract_wait_threads as _extract_wait_threads
 from cloud.peer_messages import extract_peer_messages as _extract_peer_messages
 
 MAX_STDOUT_CHARS = 2000
@@ -280,6 +285,11 @@ class LocalExecutor(Executor):
         status = "failed"
         usage = None
         task_messages = []
+        task_creates = []
+        task_polls = []
+        wait_threads = []
+        task_cancels = []
+        task_retries = []
         turn_id = None
         root_turn_id = None
         if returncode == 0:
@@ -311,6 +321,13 @@ class LocalExecutor(Executor):
                 # 版 text 尾部的块会丢），worker 在 completed 轮次
                 # 后投递；goal 路径不走这里，goal 任务只收不发。
                 task_messages = _extract_peer_messages(parsed.get("text"))
+                # 标记工具块：与 task-message 同款从完整 text 里提取，
+                # worker 在 completed 轮次后执行；goal 路径不走这里。
+                task_creates = _extract_task_creates(parsed.get("text"))
+                task_polls = _extract_task_polls(parsed.get("text"))
+                wait_threads = _extract_wait_threads(parsed.get("text"))
+                task_cancels = _extract_task_cancels(parsed.get("text"))
+                task_retries = _extract_task_retries(parsed.get("text"))
             else:
                 host_status = None
         else:
@@ -328,6 +345,11 @@ class LocalExecutor(Executor):
             "stderr": _truncate_stdout(getattr(result, "stderr", "")),
             "usage": usage,
             "task_messages": task_messages,
+            "task_creates": task_creates,
+            "task_polls": task_polls,
+            "wait_threads": wait_threads,
+            "task_cancels": task_cancels,
+            "task_retries": task_retries,
             "turn_id": turn_id,
             "root_turn_id": root_turn_id,
         }
@@ -1570,6 +1592,11 @@ def _parse_host_result(task_id, fake):
     host_status = None
     usage = None
     task_messages = []
+    task_creates = []
+    task_polls = []
+    wait_threads = []
+    task_cancels = []
+    task_retries = []
     turn_id = None
     root_turn_id = None
     if returncode == 0:
@@ -1588,6 +1615,13 @@ def _parse_host_result(task_id, fake):
                       else "failed")
             usage = _clean_usage(parsed.get("usage"))
             task_messages = _extract_peer_messages(parsed.get("text"))
+            # 标记工具块：与 task-message 同款从完整 text 里提取，
+            # worker 在 completed 轮次后执行。
+            task_creates = _extract_task_creates(parsed.get("text"))
+            task_polls = _extract_task_polls(parsed.get("text"))
+            wait_threads = _extract_wait_threads(parsed.get("text"))
+            task_cancels = _extract_task_cancels(parsed.get("text"))
+            task_retries = _extract_task_retries(parsed.get("text"))
             # 任务列表显示 turn 用：与 Local 版同款语义，取不到
             # 时保持 None，worker 写回时不覆盖旧值。
             turn_id = parsed.get("turn_id")
@@ -1602,6 +1636,11 @@ def _parse_host_result(task_id, fake):
         "stderr": _truncate_stdout(getattr(fake, "stderr", "")),
         "usage": usage,
         "task_messages": task_messages,
+        "task_creates": task_creates,
+        "task_polls": task_polls,
+        "wait_threads": wait_threads,
+        "task_cancels": task_cancels,
+        "task_retries": task_retries,
         "turn_id": turn_id,
         "root_turn_id": root_turn_id,
     }

cloud/tests/test_marker_tools.py

@@ -0,0 +1,646 @@
+"""任务内标记工具单测：解析、门禁、快照格式化与 worker 投递。
+
+不跑真实模型、不联网：解析与门禁是纯函数；worker 投递用
+init_db 建的临时 SQLite 库 + 最小任务行，直接调
+_deliver_marker_tools，不走 run_one 全量流程。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    import cloud.marker_tools as mt
+    from cloud.marker_tools import (
+        ancestry_relation,
+        check_control_gate,
+        extract_task_cancels,
+        extract_task_creates,
+        extract_task_polls,
+        extract_task_retries,
+        extract_wait_threads,
+        format_snapshot,
+    )
+    from cloud.sandbox import LocalExecutor, _parse_host_result
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    import cloud.marker_tools as mt
+    from cloud.marker_tools import (
+        ancestry_relation,
+        check_control_gate,
+        extract_task_cancels,
+        extract_task_creates,
+        extract_task_polls,
+        extract_task_retries,
+        extract_wait_threads,
+        format_snapshot,
+    )
+    from cloud.sandbox import LocalExecutor, _parse_host_result
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+FAKE_ENV = {"XCM_API_KEY": "test-placeholder"}
+
+
+def _mem_conn():
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        "CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT,"
+        " created_by TEXT, status TEXT, parent_task_id TEXT,"
+        " coordinator_part TEXT, verify_command TEXT,"
+        " goal_mode INTEGER, result_text TEXT, error_text TEXT,"
+        " executor TEXT, model TEXT, actual_model TEXT,"
+        " turn_id TEXT, last_heartbeat REAL, worktree_host TEXT,"
+        " pr_url TEXT, pr_number INTEGER, environment_id TEXT)")
+    return conn
+
+
+def _mem_task(conn, task_id, project_id="p1", status="queued",
+              created_by="tester", parent=None, part=None,
+              verify_command=None, goal_mode=None, result=None):
+    conn.execute(
+        "INSERT INTO tasks(id, project_id, created_by, status,"
+        " parent_task_id, coordinator_part, verify_command, goal_mode,"
+        " result_text, error_text, executor, model, actual_model,"
+        " turn_id, last_heartbeat, worktree_host, pr_url, pr_number,"
+        " environment_id) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, NULL,"
+        " NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)",
+        (task_id, project_id, created_by, status, parent, part,
+         verify_command, goal_mode, result))
+    conn.commit()
+
+
+class ExtractTestCase(unittest.TestCase):
+    def test_create_basic_and_caps(self):
+        blocks = "".join(
+            "```task-create\n{\"name\": \"n%d\", \"prompt\": \"do %d\"}\n```\n"
+            % (i, i) for i in range(7))
+        creates = extract_task_creates(blocks)
+        self.assertEqual(len(creates), 5)
+        self.assertEqual(creates[0]["name"], "n0")
+        self.assertEqual(creates[0]["prompt"], "do 0")
+
+    def test_create_invalid_blocks_skipped(self):
+        text = (
+            "```task-create\nnot json\n```\n"
+            "```task-create\n{\"prompt\": \"  \"}\n```\n"
+            "```task-create\n{\"prompt\": \"" + "x" * 20001 + "\"}\n```\n"
+            "```task-create\n{\"prompt\": \"ok\", \"effort\": \"huge\"}\n```\n"
+            "```task-create\n{\"prompt\": \"ok\", \"executor\": \"mars\"}\n```\n"
+            "```task-create\n{\"prompt\": \"good\"}\n```\n")
+        creates = extract_task_creates(text)
+        self.assertEqual(len(creates), 1)
+        self.assertEqual(creates[0]["prompt"], "good")
+
+    def test_create_fence_at_line_start_truncates_to_invalid(self):
+        # 行首的围栏会把块截断成非法 JSON 而被跳过（非贪婪匹配）；
+        # 同行内的围栏不构成块边界，能正常解析（见下一断言的对照）。
+        text = ("```task-create\n{\"prompt\": \"a\n```\n b\"}\n```\n"
+                "```task-create\n{\"prompt\": \"good\"}\n```\n")
+        creates = extract_task_creates(text)
+        self.assertEqual([c["prompt"] for c in creates], ["good"])
+        inline = extract_task_creates(
+            "```task-create\n{\"prompt\": \"a ``` b\"}\n```\n")
+        self.assertEqual([c["prompt"] for c in inline], ["a ``` b"])
+
+    def test_poll_clamps_and_read_alias(self):
+        text = (
+            "```task-poll\n{\"task_id\": \"T1\", \"interval_seconds\": 999,"
+            " \"timeout_seconds\": 99999, \"chars\": 1}\n```\n"
+            "```task-read\n{\"task_id\": \"T2\"}\n```\n"
+            "```task-read\n{\"task_id\": \"T3\", \"wait_seconds\": 30}\n```\n")
+        polls = extract_task_polls(text)
+        self.assertEqual(len(polls), 3)
+        self.assertEqual(polls[0]["interval_seconds"], 60)
+        self.assertEqual(polls[0]["timeout_seconds"], 3600)
+        self.assertEqual(polls[0]["chars"], 200)
+        self.assertEqual(polls[1]["timeout_seconds"], 0)
+        self.assertEqual(polls[2]["timeout_seconds"], 30)
+        self.assertEqual(polls[2]["interval_seconds"], 30)
+
+    def test_poll_max_five_and_missing_id_skipped(self):
+        blocks = "".join(
+            "```task-poll\n{\"task_id\": \"T%d\"}\n```\n" % i
+            for i in range(6))
+        blocks += "```task-poll\n{\"interval_seconds\": 5}\n```\n"
+        self.assertEqual(len(extract_task_polls(blocks)), 5)
+
+    def test_wait_cap_eight_not_ten_and_max_two_blocks(self):
+        ids = ["T%d" % i for i in range(12)]
+        text = (
+            "```wait-threads\n%s\n```\n"
+            "```wait_threads\n{\"task_ids\": [\"A\"]}\n```\n"
+            "```wait-threads\n{\"task_ids\": [\"B\"]}\n```\n"
+            % json.dumps({"task_ids": ids}))
+        waits = extract_wait_threads(text)
+        # 每块最多 8 个（与服务端 /wait 对齐，不是 spawn 原先的 10 个），
+        # 每轮最多 2 块。
+        self.assertEqual(len(waits), 2)
+        self.assertEqual(len(waits[0]["task_ids"]), 8)
+        self.assertEqual(waits[0]["task_ids"][0], "T0")
+        self.assertEqual(waits[1]["task_ids"], ["A"])
+
+    def test_wait_invalid_blocks_skipped(self):
+        text = (
+            "```wait-threads\nnot json\n```\n"
+            "```wait-threads\n{\"task_ids\": []}\n```\n"
+            "```wait-threads\n{\"task_ids\": \"not-a-list\"}\n```\n"
+            "```wait-threads\n{\"task_ids\": [\"A\", \"A\", \"  \", \"B\"]}\n```\n")
+        waits = extract_wait_threads(text)
+        self.assertEqual(len(waits), 1)
+        self.assertEqual(waits[0]["task_ids"], ["A", "B"])
+
+    def test_control_extract_caps_and_aliases(self):
+        cancels = "".join(
+            "```task-cancel\n{\"task_id\": \"C%d\"}\n```\n" % i
+            for i in range(6))
+        self.assertEqual(len(extract_task_cancels(cancels)), 5)
+        retries = extract_task_retries(
+            "```task_retry\n{\"task_id\": \"R1\"}\n```\n"
+            "```task-retry\n{\"to_task_id\": \"R2\"}\n```\n"
+            "```task-retry\n{\"no_id\": true}\n```\n")
+        self.assertEqual([r["task_id"] for r in retries], ["R1", "R2"])
+
+
+class GateTestCase(unittest.TestCase):
+    def _sender(self, task_id="PM", project_id="p1", created_by="tester"):
+        return {"id": task_id, "project_id": project_id,
+                "created_by": created_by, "verify_command": None,
+                "goal_mode": None}
+
+    def _target(self, task_id="E1", project_id="p1", status="running",
+                created_by="tester", verify_command=None, goal_mode=None):
+        return {"id": task_id, "project_id": project_id, "status": status,
+                "created_by": created_by, "verify_command": verify_command,
+                "goal_mode": goal_mode}
+
+    def test_cancel_terminal_rejected(self):
+        for status in ("completed", "failed", "interrupted", "merged",
+                       "merge-failed"):
+            ok, _ = check_control_gate(
+                self._target(status=status), self._sender(),
+                "descendant", "cancel")
+            self.assertFalse(ok, status)
+
+    def test_retry_only_three_states(self):
+        for status in ("failed", "interrupted", "merge-failed"):
+            ok, _ = check_control_gate(
+                self._target(status=status), self._sender(),
+                "descendant", "retry")
+            self.assertTrue(ok, status)
+        for status in ("completed", "merged", "queued", "running"):
+            ok, _ = check_control_gate(
+                self._target(status=status), self._sender(),
+                "descendant", "retry")
+            self.assertFalse(ok, status)
+
+    def test_cross_project_self_goal_merged_rejected(self):
+        sender = self._sender()
+        ok, _ = check_control_gate(
+            self._target(project_id="p2"), sender, "descendant",
+            "cancel")
+        self.assertFalse(ok)
+        ok, _ = check_control_gate(
+            self._target(task_id="PM"), sender, "descendant", "cancel")
+        self.assertFalse(ok)
+        ok, _ = check_control_gate(
+            self._target(verify_command="make test"), sender,
+            "descendant", "cancel")
+        self.assertFalse(ok)
+        ok, _ = check_control_gate(
+            self._target(goal_mode=1), sender, "descendant", "retry")
+        self.assertFalse(ok)
+        ok, _ = check_control_gate(
+            self._target(status="merged"), sender, "descendant",
+            "cancel")
+        self.assertFalse(ok)
+
+    def test_relation_other_rejected_same_creator_passes(self):
+        ok, _ = check_control_gate(
+            self._target(created_by="other"), self._sender(), "other",
+            "cancel")
+        self.assertFalse(ok)
+        ok, _ = check_control_gate(
+            self._target(created_by="tester"), self._sender(),
+            "same_creator", "cancel")
+        self.assertTrue(ok)
+
+    def test_ancestry_chain(self):
+        conn = _mem_conn()
+        _mem_task(conn, "PM")
+        _mem_task(conn, "E1", parent="PM")
+        _mem_task(conn, "E2", parent="E1")
+        _mem_task(conn, "X1", created_by="other")
+        sender = {"id": "PM", "project_id": "p1", "created_by": "tester"}
+        target_e2 = mt.fetch_task_snapshot(conn, "E2")
+        self.assertEqual(ancestry_relation(conn, sender, target_e2),
+                         "descendant")
+        target_x = mt.fetch_task_snapshot(conn, "X1")
+        self.assertEqual(ancestry_relation(conn, sender, target_x),
+                         "other")
+        target_e1 = mt.fetch_task_snapshot(conn, "E1")
+        self.assertEqual(ancestry_relation(conn, sender, target_e1),
+                         "descendant")
+
+
+class SnapshotTestCase(unittest.TestCase):
+    def test_format_terminal_truncates(self):
+        snap = {"id": "T1", "status": "completed",
+                "result_text": "r" * 5000, "error_text": "",
+                "executor": "docker", "model": "m", "actual_model": None,
+                "turn_id": None, "last_heartbeat": None,
+                "worktree_host": None}
+        text = format_snapshot(snap, 200)
+        self.assertIn("状态：completed", text)
+        self.assertIn("r" * 200, text)
+        self.assertNotIn("r" * 201, text)
+
+    def test_format_nonterminal_timeout_cancelled(self):
+        base = {"id": "T1", "status": "running", "result_text": "",
+                "error_text": "", "executor": None, "model": None,
+                "actual_model": None, "turn_id": None,
+                "last_heartbeat": None, "worktree_host": None}
+        self.assertIn("仍在运行或排队",
+                      format_snapshot(base, 200))
+        self.assertIn("等待超时",
+                      format_snapshot(base, 200, timed_out=True))
+        self.assertIn("父任务被取消",
+                      format_snapshot(base, 200, cancelled=True))
+
+
+class SandboxExtractTestCase(unittest.TestCase):
+    def _text(self, *blocks):
+        return "\n".join(blocks)
+
+    def test_local_and_docker_extract_all_four(self):
+        text = self._text(
+            "```task-create\n{\"prompt\": \"do x\"}\n```",
+            "```task-poll\n{\"task_id\": \"T1\"}\n```",
+            "```wait-threads\n{\"task_ids\": [\"T1\", \"T2\"]}\n```",
+            "```task-cancel\n{\"task_id\": \"T2\"}\n```",
+            "```task-retry\n{\"task_id\": \"T3\"}\n```")
+        fake = SimpleNamespace(
+            returncode=0,
+            stdout=json.dumps({"status": "completed", "text": text}),
+            stderr="")
+        local = LocalExecutor("/fake/host.py")._result_from_runner(
+            "t1", fake)
+        self.assertEqual(len(local["task_creates"]), 1)
+        self.assertEqual(len(local["task_polls"]), 1)
+        self.assertEqual(len(local["wait_threads"]), 1)
+        self.assertEqual(len(local["task_cancels"]), 1)
+        self.assertEqual(len(local["task_retries"]), 1)
+        docker = _parse_host_result("t1", fake)
+        self.assertEqual(len(docker["task_creates"]), 1)
+        self.assertEqual(len(docker["wait_threads"]), 1)
+        self.assertEqual(len(docker["task_retries"]), 1)
+
+
+class WorkerDeliverTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self._insert("PM", status="running", created_by="tester")
+        self._member("tester", "maintainer")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _insert(self, task_id, status="queued", created_by="tester",
+                project_id="p1", parent=None, **cols):
+        now = time.time()
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " idempotency_key, depends_on, priority, created_by,"
+                " created_at, updated_at, last_heartbeat)"
+                " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)",
+                (task_id, project_id, "do work", status, created_by,
+                 now, now))
+            if parent is not None:
+                conn.execute(
+                    "UPDATE tasks SET parent_task_id = ? WHERE id = ?",
+                    (parent, task_id))
+            for key, value in cols.items():
+                conn.execute(
+                    "UPDATE tasks SET %s = ? WHERE id = ?" % key,
+                    (value, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _member(self, user_id, role):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY,"
+                " username TEXT UNIQUE, api_key_hash TEXT,"
+                " created_at REAL)")
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS project_members("
+                "project_id TEXT, user_id TEXT, role TEXT,"
+                " PRIMARY KEY(project_id, user_id))")
+            conn.execute(
+                "INSERT OR REPLACE INTO project_members"
+                "(project_id, user_id, role) VALUES('p1', ?, ?)",
+                (user_id, role))
+            conn.execute(
+                "INSERT OR IGNORE INTO users(id, username, api_key_hash,"
+                " created_at) VALUES(?, ?, '', 0)", (user_id, user_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _worker(self):
+        return Worker(self.db_path, "/fake/host.py", env=FAKE_ENV)
+
+    def _sender(self, worker):
+        return worker.get_task("PM")
+
+    def _followup_bodies(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            rows = conn.execute(
+                "SELECT body FROM task_followups"
+                " WHERE task_id = 'PM'").fetchall()
+            return [row[0] for row in rows]
+        finally:
+            conn.close()
+
+    def _status(self, task_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT status, result_text FROM tasks WHERE id = ?",
+                (task_id,)).fetchone()
+            return row[0], row[1]
+        finally:
+            conn.close()
+
+    def test_deliver_create_and_poll(self):
+        worker = self._worker()
+        sender = self._sender(worker)
+        worker._deliver_marker_tools(sender, {
+            "status": "completed",
+            "task_creates": [{"name": "E1", "prompt": "do e1",
+                              "model": None, "effort": None,
+                              "executor": None}],
+            "task_polls": [],
+            "wait_threads": [],
+            "task_cancels": [],
+            "task_retries": []})
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT id, parent_task_id, status, created_by FROM tasks"
+                " WHERE parent_task_id = 'PM'").fetchone()
+        finally:
+            conn.close()
+        self.assertIsNotNone(row)
+        self.assertEqual(row[1], "PM")
+        self.assertEqual(row[2], "queued")
+        self.assertEqual(row[3], "tester")
+
+    def test_deliver_poll_reports_snapshot(self):
+        self._insert("E1", status="completed", parent="PM",
+                     result_text="done-thing")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [{"task_id": "E1", "interval_seconds": 5,
+                            "timeout_seconds": 0, "chars": 2000}],
+            "wait_threads": [], "task_cancels": [], "task_retries": []})
+        bodies = self._followup_bodies()
+        self.assertEqual(len(bodies), 1)
+        self.assertIn("done-thing", bodies[0])
+
+    def test_deliver_wait_reports_all_targets(self):
+        self._insert("E1", status="completed", parent="PM",
+                     result_text="first-done")
+        self._insert("E2", status="queued", parent="PM")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [],
+            "wait_threads": [{"task_ids": ["E1", "E2"],
+                              "interval_seconds": 5,
+                              "timeout_seconds": 0, "chars": 2000}],
+            "task_cancels": [], "task_retries": []})
+        bodies = self._followup_bodies()
+        self.assertEqual(len(bodies), 1)
+        self.assertIn("E1", bodies[0])
+        self.assertIn("E2", bodies[0])
+        self.assertIn("first-done", bodies[0])
+
+    def test_deliver_cancel_descendant(self):
+        self._insert("E1", status="running", parent="PM")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [{"task_id": "E1"}],
+            "task_retries": []})
+        self.assertEqual(self._status("E1")[0], "interrupted")
+
+    def test_deliver_cancel_cross_project_and_terminal_rejected(self):
+        self._insert("X1", status="running", project_id="p2")
+        self._insert("E1", status="completed", parent="PM")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [{"task_id": "X1"}, {"task_id": "E1"}],
+            "task_retries": []})
+        self.assertEqual(self._status("X1")[0], "running")
+        self.assertEqual(self._status("E1")[0], "completed")
+        bodies = self._followup_bodies()
+        self.assertTrue(any("拒绝" in body for body in bodies))
+
+    def test_deliver_retry_clears_old_result(self):
+        self._insert("E1", status="failed", parent="PM",
+                     result_text="old-result", error_text="old-error")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET actual_model = 'm' WHERE id = 'E1'")
+            conn.commit()
+        finally:
+            conn.close()
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [],
+            "task_retries": [{"task_id": "E1"}]})
+        status, result = self._status("E1")
+        self.assertEqual(status, "queued")
+        self.assertIsNone(result)
+
+    def test_deliver_retry_denied_without_approve_merge(self):
+        self._member("tester", "viewer")
+        self._insert("E1", status="failed", parent="PM",
+                     result_text="old-result")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [],
+            "task_retries": [{"task_id": "E1"}]})
+        self.assertEqual(self._status("E1")[0], "failed")
+        bodies = self._followup_bodies()
+        self.assertTrue(any("拒绝" in body for body in bodies))
+
+    def test_deliver_goal_and_coordinator_sender_ignored(self):
+        self._insert("G1", status="running", verify_command="make test")
+        worker = self._worker()
+        goal_sender = worker.get_task("G1")
+        worker._deliver_marker_tools(goal_sender, {
+            "status": "completed",
+            "task_creates": [{"name": "E1", "prompt": "do e1",
+                              "model": None, "effort": None,
+                              "executor": None}],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [], "task_retries": []})
+        conn = sqlite3.connect(self.db_path)
+        try:
+            count = conn.execute(
+                "SELECT COUNT(*) FROM tasks"
+                " WHERE parent_task_id = 'G1'").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(count, 0)
+
+    def test_deliver_create_quota_exceeded_skips(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS user_quotas("
+                "user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT,"
+                " updated_at DOUBLE PRECISION)")
+            conn.execute(
+                "INSERT OR REPLACE INTO user_quotas"
+                "(user_id, quota_usd_micros, updated_at)"
+                " VALUES('tester', 0, 0)")
+            conn.commit()
+        finally:
+            conn.close()
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed",
+            "task_creates": [{"name": "E1", "prompt": "do e1",
+                              "model": None, "effort": None,
+                              "executor": None}],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [], "task_retries": []})
+        conn = sqlite3.connect(self.db_path)
+        try:
+            count = conn.execute(
+                "SELECT COUNT(*) FROM tasks"
+                " WHERE parent_task_id = 'PM'").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(count, 0)
+        bodies = self._followup_bodies()
+        self.assertTrue(any("配额" in body for body in bodies))
+
+    def test_deliver_create_direct_limit_rejected(self):
+        for i in range(20):
+            self._insert("OLD%d" % i, status="completed", parent="PM")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed",
+            "task_creates": [{"name": "E1", "prompt": "do e1",
+                              "model": None, "effort": None,
+                              "executor": None}],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [], "task_retries": []})
+        conn = sqlite3.connect(self.db_path)
+        try:
+            count = conn.execute(
+                "SELECT COUNT(*) FROM tasks"
+                " WHERE parent_task_id = 'PM'").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(count, 20)
+        bodies = self._followup_bodies()
+        self.assertTrue(any("上限" in body for body in bodies))
+
+    def test_deliver_create_depth_limit_rejected(self):
+        self._insert("A", status="completed", parent="PM")
+        self._insert("B", status="completed", parent="A")
+        self._insert("C", status="running", parent="B")
+        worker = self._worker()
+        sender_c = worker.get_task("C")
+        worker._deliver_marker_tools(sender_c, {
+            "status": "completed",
+            "task_creates": [{"name": "E1", "prompt": "do e1",
+                              "model": None, "effort": None,
+                              "executor": None}],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [], "task_retries": []})
+        conn = sqlite3.connect(self.db_path)
+        try:
+            count = conn.execute(
+                "SELECT COUNT(*) FROM tasks"
+                " WHERE parent_task_id = 'C'").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(count, 0)
+
+    def test_deliver_wait_timeout_reports_pending(self):
+        self._insert("E1", status="queued", parent="PM")
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "completed", "task_creates": [],
+            "task_polls": [],
+            "wait_threads": [{"task_ids": ["E1"],
+                              "interval_seconds": 1,
+                              "timeout_seconds": 1, "chars": 2000}],
+            "task_cancels": [], "task_retries": []})
+        bodies = self._followup_bodies()
+        self.assertEqual(len(bodies), 1)
+        self.assertIn("E1", bodies[0])
+        self.assertIn("等待超时", bodies[0])
+
+    def test_deliver_control_auth_failure_falls_back_to_descendant(self):
+        from unittest import mock
+        self._insert("E1", status="running", parent="PM")
+        self._insert("S1", status="running")
+        worker = self._worker()
+        with mock.patch("cloud.worker.AuthStore",
+                        side_effect=RuntimeError("auth down")):
+            worker._deliver_marker_tools(self._sender(worker), {
+                "status": "completed", "task_creates": [],
+                "task_polls": [], "wait_threads": [],
+                "task_cancels": [{"task_id": "E1"},
+                                 {"task_id": "S1"}],
+                "task_retries": []})
+        # 权限查不到时：派生链内仍可取消，同创建人-only 被拒绝。
+        self.assertEqual(self._status("E1")[0], "interrupted")
+        self.assertEqual(self._status("S1")[0], "running")
+
+    def test_deliver_non_completed_round_ignored(self):
+        worker = self._worker()
+        worker._deliver_marker_tools(self._sender(worker), {
+            "status": "failed",
+            "task_creates": [{"name": "E1", "prompt": "do e1",
+                              "model": None, "effort": None,
+                              "executor": None}],
+            "task_polls": [], "wait_threads": [],
+            "task_cancels": [], "task_retries": []})
+        self.assertEqual(self._followup_bodies(), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -148,7 +148,19 @@ from cloud.models import ModelRouter, is_fallbackable, record_fallback
 from cloud.auto_memory import AutoMemoryStore
 from cloud.auto_memory import build_injection_prefix as _auto_prefix
 from cloud.auto_memory import process_due_tasks as _auto_process_due
+from cloud.marker_tools import ancestry_relation as _marker_ancestry
+from cloud.marker_tools import check_control_gate as _marker_gate
+from cloud.marker_tools import count_direct_spawn_children as _marker_count_direct
+from cloud.marker_tools import fetch_task_snapshot as _marker_snapshot
+from cloud.marker_tools import format_snapshot as _marker_format
+from cloud.marker_tools import marker_report_header as _marker_header
+from cloud.marker_tools import new_child_id as _marker_new_id
+from cloud.marker_tools import parent_spawn_generation as _marker_generation
+from cloud.marker_tools import MAX_DIRECT_CHILDREN as _MARKER_MAX_DIRECT
+from cloud.marker_tools import MAX_SPAWN_DEPTH as _MARKER_MAX_DEPTH
+from cloud.marker_tools import TERMINAL_STATUSES as _MARKER_TERMINAL
 from cloud.peer_messages import deliver_peer_message as _deliver_peer_message
+from cloud.peer_messages import ensure_followups_table as _ensure_followups_table
 from cloud.peer_messages import peer_message_prefix as _peer_prefix
 from cloud.peer_messages import row_get as _row_get
 from cloud.sub2api import ensure_user_gateway_key as _ensure_sub2api_key
@@ -2303,6 +2315,382 @@ class Worker:
         except Exception:
             pass
 
+    def _queue_self_followup(self, task, body):
+        """给父任务自己排一条系统后续消息（标记工具报告 / 轮询快照）。
+
+        sender_task_id 置空，排空时不加“来自任务 X 的消息”前缀，
+        正文自带 [任务工具报告] 抬头即可区分。排空循环（run_one
+        的后续消息段）会在轮次边界认领它：父任务运行中直接跑下
+        一轮，父任务已终态时下次认领走续跑——与任务间消息的复活
+        语义同款，只是不经过 peer 的“不能发给自己”拒绝。"""
+        try:
+            conn = _db_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, sender_task_id,"
+                    " created_at, updated_at)"
+                    " VALUES(?, ?, ?, ?, 'queued', 'pending', ?, NULL,"
+                    " ?, ?)",
+                    (uuid.uuid4().hex, task["id"],
+                     task["project_id"], body, task.get("created_by"),
+                     time.time(), time.time()),
+                )
+                conn.commit()
+                return True
+            finally:
+                conn.close()
+        except Exception:
+            return False
+
+    def _poll_task_snapshot(self, task, poll):
+        """轮询一个目标任务：每 interval 查一次库状态，直到终态、
+        超时或父任务被取消。等待期间不烧模型费，但占着本 worker
+        的槽位（与协调员等齐同款代价）。返回报告正文。"""
+        target_id = poll["task_id"]
+        interval = poll["interval_seconds"]
+        timeout = poll["timeout_seconds"]
+        chars = poll["chars"]
+        started = time.time()
+        snapshot = None
+        timed_out = False
+        cancelled = False
+        while True:
+            if self._was_cancelled(task["id"]):
+                cancelled = True
+                break
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    snapshot = _marker_snapshot(conn, target_id)
+                finally:
+                    conn.close()
+            except Exception:
+                snapshot = None
+            if snapshot is None:
+                return "轮询任务 %s：不存在（同项目内也查不到）。" % target_id
+            if snapshot["project_id"] != task["project_id"]:
+                return ("轮询任务 %s：拒绝，跨项目任务不可读。"
+                        % target_id)
+            if target_id == task["id"]:
+                return "轮询任务 %s：拒绝，不能轮询自己。" % target_id
+            if snapshot["status"] in _MARKER_TERMINAL:
+                break
+            if timeout <= 0:
+                break
+            elapsed = time.time() - started
+            if elapsed >= timeout:
+                timed_out = True
+                break
+            time.sleep(min(interval, max(timeout - elapsed, 0.05)))
+        if snapshot is None:
+            return "轮询任务 %s：读取失败。" % target_id
+        return _marker_format(snapshot, chars, timed_out=timed_out,
+                              cancelled=cancelled)
+
+    def _wait_task_targets(self, task, wait):
+        """等一批目标任务：每 interval 并行查一次全部目标，直到
+        全部终态、超时或父任务被取消。缺失、跨项目、自等待的
+        目标当场记为已处理（写清原因），不让它们把整批拖到超时。
+        轮询/等待目标是只读的，goal 任务也可以等；取消与重试
+        的 goal 拒绝只在控制门禁里，不在这里。返回报告行列表。"""
+        target_ids = wait["task_ids"]
+        interval = wait["interval_seconds"]
+        timeout = wait["timeout_seconds"]
+        chars = wait["chars"]
+        started = time.time()
+        timed_out = False
+        cancelled = False
+        resolved = {}  # task_id -> ("snapshot", snap) | ("note", text)
+        latest = {}  # task_id -> 最近一次快照（超时报告用）
+        while True:
+            if self._was_cancelled(task["id"]):
+                cancelled = True
+                break
+            pending = [tid for tid in target_ids if tid not in resolved]
+            if not pending:
+                break
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    for tid in pending:
+                        snap = _marker_snapshot(conn, tid)
+                        if snap is None:
+                            resolved[tid] = (
+                                "note",
+                                "等待任务 %s：不存在（同项目内也查不到）。"
+                                % tid)
+                        elif snap["project_id"] != task["project_id"]:
+                            resolved[tid] = (
+                                "note",
+                                "等待任务 %s：拒绝，跨项目任务不可读。"
+                                % tid)
+                        elif tid == task["id"]:
+                            resolved[tid] = (
+                                "note",
+                                "等待任务 %s：拒绝，不能等待自己。" % tid)
+                        else:
+                            latest[tid] = snap
+                            if snap["status"] in _MARKER_TERMINAL:
+                                resolved[tid] = ("snapshot", snap)
+                finally:
+                    conn.close()
+            except Exception:
+                pass
+            if all(tid in resolved for tid in target_ids):
+                break
+            if timeout <= 0:
+                break
+            elapsed = time.time() - started
+            if elapsed >= timeout:
+                timed_out = True
+                break
+            time.sleep(min(interval, max(timeout - elapsed, 0.05)))
+        lines = ["[等待 %d 个任务]" % len(target_ids)]
+        for tid in target_ids:
+            kind, value = resolved.get(tid, (None, None))
+            if kind == "note":
+                lines.append(value)
+            elif kind == "snapshot":
+                lines.append(_marker_format(value, chars))
+            else:
+                snap = latest.get(tid)
+                if snap is None:
+                    lines.append("等待任务 %s：读取失败。" % tid)
+                else:
+                    lines.append(_marker_format(
+                        snap, chars, timed_out=timed_out,
+                        cancelled=cancelled))
+        if cancelled:
+            lines.append("说明：等待期间父任务被取消，等待已中止。")
+        elif timed_out:
+            lines.append("说明：等待超时，仍有目标未终态，可再等一次。")
+        return lines
+
+    def _marker_auth(self, task, action):
+        """标记控制操作的权限层，返回 (ok_or_None, reason)。
+
+        ok=True：发送方创建人有该权限；False：明确无权限；
+        None：权限检查本身抛异常（查不到），调用方必须收紧到
+        仅派生链内可用，绝不能静默放宽。"""
+        creator = task.get("created_by")
+        if not creator:
+            return False, "发送方无创建人，无法查权限"
+        try:
+            store = AuthStore(self.db_path)
+            try:
+                ok = store.check_permission(
+                    creator, task["project_id"], action)
+            finally:
+                try:
+                    store.conn.close()
+                except Exception:
+                    pass
+        except Exception as exc:
+            return None, "权限检查失败：%s" % exc
+        return bool(ok), ""
+
+    def _deliver_marker_tools(self, task, result):
+        """执行本轮 agent 输出里的标记工具块（见 marker_tools）。
+
+        只执行 completed 轮次的块；goal 任务与协调员任务不解析
+        （只收不发，与 peer 同款）；父任务被取消时不执行。
+        执行顺序：创建 → 取消 → 重试 → 轮询 → 等待——创建在前
+        让同一轮刚创建的任务立刻能被轮询，控制操作在快照之前
+        让轮询/等待读到取消或重试后的新状态。全部报告拼成一条
+        系统后续消息排给父任务自己，由排空循环在轮次边界跑下
+        一轮；拒绝只记报告与审计，不炸 run_one。"""
+        if not result or result.get("status") != "completed":
+            return
+        if self._was_cancelled(task["id"]):
+            return
+        if task.get("verify_command") or task.get("goal_mode"):
+            return
+        if task.get("coordinator_parts"):
+            return
+        creates = result.get("task_creates") or []
+        polls = result.get("task_polls") or []
+        waits = result.get("wait_threads") or []
+        cancels = result.get("task_cancels") or []
+        retries = result.get("task_retries") or []
+        if not (creates or polls or waits or cancels or retries):
+            return
+        reports = []
+        created = 0
+        rejected = 0
+        if creates:
+            quota_error = None
+            try:
+                quota_error = self._quota_exceeded(task, None)
+            except Exception as exc:
+                quota_error = "quota_check_failed: %s" % exc
+            if quota_error is not None:
+                reports.append("创建跳过：父任务配额检查未通过（%s），"
+                               "本轮 %d 个创建块全部跳过。"
+                               % (quota_error, len(creates)))
+                rejected += len(creates)
+            else:
+                try:
+                    conn = _db_connect(self.db_path)
+                    try:
+                        self._ensure_result_columns(conn)
+                        generation = _marker_generation(conn, task["id"])
+                        direct = _marker_count_direct(conn, task["id"])
+                        now = time.time()
+                        for create in creates:
+                            new_depth = generation + 1
+                            if new_depth > _MARKER_MAX_DEPTH:
+                                reports.append(
+                                    "创建跳过：派生深度超限（上限 %d 代）。"
+                                    % _MARKER_MAX_DEPTH)
+                                rejected += 1
+                                continue
+                            if direct >= _MARKER_MAX_DIRECT:
+                                reports.append(
+                                    "创建跳过：直接派生子任务已达上限 %d 个。"
+                                    % _MARKER_MAX_DIRECT)
+                                rejected += 1
+                                continue
+                            child_id = _marker_new_id()
+                            conn.execute(
+                                "INSERT INTO tasks(id, project_id, prompt,"
+                                " status, idempotency_key, depends_on,"
+                                " priority, created_by, created_at,"
+                                " updated_at, last_heartbeat, executor,"
+                                " model, effort, environment_id,"
+                                " parent_task_id, coordinator_part)"
+                                " VALUES(?, ?, ?, 'queued', NULL, '[]',"
+                                " ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, NULL)",
+                                (child_id, task["project_id"],
+                                 create["prompt"],
+                                 task.get("priority", 0) or 0,
+                                 task.get("created_by"), now, now,
+                                 create["executor"]
+                                 or task.get("executor"),
+                                 create["model"] or task.get("model"),
+                                 create["effort"] or task.get("effort"),
+                                 task.get("environment_id"),
+                                 task["id"]),
+                            )
+                            direct += 1
+                            created += 1
+                            label = create["name"] or "未命名"
+                            reports.append(
+                                "已创建任务 %s（%s），状态 queued，"
+                                "可用 task-poll 按该 id 轮询。"
+                                % (child_id, label))
+                        conn.commit()
+                    finally:
+                        conn.close()
+                except Exception as exc:
+                    reports.append("创建失败：%s" % exc)
+                    rejected += len(creates) - created
+        for kind, blocks, action in (("取消", cancels, "create_task"),
+                                     ("重试", retries, "approve_merge")):
+            for block in blocks:
+                target_id = block["task_id"]
+                try:
+                    target = self.get_task(target_id)
+                except Exception as exc:
+                    target = None
+                    reports.append("%s任务 %s：读取失败（%s），已跳过。"
+                                   % (kind, target_id, exc))
+                    rejected += 1
+                    continue
+                relation = "other"
+                if target is not None:
+                    try:
+                        conn = _db_connect(self.db_path)
+                        try:
+                            relation = _marker_ancestry(
+                                conn, task, target)
+                        finally:
+                            conn.close()
+                    except Exception:
+                        relation = "other"
+                gate_kind = "cancel" if kind == "取消" else "retry"
+                ok, reason = _marker_gate(target, task, relation,
+                                          gate_kind)
+                if not ok:
+                    reports.append("%s任务 %s：拒绝（%s）。"
+                                   % (kind, target_id, reason))
+                    rejected += 1
+                    continue
+                auth_ok, auth_reason = self._marker_auth(task, action)
+                if auth_ok is False:
+                    reports.append("%s任务 %s：拒绝（发送方创建人无"
+                                   " %s 权限）。"
+                                   % (kind, target_id, action))
+                    rejected += 1
+                    continue
+                if auth_ok is None and relation != "descendant":
+                    reports.append("%s任务 %s：拒绝（%s，仅派生链内"
+                                   "可用）。"
+                                   % (kind, target_id, auth_reason))
+                    rejected += 1
+                    continue
+                try:
+                    if kind == "取消":
+                        done = self.cancel(target_id)
+                        if done:
+                            reports.append("已取消任务 %s，状态为"
+                                           " interrupted。" % target_id)
+                        else:
+                            reports.append("取消任务 %s：未执行（目标"
+                                           "状态已变或不存在）。"
+                                           % target_id)
+                            rejected += 1
+                    else:
+                        conn = _db_connect(self.db_path)
+                        try:
+                            cur = conn.execute(
+                                "UPDATE tasks SET status = 'queued',"
+                                " result_text = NULL, error_text = NULL,"
+                                " actual_model = NULL, updated_at = ?"
+                                " WHERE id = ? AND status IN"
+                                " ('failed', 'interrupted',"
+                                " 'merge-failed')",
+                                (time.time(), target_id))
+                            conn.commit()
+                            done = cur.rowcount == 1
+                        finally:
+                            conn.close()
+                        if done:
+                            self._audit(task["project_id"],
+                                        "worker_marker_retry",
+                                        target_id)
+                            reports.append("已重试任务 %s，状态回"
+                                           " queued，上一轮结果已清空。"
+                                           % target_id)
+                        else:
+                            reports.append("重试任务 %s：未执行（目标"
+                                           "状态已变）。" % target_id)
+                            rejected += 1
+                except Exception as exc:
+                    reports.append("%s任务 %s：执行失败（%s）。"
+                                   % (kind, target_id, exc))
+                    rejected += 1
+        for poll in polls:
+            try:
+                reports.append(self._poll_task_snapshot(task, poll))
+            except Exception as exc:
+                reports.append("轮询任务 %s 失败：%s"
+                               % (poll["task_id"], exc))
+        for wait in waits:
+            try:
+                reports.extend(self._wait_task_targets(task, wait))
+            except Exception as exc:
+                reports.append("等待任务失败：%s" % exc)
+        if reports:
+            body = _marker_header() + "\n" + "\n".join(reports)
+            self._queue_self_followup(task, body)
+        detail = ("%s: spawned=%d rejected=%d polls=%d waits=%d "
+                  "cancels=%d retries=%d"
+                  % (task["id"], created, rejected, len(polls),
+                     len(waits), len(cancels), len(retries)))
+        self._audit(task["project_id"], "worker_marker_tools", detail)
+
     def _deliver_peer_messages(self, task, result):
         """投递本轮 agent 输出里的互发消息标记块（见 peer_messages）。
 
@@ -3509,6 +3897,7 @@ class Worker:
             usage_total = sum_usage(usage_total,
                                     (result or {}).get("usage"))
             self._deliver_peer_messages(task, result)
+            self._deliver_marker_tools(task, result)
             if (result or {}).get("status") == "quota_exceeded":
                 quota_error = self._quota_exceeded(task, usage_total) or \
                     "quota_exceeded: 本月配额已用完"
@@ -3526,6 +3915,7 @@ class Worker:
             result, error = self._run_attempt(task, quota_check=_quota_hit)
             usage_total = sum_usage(usage_total, (result or {}).get("usage"))
             self._deliver_peer_messages(task, result)
+            self._deliver_marker_tools(task, result)
             if (result or {}).get("status") == "quota_exceeded":
                 # 轮询掐断：进程已被杀，配额原因以现查为准。
                 quota_error = self._quota_exceeded(task, usage_total) or \
@@ -3568,6 +3958,7 @@ class Worker:
                         usage_total = sum_usage(
                             usage_total, (result or {}).get("usage"))
                         self._deliver_peer_messages(task, result)
+                        self._deliver_marker_tools(task, result)
                         if (result or {}).get("status") == "quota_exceeded":
                             quota_error = self._quota_exceeded(
                                 task, usage_total) or \
@@ -3611,6 +4002,7 @@ class Worker:
                 usage_total = sum_usage(usage_total,
                                         (result or {}).get("usage"))
                 self._deliver_peer_messages(task, result)
+                self._deliver_marker_tools(task, result)
                 if (result or {}).get("status") == "quota_exceeded":
                     quota_error = self._quota_exceeded(task, usage_total) or \
                         "quota_exceeded: 本月配额已用完"
@@ -3673,6 +4065,7 @@ class Worker:
                 usage_total = sum_usage(
                     usage_total, (fu_result or {}).get("usage"))
                 self._deliver_peer_messages(task, fu_result)
+                self._deliver_marker_tools(task, fu_result)
                 if (fu_result or {}).get("status") == "quota_exceeded":
                     self._restore_followup(fu["id"])
                     quota_error = self._quota_exceeded(task, usage_total) or \

docs/marker-tools.md

@@ -0,0 +1,207 @@
+# 任务内标记工具（marker tools）
+
+适用基线：main d129f7b 上的 feat-marker-tools 分支（本地提交，未推、
+未建 PR、未部署）。本批只做四类标记工具，不要把本文档读成
+“所有 Codex app 线程工具都已实现”：handoff、worktree 创建状态、
+PR 检查诊断、automation、clock-sleep 都不在本批；MCP 真工具路
+（同轮函数返回）也不在本批。
+
+## 一句话原理
+
+agent 不用联网、不用凭证。agent 在回合输出里写标记块，本地与
+Docker 执行器都经 sandbox 的结果解析提取标记块（与 task-message
+同款两处解析），可信的 worker 在 completed 轮次后执行。
+
+异步语义（必须先讲清，否则会被当成函数调用）：
+
+- 写块 → 回合结束 → worker 执行 → 报告拼成一条后续消息、
+  下一轮才回到父任务。不是同轮函数返回。
+- 轮询与等待在 worker 执行期间循环查库、占 worker 槽位（与
+  协调员等齐同款代价）；超时或父任务被取消时中止。
+- 只有 completed 轮次的块被执行；失败 / 超时 / 取消 / 配额中断
+  的轮次不执行；父任务执行期间被取消时也不执行。
+- 拒绝只记报告与审计、不炸任务、不改目标状态。
+
+## 边界：谁能用、谁不能用
+
+- 能用：普通任务。普通任务当项目经理、自己派生员工、轮询员工、
+  等一批员工、取消或重试员工，走的就是这批工具。
+- 不能用：goal 任务与协调员任务不解析这些标记块（只收不发）。
+  协调员任务本身不跑模型回合，等齐循环也不会在中途排空后续
+  消息做判断，所以协调员 PM 不能因此获得任务内工具。
+- 容易混淆的一点：协调员的子任务本身是普通任务（不是协调员
+  任务），它写标记块会被解析；它的派生代数从 0 计（见“派生护栏”）。
+
+## 四个工具的标记格式
+
+### task-create：新建平级主线任务
+
+```text
+```task-create
+{"name": "test-xxx", "prompt": "...", "model": null,
+ "effort": null, "executor": "docker"}
+```
+```
+
+- 新任务是平级主线，不是子代理、不是 fork；不继承父任务历史；
+  环境基线继承父任务的 environment_id，但 worker 给它独立的
+  tenant/worktree，不动父任务的 cwd；父任务未提交的改动不会
+  自动带过去。
+- prompt 必填、1-20000 字；name 可选、1-100 字（仅用于派生报告，
+  tasks 表没有 name 列，不落库、不改 prompt）；model 不做白名单；
+  effort 只能是 low / medium / high / xhigh / max；executor 只能是
+  local / docker / cloud，未指定时继承父任务。
+- 只建普通任务：不通过标记建协调员任务与 goal 任务。
+- 每轮最多 5 块，多余的忽略；非法 JSON、prompt 缺失/空串/超长、
+  name 超长、effort/executor 非法时整块跳过；正文里的行首围栏
+  会把块截断成非法 JSON 而被跳过（与 task-message 同款非贪婪匹配）。
+
+### task-poll / task-read：按任务 id 轮询阅读
+
+```text
+```task-poll
+{"task_id": "<目标任务 id>", "interval_seconds": 5,
+ "timeout_seconds": 300, "chars": 2000}
+```
+
+```task-read
+{"task_id": "<目标任务 id>", "wait_seconds": 30}
+```
+```
+
+- task-read 是 task-poll 的别名；read-thread / read_thread 也是
+  同款别名。不传 wait_seconds 时立即读一次快照；传了则等这么久
+  再读一次（等价于 interval = timeout = wait_seconds）。
+- interval 钳制 1-60 秒，timeout 钳制 0-3600 秒（0 = 立即快照），
+  chars 钳制 200-8000 字；各类块合计每轮最多 5 块。
+- 轮询目标不存在、跨项目、轮询自己时，报告里写清拒绝原因，
+  不改任何状态。轮询目标是只读的，goal 任务也可以被轮询。
+
+### wait-threads：一次等多个目标任务
+
+```text
+```wait-threads
+{"task_ids": ["B", "C"], "interval_seconds": 5,
+ "timeout_seconds": 300, "chars": 2000}
+```
+```
+
+- wait_threads 是同款写法；task_ids 也接受 thread_ids / threads
+  两个别名；去重保序。
+- 每块最多 8 个目标，每轮最多 2 块。8 这个数是刻意与服务端
+  POST /tasks/{id}/wait 的“一次最多 8 个线程”对齐的；spawn
+  分支原先的每块 10 个在本批收紧到 8 个，不要再用 10。
+- 并行查库：每轮查一次全部目标，不是逐个串行等超时相加；
+  全部终态或超时后把每目标快照一起排回父任务。
+- 缺失、跨项目、等待自己的目标当场记为已处理（写清原因），
+  不让它们把整批拖到超时；其余目标继续等到终态或超时。
+- 超时口径与服务端 /wait 不同，且是故意的：服务端 /wait 是给
+  外部调用方持 HTTP 连接等的，上限 120 秒；标记 wait 是 worker
+  在回合后直连库循环查，不持 HTTP 连接，超时钳制与 task-poll
+  同款（0-3600 秒），占的是 worker 槽位而不是 HTTP 连接。
+  两者不要混用、不要互抄上限。
+
+### task-cancel / task-retry：取消 / 重试已存在的任务
+
+```text
+```task-cancel
+{"task_id": "<目标任务 id>"}
+```
+
+```task-retry
+{"task_id": "<目标任务 id>"}
+```
+```
+
+- task_cancel / task_retry 是同款写法；task_id 也接受 to /
+  to_task_id / thread_id 别名；每种每轮最多 5 块。
+- 取消复用 worker 现有的取消路径语义：取消后状态为 interrupted，
+  正在跑的进程组 / 容器按现有取消路径真杀，不只改库状态。
+- 重试与服务端 POST /tasks/{id}/retry 对齐：仅 failed /
+  interrupted / merge-failed 可重试，重试时清空上一轮
+  result_text / error_text / actual_model、状态回 queued、
+  prompt 与模型等原样保留、记审计。重试本身不花模型钱，
+  重试后的新一轮照常计费。
+- 执行顺序：创建 → 取消 → 重试 → 轮询 → 等待。创建在前让同一轮
+  刚创建的任务立刻能被轮询；控制操作在快照之前让轮询/等待读到
+  取消或重试后的新状态。
+
+## 门禁（task-cancel / task-retry，缺一不可）
+
+纯函数层（cloud/marker_tools.py check_control_gate）与 worker
+权限层（worker.py _deliver_marker_tools）两道，拒绝时只记报告
+与审计、不改目标状态：
+
+1. 目标存在；同项目；不是自己。
+2. 非 goal 任务（verify_command 非空或 goal_mode 为真都算 goal）。
+3. 取消要求目标非终态；重试要求目标仅为 failed / interrupted /
+   merge-failed。已合并（merged）两者都拒绝；merge-failed 本身
+   是终态，所以取消拒绝、重试接受。
+4. 关系二选一：目标在发送方派生链上（沿 parent_task_id 链最多
+   32 步能追溯到发送方），或目标 created_by 与发送方 created_by
+   相同。“同项目猜到 id 就能取消任何人的任务”在此被挡住。
+5. 权限层：取消要求发送方创建人有项目 create_task 权限（与
+   服务端取消口径同款），重试要求有 approve_merge 权限（与
+   服务端重试口径同款）。权限检查本身抛异常时收紧到仅派生链内
+   可用，同创建人-only 的目标一律拒绝，并把原因写进报告。
+
+## 派生护栏与费用
+
+- 派生深度上限 3 代（父任务本身是派生任务时逐代计；协调员子任务
+  的代数从 0 计，见“边界”）；单个父任务直接派生上限 20 个
+  （不含协调员子任务）；超限的块跳过。
+- 派生任务是独立主线：父任务完成或取消都不自动杀派生任务；
+  想停派生任务，用 task-cancel 或控制台 / API 显式取消它。
+  协调员孤儿收口只扫 coordinator_part 非空的协调员子任务，
+  不扫派生任务。
+- 费用与配额：派生任务 created_by 跟父任务走，认领时照常过
+  key 门与配额门；创建时 worker 先查一次父任务配额，超限时
+  整轮创建跳过（排队本身不花钱，但不让超限方再批量排队）。
+
+## 诚实边界（与服务端口径不一致处，逐条列出）
+
+- 重试的 goal 拒绝比服务端严：服务端 POST /tasks/{id}/retry
+  没有单独的 goal 检查，标记重试为了不破坏 goal 循环语义，
+  明确拒绝 goal 目标。这是标记口径比服务端口径严，不是遗漏。
+- 取消的 goal 拒绝同理：服务端取消也没有单独的 goal 检查，
+  标记取消同样拒绝 goal 目标。
+- 服务端重试是双人门（触发人与原任务创建人任一超限即拦，
+  且触发人须有 approve_merge 权限）；标记重试的触发方是发送
+  方任务本身，其 created_by 与派生任务的 created_by 天然相同
+  （派生时跟父任务走），所以 worker 侧只查发送方创建人这一道
+  approve_merge，效果与双人门在派生链内等价；同创建人-only
+  的目标同样只查这一道，权限查不到时则收紧到仅派生链内。
+- wait 的超时口径与服务端 /wait 不同（见上文），数量口径已
+  对齐到 8 个；不要把标记 wait 的 3600 秒上限套到 HTTP /wait 上。
+- 本批没有 MCP 真工具：标记工具永远是异步的（下一轮才拿到
+  报告），做不到同轮函数返回；想做同轮返回要另搭任务级鉴权的
+  MCP 服务器，不在本批。
+- 签名门残留：worker 的 _code_signature 只签 worker / sandbox /
+  coordinator / cost / server / host 共 6 个文件，新模块
+  marker_tools.py 不在签名候选里。本批部署时 worker.py 与
+  sandbox.py 本身都变了，签名门照常生效；但将来只改
+  marker_tools.py 的补丁不会单独触发签名门，届时要么把
+  marker_tools.py 加进签名候选并同步改单测的 6 个断言，要么
+  接受这一处缺口——本批不顺手改签名，避免动无关安全测试的
+  预期。
+- spawn 分支的未提交半成品没有原样合并进来：本批只移植了
+  已提交的 task-create / task-poll / task-read 实现与未提交的
+  wait 解析（收紧到 8 个后接通执行），handoff / worktree /
+  PR 检查 / automation / clock-sleep 的解析与 send-message-to-thread
+  别名均未移植；spawn 工作区本身未动。
+
+## 关键实现位置
+
+- cloud/marker_tools.py：纯逻辑（解析、校验、快照格式化、
+  门禁、派生代数与直接上限计数），单测不需要 worker/server。
+- cloud/sandbox.py：本地 _result_from_runner 与 Docker
+  _parse_host_result 两处提取五种标记块，worker 侧按结果
+  dict 里的 task_creates / task_polls / wait_threads /
+  task_cancels / task_retries 执行。
+- cloud/worker.py：_deliver_marker_tools（5 处投递点，与
+  _deliver_peer_messages 同款位置）、_poll_task_snapshot、
+  _wait_task_targets、_queue_self_followup、_marker_auth。
+- cloud/tests/test_marker_tools.py：解析上限与钳制、非法块跳过、
+  跨项目拒绝、自操作拒绝、已终态拒绝、goal / 已合并拒绝、
+  重试仅三态且清空旧结果、wait 多目标终态与超时快照、派生
+  深度与直接上限、权限检查失败收紧。