feat: 协调员任务（并行子任务、等齐、合流汇总）

建任务时传 coordinator_parts（分工列表，1-20 项，分工名唯一）即为协调员任务：本身不跑模型回合，worker 认领后一次建齐子任务（普通任务，并行被各 worker 认领），等全部终态后合流成汇总报告写回。子任务带 parent_task_id 与 coordinator_part，控制台可用 parent_task_id 查到；协调员重跑按分工名复用已完成子任务、失败/中断的回 queued 重跑；取消与超时尽力取消未终态子任务。子任务的配额/key/执行器门照常生效，不构成绕过。协调员任务拒绝与 goal/plan/验收命令同用。控制台：任务列表 [协调]/[协调子任务] 标记、详情分工与子任务区、新建表单子任务 JSON 输入。新单测 26 个。

cloud/worker.py

@@ -130,9 +130,15 @@ import subprocess
 import sys
 import threading
 import time
+import uuid
 from types import SimpleNamespace
 
 from cloud.auth import AuthStore
+from cloud.coordinator import build_report as _coord_build_report
+from cloud.coordinator import DEFAULT_POLL_INTERVAL as _COORD_POLL_INTERVAL
+from cloud.coordinator import DEFAULT_TIMEOUT_SECONDS as _COORD_TIMEOUT
+from cloud.coordinator import ensure_columns as _coord_ensure_columns
+from cloud.coordinator import parse_parts as _coord_parse_parts
 from cloud.cost import CostStore, estimate_tokens, price_usd_micros, sum_usage
 from cloud.db import connect as _db_connect
 from cloud.logging_config import get_logger, log_event, setup_logging
@@ -263,6 +269,11 @@ def _task_from_row(row):
         "environment_id": _col("environment_id"),
         # 续跑消息：老行缺列时 None = 没有待续跑消息。
         "resume_message": _col("resume_message"),
+        # 协调员：老行缺列时 coordinator_parts 为 []（= 普通任务），
+        # parent_task_id/coordinator_part 缺列时为 None。
+        "coordinator_parts": _coord_parse_parts(_col("coordinator_parts")),
+        "parent_task_id": _col("parent_task_id"),
+        "coordinator_part": _col("coordinator_part"),
     }
 
 
@@ -612,7 +623,10 @@ class Worker:
                 # turn_id 的回合编号（local/Docker 两路都经
                 # _result_from_runner/_parse_host_result 提取）。
                 "turn_id": "TEXT",
-                "root_turn_id": "TEXT"}.items():
+                "root_turn_id": "TEXT",
+                "coordinator_parts": "TEXT",
+                "parent_task_id": "TEXT",
+                "coordinator_part": "TEXT"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -2031,6 +2045,215 @@ class Worker:
             return match.group(1)
         return None
 
