feat: bounded subagent delegation (v5)

Coordinator parts gain three enforced bounds, validated in
validate_parts and carried through the coordinator_parts JSON
(no schema change): timeout_seconds (1..86400, per-attempt
wall clock from coordinator start; the coordinator poll loop
cancels only that child with the same cancel mechanism as
the global timeout, other parts keep waiting, audit
coordinator_child_bound_exceeded), on_failure fail|tolerate
(tolerated failures do not fail the coordinator, are named
in the report head, and their failed children are not
requeued on coordinator re-run while interrupted children
still are), and result_chars (0..2000, default 500, enforced
in build_report; the 4000-char report writeback cap still
applies). evaluate_outcome is the single outcome rule used
by both build_report and the worker error text, so tolerated
failures never leak into the coordinator failure reason.
Unknown part names fail closed.

Deliberately NOT added, after reading the execution layer:
per-part token/cost budget (cost is recorded only on
completion, so a running child's spend is unobservable and
creator quota is monthly, not per part), per-part round cap
(normal tasks have no per-task round cap in the worker),
and write-scope isolation (sandbox is per-project worktree,
a scope string in a prompt would be an instruction, not a
guarantee). Tests: test_delegation_bounds.py (8 tests).

cloud/coordinator.py

@@ -26,6 +26,37 @@
   （父仍在跑时绝不动子任务）。本模块只提供纯函数
   nonterminal_children 供 worker 判断，DB 写回在 worker 侧。
 
+- 有界委派（v5，Foundry“有界 subagent 委派”的对应物）：
+  每个分工除 name/prompt/model/effort/executor 外，还可带
+  三个显式边界，全部在 validate_parts 校验、经 DB 的
+  coordinator_parts JSON 原样流转、由 worker 的协调员循环
+  真正执行，不是只写进 prompt 的愿望：
+  ① timeout_seconds：该分工单次委派的墙钟上限，从协调员
+  本轮开始计时，超时由协调员循环直接 cancel 该子任务
+  （与全局超时同款 cancel 机制），其他分工继续等，不因
+  一个慢分工拖住整批到全局超时；② on_failure：fail（默认，
+  旧行为，任一非容忍分工未完成则协调员失败）或 tolerate
+  （该分工失败不拖垮协调员，汇总里点名已容忍，协调员
+  重跑时其 failed 子任务不再重排、interrupted 的仍重排，
+  因为后者可能是父任务取消所致、未获公平一轮）；③
+  result_chars：回传给父任务的单项输出上限，默认 500，
+  汇总报告本身仍受 worker 写回的 4000 字上限约束。
+  三个边界的执行点都在协调员层：超时在循环里 cancel，
+  容忍在合流评价里算，截断在汇总报告里切。
+
+- v5 明确不宣称的边界（查代码后的诚实口径，勿以字段
+  冒充执行）：① 每分工 token/费用预算：成本记录只在任务
+  完成时写，运行中的子任务花费在循环里观测不到，创建人
+  配额是按月不是按分工，所以不存在可执行的分工费用门，
+  v5 不加 max_tokens/max_cost 字段；② 每分工轮数上限：
+  普通任务在 worker 侧没有按任务的轮数上限（只有 goal
+  任务的 goal_max_rounds 与 worker 进程级的 max_turns），
+  所以 max_rounds 字段在子任务执行层无法执行，v5 不加；
+  ③ 写范围隔离：沙箱按项目 worktree 隔离、不按分工
+  量子目录隔离，prompt 里写“只写 docs/”是指令不是保证，
+  所以 v5 不加 scope 字段冒充隔离。这三项要做需先改
+  执行层（成本流、轮数帽、沙箱子目录），不在本轨范围。
+
 本模块只放纯逻辑（校验、解析、汇总、建表），不碰 DB 之外的
 任何东西，单测不需要 worker/server。
 """
@@ -36,15 +67,25 @@ MAX_PARTS = 20
 DEFAULT_TIMEOUT_SECONDS = 3600
 DEFAULT_POLL_INTERVAL = 5.0
 REPORT_RESULT_CHARS = 500
+MAX_RESULT_CHARS = 2000
+MAX_TIMEOUT_SECONDS = 86400
+ON_FAILURES = ("fail", "tolerate")
 EFFORTS = {"low", "medium", "high", "xhigh", "max"}
 EXECUTORS = {"local", "docker", "cloud"}
 
 
+def _is_int(value):
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
 def validate_parts(parts):
     """校验 coordinator_parts，返回 (normalized, error)。
 
