From 7c46ac8e5b62b8377b89a5d6ebff1809f1d7b2bd Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Tue, 27 Oct 2026 15:35:51 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8D=8F=E8=B0=83=E5=91=98=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=EF=BC=88=E5=B9=B6=E8=A1=8C=E5=AD=90=E4=BB=BB=E5=8A=A1?=
 =?UTF-8?q?=E3=80=81=E7=AD=89=E9=BD=90=E3=80=81=E5=90=88=E6=B5=81=E6=B1=87?=
 =?UTF-8?q?=E6=80=BB=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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

diff --git a/cloud/console.html b/cloud/console.html
index f2bdd7c..8c87173 100644
--- a/cloud/console.html
+++ b/cloud/console.html
@@ -160,6 +160,8 @@
   <label><input id="planMode" type="checkbox"> 计划模式（只读）</label>
 </div>
 <textarea id="prompt" placeholder="任务 prompt（必填）"></textarea>
+<textarea id="coordParts" placeholder="协调员子任务 JSON（可选），留空为普通任务"></textarea>
+<div class="muted">协调员格式：[{"name":"后端","prompt":"做后端"},{"name":"前端","prompt":"做前端"}]；提交后 worker 并行建子任务、等齐、合流汇总。不能与 goal/计划模式同用。</div>
 <div class="bar">
   <select id="modelSelect">
     <option value="">模型：默认（跟随配置）</option>
@@ -248,6 +250,8 @@
   <div class="bar"><button id="btnContinueConfirm" class="primary">继续会话</button><span id="continueMsgSpan" class="muted"></span></div>
 </div>
 <pre id="detailBody" style="display:none"></pre>
+<h3>协调子任务</h3>
+<div id="coordBox" class="muted">点任务行查看详情后，这里显示协调员的子任务</div>
 <h3>子代理链路</h3>
 <div id="agentBox" class="muted">点任务行查看详情后，这里显示子代理链路</div>
 <h3>实时链路</h3>
