diff --git a/cloud/coordinator.py b/cloud/coordinator.py index 4c13b23..7b3211d 100644 --- a/cloud/coordinator.py +++ b/cloud/coordinator.py @@ -239,11 +239,19 @@ def evaluate_outcome(children, parts=None): """合流评价（纯函数）：返回 (overall, failed_names, tolerated_names)。 - completed 计完成；未完成的项中，on_failure 为 tolerate - 的分工计容忍、不拖垮整体，其余计失败。未知分工名按 - fail 处理（fail-closed，绝不把查不到的分工当容忍）。 - 空 children 时 overall 为 failed（与旧 build_report 一致， - 失败名为空——调用方按“无子任务”理解，不按名理解）。 + completed 与 merged 都计完成：merged 表示工作已落地（PR + 已合），是完成的更强形态，不是失败——与本模块的 + COMPLETED_STATUSES（{completed, merged}）以及依赖满足、 + pick_next 等处“merged 视同完成”的口径一致。协调员审计 + 循环复审 I1 的对齐点：此前 evaluate_outcome 只认 + completed，审计门却把 merged 算完成，导致审计 pass、 + 包清掉、父任务却以 failed 终局；改成两边都认 merged + 后，审计 pass 与父任务 completed 对齐。未完成的项中， + on_failure 为 tolerate 的分工计容忍、不拖垮整体，其余 + 计失败。未知分工名按 fail 处理（fail-closed，绝不把查 + 不到的分工当容忍）。空 children 时 overall 为 failed + （与旧 build_report 一致，失败名为空——调用方按“无子 + 任务”理解，不按名理解）。 """ by_name = parts_by_name(parts) has_parts = parts is not None @@ -252,7 +260,7 @@ def evaluate_outcome(children, parts=None): tolerated = [] for ch in children or []: name = ch.get("name") - if ch.get("status") == "completed": + if ch.get("status") in ("completed", "merged"): done += 1 continue part = by_name.get(name) if has_parts else None @@ -282,7 +290,7 @@ def build_report(children, parts=None): done = 0 for ch in children: st = ch.get("status") or "?" - if st == "completed": + if st in ("completed", "merged"): done += 1 part = by_name.get(ch.get("name")) chars = part.get("result_chars") if part is not None else \ diff --git a/cloud/tests/test_audit_loop_coord_fix.py b/cloud/tests/test_audit_loop_coord_fix.py new file mode 100644 index 0000000..20fb5a5 --- /dev/null +++ b/cloud/tests/test_audit_loop_coord_fix.py @@ -0,0 +1,283 @@ +"""协调员审计循环复审修复测试：B1 + I1 + I2 + I3。 + +只用 init_db 的本地库与注入的假 runner，不跑真实模型回合、 +不联网。覆盖（按任务书）： +- B1：旧门拒绝审计行（早于本轮 started_at）不再误判新失败， + 子任务执行失败应推回而非升级；本轮门拒绝审计行仍升级； + started_at 未知时保守保留门拒绝（fail-closed 向升级）。 +- I1：merged 计完成——evaluate_outcome / build_report / + 合流 pass 三处对齐；sync 不把 merged 子任务重排。 +- I2：合流推回事务里父 CAS 失败时，子任务不被孤儿重排、 + 包列不写、且有明确审计；推回成功时父任务已由合流 + CAS 回 queued（hook 不再另做父 CAS）。 +- I3：推回存库分支异常时走升级而非回正常终局，error + 写明存库失败、记 escalated 审计；已消耗的轮次不恢复。 +""" + +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.coordinator import build_report, evaluate_outcome + 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.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, 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," + " coordinator_parts)" + " VALUES(?, 'p1', 'do', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, ?, ?, ?)", + (task_id, status, now, now, audit_loop, parent, part, + result, error, + 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 _set_started(db_path, task_id, started_at): + conn = sqlite3.connect(db_path) + try: + conn.execute("UPDATE tasks SET started_at = ? WHERE id = ?", + (started_at, task_id)) + conn.commit() + finally: + conn.close() + + +def _gate_audit(db_path, detail, created_at): + conn = sqlite3.connect(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', ?, ?)", + (detail, created_at)) + conn.commit() + finally: + conn.close() + + +def _completed_runner(cmd): + return SimpleNamespace(returncode=0, stdout="{}", stderr="") + + +class CoordFixTestCase(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 = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, env=self.env, + max_retries=0, retry_backoff_seconds=0) + self.worker._audit = lambda p, a, d="": self.audits.append((p, a, d)) + 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 _parent(self, task_id="P1"): + return {"id": task_id, "project_id": "p1", "audit_loop": 1} + + def _parts(self, *names): + return [{"name": n, "on_failure": "fail"} for n in names] + + # ---- B1 ---- + def test_b1_old_gate_audit_does_not_block_pushback(self): + # 门拒绝 → 修复 → 执行失败：旧审计行早于本轮 started_at， + # 必须被作用域过滤掉，子任务按执行失败推回而非升级。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + now = time.time() + _set_started(self.db_path, "C1", now) + _gate_audit(self.db_path, "C1: old gate rejection", now - 100) + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "pushback") + self.assertEqual(_row(self.db_path, "C1")[0], "queued") + + def test_b1_current_gate_audit_still_escalates(self): + # 本轮门拒绝审计行晚于 started_at，仍应升级、不耗轮次。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="gateway exploded") + now = time.time() + _set_started(self.db_path, "C1", now) + _gate_audit(self.db_path, "C1: new gate rejection", now + 10) + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "escalate") + self.assertEqual(_row(self.db_path, "C1")[0], "failed") + + def test_b1_unknown_started_at_is_conservative(self): + # started_at 未知（老行/缺列）时保守保留门拒绝，向升级 + # fail-closed，绝不因起始未知就把门拒绝任务推回。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="gateway exploded") + _gate_audit(self.db_path, "C1: gate rejection", + time.time()) + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "escalate") + + # ---- I1 ---- + def test_i1_merged_counts_as_completed_everywhere(self): + overall, failed, _tol = evaluate_outcome( + [{"name": "a", "status": "merged"}, + {"name": "b", "status": "completed"}], + [{"name": "a"}, {"name": "b"}]) + self.assertEqual(overall, "completed") + self.assertEqual(failed, []) + _report, overall2 = build_report( + [{"name": "a", "id": "C1", "status": "merged", + "result_text": "done", "error_text": ""}, + {"name": "b", "id": "C2", "status": "completed", + "result_text": "done", "error_text": ""}], + [{"name": "a"}, {"name": "b"}]) + self.assertEqual(overall2, "completed") + # 合流：merged 子任务按完成计，审计 pass。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="merged", parent="P1", + part="a", result="done") + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "pass") + + def test_i1_sync_does_not_requeue_merged_child(self): + # sync 把 merged 当完成直接复用，绝不回 queued 清结果。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="merged", parent="P1", + part="a", result="done") + self.worker._quota_exceeded = lambda *a: None + children = self.worker._coord_sync_children( + {"id": "P1", "project_id": "p1", "audit_loop": 1, + "created_by": "tester", "priority": 0}, + [{"name": "a", "prompt": "do a"}]) + self.assertEqual(children[0]["id"], "C1") + row = _row(self.db_path, "C1") + self.assertEqual(row[0], "merged") + self.assertEqual(row[1], "done") + + # ---- I2 ---- + def test_i2_parent_cas_failure_rolls_back_children(self): + # 父任务状态已被改动（模拟并发取消）：合流事务回滚， + # 子任务不被孤儿重排、包列不写、且有明确审计。 + _insert(self.db_path, "P1", status="interrupted", + audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "pushback_parent_cas_failed") + self.assertEqual(_row(self.db_path, "C1")[0], "failed") + self.assertIsNone(_packet(self.db_path, "C1")) + self.assertIsNone(_packet(self.db_path, "P1")) + self.assertTrue(any( + a == "worker_coord_audit_loop_pushback_parent_cas_failed" + for _, a, _ in self.audits)) + + def test_i2_pushback_parent_already_queued_by_merge(self): + # 推回成功时父任务已由合流事务 CAS 回 queued（hook + # 不再另做父 CAS）；子任务重排、包可读。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "pushback") + self.assertEqual(_row(self.db_path, "P1")[0], "queued") + self.assertEqual(_row(self.db_path, "C1")[0], "queued") + self.assertIsNotNone(_packet(self.db_path, "C1")) + + # ---- I3 ---- + def test_i3_pushback_store_failure_escalates(self): + # 推回存库分支内任何异常都走升级而非回正常终局： + # 用拆分包抛异常模拟存库分支异常（该异常位于同一 + # try 块内，与真实 DB 写失败走同一条 except 路径）。 + _insert(self.db_path, "P1", status="running", audit_loop=1) + _insert(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + orig = Worker._coord_split_packet + try: + Worker._coord_split_packet = staticmethod( + lambda *a: (_ for _ in ()).throw( + RuntimeError("db write failed"))) + out = self.worker._coord_audit_loop_merge( + self._parent(), self._parts("a"), + [{"name": "a", "id": "C1"}]) + finally: + Worker._coord_split_packet = orig + self.assertEqual(out["decision"], "escalate") + self.assertIn("推回存库失败", out["error"]) + self.assertTrue(any(a == "worker_coord_audit_loop_escalated" + for _, a, _ in self.audits)) + # 升级清包：父包与子包都不应残留。 + self.assertIsNone(_packet(self.db_path, "P1")) + self.assertIsNone(_packet(self.db_path, "C1")) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 3ca998b..b6ded27 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -5610,8 +5610,14 @@ class Worker: name = part["name"] prev = existing.get(name) if prev is not None: + # 协调员审计循环复审 I1：merged 视同完成， + # 直接复用、绝不重排——旧版把 merged 当成 + # “终态非 completed”重排，会把已落地（PR + # 已合）的子任务清结果回 queued，等于销毁 + # 已完成的工作；与 evaluate_outcome / 审计 + # 门把 merged 计为完成保持一致。 if prev["status"] in TERMINAL_STATUSES and \ - prev["status"] != "completed": + prev["status"] not in ("completed", "merged"): # 有界委派：容忍分工的 failed 子任务接受 # 为结果，不重排；interrupted 仍重排。 if prev["status"] == "failed" and \ @@ -5729,7 +5735,7 @@ class Worker: def _coord_child_rows(self, child_ids): """子任务当前行：{id: {status, result_text, error_text, - paused_by_user, interrupted_reason}}。 + paused_by_user, interrupted_reason, started_at}}。 已删或查不到的子任务按 interrupted 终态合成一行（error_text 注明行已消失）：等齐判断里缺失不再等到超时，合流报告里 @@ -5742,6 +5748,16 @@ class Worker: 老库缺列时回落到不带这两列的查询、按 0/None 计，绝不 因缺列把整个子任务从行集里丢掉（与 _coord_sync_children 的快照容错同款口径）。 + + started_at 是子任务本轮尝试的开始时间：run_one 认领时 + 以 time.time() 为 started_at，最终写回时原样写进任务行， + 因此终态行的 started_at 就是本轮尝试的开始、不是创建 + 时间；重排后下一轮认领会覆盖它。协调员审计循环 B1 + 的门拒绝作用域就靠它：门拒绝审计发生在认领之后、终 + 局之前，其 created_at 必然晚于本轮 started_at；上一 + 轮的门拒绝审计必然早于本轮 started_at（上一轮终局 + 在本轮认领之前）。老库缺列时 started_at 按 None 计， + 调用方对 None 一律按“起始未知、保守保留门拒绝”处理。 """ rows = {} conn = _db_connect(self.db_path) @@ -5751,16 +5767,26 @@ class Worker: for cid in child_ids: row = conn.execute( "SELECT id, status, result_text, error_text," - " paused_by_user, interrupted_reason FROM tasks" + " paused_by_user, interrupted_reason," + " started_at 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)) + 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] = { @@ -5768,7 +5794,8 @@ class Worker: "result_text": None, "error_text": "子任务行已消失（已删除）", "paused_by_user": 0, - "interrupted_reason": None} + "interrupted_reason": None, + "started_at": None} continue try: entry = { @@ -5786,6 +5813,10 @@ class Worker: row["interrupted_reason"] except (KeyError, IndexError, TypeError): entry["interrupted_reason"] = None + try: + entry["started_at"] = row["started_at"] + except (KeyError, IndexError, TypeError): + entry["started_at"] = None rows[row["id"]] = entry finally: conn.close() @@ -6061,7 +6092,8 @@ class Worker: for f in findings if f["id"] in criteria}, } - def _coord_gate_rejected_child_ids(self, project_id, child_ids): + def _coord_gate_rejected_child_ids(self, project_id, child_ids, + attempt_started_at=None): """前置门拒绝的子任务 id 集（合流分类用）。 依据只有审计行：能力门、工具门、审批门的拒绝审计 detail @@ -6071,26 +6103,57 @@ class Worker: 故不在此列——那两类只走下面的 error_text 强标记回落。 审计是 best-effort，查不到不等于没拒绝，调用方不得把 “无审计行”当成“没被门拒绝”的证据（回落标记见分类）。 - 查询按 created_at 倒序取 5000 行封顶：门拒绝审计是本轮 - 子任务的近期记录，超限时回落标记仍可兜底，绝不全表扫。 - 注意 audit_log 表没有 id 列（只有 user_id/project_id/ - action/detail/created_at），排序只能按 created_at，写 - id 排序会在这里炸掉整条查询——本方法查库异常时记警告 - 日志再返回空集，绝不静默吞掉（静默吞掉会让门拒绝 - 分类悄悄失效、推回本应升级的任务）。 + + 作用域（协调员审计循环复审 B1）：审计行只追加永不删 + 除，无作用域的查询会把上一轮的门拒绝永久误判为本轮 + 的门拒绝——子任务第 1 次门拒绝→父升级→运营方修门→ + 手动重试→同一子任务重排后执行失败，旧门拒绝审计行 + 仍按 id 前缀命中，执行 finding 被吞、推回永远走不到。 + 因此本方法按子任务本轮尝试开始时间过滤：只认 + created_at >= 该子任务行 started_at 的审计行。 + started_at 由调用方从 _coord_child_rows 取（终态行的 + started_at 就是本轮尝试的开始，见该方法注释）；重排 + 时 result/error 被清掉对作用域是利好——error_text + 强标记回落天然只剩本轮错误，与审计行作用域互补。 + 起始未知（attempt_started_at 未给、该子任务无此项、 + 或 started_at 为 None）时保守保留：该子任务的旧审计 + 行仍计入门拒绝（fail-closed 向升级，绝不因起始未知 + 就把门拒绝任务推回）。audit_log 的 created_at 与 + started_at 同为 time.time() 浮点秒（同一 worker + 时钟），可比；浮点相等时计入（>=，防同毫秒误杀）。 + + 排序：audit_log 表确实有 id 列（SQLite 建表含 id + INTEGER PRIMARY KEY AUTOINCREMENT，PG 含 id BIGINT + IDENTITY——此前注释称“没有 id 列”是错的，真正无 id + 的只是测试手工建的表）。排序用 created_at 倒序、id + 作 tie-break；缺 id 列的库（如手工建表）会炸掉第一 + 条查询，此时回落到仅按 created_at 排序再查一次， + 仍失败才记警告日志返回空集，绝不静默吞掉。 + 查询取 5000 行封顶：超限时回落标记仍可兜底，绝不 + 全表扫；封顶漏行只会导致执行失败走推回而非升级， + 是 fail-open 向推回，不是向放行。 """ actions = ("worker_capability_gate_rejected", "worker_tool_gate_rejected", "worker_tool_approval_gate_rejected") + rows = None 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() + try: + rows = conn.execute( + "SELECT detail, created_at FROM audit_log" + " WHERE project_id = ? AND action IN (?, ?, ?)" + " ORDER BY created_at DESC, id DESC LIMIT 5000", + (project_id, actions[0], actions[1], + actions[2])).fetchall() + except Exception: + rows = conn.execute( + "SELECT detail, created_at 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: @@ -6101,14 +6164,24 @@ class Worker: pass return set() wanted = set(child_ids or []) + started = attempt_started_at or {} found = set() - for row in rows: + for row in rows or []: try: detail = row[0] or "" + created = row[1] except (KeyError, IndexError, TypeError): continue for cid in wanted: - if detail == cid or detail.startswith(cid + ":"): + if not (detail == cid or + detail.startswith(cid + ":")): + continue + begin = started.get(cid) if isinstance(started, dict) \ + else None + if begin is None or created is None: + # 起始未知或审计时间未知：保守保留。 + found.add(cid) + elif created >= begin: found.add(cid) return found @@ -6122,11 +6195,15 @@ class Worker: 分类优先级写死，绝不凭 error_text 猜测覆盖更强的证据： 完成 > 容忍 > 用户取消/暂停 > 门拒绝（审计行）> 门拒绝（error_text 强标记回落）> 执行失败。 - 注意 merged 状态：evaluate_outcome 只认 completed 为 - 完成，但子任务 merged 表示工作已落地（PR 已合），审计 - 门按完成计；这与合流总体判定无关——总体仍按 - evaluate_outcome 的旧口径算，审计 pass 不强制父任务 - completed，父任务终局仍走正常路径。 + 注意 merged 状态：merged 表示工作已落地（PR 已合）， + 是完成的更强形态——审计门按完成计，且 + coordinator.evaluate_outcome / build_report 在协调员 + 审计循环复审 I1 后同样把 merged 计为完成，两边口径 + 已对齐；审计 pass 不再与父任务 failed 并存（merged + 子任务不再拖垮 overall）。pass 本身仍不强制父任务 + completed：用户取消等 excluded 情形下，审计 pass + 只表示“无可重做的确定性失败”，父任务终局仍由 + overall 定（取消的子任务仍会使 overall 为 failed）。 """ status = (row or {}).get("status") if status == "completed" or status == "merged": @@ -6196,13 +6273,27 @@ class Worker: 失败）。返回 dict，decision 为： - skip：开关关闭、无可列门、无子任务——走正常终局。 - pass：无未解决的确定性问题——清包，走正常终局 - （注意：pass 不强制父任务 completed，总体仍按 - evaluate_outcome 的旧口径算）。 - - pushback：包已写进父任务行、按分工拆分的包已写进 - 各执行失败子任务行、失败子任务已 CAS 回 queued； - 调用方只需把父任务 CAS 回 queued 并返回 queued 结果。 + （注意：pass 不强制父任务 completed——用户取消等 + excluded 情形下，pass 只表示“无可重做的确定性失 + 败”，终局仍由 overall 定；但 merged 子任务不再 + 造成 pass 与 failed 并存，见 I1 对齐）。 + - pushback：父包、按分工拆分的子包、失败子任务重排 + 与父任务 CAS 回 queued 已在同一连接、同一事务里 + 一次提交；调用方只需返回 queued 结果，不再另做 + 父 CAS（协调员审计循环复审 I2：旧版先写子任务 + 重排、后由 hook 另做父 CAS，父 CAS 失败时子任务 + 已成孤儿重排；现版父 CAS 失败即整个事务回滚， + 子任务与包列原样不动）。 + - pushback_parent_cas_failed：同一事务里父 CAS + rowcount 不为 1（父任务状态已被改动，例如并发取 + 消）——事务已回滚，子任务未重排、包列未写；调用 + 方按并发取消处理为 interrupted，绝不谎称已推回。 - escalate：调用方把父任务置 failed、error 取返回的 - error，走正常终局路径（写回、审计、通知）。 + error，走正常终局路径（写回、审计、通知）。推回 + 存库失败也走本分支（复审 I3）：存库异常时事务回 + 滚、记 escalated 审计，error 写明是存库失败升级； + 已消耗的一轮不再恢复——升级即终局，轮次不再有 + 意义，不要为了“补轮次”再改状态表。 - needs_review：防御分支（合流只产生确定性发现，正常 流程到不了这里）——只记审计，走正常终局。 - error：合流本身异常——记警告，走正常终局，审计 @@ -6225,7 +6316,9 @@ class Worker: 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) + project_id, child_ids, + {cid: (rows.get(cid) or {}).get("started_at") + for cid in child_ids}) gate_results = {} findings = [] gate_rejected = [] @@ -6260,7 +6353,7 @@ class Worker: # 配额跳过的分工在合流里根本没有子任务行：这是 # 创建时门拒绝（运营方侧），不是 agent 可重做的 # 失败，按前置门同款理由不列门、不成 finding；只 - # 记审计，总体仍按 evaluate_outcome 的旧口径算。 + # 记审计，总体仍按 evaluate_outcome 算。 excluded_notes.append( "missing(quota_skipped):%s" % ",".join(missing)) deterministic, subjective = self._coord_split_findings(findings) @@ -6323,6 +6416,20 @@ class Worker: packet_dict = packet.to_dict() else: packet_dict = packet + # 协调员审计循环复审 I2 + I3：父包、子包、子任务重排 + # 与父任务 CAS 回 queued 必须在同一连接、同一事务 + # 里一次提交。旧版先在本连接写子任务重排、后由 + # hook 在另一连接做父 CAS，父 CAS 失败时子任务已 + # 成孤儿重排（会被别的 worker 认领跑起来、却无父 + # 任务等结果）；现版父 CAS 也在本事务里，rowcount + # 不为 1 即整个事务回滚——子任务与包列原样不动， + # 调用方按并发取消处理为 interrupted，绝不谎称已 + # 推回。事务内任何异常同样回滚，并按 I3 走升级 + # （不是回正常终局）：存库失败时轮次已由 evaluate + # 消耗、包未投递，升级即终局，轮次不再有意义， + # 不要为了“补轮次”再改状态表；升级审计与清包 + # 均为 best-effort，失败不改升级决策本身。 + conn = None try: conn = _db_connect(self.db_path) try: @@ -6359,10 +6466,32 @@ class Worker: project_id, "worker_coord_audit_loop_pushback_skipped", "%s -> %s status changed" % (name, cid)) + parent_cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " updated_at = ? WHERE id = ?" + " AND status = 'running'", + (time.time(), task_id)) + if parent_cur.rowcount != 1: + try: + conn.rollback() + except Exception: + pass + self._audit( + project_id, + "worker_coord_audit_loop_pushback_" + "parent_cas_failed", + "%s parent status changed" % task_id) + return {"decision": + "pushback_parent_cas_failed"} conn.commit() finally: conn.close() except Exception as exc: + if conn is not None: + try: + conn.rollback() + except Exception: + pass try: log_event( self.logger, @@ -6372,7 +6501,20 @@ class Worker: message="coord pushback store failed: %s" % exc) except Exception: pass - return {"decision": "error"} + try: + self._audit( + project_id, "worker_coord_audit_loop_escalated", + "%s store_failed error=%s" % (task_id, exc)) + except Exception: + pass + try: + self._coord_clear_packets(task, child_ids) + except Exception: + pass + return { + "decision": "escalate", + "error": "审计循环升级（推回存库失败：%s，轮次已" + "消耗、包未投递，升级即终局）" % exc} self._audit(project_id, "worker_coord_audit_loop_pushback", "%s round=%s findings=%s excluded=%s" % ( task_id, decision.get("round"), @@ -6516,13 +6658,17 @@ class Worker: # 协调员审计循环合流点（只在正常合流时跑，不在终局 # 块里跑——父任务根本不走终局块；取消与超时是父任 # 务自身的结局，不是子任务可重做的失败，不接循环）。 - # pushback 时父任务不写终局：子任务包与重排已由合 - # 流处理，这里只需把父任务 CAS 回 queued；CAS 失败 - # 说明父任务状态已被改动（例如并发取消），按 + # pushback 时父任务不写终局：父包、子包、子任务重 + # 排与父任务 CAS 回 queued 已由合流在同一事务里 + # 一次提交（复审 I2），这里不再另做父 CAS——旧版 + # 在此另做 CAS 时，若合流已写子任务重排、父 CAS + # 再失败，子任务就成孤儿重排；现版合流返回 + # pushback 即表示父任务已 queued，直接返回 queued + # 结果。合流返回 pushback_parent_cas_failed 表示 + # 同一事务已回滚（父任务状态已被改动，例如并发取 + # 消，子任务未重排、包列未写），按 # final_written=False 同款处理为 interrupted，绝不 - # 在状态已变的情况下谎称已推回。合流前已查过取消 - # （循环内每轮查），CAS 窗口只剩并发取消这一窄缝， - # 与 I7 的 CAS 窗口同款残留、rowcount 兜底。 + # 在状态已变的情况下谎称已推回。 try: coord_audit = self._coord_audit_loop_merge( task, parts, children) @@ -6537,34 +6683,20 @@ class Worker: 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} + 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} + if coord_audit.get("decision") == \ + "pushback_parent_cas_failed": + self._cleanup_task_seed(task) final = "interrupted" error = "cancelled" final_written = False