-    normalized 每项：{name, prompt, model, effort, executor}，
-    未指定的三项为 None（子任务跟随协调员任务或 worker 默认）。
+    normalized 每项：{name, prompt, model, effort, executor,
+    timeout_seconds, on_failure, result_chars}，未指定的模型/
+    强度/执行器为 None（子任务跟随协调员任务或 worker 默认），
+    未指定的 timeout_seconds 为 None（只受全局超时约束），
+    on_failure 默认 fail，result_chars 默认 500。
     """
     if not isinstance(parts, list):
         return None, "coordinator_parts must be a list"
@@ -80,12 +121,34 @@ def validate_parts(parts):
         if executor is not None and executor not in EXECUTORS:
             return None, (
                 "part %s: executor must be local, docker or cloud" % name)
+        timeout = part.get("timeout_seconds")
+        if timeout is not None:
+            if not _is_int(timeout) or not 1 <= timeout <= \
+                    MAX_TIMEOUT_SECONDS:
+                return None, (
+                    "part %s: timeout_seconds must be an integer in"
+                    " 1..%d" % (name, MAX_TIMEOUT_SECONDS))
+        on_failure = part.get("on_failure") or "fail"
+        if on_failure not in ON_FAILURES:
+            return None, (
+                "part %s: on_failure must be fail or tolerate" % name)
+        result_chars = part.get("result_chars")
+        if result_chars is None:
+            result_chars = REPORT_RESULT_CHARS
+        if not _is_int(result_chars) or not 0 <= result_chars <= \
+                MAX_RESULT_CHARS:
+            return None, (
+                "part %s: result_chars must be an integer in"
+                " 0..%d" % (name, MAX_RESULT_CHARS))
         out.append({
             "name": name,
             "prompt": prompt.strip(),
             "model": model.strip() if model else None,
             "effort": effort,
             "executor": executor,
+            "timeout_seconds": timeout,
+            "on_failure": on_failure,
+            "result_chars": result_chars,
         })
     return out, None
 
@@ -163,25 +226,71 @@ def nonterminal_children(children):
     return out
 
 
-def build_report(children):
+def parts_by_name(parts):
+    """分工列表 → {name: part}；parts 为 None 时返回 {}。"""
+    out = {}
+    for part in parts or []:
+        if isinstance(part, dict) and part.get("name"):
+            out[part["name"]] = part
+    return out
+
+
+def evaluate_outcome(children, parts=None):
+    """合流评价（纯函数）：返回 (overall, failed_names,
+    tolerated_names)。
+
+    completed 计完成；未完成的项中，on_failure 为 tolerate
+    的分工计容忍、不拖垮整体，其余计失败。未知分工名按
+    fail 处理（fail-closed，绝不把查不到的分工当容忍）。
+    空 children 时 overall 为 failed（与旧 build_report 一致，
+    失败名为空——调用方按“无子任务”理解，不按名理解）。
+    """
+    by_name = parts_by_name(parts)
+    has_parts = parts is not None
+    done = 0
+    failed = []
+    tolerated = []
+    for ch in children or []:
+        name = ch.get("name")
+        if ch.get("status") == "completed":
+            done += 1
+            continue
+        part = by_name.get(name) if has_parts else None
+        if has_parts and part is not None and \
+                part.get("on_failure") == "tolerate":
+            tolerated.append(name)
+        else:
+            failed.append(name)
+    overall = "completed" if children and not failed else "failed"
+    return overall, failed, tolerated
+
+
+def build_report(children, parts=None):
     """合流汇总：children 为 [{name, id, status, result_text,
     error_text}]，返回 (report_text, overall_status)。
 
