diff --git a/cloud/server.py b/cloud/server.py index 927d11c..8b73837 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -4915,10 +4915,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): " goal/coordinator tasks"}) return # 审计循环接线：audit_loop 只认 True/1（其它真值一律 - # 400，不猜）；协调员父任务不跑模型回合、接线覆盖不到 - # 它，带 audit_loop 建协调员任务属于静默丢弃，同款 - # v9 拒绝逻辑，400 且不建任务。goal 任务允许（goal - # 校验门是接线要消费的确定性门之一）。 + # 400，不猜）；协调员父任务自协调员接线轨起允许携带 + # 开关（合流点在子任务汇总后、父任务终局前调审计循环， + # 不再是“接线覆盖不到”的情形——此前拒绝的理由已消除， + # 见 worker._coord_audit_loop_merge）。子任务仍不直接 + # 接受外部开关：子任务由 worker 内部 INSERT 创建，本来 + # 就不写 audit_loop 列（默认 0），父任务的循环只记在 + # 父任务上，子任务包只是前言载体。goal 任务允许（goal + # 校验门是单任务接线要消费的确定性门之一）。 audit_loop_val = 0 if body.get("audit_loop") is not None: if body.get("audit_loop") not in (True, 1): @@ -4926,11 +4930,6 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "error": "audit_loop must be true or 1"}) return audit_loop_val = 1 - if coord_parts: - self._send_json(400, { - "error": "audit_loop cannot be used with" - " coordinator tasks"}) - return agent_name_val = definition.name if definition else None agent_version_val = definition.version if definition else None agent_instructions_val = ( diff --git a/cloud/tests/test_audit_loop_coord.py b/cloud/tests/test_audit_loop_coord.py new file mode 100644 index 0000000..453a48a --- /dev/null +++ b/cloud/tests/test_audit_loop_coord.py @@ -0,0 +1,444 @@ +"""协调员审计循环接线测试：合流点 + 包拆分 + 开关落库调整。 + +只用 init_db 的本地库与注入的假 runner，不跑真实模型回合、 +不联网。覆盖（按任务书）：合流推回（失败子任务重排、包只含 +自己的 finding、父任务包可读、其余子任务不动）；合流通过 +（包清理）；前置门拒绝直接升级、不耗轮次、不重排子任务； +容忍/用户取消/配额缺失子任务不进门、不成 finding；3 轮仍 +失败升级；主观与确定性拆分（helper 层）；sync 创建与重排时 +的包拆分投递；POST /tasks 协调员带 audit_loop 改为接受。 + +注意：合流逻辑测试直接调 _coord_audit_loop_merge（合流点本身）， +不跑完整 _run_coordinator 轮询循环——轮询要等子任务终态， +这里子任务行已是终态，合流逻辑与 _run_coordinator 调用的 +是同一函数；另有两例线程测试跑完整 _run_coordinator（子任务 +在本轮轮询期间终态，模拟 worker 完成），覆盖父任务 CAS 回 +queued 与升级终局两条分支。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.audit_loop import Finding + 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.audit_loop import Finding + from cloud.server import init_db + from cloud.worker import Worker + + +def _insert(db_path, task_id, status="queued", parent=None, part=None, + result=None, error=None, audit_loop=0, paused=0, reason=None, + parts=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, audit_loop," + " parent_task_id, coordinator_part, result_text, error_text," + " paused_by_user, interrupted_reason, coordinator_parts)" + " VALUES(?, 'p1', 'do', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)", + (task_id, status, now, now, audit_loop, parent, part, + result, error, paused, reason, + json.dumps(parts) if parts is not None else None)) + conn.commit() + finally: + conn.close() + + +def _row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT status, result_text, error_text, audit_loop_packet" + " FROM tasks WHERE id = ?", (task_id,)).fetchone() + finally: + conn.close() + + +def _packet(db_path, task_id): + row = _row(db_path, task_id) + if row is None or row[3] is None: + return None + return json.loads(row[3]) + + +def _state(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT round, last_decision FROM task_audit_loop_state" + " WHERE task_id = ?", (task_id,)).fetchone() + except sqlite3.OperationalError: + return None + finally: + conn.close() + return row + + +def _completed_runner(cmd): + return SimpleNamespace(returncode=0, stdout="{}", stderr="") + + +class CoordAuditTestCase(unittest.TestCase): + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + 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"} + self.audits = [] + self.worker = self._worker() + conn = sqlite3.connect(self.db_path) + try: + self.worker._ensure_result_columns(conn) + conn.commit() + finally: + conn.close() + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = self._old + self.tmp.cleanup() + + def _worker(self): + worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, env=self.env, + max_retries=0, retry_backoff_seconds=0) + worker._audit = lambda p, a, d="": self.audits.append((p, a, d)) + return worker + + def _parent(self, task_id="P1"): + return {"id": task_id, "project_id": "p1", "audit_loop": 1} + + def _parts(self, *names, tolerated=()): + return [{"name": n, + "on_failure": "tolerate" if n in tolerated else "fail"} + for n in names] + + def test_split_findings(self): + det = Finding("child:a", "high", "e", "r", "a", + kind="deterministic", gate="child:a") + sub = Finding("s1", "low", "e", "r", "a", kind="subjective") + d, s = Worker._coord_split_findings([det, sub]) + self.assertEqual([f.id for f in d], ["child:a"]) + self.assertEqual([f.id for f in s], ["s1"]) + + def test_split_packet_only_own_finding(self): + packet = {"task_id": "P1", "round": 2, + "findings": [ + {"id": "child:a", "severity": "high", + "evidence": "ea", "reproduction": "ra", + "acceptance": "aa", "kind": "deterministic", + "gate": "child:a"}, + {"id": "child:b", "severity": "high", + "evidence": "eb", "reproduction": "rb", + "acceptance": "ab", "kind": "deterministic", + "gate": "child:b"}], + "criteria": {"child:a": "child:a", + "child:b": "child:b"}} + split = Worker._coord_split_packet(packet, "a") + self.assertEqual([f["id"] for f in split["findings"]], + ["child:a"]) + self.assertEqual(split["criteria"], {"child:a": "child:a"}) + self.assertEqual(split["round"], 2) + self.assertIsNone(Worker._coord_split_packet(packet, "zzz")) + + def test_merge_pushback(self): + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + _insert(self.db_path, "C2", status="completed", parent="P1", + part="b", result="done") + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a", "b"), + [{"name": "a", "id": "C1"}, {"name": "b", "id": "C2"}]) + self.assertEqual(out["decision"], "pushback") + row_a = _row(self.db_path, "C1") + self.assertEqual(row_a[0], "queued") + self.assertIsNone(row_a[1]) + packet_a = _packet(self.db_path, "C1") + self.assertEqual([f["id"] for f in packet_a["findings"]], + ["child:a"]) + self.assertEqual(_row(self.db_path, "C2")[0], "completed") + parent_packet = _packet(self.db_path, "P1") + self.assertEqual([f["id"] for f in parent_packet["findings"]], + ["child:a"]) + self.assertEqual(_state(self.db_path, "P1")[0], 1) + self.assertTrue(any(a == "worker_coord_audit_loop_pushback" + for _, a, _ in self.audits)) + + def test_merge_pass_clears_packets(self): + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="completed", parent="P1", + part="a", result="done") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET audit_loop_packet = ? WHERE id = ?", + ('{"round": 1, "findings": []}', "C1")) + conn.execute( + "UPDATE tasks SET audit_loop_packet = ? WHERE id = ?", + ('{"round": 1, "findings": []}', "P1")) + conn.commit() + finally: + conn.close() + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "pass") + self.assertIsNone(_packet(self.db_path, "C1")) + self.assertIsNone(_packet(self.db_path, "P1")) + + def test_merge_gate_rejected_escalates_without_round(self): + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="gateway exploded") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS audit_log(user_id TEXT," + " project_id TEXT, action TEXT, detail TEXT," + " created_at DOUBLE PRECISION)") + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES('system', 'p1'," + " 'worker_capability_gate_rejected', ?, ?)", + ("C1: tier gateway model X; refusing to dispatch", time.time())) + conn.commit() + finally: + conn.close() + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out["error"]) + # 不耗轮次：状态表里没有该父任务的行；子任务不重排。 + self.assertIsNone(_state(self.db_path, "P1")) + self.assertEqual(_row(self.db_path, "C1")[0], "failed") + self.assertTrue(any(a == "worker_coord_audit_loop_escalated" + for _, a, _ in self.audits)) + + def test_merge_gate_rejected_by_error_marker(self): + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="Missing environment variable: XCM_API_KEY") + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "escalate") + + def test_merge_excluded_children_pass(self): + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="tolerated boom") + _insert(self.db_path, "C2", status="failed", parent="P1", + part="b", error="user stopped", paused=1) + _insert(self.db_path, "C3", status="completed", parent="P1", + part="c", result="done") + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a", "b", "c", tolerated=("a",)), + [{"name": "a", "id": "C1"}, {"name": "b", "id": "C2"}, + {"name": "c", "id": "C3"}]) + self.assertEqual(out["decision"], "pass") + # 配额缺失的分工不在 children 里：不列门、不成 finding。 + out2 = self.worker._coord_audit_loop_merge( + self._parent("P2"), + self._parts("c"), + [{"name": "c", "id": "C3"}]) + self.assertEqual(out2["decision"], "pass") + + def test_merge_max_rounds_escalates(self): + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom again") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO task_audit_loop_state(task_id, round," + " passed_gates, criteria, last_decision, updated_at)" + " VALUES('P1', 3, '[]', ?, 'pushback', ?)", + ('{"child:a": "child:a"}', time.time())) + conn.commit() + finally: + conn.close() + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "escalate") + self.assertIsNone(_packet(self.db_path, "P1")) + self.assertIsNone(_packet(self.db_path, "C1")) + + def test_merge_switch_off_skips(self): + _insert(self.db_path, "P1", status="running", audit_loop=0) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + out = self.worker._coord_audit_loop_merge( + {"id": "P1", "project_id": "p1", "audit_loop": 0}, + self._parts("a"), [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "skip") + self.assertEqual(_row(self.db_path, "C1")[0], "failed") + + def test_sync_creation_and_requeue_attach_packet(self): + parent_packet = { + "task_id": "P1", "round": 1, + "findings": [{"id": "child:a", "severity": "high", + "evidence": "ea", "reproduction": "ra", + "acceptance": "aa", "kind": "deterministic", + "gate": "child:a"}], + "criteria": {"child:a": "child:a"}} + parent = {"id": "P1", "project_id": "p1", "audit_loop": 1, + "audit_loop_packet": json.dumps(parent_packet), + "created_by": "tester", "priority": 0} + self.worker._quota_exceeded = lambda *a: None + children = self.worker._coord_sync_children( + parent, [{"name": "a", "prompt": "do a"}]) + self.assertEqual(len(children), 1) + packet = _packet(self.db_path, children[0]["id"]) + self.assertIsNotNone(packet) + self.assertEqual([f["id"] for f in packet["findings"]], + ["child:a"]) + # 重排分支：已失败的子任务 + 父包 → 重排并贴拆分包。 + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'failed', error_text = 'boom'" + " WHERE id = ?", (children[0]["id"],)) + conn.execute( + "UPDATE tasks SET audit_loop_packet = NULL WHERE id = ?", + (children[0]["id"],)) + conn.commit() + finally: + conn.close() + children2 = self.worker._coord_sync_children( + parent, [{"name": "a", "prompt": "do a"}]) + self.assertEqual(children2[0]["id"], children[0]["id"]) + row = _row(self.db_path, children[0]["id"]) + self.assertEqual(row[0], "queued") + packet2 = _packet(self.db_path, children[0]["id"]) + self.assertEqual([f["id"] for f in packet2["findings"]], + ["child:a"]) + + + def test_run_coordinator_pushback_full_path(self): + # 完整 _run_coordinator 路径：子任务在本轮轮询期间终态 + # （线程里翻状态模拟 worker 完成），合流推回后父任务回 + # queued、失败子任务回 queued 且带拆分包。直接插 failed + # 行测不到合流——sync 会在轮询前就把它重排走。 + import threading + _insert(self.db_path, "P1", status="running", audit_loop=1, + parts=[{"name": "a", "prompt": "do a"}, + {"name": "b", "prompt": "do b"}]) + _insert(self.db_path, "C1", status="queued", parent="P1", + part="a") + _insert(self.db_path, "C2", status="queued", parent="P1", + part="b") + self.worker._quota_exceeded = lambda *a: None + box = {} + + def _run(): + box["result"] = self.worker._run_coordinator( + self.worker.get_task("P1"), time.time(), + poll_interval=0.01, timeout=10) + + thread = threading.Thread(target=_run) + thread.start() + deadline = time.time() + 5 + while time.time() < deadline: + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'failed'," + " error_text = 'boom' WHERE id = 'C1'") + conn.execute( + "UPDATE tasks SET status = 'completed'," + " result_text = 'done' WHERE id = 'C2'") + conn.commit() + finally: + conn.close() + time.sleep(0.05) + if not thread.is_alive(): + break + thread.join(timeout=5) + self.assertFalse(thread.is_alive()) + result = box["result"] + self.assertEqual(result["status"], "queued") + self.assertEqual(_row(self.db_path, "P1")[0], "queued") + self.assertEqual(_row(self.db_path, "C1")[0], "queued") + packet = _packet(self.db_path, "C1") + self.assertEqual([f["id"] for f in packet["findings"]], + ["child:a"]) + + def test_run_coordinator_escalate_full_path(self): + # 门拒绝子任务在本轮终态 → 合流直接升级：父任务 failed、 + # 错误点名分工、子任务不重排。 + import threading + _insert(self.db_path, "P1", status="running", audit_loop=1, + parts=[{"name": "a", "prompt": "do a"}]) + _insert(self.db_path, "C1", status="queued", parent="P1", + part="a") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS audit_log(user_id TEXT," + " project_id TEXT, action TEXT, detail TEXT," + " created_at DOUBLE PRECISION)") + conn.execute( + "INSERT INTO audit_log VALUES('system', 'p1'," + " 'worker_tool_gate_rejected'," + " 'C1: agent 声明必需工具缺失：websearch', ?)", + (time.time(),)) + conn.commit() + finally: + conn.close() + self.worker._quota_exceeded = lambda *a: None + box = {} + + def _run(): + box["result"] = self.worker._run_coordinator( + self.worker.get_task("P1"), time.time(), + poll_interval=0.01, timeout=10) + + thread = threading.Thread(target=_run) + thread.start() + deadline = time.time() + 5 + while time.time() < deadline: + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'failed'," + " error_text = 'agent 声明必需工具缺失：websearch'" + " WHERE id = 'C1'") + conn.commit() + finally: + conn.close() + time.sleep(0.05) + if not thread.is_alive(): + break + thread.join(timeout=5) + self.assertFalse(thread.is_alive()) + result = box["result"] + self.assertEqual(result["status"], "failed") + self.assertIn("审计循环升级", result["error"]) + self.assertEqual(_row(self.db_path, "P1")[0], "failed") + self.assertEqual(_row(self.db_path, "C1")[0], "failed") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_audit_loop_wire.py b/cloud/tests/test_audit_loop_wire.py index 99a1019..eed2810 100644 --- a/cloud/tests/test_audit_loop_wire.py +++ b/cloud/tests/test_audit_loop_wire.py @@ -347,16 +347,19 @@ class ServerCreateTests(unittest.TestCase): {"project_id": pid, "prompt": "bad", "audit_loop": "yes"}, self.owner_key) self.assertEqual(status, 400) - status, body = self._req( + # 协调员接线轨：父任务带 audit_loop 不再 400——此前拒绝 + # 的理由是“父任务接线覆盖不到、接受即静默丢弃”，合流 + # 接线（worker._coord_audit_loop_merge）落地后该理由已 + # 消除；子任务仍不直接接受外部开关（worker 内部 INSERT + # 不写该列）。非法值仍 400。 + status, task = self._req( "POST", "/tasks", {"project_id": pid, "prompt": "coord", "coordinator_parts": [{"name": "a", "prompt": "do a"}], "audit_loop": True}, self.owner_key) - self.assertEqual(status, 400) - self.assertEqual( - body["error"], - "audit_loop cannot be used with coordinator tasks") + self.assertEqual(status, 201, task) + self.assertEqual(task["audit_loop"], 1) if __name__ == "__main__": diff --git a/cloud/worker.py b/cloud/worker.py index 0dce2a2..3ca998b 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -5536,10 +5536,26 @@ class Worker: 或排队）；缺失的才新建。重跑协调员因此不会重复建子 任务。子任务是普通任务：执行器过滤、配额门、key 门都 在子任务认领时照常生效。 + + 协调员审计循环的包投递也走这里（与合流推回互补）： + 父任务行上的推回包在子任务创建/重排时按分工拆分后 + 写进子任务行，子任务认领时已有的 _apply_audit_loop_packet + 按包列贴前言——这覆盖了合流推回 CAS 失败、以及子任务 + 行消失后重建这两种合流拆分够不到的情形。父包无效或 + 父任务未启用循环时不贴包；子任务 audit_loop 开关本身 + 仍不继承（INSERT 不写该列，默认 0），包只是前言载体。 """ conn = _db_connect(self.db_path) try: _coord_ensure_columns(conn) + # 审计循环包列：INSERT 现在写 audit_loop_packet， + # 老库缺列时这里补上（否则 INSERT 会炸）；认领路径 + # 本来就会调 _ensure_result_columns，这里是双保险， + # 直接调 sync 的测试与旧库同样安全。 + try: + self._ensure_result_columns(conn) + except Exception: + pass try: conn.commit() except Exception: @@ -5574,6 +5590,20 @@ class Worker: continue now = time.time() children = [] + parent_packet = None + if task.get("audit_loop") and task.get("audit_loop_packet"): + parent_packet = _parse_audit_packet( + task.get("audit_loop_packet")) + if parent_packet is None: + try: + log_event(self.logger, + "worker_coord_audit_loop_packet_invalid", + task_id=task["id"], + project_id=task.get("project_id"), + status="warning", + message="parent packet unparsable, no split") + except Exception: + pass conn = _db_connect(self.db_path) try: for part in parts: @@ -5608,12 +5638,29 @@ class Worker: # 人工重试并跑完，无条件 UPDATE 会 # 把已完成任务清结果重排；rowcount # 为 0 说明状态已变，跳过重排与审计。 - _cur = conn.execute( - "UPDATE tasks SET status = 'queued'," - " result_text = NULL, error_text = NULL," - " updated_at = ? WHERE id = ?" - " AND status = ?", - (now, prev["id"], prev["status"])) + # 协调员审计循环：父包里有属于该分 + # 工的 finding 时，把拆分包一并写进 + # 子任务行（合流推回 CAS 失败时的兜底， + # 否则该子任务重跑时看不到推回包）。 + split = self._coord_split_packet( + parent_packet, name) \ + if parent_packet is not None else None + if split is not None: + _cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " result_text = NULL, error_text = NULL," + " audit_loop_packet = ?," + " updated_at = ? WHERE id = ?" + " AND status = ?", + (json.dumps(split), now, prev["id"], + prev["status"])) + else: + _cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " result_text = NULL, error_text = NULL," + " updated_at = ? WHERE id = ?" + " AND status = ?", + (now, prev["id"], prev["status"])) if _cur.rowcount: self._audit( task["project_id"], @@ -5640,14 +5687,18 @@ class Worker: "%s: %s" % (name, gate)) continue child_id = uuid.uuid4().hex + split = self._coord_split_packet( + parent_packet, name) \ + if parent_packet is not None else None 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, child_model, child_effort)" + " coordinator_part, child_model, child_effort," + " audit_loop_packet)" " VALUES(?, ?, ?, 'queued', NULL, '[]', ?, ?, ?, ?," - " NULL, ?, ?, ?, ?, ?, ?, ?, ?)", + " NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (child_id, task["project_id"], part["prompt"], task.get("priority", 0) or 0, task["created_by"], now, now, @@ -5665,7 +5716,8 @@ class Worker: or task.get("effort"), task.get("environment_id"), task["id"], name, task.get("child_model"), - task.get("child_effort"))) + task.get("child_effort"), + json.dumps(split) if split is not None else None)) self._audit( task["project_id"], "coordinator_child_created", "%s -> %s" % (name, child_id)) @@ -5676,34 +5728,65 @@ class Worker: return children def _coord_child_rows(self, child_ids): - """子任务当前行：{id: {status, result_text, error_text}}。 + """子任务当前行：{id: {status, result_text, error_text, + paused_by_user, interrupted_reason}}。 已删或查不到的子任务按 interrupted 终态合成一行（error_text 注明行已消失）：等齐判断里缺失不再等到超时，合流报告里 该子任务标 interrupted 且整体不为 completed；孤儿标记不 会对不存在的行再 UPDATE（rowcount 本来就是 0）。行消失 不是完成，绝不能静默当 completed 计。 + + paused_by_user / interrupted_reason 是协调员审计循环合流 + 分类要用的字段（用户取消的子任务绝不能变成审计推回）： + 老库缺列时回落到不带这两列的查询、按 0/None 计，绝不 + 因缺列把整个子任务从行集里丢掉（与 _coord_sync_children + 的快照容错同款口径）。 """ 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() + try: + fetched = [] + for cid in child_ids: + row = conn.execute( + "SELECT id, status, result_text, error_text," + " paused_by_user, interrupted_reason FROM tasks" + " WHERE id = ?", (cid,)).fetchone() + fetched.append((cid, row)) + except Exception: + fetched = [] + for cid in child_ids: + row = conn.execute( + "SELECT id, status, result_text, error_text" + " FROM tasks WHERE id = ?", (cid,)).fetchone() + fetched.append((cid, row)) + for cid, row in fetched: if row is None: rows[cid] = { "status": "interrupted", "result_text": None, - "error_text": "子任务行已消失（已删除）"} + "error_text": "子任务行已消失（已删除）", + "paused_by_user": 0, + "interrupted_reason": None} continue try: - rows[row["id"]] = { + entry = { "status": row["status"], "result_text": row["result_text"], "error_text": row["error_text"]} except (KeyError, IndexError, TypeError): continue + try: + entry["paused_by_user"] = row["paused_by_user"] + except (KeyError, IndexError, TypeError): + entry["paused_by_user"] = 0 + try: + entry["interrupted_reason"] = \ + row["interrupted_reason"] + except (KeyError, IndexError, TypeError): + entry["interrupted_reason"] = None + rows[row["id"]] = entry finally: conn.close() return rows @@ -5932,6 +6015,378 @@ class Worker: message="orphan sweep failed: %s" % exc) return marked + @staticmethod + def _coord_split_findings(findings): + """主观与确定性发现拆分（协调员审计循环，单任务复审 M3）。 + + evaluate() 只要混入一条主观发现就整体返回 needs_review， + 确定性推回会被主观发现吞掉——所以合流调用方必须先拆分 + 再调 evaluate：确定性进循环，主观只记审计。返回 + (deterministic, subjective)。本方法本身不产生发现， + 合流代码产生的发现全是确定性的，拆分是防御层。 + """ + deterministic = [] + subjective = [] + for finding in findings or []: + if getattr(finding, "kind", None) == "deterministic": + deterministic.append(finding) + else: + subjective.append(finding) + return deterministic, subjective + + @staticmethod + def _coord_split_packet(packet, part_name): + """父任务推回包按分工拆分：只取属于该分工的那一份。 + + packet 为推回包 dict（round/findings/criteria）；finding + 的 id 就是门名 "child:<分工名>"，拆分按 id 精确匹配， + 绝不把别的分工的失败塞进这个子任务的 prompt。无属于 + 该分工的 finding 时返回 None（调用方不写包、不谎称 + 已投递）。返回的新包只含该分工的 findings 与对应的 + criteria 子集，round 与父包一致（轮次只记在父任务上）。 + """ + if not isinstance(packet, dict): + return None + findings = [f for f in (packet.get("findings") or []) + if isinstance(f, dict) + and f.get("id") == "child:" + part_name] + if not findings: + return None + criteria = packet.get("criteria") or {} + return { + "task_id": packet.get("task_id"), + "round": packet.get("round"), + "findings": findings, + "criteria": {f["id"]: criteria.get(f["id"]) + for f in findings if f["id"] in criteria}, + } + + def _coord_gate_rejected_child_ids(self, project_id, child_ids): + """前置门拒绝的子任务 id 集（合流分类用）。 + + 依据只有审计行：能力门、工具门、审批门的拒绝审计 detail + 是 "<子任务 id>: <错误>" 格式，可按 id 前缀归属；缺 key + 门（worker_missing_key）与仓库门（worker_repo_failed） + 的 detail 只有错误文本、没有任务 id，审计行无法归属， + 故不在此列——那两类只走下面的 error_text 强标记回落。 + 审计是 best-effort，查不到不等于没拒绝，调用方不得把 + “无审计行”当成“没被门拒绝”的证据（回落标记见分类）。 + 查询按 created_at 倒序取 5000 行封顶：门拒绝审计是本轮 + 子任务的近期记录，超限时回落标记仍可兜底，绝不全表扫。 + 注意 audit_log 表没有 id 列（只有 user_id/project_id/ + action/detail/created_at），排序只能按 created_at，写 + id 排序会在这里炸掉整条查询——本方法查库异常时记警告 + 日志再返回空集，绝不静默吞掉（静默吞掉会让门拒绝 + 分类悄悄失效、推回本应升级的任务）。 + """ + actions = ("worker_capability_gate_rejected", + "worker_tool_gate_rejected", + "worker_tool_approval_gate_rejected") + try: + conn = _db_connect(self.db_path) + try: + rows = conn.execute( + "SELECT detail FROM audit_log" + " WHERE project_id = ? AND action IN (?, ?, ?)" + " ORDER BY created_at DESC LIMIT 5000", + (project_id, actions[0], actions[1], + actions[2])).fetchall() + finally: + conn.close() + except Exception as exc: + try: + self.logger.warning( + "coord gate audit lookup failed: %s", exc) + except Exception: + pass + return set() + wanted = set(child_ids or []) + found = set() + for row in rows: + try: + detail = row[0] or "" + except (KeyError, IndexError, TypeError): + continue + for cid in wanted: + if detail == cid or detail.startswith(cid + ":"): + found.add(cid) + return found + + def _coord_classify_child(self, part, row, gate_rejected_ids): + """单个子任务的合流分类，返回 (kind, reason)。 + + kind：completed（完成，门通过）、excluded（不进门、不成 + finding：容忍、用户取消/暂停、孤儿中断、配额缺失由调用 + 方另行处理）、gate_rejected（前置门拒绝，走升级路径， + 绝不推回）、execution（可重做的执行失败，成 finding）。 + 分类优先级写死，绝不凭 error_text 猜测覆盖更强的证据： + 完成 > 容忍 > 用户取消/暂停 > 门拒绝（审计行）> + 门拒绝（error_text 强标记回落）> 执行失败。 + 注意 merged 状态：evaluate_outcome 只认 completed 为 + 完成，但子任务 merged 表示工作已落地（PR 已合），审计 + 门按完成计；这与合流总体判定无关——总体仍按 + evaluate_outcome 的旧口径算，审计 pass 不强制父任务 + completed，父任务终局仍走正常路径。 + """ + status = (row or {}).get("status") + if status == "completed" or status == "merged": + return "completed", "completed" + if part is not None and part.get("on_failure") == "tolerate": + return "excluded", "tolerated" + if row.get("paused_by_user") or row.get("interrupted_reason") in ( + "user_cancel", "user_pause"): + return "excluded", "user_cancelled" + if row.get("interrupted_reason") == "coordinator_orphan": + return "excluded", "orphaned" + child_id = row.get("id") + if child_id in (gate_rejected_ids or set()): + return "gate_rejected", "gate_audit" + err = (row.get("error_text") or "") + for marker in ("Missing environment variable:", + "网关准备失败", + "必需工具缺失", + "工具审批声明矛盾", + "refusing to dispatch an incapable or" + " unverifiable model", + "fails capability"): + if marker in err: + return "gate_rejected", "gate_error_marker" + if status == "failed" or status == "interrupted": + return "execution", "execution_failed" + return "excluded", "non_terminal" + + def _coord_clear_packets(self, task, child_ids): + """pass/escalate 时清包（best-effort）：父任务包 + 各子 + 任务包。清失败只记警告，绝不因此改变已定的父任务结局； + 子任务包不清干净的后果只是下次手动重跑该子任务时旧包 + 可能再贴一次前言，不影响审计循环的轮次（轮次只在父 + 任务状态表里）。 + """ + try: + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + conn.execute( + "UPDATE tasks SET audit_loop_packet = NULL," + " updated_at = ? WHERE id = ?", + (time.time(), task["id"])) + for cid in child_ids or []: + conn.execute( + "UPDATE tasks SET audit_loop_packet = NULL," + " updated_at = ? WHERE id = ?", + (time.time(), cid)) + conn.commit() + finally: + conn.close() + except Exception as exc: + try: + log_event(self.logger, "worker_coord_audit_loop_clear_failed", + task_id=task["id"], + project_id=task.get("project_id"), + status="warning", + message="clear packets failed: %s" % exc) + except Exception: + pass + + def _coord_audit_loop_merge(self, task, parts, children): + """协调员审计循环合流：子任务汇总后、父任务终局前调用。 + + 只由 _run_coordinator 的正常合流路径调用（已排除取消与 + 超时；那两条是父任务自身的结局，不是子任务可重做的 + 失败）。返回 dict，decision 为： + - skip：开关关闭、无可列门、无子任务——走正常终局。 + - pass：无未解决的确定性问题——清包，走正常终局 + （注意：pass 不强制父任务 completed，总体仍按 + evaluate_outcome 的旧口径算）。 + - pushback：包已写进父任务行、按分工拆分的包已写进 + 各执行失败子任务行、失败子任务已 CAS 回 queued； + 调用方只需把父任务 CAS 回 queued 并返回 queued 结果。 + - escalate：调用方把父任务置 failed、error 取返回的 + error，走正常终局路径（写回、审计、通知）。 + - needs_review：防御分支（合流只产生确定性发现，正常 + 流程到不了这里）——只记审计，走正常终局。 + - error：合流本身异常——记警告，走正常终局，审计 + 循环失败绝不改变父任务结局。 + 轮次只记在父任务的 task_audit_loop_state 里（经 + audit_loop_evaluate）；子任务各记一套、父子各记一套 + 都禁止。子任务不继承 audit_loop 开关（worker INSERT + 本来就不写该列，默认 0），子任务包只是 prompt 前言 + 载体（_apply_audit_loop_packet 按包列贴，不认开关）， + 子任务自身的 _audit_loop_wire 永远不会跑（开关为 0）。 + """ + task_id = task["id"] + project_id = task.get("project_id") + if not task.get("audit_loop"): + return {"decision": "skip"} + parts_by_name = _coord_parts_by_name(parts) + child_ids = [c["id"] for c in (children or [])] + child_names = {} + for child in children or []: + child_names[child["id"]] = child["name"] + rows = self._coord_child_rows(child_ids) + gate_rejected_ids = self._coord_gate_rejected_child_ids( + project_id, child_ids) + gate_results = {} + findings = [] + gate_rejected = [] + excluded_notes = [] + for cid in child_ids: + name = child_names.get(cid) + part = parts_by_name.get(name) + row = dict(rows.get(cid) or {}) + row["id"] = cid + kind, reason = self._coord_classify_child( + part, row, gate_rejected_ids) + if kind == "completed": + gate_results["child:" + name] = True + elif kind == "excluded": + excluded_notes.append("%s:%s" % (name, reason)) + elif kind == "gate_rejected": + gate_rejected.append(name) + elif kind == "execution": + gate_results["child:" + name] = False + findings.append(_AuditFinding( + "child:" + name, "high", + "child %s %s: %s" % ( + name, row.get("status"), + (row.get("error_text") or ""))[:1000], + "re-run child %s" % name, + "child %s completes" % name, + kind="deterministic", gate="child:" + name)) + missing = [p.get("name") for p in (parts or []) + if isinstance(p, dict) and p.get("name") + and p["name"] not in set(child_names.values())] + if missing: + # 配额跳过的分工在合流里根本没有子任务行：这是 + # 创建时门拒绝（运营方侧），不是 agent 可重做的 + # 失败，按前置门同款理由不列门、不成 finding；只 + # 记审计，总体仍按 evaluate_outcome 的旧口径算。 + excluded_notes.append( + "missing(quota_skipped):%s" % ",".join(missing)) + deterministic, subjective = self._coord_split_findings(findings) + if subjective: + self._audit(project_id, "worker_coord_audit_loop_needs_review", + "%s subjective=%s" % ( + task_id, + ",".join(f.id for f in subjective))) + if gate_rejected: + # 前置门拒绝走升级路径，不进 evaluate、不耗轮次： + # 修复方在运营方（目录、工具配置、网关、key）， + # agent 重跑会确定性地再撞同一道门；轮次不推进， + # 运营方修好门后手动重试父任务仍从同一轮次起算， + # 不因运营方问题被扣轮次。父任务包与子任务包清掉， + # 循环就此终结（终态由调用方写 failed）。 + self._coord_clear_packets(task, child_ids) + self._audit(project_id, "worker_coord_audit_loop_escalated", + "%s gate_rejected=%s excluded=%s" % ( + task_id, ",".join(gate_rejected), + ",".join(excluded_notes))) + return { + "decision": "escalate", + "error": "审计循环升级（前置门拒绝：%s，修复方在运营方，" + "自动重排只会再撞同一道门）" % "、".join(gate_rejected), + "gate_rejected": gate_rejected} + if not gate_results and not deterministic: + self._audit(project_id, "worker_coord_audit_loop_skipped", + "%s excluded=%s" % ( + task_id, ",".join(excluded_notes))) + return {"decision": "skip"} + try: + decision = self.audit_loop_evaluate( + task, gate_results, findings=deterministic) + except Exception as exc: + try: + log_event(self.logger, "worker_coord_audit_loop_error", + task_id=task_id, project_id=project_id, + status="warning", + message="coord audit loop failed: %s" % exc) + except Exception: + pass + return {"decision": "error"} + kind = decision.get("decision") + if kind == "needs_review": + self._audit(project_id, "worker_coord_audit_loop_needs_review", + "%s" % task_id) + return {"decision": "needs_review"} + if kind == "pass": + self._coord_clear_packets(task, child_ids) + self._audit(project_id, "worker_coord_audit_loop_passed", + "%s round=%s excluded=%s" % ( + task_id, decision.get("round"), + ",".join(excluded_notes))) + return {"decision": "pass"} + if kind == "pushback": + packet = decision.get("packet") + if packet is None: + packet_dict = {} + elif hasattr(packet, "to_dict"): + packet_dict = packet.to_dict() + else: + packet_dict = packet + try: + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + conn.execute( + "UPDATE tasks SET audit_loop_packet = ?," + " updated_at = ? WHERE id = ?", + (json.dumps(packet_dict), time.time(), task_id)) + for finding in deterministic: + name = finding.id[len("child:"):] \ + if finding.id.startswith("child:") else None + if name is None: + continue + cid = None + for child in children or []: + if child["name"] == name: + cid = child["id"] + break + if cid is None: + continue + split = self._coord_split_packet( + packet_dict, name) + if split is None: + continue + cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " result_text = NULL, error_text = NULL," + " audit_loop_packet = ?, updated_at = ?" + " WHERE id = ? AND status IN" + " ('failed', 'interrupted')", + (json.dumps(split), time.time(), cid)) + if not cur.rowcount: + self._audit( + project_id, + "worker_coord_audit_loop_pushback_skipped", + "%s -> %s status changed" % (name, cid)) + conn.commit() + finally: + conn.close() + except Exception as exc: + try: + log_event( + self.logger, + "worker_coord_audit_loop_pushback_failed", + task_id=task_id, project_id=project_id, + status="warning", + message="coord pushback store failed: %s" % exc) + except Exception: + pass + return {"decision": "error"} + self._audit(project_id, "worker_coord_audit_loop_pushback", + "%s round=%s findings=%s excluded=%s" % ( + task_id, decision.get("round"), + ",".join(f.id for f in deterministic), + ",".join(excluded_notes))) + return {"decision": "pushback", "packet": packet_dict} + if kind == "escalate": + self._coord_clear_packets(task, child_ids) + esc_error = ( + "审计循环升级（第 %s 轮，原因 %s）" % ( + decision.get("round"), decision.get("reason"))) + return {"decision": "escalate", "error": esc_error} + return {"decision": "error"} + def _run_coordinator(self, task, started_at, poll_interval=None, timeout=None): """协调员执行：建子任务 → 等齐 → 合流汇总写回。 @@ -6056,7 +6511,87 @@ class Worker: if orphaned: report += ("\n\n孤儿子任务已标 interrupted：%s" % "、".join(orphaned)) - if cancelled: + coord_audit = None + if not cancelled and not timed_out and task.get("audit_loop"): + # 协调员审计循环合流点（只在正常合流时跑，不在终局 + # 块里跑——父任务根本不走终局块；取消与超时是父任 + # 务自身的结局，不是子任务可重做的失败，不接循环）。 + # pushback 时父任务不写终局：子任务包与重排已由合 + # 流处理，这里只需把父任务 CAS 回 queued；CAS 失败 + # 说明父任务状态已被改动（例如并发取消），按 + # final_written=False 同款处理为 interrupted，绝不 + # 在状态已变的情况下谎称已推回。合流前已查过取消 + # （循环内每轮查），CAS 窗口只剩并发取消这一窄缝， + # 与 I7 的 CAS 窗口同款残留、rowcount 兜底。 + try: + coord_audit = self._coord_audit_loop_merge( + task, parts, children) + except Exception as exc: + try: + log_event(self.logger, + "worker_coord_audit_loop_error", + task_id=task_id, project_id=project_id, + status="warning", + message="coord audit loop failed: %s" % exc) + except Exception: + pass + coord_audit = {"decision": "error"} + if coord_audit.get("decision") == "pushback": + parent_queued = False + try: + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " updated_at = ? WHERE id = ?" + " AND status = 'running'", + (time.time(), task_id)) + conn.commit() + parent_queued = cur.rowcount == 1 + finally: + conn.close() + except Exception: + parent_queued = False + self._cleanup_task_seed(task) + if parent_queued: + report += ("\n\n审计循环：第 %s 轮推回，失败分工已重排" + % coord_audit.get("packet", {}).get("round")) + 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": "queued", + "executor": self._executor_name(), + "stdout": report, "error": None, + "events": [], "cost_record_id": None} + final = "interrupted" + error = "cancelled" + final_written = False + 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} + if coord_audit.get("decision") == "escalate": + # 升级不走推回：父任务按 failed 终局，正常终局 + # 路径写回、审计、通知；子任务保持各自终态不动 + # （升级的是父任务整体，不是某个子任务）。 + final = "failed" + error = coord_audit.get("error") or \ + "审计循环升级" + report = error + "\n\n" + report + if coord_audit is not None and \ + coord_audit.get("decision") == "escalate": + # 审计循环升级已定好 final/error：跳过下面的正常 + # 终局判定，绝不能让 overall 把 failed 覆盖回去； + # 终局写回、审计、通知走正常路径。 + pass + elif cancelled: final = "interrupted" error = "cancelled" elif timed_out: