diff --git a/cloud/coordinator.py b/cloud/coordinator.py index e3dd37a..4c13b23 100644 --- a/cloud/coordinator.py +++ b/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 diff --git a/cloud/tests/test_coordinator.py b/cloud/tests/test_coordinator.py index f378750..9398a4e 100644 --- a/cloud/tests/test_coordinator.py +++ b/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([{ diff --git a/cloud/tests/test_delegation_bounds.py b/cloud/tests/test_delegation_bounds.py new file mode 100644 index 0000000..5871632 --- /dev/null +++ b/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 "") diff --git a/cloud/worker.py b/cloud/worker.py index 31f7529..6814f53 100644 --- a/cloud/worker.py +++ b/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: