feat: coordinator audit-loop wiring at merge point

Parent audit loop sits after all children are aggregated and
before the parent final state; the single-task _audit_loop_wire
is never applied to parents or children. Parent gates are
per-child gates child:<name> (completed only); execution
failures map to one deterministic finding per child; subjective
findings are split off before evaluate and only audited as
needs_review. Gate-rejected children (capability/tool/approval
audit rows, or distinctive error markers) bypass evaluate and
escalate directly without consuming a round; tolerated, paused
and quota-missing parts are excluded, never findings. Rounds
are counted only in the parent task_audit_loop_state; on
pushback the packet is stored on the parent and split per failed
child into that child's audit_loop_packet column, consumed once
by the existing packet-injection path at next claim; pass and
escalate clear parent and child packets. POST /tasks now allows
audit_loop on coordinator parents (the old 400 rationale is
gone); children still take no external switch.

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: