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 @@
+
+协调员格式：[{"name":"后端","prompt":"做后端"},{"name":"前端","prompt":"做前端"}]；提交后 worker 并行建子任务、等齐、合流汇总。不能与 goal/计划模式同用。
+协调子任务
+点任务行查看详情后，这里显示协调员的子任务
子代理链路
点任务行查看详情后，这里显示子代理链路
实时链路
@@ -485,7 +489,7 @@ async function loadTasks() {
"" + esc(shortId(t.id)) + " | " +
"" + esc(shortId(t.turn_id)) + " | " +
"" + esc(shortId(t.root_turn_id)) + " | " +
- "" + (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)) + " | " +
+ "" + (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)) + " | " +
"" + esc(t.status) + " | " +
"" + esc(t.executor || "") + " | " +
"" + esc(t.worktree_host || "") + " | " +
@@ -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 = "（非协调员任务）"; 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 = "（子任务尚未创建）"; return; }
+ box.innerHTML = kids.map(function (k) {
+ return "分工 " + esc(k.coordinator_part || "") + " ｜ 状态 " + esc(k.status) + " ｜ " + esc(shortId(k.id)) + " ｜ " + esc(String(k.prompt || "").slice(0, 60)) + "
";
+ }).join("");
+ } catch (e) { box.innerHTML = "协调子任务加载失败：" + esc(e.message) + ""; }
+}
async function startTrace(taskId, status) {
stopTrace();
var box = $("#traceBox");
@@ -655,6 +671,8 @@ async function showDetail(taskId) {
(t.plan ? " ｜ 计划模式（只读）" : "") +
(t.forked_from ? " ｜ 分支自 " + esc(shortId(t.forked_from)) : "") +
(t.goal_mode ? " ｜ goal 模式（不限轮次，模型宣告完成）" : (t.verify_command ? " ｜ goal 模式（旧语义，验收命令）" : "")) +
+ (t.coordinator_parts ? " ｜ 协调员：" + t.coordinator_parts.length + " 个子任务（" + esc(t.coordinator_parts.map(function (p) { return p.name; }).join("、")) + "）" : "") +
+ (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 ? " ｜ PR #" + esc(t.pr_number) + "" : "") +
"
prompt：
" + 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 = "任务已建：" + esc(shortId(t.id)) + "";
$("#prompt").value = "";
+ $("#coordParts").value = "";
$("#attFiles").value = ""; $("#attDir").value = ""; $("#attImg").value = "";
loadTasks();
} catch (e) { $("#taskMsg").innerHTML = "提交失败：" + esc(e.message) + ""; }
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