+    # -- 协调员 ---------------------------------------------------------
+    def _coord_sync_children(self, task, parts):
+        """按分工建/复用子任务，返回 [{name, id}]。
+
+        已有子任务按 coordinator_part（分工名）匹配：已完成的直接
+        复用；失败/中断的回 queued 重跑（旧输出清掉）；未终态的
+        不动（正被别的 worker 跑或排队）；缺失的才新建。重跑协调
+        员因此不会重复建子任务。子任务是普通任务：执行器过滤、
+        配额门、key 门都在子任务认领时照常生效。
+        """
+        conn = _db_connect(self.db_path)
+        try:
+            _coord_ensure_columns(conn)
+            try:
+                conn.commit()
+            except Exception:
+                pass
+            rows = conn.execute(
+                "SELECT id, coordinator_part, status FROM tasks"
+                " WHERE parent_task_id = ?",
+                (task["id"],)).fetchall()
+        finally:
+            conn.close()
+        existing = {}
+        for row in rows:
+            try:
+                existing[row["coordinator_part"]] = {
+                    "id": row["id"], "status": row["status"]}
+            except (KeyError, IndexError, TypeError):
+                continue
+        now = time.time()
+        children = []
+        conn = _db_connect(self.db_path)
+        try:
+            for part in parts:
+                name = part["name"]
+                prev = existing.get(name)
+                if prev is not None:
+                    if prev["status"] in TERMINAL_STATUSES and \
+                            prev["status"] != "completed":
+                        conn.execute(
+                            "UPDATE tasks SET status = 'queued',"
+                            " result_text = NULL, error_text = NULL,"
+                            " updated_at = ? WHERE id = ?",
+                            (now, prev["id"]))
+                        self._audit(
+                            task["project_id"],
+                            "coordinator_child_requeued",
+                            "%s -> %s" % (name, prev["id"]))
+                    children.append({"name": name, "id": prev["id"]})
+                    continue
+                child_id = uuid.uuid4().hex
+                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, ?, ?, ?, ?, ?, ?)",
+                    (child_id, task["project_id"], part["prompt"],
+                     task.get("priority", 0) or 0, task["created_by"],
+                     now, now,
+                     part.get("executor") or task.get("executor"),
+                     part.get("model"), part.get("effort"),
+                     task.get("environment_id"), task["id"], name))
+                self._audit(
+                    task["project_id"], "coordinator_child_created",
+                    "%s -> %s" % (name, child_id))
+                children.append({"name": name, "id": child_id})
+            conn.commit()
+        finally:
+            conn.close()
+        return children
+
+    def _coord_child_rows(self, child_ids):
+        """子任务当前行：{id: {status, result_text, error_text}}。
+
+        已删或查不到的子任务不在结果里，调用方按缺失处理（等齐
+        判断里缺失不算终态，会一直等到超时——子任务行不应被删，
+        真删了就是超时失败，不静默当完成）。
+        """
+        rows = {}
+        conn = _db_connect(self.db_path)
+        try:
+            for cid in child_ids:
+                row = conn.execute(
+                    "SELECT id, status, result_text, error_text FROM tasks"
+                    " WHERE id = ?", (cid,)).fetchone()
+                if row is None:
+                    continue
+                try:
+                    rows[row["id"]] = {
+                        "status": row["status"],
+                        "result_text": row["result_text"],
+                        "error_text": row["error_text"]}
+                except (KeyError, IndexError, TypeError):
+                    continue
+        finally:
+            conn.close()
+        return rows
+
+    def _run_coordinator(self, task, started_at, poll_interval=None,
+                         timeout=None):
+        """协调员执行：建子任务 → 等齐 → 合流汇总写回。
+
+        占一个 worker 槽位直到子任务全部终态：等齐期间每轮刷心跳
+        （sweep 不会误回收）；子任务本身由各 worker 并行认领。
+        取消与超时都会尽力取消未终态的子任务，不留孤儿继续烧
+        模型费用。返回 run_one 同款结果 dict（无模型花费，
+        cost_record_id 为 None）。
+        """
+        task_id = task["id"]
+        project_id = task["project_id"]
+        parts = task.get("coordinator_parts") or []
+        if poll_interval is None:
+            poll_interval = _COORD_POLL_INTERVAL
+        if timeout is None:
+            timeout = _COORD_TIMEOUT
+        children = self._coord_sync_children(task, parts)
+        child_ids = [c["id"] for c in children]
+        deadline = time.time() + timeout
+        cancelled = False
+        timed_out = False
+        while True:
+            try:
+                self.heartbeat(task_id)
+            except Exception:
+                pass
+            if self._was_cancelled(task_id):
+                cancelled = True
+                break
+            rows = self._coord_child_rows(child_ids)
+            if all((rows.get(cid) or {}).get("status") in TERMINAL_STATUSES
+                   for cid in child_ids):
+                break
+            if time.time() >= deadline:
+                timed_out = True
+                break
+            time.sleep(poll_interval)
+        if cancelled or timed_out:
+            rows = self._coord_child_rows(child_ids)
+            for cid in child_ids:
+                if (rows.get(cid) or {}).get("status") \
+                        not in TERMINAL_STATUSES:
+                    try:
+                        self.cancel(cid)
+                    except Exception:
+                        pass
+        rows = self._coord_child_rows(child_ids)
+        merged = []
+        for ch in children:
+            row = rows.get(ch["id"]) or {}
+            merged.append({
+                "name": ch["name"], "id": ch["id"],
+                "status": row.get("status") or "?",
+                "result_text": row.get("result_text"),
+                "error_text": row.get("error_text")})
+        report, overall = _coord_build_report(merged)
+        if cancelled:
+            final = "interrupted"
+            error = "cancelled"
+        elif timed_out:
+            final = "failed"
+            pending = [m["name"] for m in merged
+                       if m["status"] not in TERMINAL_STATUSES]
+            error = "coordinator timeout: %s 未完成" % "、".join(pending)
+            report = error + "\n\n" + report
+        else:
+            final = overall
+            error = None
+            if final != "completed":
+                failed = [m["name"] for m in merged
+                          if m["status"] != "completed"]
+                error = "coordinator: %s 未完成" % "、".join(failed)
+        final_written = self._set_status_if_running(task_id, final)
+        if not final_written:
+            final = "interrupted"
+        finished_at = time.time()
+        self._write_execution_result(
+            task_id, result_text=report, error_text=error,
+            actual_model=None, started_at=started_at,
+            finished_at=finished_at)
+        self._cleanup_task_seed(task)
+        if final_written:
+            self._audit(project_id,
+                        "worker_completed" if final == "completed"
+                        else ("worker_interrupted"
+                              if final == "interrupted"
+                              else "worker_failed"),
+                        task_id)
+            log_event(self.logger,
+                      "worker_completed" if final == "completed"
+                      else ("worker_interrupted"
+                            if final == "interrupted"
+                            else "worker_failed"),
+                      task_id=task_id, project_id=project_id, status=final)
+            if final in ("completed", "failed"):
+                self._notify_task_finished(task_id, project_id, final,
+                                           error=error)
+        else:
+            self._audit(project_id, "worker_interrupted", task_id)
+        host_name = self.executor._host_name(task) \
+            if hasattr(self.executor, "_host_name") else task_id
+        return {"dispatched": True, "task_id": task_id,
+                "host_name": host_name, "status": final,
+                "executor": self._executor_name(), "stdout": report,
+                "error": error, "events": [], "cost_record_id": None}
+
     def run_one(self, task=None):
         """认领并执行一个任务，返回结果 dict（见模块 docstring）。
 
@@ -2137,6 +2360,13 @@ class Worker:
                     "stdout": "", "error": quota_error, "events": [],
                     "cost_record_id": None}
 
+        # 协调员分支：不跑模型回合，改为按分工建并行子任务、等齐、
+        # 合流汇总。已过 key 门与配额门；仓库门已对协调员任务本身
+        # 跑过（项目仓库解析失败会在前面直接 failed，不会建一堆
+        # 注定失败的子任务）；子任务各自再过 key/配额/仓库门。
+        if task.get("coordinator_parts"):
+            return self._run_coordinator(task, started_at)
+
         events = []
         fallback_count = 0
         result = None