fix: coord audit-loop review B1 + I1-I3 (gate scope, merged alignment, single-tx pushback, store-failure escalate)

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 \

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()

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