Merge pull request #87 from openqj/feat-coordinator

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

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
@@ -772,6 +774,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:
@@ -1097,6 +1106,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"),
     }
 
 
@@ -2502,6 +2516,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"})
@@ -2623,14 +2653,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(
@@ -2679,13 +2712,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()