-    overall：全部 completed 才 completed，否则 failed（取消与超时
-    路径由调用方直接定 interrupted/failed，不走这里）。单项输出
-    截断到 REPORT_RESULT_CHARS，汇总报告本身仍受 worker 写回的
-    4000 字上限约束（超长时尾部项会被截断，头部计数不受影响）。
+    不传 parts 时为旧行为：全部 completed 才 completed，
+    单项输出截断到 REPORT_RESULT_CHARS。传 parts 时按
+    evaluate_outcome 算总体（容忍分工的失败不拖垮整体，
+    头部点名已容忍项），单项截断按该分工的 result_chars。
+    取消与超时路径由调用方直接定 interrupted/failed，不走
+    这里。汇总报告本身仍受 worker 写回的 4000 字上限约束
+    （超长时尾部项会被截断，头部计数不受影响）。
     """
+    by_name = parts_by_name(parts)
     lines = []
     done = 0
     for ch in children:
         st = ch.get("status") or "?"
         if st == "completed":
             done += 1
+        part = by_name.get(ch.get("name"))
+        chars = part.get("result_chars") if part is not None else \
+            REPORT_RESULT_CHARS
         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]
+        res = (ch.get("result_text") or "")[:chars]
+        err = (ch.get("error_text") or "")[:chars]
         if res:
             lines.append(res)
         if err:
@@ -189,7 +298,16 @@ def build_report(children):
         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)
+    if parts is None:
+        overall = "completed" if children and \
+            done == len(children) else "failed"
+        head = "协调员汇总：%d/%d 个子任务完成，总体 %s" % (
+            done, len(children), overall)
+    else:
+        overall, _failed, tolerated = evaluate_outcome(children, parts)
+        head = "协调员汇总：%d/%d 个子任务完成，总体 %s" % (
+            done, len(children), overall)
+        if tolerated:
+            head += "（%d 项失败已容忍：%s）" % (
+                len(tolerated), "、".join(tolerated))
     return head + "\n\n" + "\n".join(lines), overall

cloud/tests/test_coordinator.py

@@ -95,7 +95,10 @@ class ValidatePartsTestCase(unittest.TestCase):
         self.assertIsNone(err)
         self.assertEqual(parts, [{"name": "a", "prompt": "做 a",
                                   "model": None, "effort": None,
-                                  "executor": None}])
+                                  "executor": None,
+                                  "timeout_seconds": None,
+                                  "on_failure": "fail",
+                                  "result_chars": 500}])
 
     def test_valid_with_options(self):
         parts, err = validate_parts([{

cloud/tests/test_delegation_bounds.py

@@ -0,0 +1,241 @@
+"""有界委派 v5 单测：分工边界校验、容忍评价、回传截断、
+单分工超时取消、容忍分工重跑不重排。
+
+运行（在仓库根目录下）：
+    python3 -m unittest discover -s cloud/tests -p "test_delegation_bounds.py" -v
+
+说明：纯函数部分不碰库；worker 部分用临时 SQLite + 假
+runner，子任务终态由测试直接写库模拟（与
+test_coordinator 同款骨架），不跑真实模型、不联网。
+关键断言：慢分工超时只取消该分工、不拖到全局超时；
+容忍分工失败不拖垮协调员、且重跑时不重排；未知分工
+按 fail 处理；token/轮数/写范围字段不在本轨——查代码
+后确认执行层无法执行，故不以字段冒充（见
+cloud/coordinator.py 模块注释）。
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.coordinator import build_report, evaluate_outcome
+    from cloud.coordinator import validate_parts
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.coordinator import build_report, evaluate_outcome
+    from cloud.coordinator import validate_parts
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+
+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:
+        import json
+        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 status, result_text, error_text FROM tasks"
+            " WHERE id = ?", (task_id,)).fetchone()
+    finally:
+        conn.close()
+    return {"status": row[0], "result_text": row[1],
+            "error_text": row[2]} if row else None
+
+
+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 ValidateBoundsTestCase(unittest.TestCase):
+    def test_valid_bounds(self):
+        parts, err = validate_parts([{
+            "name": "a", "prompt": "x", "timeout_seconds": 60,
+            "on_failure": "tolerate", "result_chars": 100}])
+        self.assertIsNone(err)
+        self.assertEqual(parts[0]["timeout_seconds"], 60)
+        self.assertEqual(parts[0]["on_failure"], "tolerate")
+        self.assertEqual(parts[0]["result_chars"], 100)
+
+    def test_rejects_bad_bounds(self):
+        for bad in (
+                {"timeout_seconds": 0},
+                {"timeout_seconds": 86401},
+                {"timeout_seconds": True},
+                {"timeout_seconds": 1.5},
+                {"on_failure": "retry"},
+                {"result_chars": -1},
+                {"result_chars": 2001},
+        ):
+            _parts, err = validate_parts(
+                [{"name": "a", "prompt": "x", **bad}])
+            self.assertIsNotNone(err, bad)
+
+    def test_old_parts_get_defaults(self):
+        parts, err = validate_parts([{"name": "a", "prompt": "x"}])
+        self.assertIsNone(err)
+        self.assertIsNone(parts[0]["timeout_seconds"])
+        self.assertEqual(parts[0]["on_failure"], "fail")
+        self.assertEqual(parts[0]["result_chars"], 500)
+
+
+class OutcomeTestCase(unittest.TestCase):
+    def test_tolerated_failure_completes(self):
+        parts, _err = validate_parts([
+            {"name": "a", "prompt": "x"},
+            {"name": "b", "prompt": "x", "on_failure": "tolerate"}])
+        overall, failed, tolerated = evaluate_outcome([
+            {"name": "a", "status": "completed"},
+            {"name": "b", "status": "failed"}], parts)
+        self.assertEqual(overall, "completed")
+        self.assertEqual(failed, [])
+        self.assertEqual(tolerated, ["b"])
+
+    def test_unknown_part_fails_closed(self):
+        parts, _err = validate_parts([{"name": "a", "prompt": "x"}])
+        overall, failed, _tol = evaluate_outcome([
+            {"name": "ghost", "status": "failed"}], parts)
+        self.assertEqual(overall, "failed")
+        self.assertEqual(failed, ["ghost"])
+
+    def test_per_part_result_chars(self):
+        parts, _err = validate_parts([
+            {"name": "a", "prompt": "x", "result_chars": 10}])
+        report, _overall = build_report([
+            {"name": "a", "id": "1", "status": "completed",
+             "result_text": "x" * 100, "error_text": None}], parts)
+        self.assertIn("x" * 10, report)
+        self.assertNotIn("x" * 11, report)
+
+
+class WorkerBoundsTestCase(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):
+        from types import SimpleNamespace
+
+        def _runner(cmd):
+            return SimpleNamespace(returncode=0, stdout="{}",
+                                   stderr="")
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=_runner, env=self.env)
+
+    def _parts(self):
+        return [
+            {"name": "快", "prompt": "做快活", "timeout_seconds": 60},
+            {"name": "慢", "prompt": "做慢活", "timeout_seconds": 1},
+        ]
+
+    def test_per_part_timeout_cancels_only_slow_part(self):
+        # 快分工已完成，慢分工排队未终态：协调员本轮开始已
+        # 超过慢分工的 1 秒上限（started_at 往前推 2 秒），
+        # 循环应只取消慢分工、不等到 60 秒全局超时。
+        _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")
+        children = worker._coord_sync_children(task, task["coordinator_parts"])
+        self.assertEqual(len(children), 2)
+        # 快分工直接标完成，慢分工保持排队：下一轮循环里
+        # 慢分工被单分工超时取消、快分工不动，等齐立即结束。
+        by_name = {}
+        conn = sqlite3.connect(self.db_path)
+        try:
+            for cid, part in conn.execute(
+                    "SELECT id, coordinator_part FROM tasks"
+                    " WHERE parent_task_id = 'coord'").fetchall():
+                by_name[part] = cid
+            conn.execute(
+                "UPDATE tasks SET status = 'completed',"
+                " result_text = 'ok' WHERE id = ?",
+                (by_name["快"],))
+            conn.commit()
+        finally:
+            conn.close()
+        started = time.time()
+        result = worker._run_coordinator(
+            worker.get_task("coord"), time.time() - 2,
+            poll_interval=0.01, timeout=60)
+        self.assertLess(time.time() - started, 10)
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("有界委派超时", result["stdout"] or "")
+        self.assertEqual(_row(self.db_path, by_name["快"])["status"],
+                         "completed")
+        self.assertEqual(_row(self.db_path, by_name["慢"])["status"],
+                         "interrupted")
+
+    def test_tolerated_failed_not_requeued(self):
+        parts = [
+            {"name": "主", "prompt": "做主活"},
+            {"name": "副", "prompt": "做副活", "on_failure": "tolerate"},
+        ]
+        _insert_task(self.db_path, "coord", coordinator_parts=parts)
+        _insert_task(self.db_path, "main", status="completed",
+                     parent_task_id="coord", coordinator_part="主",
+                     result_text="ok")
+        _insert_task(self.db_path, "side", status="failed",
+                     parent_task_id="coord", coordinator_part="副",
+                     error_text="炸了")
+        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")
+        children = worker._coord_sync_children(
+            task, task["coordinator_parts"])
+        self.assertEqual(len(children), 2)
+        # 副分工的 failed 子任务未被回 queued，仍是 failed。
+        self.assertEqual(_row(self.db_path, "side")["status"], "failed")
+        result = worker._run_coordinator(
+            task, time.time(), poll_interval=0.01, timeout=60)
+        self.assertEqual(result["status"], "completed")
+        self.assertIn("已容忍", result["stdout"] or "")

