diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py new file mode 100644 index 0000000..19e351f --- /dev/null +++ b/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 "[任务工具报告]" diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 7a5a060..352fb42 100644 --- a/cloud/sandbox.py +++ b/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, } diff --git a/cloud/tests/test_marker_tools.py b/cloud/tests/test_marker_tools.py new file mode 100644 index 0000000..427e77b --- /dev/null +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index 0e943b2..9046a54 100644 --- a/cloud/worker.py +++ b/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 \ diff --git a/docs/marker-tools.md b/docs/marker-tools.md new file mode 100644 index 0000000..5963bcd --- /dev/null +++ b/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 多目标终态与超时快照、派生 + 深度与直接上限、权限检查失败收紧。