@@ -485,7 +489,7 @@ async function loadTasks() {
         "<td>" + esc(shortId(t.id)) + "</td>" +
         "<td title='" + esc(t.turn_id || "") + "'>" + esc(shortId(t.turn_id)) + "</td>" +
         "<td title='" + esc(t.root_turn_id || "") + "'>" + esc(shortId(t.root_turn_id)) + "</td>" +
-        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + (t.environment_id ? "[环境 " + esc(shortId(t.environment_id)) + "] " : "") + ((t.attachments && t.attachments.length) ? "[附件×" + t.attachments.length + "] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
+        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + (t.coordinator_parts ? "[协调] " : "") + (t.parent_task_id ? "[协调子任务] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + (t.environment_id ? "[环境 " + esc(shortId(t.environment_id)) + "] " : "") + ((t.attachments && t.attachments.length) ? "[附件×" + t.attachments.length + "] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
         "<td class='st-" + esc(t.status) + "'>" + esc(t.status) + "</td>" +
         "<td>" + esc(t.executor || "") + "</td>" +
         "<td>" + esc(t.worktree_host || "") + "</td>" +
@@ -568,6 +572,18 @@ function scheduleAgents(taskId) {
   if (agentTimer) clearTimeout(agentTimer);
   agentTimer = setTimeout(function () { loadAgents(taskId); }, 1000);
 }
+async function loadCoordChildren(taskId) {
+  var box = $("#coordBox");
+  if (!currentTask || !currentTask.coordinator_parts) { box.innerHTML = "<span class='muted'>（非协调员任务）</span>"; return; }
+  try {
+    var d = await api("/tasks?project_id=" + encodeURIComponent(currentTask.project_id) + "&parent_task_id=" + encodeURIComponent(taskId), "GET");
+    var kids = d.tasks || [];
+    if (!kids.length) { box.innerHTML = "<span class='muted'>（子任务尚未创建）</span>"; return; }
+    box.innerHTML = kids.map(function (k) {
+      return "<div>分工 <b>" + esc(k.coordinator_part || "") + "</b> ｜ 状态 <span class='st-" + esc(k.status) + "'>" + esc(k.status) + "</span> ｜ " + esc(shortId(k.id)) + " ｜ " + esc(String(k.prompt || "").slice(0, 60)) + "</div>";
+    }).join("");
+  } catch (e) { box.innerHTML = "<span class='err'>协调子任务加载失败：" + esc(e.message) + "</span>"; }
+}
 async function startTrace(taskId, status) {
   stopTrace();
   var box = $("#traceBox");
@@ -655,6 +671,8 @@ async function showDetail(taskId) {
       (t.plan ? " ｜ <b>计划模式（只读）</b>" : "") +
       (t.forked_from ? " ｜ 分支自 " + esc(shortId(t.forked_from)) : "") +
       (t.goal_mode ? " ｜ <b>goal 模式（不限轮次，模型宣告完成）</b>" : (t.verify_command ? " ｜ goal 模式（旧语义，验收命令）" : "")) +
+      (t.coordinator_parts ? " ｜ <b>协调员：" + t.coordinator_parts.length + " 个子任务（" + esc(t.coordinator_parts.map(function (p) { return p.name; }).join("、")) + "）</b>" : "") +
+      (t.parent_task_id ? " ｜ 协调子任务，父任务 " + esc(shortId(t.parent_task_id)) + (t.coordinator_part ? "（" + esc(t.coordinator_part) + "）" : "") : "") +
       ((t.attachments && t.attachments.length) ? " ｜ 附件：" + t.attachments.map(esc).join("、") : "") +
       (t.pr_url ? " ｜ <a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>PR #" + esc(t.pr_number) + "</a>" : "") +
       "<br>prompt：<br>" + esc(t.prompt || "");
@@ -678,6 +696,8 @@ async function showDetail(taskId) {
     startTrace(taskId, t.status);
     $("#agentBox").innerHTML = "子代理链路加载中…";
     loadAgents(taskId);
+    $("#coordBox").innerHTML = "协调子任务加载中…";
+    loadCoordChildren(taskId);
     loadFollowups(taskId);
     loadPeerMessages(taskId);
     $("#peerMsg").textContent = "";
@@ -853,6 +873,12 @@ $("#createTask").onclick = async function () {
     // 验收命令与轮次（服务端 goal_mode=1 时一律不限轮次）。
     body.goal_mode = 1;
   }
+  var coordRaw = $("#coordParts").value.trim();
+  if (coordRaw) {
+    try { body.coordinator_parts = JSON.parse(coordRaw); }
+    catch (e) { $("#taskMsg").textContent = "协调员子任务 JSON 解析失败"; return; }
+    if (!Array.isArray(body.coordinator_parts)) { $("#taskMsg").textContent = "协调员子任务必须是数组"; return; }
+  }
   var atts = gatherAttachments();
   if (atts.length) {
     try {
@@ -867,6 +893,7 @@ $("#createTask").onclick = async function () {
     var t = await api("/tasks", "POST", body);
     $("#taskMsg").innerHTML = "<span class='ok'>任务已建：" + esc(shortId(t.id)) + "</span>";
     $("#prompt").value = "";
+    $("#coordParts").value = "";
     $("#attFiles").value = ""; $("#attDir").value = ""; $("#attImg").value = "";
     loadTasks();
   } catch (e) { $("#taskMsg").innerHTML = "<span class='err'>提交失败：" + esc(e.message) + "</span>"; }
diff --git a/cloud/coordinator.py b/cloud/coordinator.py
new file mode 100644
index 0000000..f3405e0
--- /dev/null
+++ b/cloud/coordinator.py
@@ -0,0 +1,160 @@
+"""协调员：把一个任务拆成多个并行子任务、等齐、合流汇总。
+
+设计（与现有任务模型对齐，不另起一套执行器）：
+
+- 协调员本身是一个云端任务：建任务时传 coordinator_parts（分工
+  列表），它不跑模型回合；worker 认领后按分工一次建齐子任务
+  （普通任务，并行被各 worker 认领），等全部终态后合流成一份
+  汇总报告写回协调员任务。
+- 子任务是普通任务：配额门、key 门、执行器过滤、取消语义全部
+  走现有链路，协调员不构成绕过；子任务带 parent_task_id 与
+  coordinator_part（分工名），控制台可用 parent_task_id 查到它们。
+- 幂等与续跑：协调员任务重跑（重试/续跑）时按分工名复用已有
+  子任务，不重复建；已完成的子任务直接复用，失败/中断的回
+  queued 重跑——重试协调员即重试未完成的分工。
+- 取消与超时：协调员被取消时尽力取消未终态的子任务；等齐
+  超时时同样取消挂起的子任务并标 failed，报告里写清哪几项
+  没完成。取消路径的协调员标 interrupted，不发完成/失败推送
+  （与普通任务的取消口径一致）。
+
+本模块只放纯逻辑（校验、解析、汇总、建表），不碰 DB 之外的
+任何东西，单测不需要 worker/server。
+"""
+
+import json
+
+MAX_PARTS = 20
+DEFAULT_TIMEOUT_SECONDS = 3600
+DEFAULT_POLL_INTERVAL = 5.0
+REPORT_RESULT_CHARS = 500
+EFFORTS = {"low", "medium", "high", "xhigh", "max"}
+EXECUTORS = {"local", "docker", "cloud"}
+
+
+def validate_parts(parts):
+    """校验 coordinator_parts，返回 (normalized, error)。
+
+    normalized 每项：{name, prompt, model, effort, executor}，
+    未指定的三项为 None（子任务跟随协调员任务或 worker 默认）。
+    """
+    if not isinstance(parts, list):
+        return None, "coordinator_parts must be a list"
+    if not parts:
+        return None, "coordinator_parts must not be empty"
+    if len(parts) > MAX_PARTS:
+        return None, "coordinator_parts exceeds %d parts" % MAX_PARTS
+    seen = set()
+    out = []
+    for part in parts:
+        if not isinstance(part, dict):
+            return None, "each coordinator part must be an object"
+        name = part.get("name")
+        if not isinstance(name, str) or not name.strip():
+            return None, "each coordinator part needs a non-empty name"
+        name = name.strip()
+        if name in seen:
+            return None, "duplicate coordinator part name: %s" % name
+        seen.add(name)
+        prompt = part.get("prompt")
+        if not isinstance(prompt, str) or not prompt.strip():
+            return None, "part %s needs a non-empty prompt" % name
+        model = part.get("model")
+        if model is not None and (
+                not isinstance(model, str) or not model.strip()):
+            return None, "part %s: model must be a non-empty string" % name
+        effort = part.get("effort")
+        if effort is not None and effort not in EFFORTS:
+            return None, (
+                "part %s: effort must be low, medium, high, xhigh or max"
+                % name)
+        executor = part.get("executor")
+        if executor is not None and executor not in EXECUTORS:
+            return None, (
+                "part %s: executor must be local, docker or cloud" % name)
+        out.append({
+            "name": name,
+            "prompt": prompt.strip(),
+            "model": model.strip() if model else None,
+            "effort": effort,
+            "executor": executor,
+        })
+    return out, None
+
+
+def parse_parts(raw):
+    """DB 列里的 coordinator_parts → list；坏值给 []，绝不炸。
+
+    raw 可能是 JSON 文本（DB 列）、已解析的 list（内存里），或
+    None（普通任务）。非法分工一律按空处理，调用方据此判定
+    “不是协调员任务”，而不是把脏数据当分工执行。
+    """
+    if not raw:
+        return []
+    if isinstance(raw, list):
+        parts, err = validate_parts(raw)
+        return parts if err is None else []
+    try:
+        data = json.loads(raw)
+    except (TypeError, ValueError):
+        return []
+    if not isinstance(data, list):
+        return []
+    parts, err = validate_parts(data)
+    return parts if err is None else []
+
+
+def ensure_columns(conn):
+    """给 tasks 表补协调员三列（与 migrate_tasks_table 同款写法）。
+
+    coordinator_parts：协调员任务的分工 JSON（普通任务为 NULL）。
+    parent_task_id：子任务的协调员父任务 id（非子任务为 NULL）。
+    coordinator_part：子任务的分工名（复用与续跑按它匹配）。
+    SQLite 与 Postgres 两支都走它；已存在时 no-op。
+    """
+    try:
+        cols = {row[1] for row in conn.execute(
+            "PRAGMA table_info(tasks)").fetchall()}
+    except Exception:
+        return
+    for name, coltype in {
+            "coordinator_parts": "TEXT",
+            "parent_task_id": "TEXT",
+            "coordinator_part": "TEXT"}.items():
+        if name not in cols:
+            try:
+                conn.execute(
+                    "ALTER TABLE tasks ADD COLUMN %s %s" % (name, coltype))
+            except Exception:
+                pass
+
+
+def build_report(children):
+    """合流汇总：children 为 [{name, id, status, result_text,
+    error_text}]，返回 (report_text, overall_status)。
+
+    overall：全部 completed 才 completed，否则 failed（取消与超时
+    路径由调用方直接定 interrupted/failed，不走这里）。单项输出
+    截断到 REPORT_RESULT_CHARS，汇总报告本身仍受 worker 写回的
+    4000 字上限约束（超长时尾部项会被截断，头部计数不受影响）。
+    """
+    lines = []
+    done = 0
+    for ch in children:
+        st = ch.get("status") or "?"
+        if st == "completed":
+            done += 1
+        lines.append("## %s (%s)" % (ch.get("name"), st))
+        lines.append("子任务：%s" % ch.get("id"))
+        res = (ch.get("result_text") or "")[:REPORT_RESULT_CHARS]
+        err = (ch.get("error_text") or "")[:REPORT_RESULT_CHARS]
+        if res:
+            lines.append(res)
+        if err:
+            lines.append("错误：%s" % err)
+        if not res and not err:
+            lines.append("（无输出）")
+        lines.append("")
+    overall = "completed" if children and done == len(children) else "failed"
+    head = "协调员汇总：%d/%d 个子任务完成，总体 %s" % (
+        done, len(children), overall)
+    return head + "\n\n" + "\n".join(lines), overall
diff --git a/cloud/db.py b/cloud/db.py
index 177a73e..80345c6 100644
--- a/cloud/db.py
+++ b/cloud/db.py
@@ -841,6 +841,9 @@ POSTGRES_SCHEMA = (
     "resume_message TEXT, "
     "turn_id TEXT, "
     "root_turn_id TEXT, "
+    "coordinator_parts TEXT, "
+    "parent_task_id TEXT, "
+    "coordinator_part TEXT, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
     # 附件包与任务附件：与 server._ensure_uploads_tables 同款，
diff --git a/cloud/server.py b/cloud/server.py
index 7c59495..53ef911 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -56,6 +56,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, quote, urlparse
 
 from cloud.auth import AuthStore
+from cloud.coordinator import parse_parts as _coord_parse_parts
+from cloud.coordinator import validate_parts as _coord_validate_parts
 from cloud.cost import CostStore
 from cloud.db import connect as _db_connect
 from cloud.db import create_database as _create_database
@@ -761,6 +763,13 @@ def migrate_tasks_table(conn):
         # root_turn_id（worker 写回时只在新值非空时覆盖）。
         "turn_id": "TEXT",
         "root_turn_id": "TEXT",
+        # 协调员：coordinator_parts 记分工 JSON（普通任务为 NULL，
+        # worker 认领后按它建并行子任务、等齐、合流）；parent_task_id
+        # 与 coordinator_part 记在子任务行上（父任务 id 与分工名，
+        # 续跑按分工名复用已有子任务，不重复建）。
+        "coordinator_parts": "TEXT",
+        "parent_task_id": "TEXT",
+        "coordinator_part": "TEXT",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -1086,6 +1095,11 @@ def _task_to_dict(row, attachments=None):
         "environment_id": _row_get(row, "environment_id"),
         "turn_id": _row_get(row, "turn_id"),
         "root_turn_id": _row_get(row, "root_turn_id"),
+        "coordinator_parts": (
+            _coord_parse_parts(_row_get(row, "coordinator_parts"))
+            if _row_get(row, "coordinator_parts") else None),
+        "parent_task_id": _row_get(row, "parent_task_id"),
+        "coordinator_part": _row_get(row, "coordinator_part"),
     }
 
 
@@ -2432,6 +2446,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._send_json(400, {
                     "error": "goal_max_rounds must be >= 1"})
                 return
+        # 协调员：coordinator_parts 非空时任务本身不跑模型回合，
+        # worker 认领后按分工建并行子任务、等齐、合流；与 goal/
+        # plan/验收命令语义冲突（那些都是“本任务跑模型”的意思），
+        # 同用时 400 且不建任务。
+        coord_parts = None
+        if body.get("coordinator_parts") is not None:
+            coord_parts, coord_err = _coord_validate_parts(
+                body.get("coordinator_parts"))
+            if coord_err:
+                self._send_json(400, {"error": coord_err})
+                return
+            if goal_mode or plan or verify_command:
+                self._send_json(400, {
+                    "error": "协调员任务不跑模型回合，不能与"
+                             " goal/plan/验收命令同用"})
+                return
         idempotency_key = body.get("idempotency_key")
         if idempotency_key is not None and not isinstance(idempotency_key, str):
             self._send_json(400, {"error": "idempotency_key must be a string"})
@@ -2553,14 +2583,17 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     " created_at, updated_at, last_heartbeat, executor,"
                     " goal, verify_command, goal_max_rounds,"
                     " model, effort, approval, plan, goal_mode,"
-                    " environment_id)"
+                    " environment_id, coordinator_parts,"
+                    " parent_task_id, coordinator_part)"
                     " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?,"
-                    " ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                     (task_id, project_id, prompt, idempotency_key,
                      json.dumps(depends_on), priority, user_id, now, now,
                      executor, goal, verify_command, goal_max_rounds,
                      model, effort, approval, plan, goal_mode,
-                     environment_id),
+                     environment_id,
+                     json.dumps(coord_parts) if coord_parts else None,
+                     None, None),
                 )
                 for rel_path, size in bundle_files:
                     conn.execute(
@@ -2609,13 +2642,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             if not store.check_permission(user_id, project_id, "read"):
                 self._send_json(403, {"error": "forbidden"})
                 return
+        parent_task_id = None
+        parent_ids = query.get("parent_task_id")
+        if parent_ids:
+            parent_task_id = parent_ids[0]
         conn = self._db()
         try:
-            rows = conn.execute(
-                "SELECT * FROM tasks WHERE project_id = ?"
-                " ORDER BY created_at, id",
-                (project_id,),
-            ).fetchall()
+            if parent_task_id:
+                rows = conn.execute(
+                    "SELECT * FROM tasks WHERE project_id = ?"
+                    " AND parent_task_id = ?"
+                    " ORDER BY created_at, id",
+                    (project_id, parent_task_id),
+                ).fetchall()
+            else:
+                rows = conn.execute(
+                    "SELECT * FROM tasks WHERE project_id = ?"
+                    " ORDER BY created_at, id",
+                    (project_id,),
+                ).fetchall()
             amap = _attachments_map(conn, [r["id"] for r in rows])
         finally:
             conn.close()
diff --git a/cloud/tests/test_coordinator.py b/cloud/tests/test_coordinator.py
new file mode 100644
index 0000000..d20a8e5
--- /dev/null
+++ b/cloud/tests/test_coordinator.py
@@ -0,0 +1,417 @@
+"""协调员单测：分工校验、汇总报告、worker 分发/等齐/合流、服务端接口。
+
+不跑真实模型回合、不联网：worker 侧用临时 SQLite + 假 runner，
+子任务终态由测试直接写库模拟（等齐逻辑只读子任务行，不跑模型）；
+服务端侧用本地临时端口 + 临时 SQLite，与 test_peer_messages_server
+同款骨架。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.coordinator import build_report, parse_parts, validate_parts
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.coordinator import build_report, parse_parts, validate_parts
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+
+
+def _completed_runner(cmd):
+    return SimpleNamespace(returncode=0, stdout="{}", stderr="")
+
+
+def _insert_task(db_path, task_id, prompt="do work", status="queued",
+                 project_id="p1", coordinator_parts=None,
+                 parent_task_id=None, coordinator_part=None,
+                 result_text=None, error_text=None):
+    now = time.time()
+    conn = sqlite3.connect(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, coordinator_parts,"
+            " parent_task_id, coordinator_part, result_text, error_text)"
+            " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL,"
+            " ?, ?, ?, ?, ?)",
+            (task_id, project_id, prompt, status, now, now,
+             json.dumps(coordinator_parts) if coordinator_parts else None,
+             parent_task_id, coordinator_part, result_text, error_text),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT id, status, result_text, error_text, parent_task_id,"
+            " coordinator_part, executor, model, effort FROM tasks"
+            " WHERE id = ?", (task_id,)).fetchone()
+    finally:
+        conn.close()
+    if row is None:
+        return None
+    return {
+        "id": row[0], "status": row[1], "result_text": row[2],
+        "error_text": row[3], "parent_task_id": row[4],
+        "coordinator_part": row[5], "executor": row[6],
+        "model": row[7], "effort": row[8],
+    }
+
+
+def _children(db_path, parent_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT id FROM tasks WHERE parent_task_id = ?"
+            " ORDER BY coordinator_part",
+            (parent_id,)).fetchall()
+    finally:
+        conn.close()
+    return [r[0] for r in rows]
+
+
+class ValidatePartsTestCase(unittest.TestCase):
+    def test_valid_minimal(self):
+        parts, err = validate_parts([{"name": "a", "prompt": "做 a"}])
+        self.assertIsNone(err)
+        self.assertEqual(parts, [{"name": "a", "prompt": "做 a",
+                                  "model": None, "effort": None,
+                                  "executor": None}])
+
+    def test_valid_with_options(self):
+        parts, err = validate_parts([{
+            "name": "b", "prompt": "做 b", "model": "m1",
+            "effort": "high", "executor": "docker"}])
+        self.assertIsNone(err)
+        self.assertEqual(parts[0]["model"], "m1")
+        self.assertEqual(parts[0]["effort"], "high")
+        self.assertEqual(parts[0]["executor"], "docker")
+
+    def test_rejects_non_list(self):
+        _parts, err = validate_parts({"name": "a"})
+        self.assertIsNotNone(err)
+
+    def test_rejects_empty(self):
+        _parts, err = validate_parts([])
+        self.assertIsNotNone(err)
+
+    def test_rejects_too_many(self):
+        parts = [{"name": "p%d" % i, "prompt": "x"} for i in range(21)]
+        _parts, err = validate_parts(parts)
+        self.assertIsNotNone(err)
+
+    def test_rejects_duplicate_name(self):
+        _parts, err = validate_parts([
+            {"name": "a", "prompt": "x"}, {"name": "a", "prompt": "y"}])
+        self.assertIsNotNone(err)
+
+    def test_rejects_empty_name_or_prompt(self):
+        _parts, err = validate_parts([{"name": " ", "prompt": "x"}])
+        self.assertIsNotNone(err)
+        _parts, err = validate_parts([{"name": "a", "prompt": " "}])
+        self.assertIsNotNone(err)
+
+    def test_rejects_bad_effort_executor_model(self):
+        _parts, err = validate_parts(
+            [{"name": "a", "prompt": "x", "effort": "ultra"}])
+        self.assertIsNotNone(err)
+        _parts, err = validate_parts(
+            [{"name": "a", "prompt": "x", "executor": "k8s"}])
+        self.assertIsNotNone(err)
+        _parts, err = validate_parts(
+            [{"name": "a", "prompt": "x", "model": " "}])
+        self.assertIsNotNone(err)
+
+    def test_rejects_non_dict_part(self):
+        _parts, err = validate_parts(["a"])
+        self.assertIsNotNone(err)
+
+
+class ParsePartsTestCase(unittest.TestCase):
+    def test_none_and_empty(self):
+        self.assertEqual(parse_parts(None), [])
+        self.assertEqual(parse_parts(""), [])
+
+    def test_bad_json(self):
+        self.assertEqual(parse_parts("{not json"), [])
+        self.assertEqual(parse_parts('{"name": "a"}'), [])
+
+    def test_valid_json(self):
+        parts = parse_parts('[{"name": "a", "prompt": "x"}]')
+        self.assertEqual(len(parts), 1)
+        self.assertEqual(parts[0]["name"], "a")
+
+    def test_list_input(self):
+        self.assertEqual(len(parse_parts([{"name": "a", "prompt": "x"}])), 1)
+        self.assertEqual(parse_parts([{"name": "a"}]), [])
+
+
+class BuildReportTestCase(unittest.TestCase):
+    def test_all_completed(self):
+        report, overall = build_report([
+            {"name": "a", "id": "1", "status": "completed",
+             "result_text": "A 好了", "error_text": None},
+            {"name": "b", "id": "2", "status": "completed",
+             "result_text": "B 好了", "error_text": None}])
+        self.assertEqual(overall, "completed")
+        self.assertIn("2/2", report)
+        self.assertIn("A 好了", report)
+
+    def test_mixed_failed(self):
+        report, overall = build_report([
+            {"name": "a", "id": "1", "status": "completed",
+             "result_text": "ok", "error_text": None},
+            {"name": "b", "id": "2", "status": "failed",
+             "result_text": None, "error_text": "炸了"}])
+        self.assertEqual(overall, "failed")
+        self.assertIn("1/2", report)
+        self.assertIn("炸了", report)
+
+    def test_truncates_long_output(self):
+        report, _overall = build_report([
+            {"name": "a", "id": "1", "status": "completed",
+             "result_text": "x" * 600, "error_text": None}])
+        self.assertNotIn("x" * 600, report)
+        self.assertIn("x" * 500, report)
+
+    def test_empty_output_note(self):
+        report, _overall = build_report([
+            {"name": "a", "id": "1", "status": "completed",
+             "result_text": None, "error_text": None}])
+        self.assertIn("无输出", report)
+
+
+class WorkerCoordinatorTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self):
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=_completed_runner, env=self.env)
+
+    def _parts(self):
+        return [{"name": "后端", "prompt": "做后端", "model": "m1",
+                 "effort": "high"},
+                {"name": "前端", "prompt": "做前端"}]
+
+    def test_sync_creates_children_once(self):
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        worker = self._worker()
+        task = worker.get_task("coord")
+        first = worker._coord_sync_children(task, task["coordinator_parts"])
+        second = worker._coord_sync_children(task, task["coordinator_parts"])
+        self.assertEqual([c["id"] for c in first],
+                         [c["id"] for c in second])
+        self.assertEqual(len(_children(self.db_path, "coord")), 2)
+        backend = _row(self.db_path, first[0]["id"])
+        self.assertEqual(backend["parent_task_id"], "coord")
+        self.assertEqual(backend["coordinator_part"], "后端")
+        self.assertEqual(backend["model"], "m1")
+        self.assertEqual(backend["effort"], "high")
+        self.assertEqual(backend["status"], "queued")
+
+    def test_sync_requeues_failed_keeps_completed(self):
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c-old", status="failed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="旧输出", error_text="旧错")
+        _insert_task(self.db_path, "c-done", status="completed",
+                     parent_task_id="coord", coordinator_part="前端",
+                     result_text="已做完")
+        worker = self._worker()
+        task = worker.get_task("coord")
+        children = worker._coord_sync_children(task, task["coordinator_parts"])
+        self.assertEqual([c["id"] for c in children], ["c-old", "c-done"])
+        old = _row(self.db_path, "c-old")
+        self.assertEqual(old["status"], "queued")
+        self.assertIsNone(old["result_text"])
+        self.assertIsNone(old["error_text"])
+        done = _row(self.db_path, "c-done")
+        self.assertEqual(done["status"], "completed")
+        self.assertEqual(done["result_text"], "已做完")
+
+    def test_run_one_all_completed(self):
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="后端好了")
+        _insert_task(self.db_path, "c2", status="completed",
+                     parent_task_id="coord", coordinator_part="前端",
+                     result_text="前端好了")
+        worker = self._worker()
+        result = worker.run_one(worker.get_task("coord"))
+        self.assertEqual(result["status"], "completed")
+        self.assertIn("2/2", result["stdout"])
+        self.assertIn("后端好了", result["stdout"])
+        self.assertEqual(_row(self.db_path, "coord")["status"], "completed")
+
+    def test_run_timeout_cancels_pending(self):
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("UPDATE tasks SET status = 'running' WHERE id = 'coord'")
+        conn.commit()
+        conn.close()
+        worker = self._worker()
+        task = worker.get_task("coord")
+        result = worker._run_coordinator(
+            task, time.time(), poll_interval=0.01, timeout=0.05)
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("timeout", result["error"])
+        self.assertEqual(_row(self.db_path, "coord")["status"], "failed")
+        for cid in _children(self.db_path, "coord"):
+            self.assertEqual(_row(self.db_path, cid)["status"], "interrupted")
+
+    def test_run_cancelled(self):
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="ok")
+        _insert_task(self.db_path, "c2", status="completed",
+                     parent_task_id="coord", coordinator_part="前端",
+                     result_text="ok")
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("UPDATE tasks SET status = 'running' WHERE id = 'coord'")
+        conn.commit()
+        conn.close()
+        worker = self._worker()
+        worker._cancel_requested.add("coord")
+        result = worker._run_coordinator(
+            worker.get_task("coord"), time.time(),
+            poll_interval=0.01, timeout=60)
+        self.assertEqual(result["status"], "interrupted")
+        self.assertEqual(result["error"], "cancelled")
+        self.assertEqual(_row(self.db_path, "coord")["status"], "interrupted")
+
+
+class ServerCoordinatorTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _create_coord(self, parts=None, extra=None):
+        body = {"project_id": self.project["id"], "prompt": "协调干活",
+                "coordinator_parts": parts if parts is not None else [
+                    {"name": "后端", "prompt": "做后端"},
+                    {"name": "前端", "prompt": "做前端"}]}
+        if extra:
+            body.update(extra)
+        return self._req("POST", "/tasks", body, self.owner_key)
+
+    def test_create_coordinator_ok(self):
+        status, task = self._create_coord()
+        self.assertEqual(status, 201)
+        self.assertEqual(len(task["coordinator_parts"]), 2)
+        self.assertEqual(task["coordinator_parts"][0]["name"], "后端")
+        self.assertIsNone(task["parent_task_id"])
+        status, fetched = self._req(
+            "GET", "/tasks/" + task["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(fetched["coordinator_parts"]), 2)
+
+    def test_create_coordinator_invalid(self):
+        status, body = self._create_coord(parts=[])
+        self.assertEqual(status, 400)
+        status, _body = self._create_coord(parts=[
+            {"name": "a", "prompt": "x"}, {"name": "a", "prompt": "y"}])
+        self.assertEqual(status, 400)
+        status, _body = self._create_coord(parts=[
+            {"name": "a", "prompt": "x", "effort": "ultra"}])
+        self.assertEqual(status, 400)
+
+    def test_create_coordinator_conflicts(self):
+        status, _body = self._create_coord(extra={"goal_mode": 1})
+        self.assertEqual(status, 400)
+        status, _body = self._create_coord(extra={"plan": 1})
+        self.assertEqual(status, 400)
+        status, _body = self._create_coord(
+            extra={"verify_command": "true"})
+        self.assertEqual(status, 400)
+
+    def test_list_children_filter(self):
+        status, coord = self._create_coord()
+        self.assertEqual(status, 201)
+        _insert_task(self.db_path, "kid1", parent_task_id=coord["id"],
+                     coordinator_part="后端",
+                     project_id=self.project["id"])
+        _insert_task(self.db_path, "other", project_id=self.project["id"])
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + self.project["id"] +
+            "&parent_task_id=" + coord["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual([t["id"] for t in data["tasks"]], ["kid1"])
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(data["tasks"]), 3)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/worker.py b/cloud/worker.py
index 7b624c9..10bfee8 100644
--- a/cloud/worker.py
+++ b/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