cloud/worker.py

@@ -142,7 +142,9 @@ 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 evaluate_outcome as _coord_evaluate
 from cloud.coordinator import parse_parts as _coord_parse_parts
+from cloud.coordinator import parts_by_name as _coord_parts_by_name
 from cloud.coordinator import nonterminal_children as _coord_nonterminal
 from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \
     sum_usage, unpriced_token_cap
@@ -5007,10 +5009,14 @@ class Worker:
         """按分工建/复用子任务，返回 [{name, id}]。
 
         已有子任务按 coordinator_part（分工名）匹配：已完成的直接
-        复用；失败/中断的回 queued 重跑（旧输出清掉）；未终态的
-        不动（正被别的 worker 跑或排队）；缺失的才新建。重跑协调
-        员因此不会重复建子任务。子任务是普通任务：执行器过滤、
-        配额门、key 门都在子任务认领时照常生效。
+        复用；失败/中断的回 queued 重跑（旧输出清掉），但有界
+        委派例外：on_failure=tolerate 分工的 failed 子任务不重
+        排——它的失败已被接受为容忍结果，重排等于撤销容忍；
+        其 interrupted 子任务仍重排，因为中断可能是父任务取消
+        所致、未获公平一轮；未终态的不动（正被别的 worker 跑
+        或排队）；缺失的才新建。重跑协调员因此不会重复建子
+        任务。子任务是普通任务：执行器过滤、配额门、key 门都
+        在子任务认领时照常生效。
         """
         conn = _db_connect(self.db_path)
         try:
@@ -5042,15 +5048,24 @@ class Worker:
                 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"]))
+                        # 有界委派：容忍分工的 failed 子任务接受
+                        # 为结果，不重排；interrupted 仍重排。
+                        if prev["status"] == "failed" and \
+                                part.get("on_failure") == "tolerate":
+                            self._audit(
+                                task["project_id"],
+                                "coordinator_child_tolerated",
+                                "%s -> %s" % (name, prev["id"]))
+                        else:
+                            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
                 # 自动化入闸（创建时门）：子任务认领时虽有配额门，
@@ -5379,15 +5394,20 @@ class Worker:
         task_id = task["id"]
         project_id = task["project_id"]
         parts = task.get("coordinator_parts") or []
+        parts_by_name = _coord_parts_by_name(parts)
+        child_by_id = {}
         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]
+        for ch in children:
+            child_by_id[ch["id"]] = ch["name"]
         deadline = time.time() + timeout
         cancelled = False
         timed_out = False
+        bound_exceeded = []
         while True:
             try:
                 self.heartbeat(task_id)
@@ -5410,6 +5430,33 @@ class Worker:
             if all((rows.get(cid) or {}).get("status") in TERMINAL_STATUSES
                    for cid in child_ids):
                 break
+            # 有界委派：单分工超时在本轮内 cancel 该子任务，
+            # 其他分工继续等，不因一个慢分工拖到全局超时。
+            # 计时起点是协调员本轮开始（started_at），重跑协
+            # 调员即新一轮委派、时钟重算。
+            now = time.time()
+            for cid in child_ids:
+                part = parts_by_name.get(child_by_id.get(cid))
+                if part is None:
+                    continue
+                limit = part.get("timeout_seconds")
+                if limit is None:
+                    continue
+                if (rows.get(cid) or {}).get("status") \
+                        in TERMINAL_STATUSES:
+                    continue
+                if now - started_at >= limit:
+                    try:
+                        self.cancel(cid)
+                    except Exception:
+                        pass
+                    name = child_by_id.get(cid)
+                    if name not in bound_exceeded:
+                        bound_exceeded.append(name)
+                        self._audit(
+                            project_id,
+                            "coordinator_child_bound_exceeded",
+                            "%s: timeout %ds" % (name, limit))
             if time.time() >= deadline:
                 timed_out = True
                 break
@@ -5442,7 +5489,10 @@ class Worker:
                 "status": row.get("status") or "?",
                 "result_text": row.get("result_text"),
                 "error_text": row.get("error_text")})
-        report, overall = _coord_build_report(merged)
+        report, overall = _coord_build_report(merged, parts)
+        if bound_exceeded:
+            report += ("\n\n有界委派超时：%s（已取消该子任务，"
+                       "其他分工继续等齐）" % "、".join(bound_exceeded))
         if orphaned:
             report += ("\n\n孤儿子任务已标 interrupted：%s"
                        % "、".join(orphaned))
@@ -5459,8 +5509,10 @@ class Worker:
             final = overall
             error = None
             if final != "completed":
-                failed = [m["name"] for m in merged
-                          if m["status"] != "completed"]
+                # 错误里只列非容忍失败名：容忍分工的失败已在
+                # 汇总头部点名，不应再进协调员的失败原因。
+                _overall, failed, _tolerated = _coord_evaluate(
+                    merged, parts)
                 error = "coordinator: %s 未完成" % "、".join(failed)
         final_written = self._set_status_if_running(task_id, final)
         if not final_written: