Audit batch 6: B1 zero-gate sentinel, I1 known-to-lack marker, I2 wire error returns row status, M1 max-rounds label, M2 chosen-last convention, M5 escalate write retry, M6/M7 clear/pass CAS tests

cloud/audit_loop.py

@@ -345,6 +345,19 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
             "regression": [], "missing_gates": [], "weakened": [],
         }
 
+    # 审计批 6（M1）：轮次耗尽的标签优先——max_rounds 分支
+    # 必须排在 gate_failed_without_finding 分支之前，否则
+    # 轮次耗尽 + 无 finding 时 reason 会被后者掩盖（决策同
+    # 为 escalate、轮次不变，只是原因标签变化；交换顺序不
+    # 改变任何决策，只改标签）。本支不要求 findings 非空，
+    # 但注意它在 pass 分支之后：无 finding、无失败门时已在
+    # 上面返回 pass，不会误升级。
+    if prev_round >= MAX_PUSHBACKS:
+        return _escalate(
+            task_id, prev_round, passed_now, new_criteria,
+            "max_rounds_exceeded",
+            regression=[f.to_dict() for f in regression])
+
     # 有 finding 但门全过、finding 为空的矛盾态：门说通过、发现说有
     # 问题，以 finding 为准（finding 是审计的直接证据）。
     # 审计批 4（I-11）：有失败门但无 finding 时，旧代码走到
@@ -364,12 +377,6 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
             "gate_failed_without_finding",
             regression=[f.to_dict() for f in regression])
 
-    if prev_round >= MAX_PUSHBACKS:
-        return _escalate(
-            task_id, prev_round, passed_now, new_criteria,
-            "max_rounds_exceeded",
-            regression=[f.to_dict() for f in regression])
-
     new_round = prev_round + 1
     packet = PushbackPacket(task_id, new_round, all_findings, new_criteria)
     new_state = LoopState(

cloud/routing_shadow.py

@@ -69,6 +69,13 @@ _DETAIL_RE = re.compile(
     # parse, and legacy suggested-with-whitespace rows still land
     # in the unparseable bucket (counted, batch 2) rather than
     # being silently lost.
+    # Audit batch 6 (M2): the "chosen is last" placement is a
+    # standing convention, not an accident — (.+?) runs to end
+    # of line, so any field appended after chosen would be
+    # swallowed into chosen and misparsed. Writers (the two
+    # worker_routing audit sites in cloud/worker.py) must keep
+    # chosen as the final field; a test in test_audit_batch6.py
+    # locks both sites to that shape.
     r"(?:\s+chosen=(?P<chosen>.+?))?\s*$")
 
 

cloud/tests/test_audit_batch5.py

@@ -231,6 +231,11 @@ class WireTxTestCase(_WorkerCase):
     def test_escalate_cas_miss_audits_skipped(self):
         # N-2：升级 CAS 未命中（状态已被改动）时记
         # escalate_skipped 审计、不谎称已升级，任务与状态原样不动。
+        # 审计批 6（I2）：接线 error 时返回的不再是输入终局，
+        # 而是回滚后读回的行实际状态（queued）；run_one 终局块
+        # 据此抑制终态审计与通知。本用例断言意图随之改写：
+        # 旧断言 final == "failed"（输入终局）已是 I2 要消除的
+        # 矛盾，新断言 final == 行状态 "queued"。
         _insert_task(self.db_path, "TC", status="failed", audit_loop=1)
         _seed_state(self.db_path, "TC", 3, "pushback")
         _set_status(self.db_path, "TC", "queued")
@@ -239,7 +244,7 @@ class WireTxTestCase(_WorkerCase):
         final, _, decision = worker._audit_loop_wire(
             task, "failed", {"stdout": ""}, "boom")
         self.assertEqual(decision, "error")
-        self.assertEqual(final, "failed")
+        self.assertEqual(final, "queued")
         self.assertEqual(_row(self.db_path, "TC")[0], "queued")
         self.assertEqual(_state_row(self.db_path, "TC")[0], 3)
         self.assertIn("worker_audit_loop_escalate_skipped",

cloud/tests/test_audit_batch6.py

@@ -0,0 +1,536 @@
+"""审计修复批 6 测试：B1、I1、I2、M1、M2、M5、M6、M7。
+
+只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、
+不联网。按任务书：M3（报告措辞）、M4（观察项）、M8/M9
+（已知边界）不需要代码，本批只在报告里写明不做；N-12 的
+goal 包前言续跑缺口不在本批范围。
+
+行为变更锁定（本批有意改语义，测试即契约）：
+- B1：run_one 入口初始化 _gates_checked=[]，零门任务两轮后
+  passed_gates 不再含硬编码三门；政策任务标记不受影响。
+- I1：_GATE_ERROR_MARKERS 补全长措辞
+  "known to lack a required capability"，无结构化行时显式拦
+  与复核拒绝仍归 gate_rejected 且有回落审计行。
+- I2：接线 error 时返回行实际状态（不再是输入终局），
+  run_one 终局块在行已不是终态时抑制终态审计与通知；
+  批 5 的 test_escalate_cas_miss_audits_skipped 断言意图
+  随之改写（final == 行状态 queued，不再是输入终局 failed）。
+- M1：轮次耗尽 + 无 finding 时 reason 为 max_rounds_exceeded
+  （不再被 gate_failed_without_finding 掩盖），决策不变。
+- M2：两处 worker_routing 写点均以 chosen 结尾（行尾约定）。
+- M5：升级结果写回有界重试一次（仍带 only_if_status 守卫）。
+- M6：_coord_clear_packets 在清包前即失败时包列原样不动。
+- M7：pass 清包 CAS 未命中按 I2 新行为断言。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+try:
+    from cloud.audit_loop import LoopState, evaluate
+    from cloud.auth import AuthStore
+    from cloud.routing_shadow import parse_worker_routing_detail
+    from cloud.server import init_db
+    from cloud.worker import Worker, _GATE_ERROR_MARKERS
+    import cloud.worker as worker_module
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.audit_loop import LoopState, evaluate
+    from cloud.auth import AuthStore
+    from cloud.routing_shadow import parse_worker_routing_detail
+    from cloud.server import init_db
+    from cloud.worker import Worker, _GATE_ERROR_MARKERS
+    import cloud.worker as worker_module
+
+
+def _insert_task(db_path, task_id, model=None, status="queued",
+                 project_id="p1", audit_loop=0, parent=None, part=None,
+                 error=None, result=None, packet=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,"
+            " model, audit_loop_packet)"
+            " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester',"
+            " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?)",
+            (task_id, project_id, status, now, now, audit_loop,
+             parent, part, result, error, model, packet))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _set_started_at(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 _insert_audit(db_path, project_id, action, detail, created_at=None):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO audit_log(user_id, project_id, action, detail,"
+            " created_at) VALUES(NULL, ?, ?, ?, ?)",
+            (project_id, action, detail,
+             created_at if created_at is not None else time.time()))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _task_row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(
+            "SELECT status, audit_loop_packet FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+
+
+def _state_row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(
+            "SELECT round, passed_gates, last_decision"
+            " FROM task_audit_loop_state WHERE task_id = ?",
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+
+
+def _seed_state(db_path, task_id, round_no, last_decision,
+                passed='["execution"]'):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS task_audit_loop_state("
+            "task_id TEXT PRIMARY KEY, round INTEGER,"
+            " passed_gates TEXT, criteria TEXT, last_decision TEXT,"
+            " updated_at DOUBLE PRECISION)")
+        conn.execute(
+            "INSERT OR REPLACE INTO task_audit_loop_state"
+            "(task_id, round, passed_gates, criteria, last_decision,"
+            " updated_at) VALUES(?, ?, ?, ?, ?, ?)",
+            (task_id, round_no, passed, '{}', last_decision,
+             time.time()))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _routing_details(db_path):
+    conn = sqlite3.connect(db_path)
+    try:
+        return [r[0] for r in conn.execute(
+            "SELECT detail FROM audit_log WHERE action = 'worker_routing'"
+            " ORDER BY id").fetchall()]
+    finally:
+        conn.close()
+
+
+def _completed_runner(cmd):
+    name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
+    return SimpleNamespace(
+        returncode=0,
+        stdout=json.dumps({"task": name, "status": "completed",
+                           "text": "done"}),
+        stderr="")
+
+
+def _failing_runner(cmd):
+    name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
+    return SimpleNamespace(
+        returncode=0,
+        stdout=json.dumps({"task": name, "status": "failed",
+                           "text": "boom"}),
+        stderr="")
+
+
+def _worker(db_path, policy_doc=None, env=None, gateway=None,
+            runner=None):
+    calls = []
+
+    def recording(cmd):
+        calls.append(list(cmd))
+        return (runner or _completed_runner)(cmd)
+
+    worker = Worker(db_path, "/fake/host.py", runner=recording,
+                    env=env or {"XCM_API_KEY": "test-placeholder"},
+                    routing_policy=policy_doc)
+    if gateway is not None:
+        worker._ensure_gateway = lambda task: gateway
+        worker._sub2api_default_model = lambda e: "gateway-model"
+    worker._calls = calls
+    return worker
+
+
+class _RunOneCase(unittest.TestCase):
+    """run_one 级夹具（批 4 GateResultsCheckedTestCase 同款口径）。"""
+
+    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)
+
+    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()
+
+
+class B1ZeroGateTestCase(_RunOneCase):
+    """B1：run_one 入口初始化 _gates_checked，零门不再硬编码三门。"""
+
+    def test_explicit_zero_gate_two_rounds_passed_gates_empty(self):
+        # 显式任务、无网关、无策略、无工具、开审计循环：
+        # 旧代码两轮后 passed_gates 是三门硬编码（漏网的直接
+        # 原因正是本例缺 passed_gates 断言）；新代码应为空集。
+        _insert_task(self.db_path, "T3", model="explicit-model",
+                     audit_loop=1)
+        worker = _worker(self.db_path, gateway=(None, {}),
+                         runner=_failing_runner)
+        for _ in range(2):
+            out = worker.run_one()
+            self.assertEqual(out["status"], "queued")
+        row = _state_row(self.db_path, "T3")
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], 2)
+        self.assertEqual(json.loads(row[1]), [])
+        self.assertEqual(row[2], "pushback")
+
+    def test_policy_two_rounds_marks_survive_init(self):
+        # 对照：政策任务本轮实际跑过的 capability 标记不能被
+        # 入口初始化冲掉，两轮后 passed_gates 仍为 ["capability"]。
+        _insert_task(self.db_path, "T2", audit_loop=1)
+        worker = _worker(self.db_path, {
+            "mode": "active",
+            "tiers": {"default": {"model": "policy-model",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "policy-model": {"tool_calling": True},
+            },
+        }, gateway=(None, {}), runner=_failing_runner)
+        for _ in range(2):
+            out = worker.run_one()
+            self.assertEqual(out["status"], "queued")
+        row = _state_row(self.db_path, "T2")
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], 2)
+        self.assertEqual(json.loads(row[1]), ["capability"])
+        self.assertEqual(row[2], "pushback")
+
+
+class I1MarkerTestCase(unittest.TestCase):
+    """I1：无结构化行时显式拦与复核拒绝仍归 gate_rejected。"""
+
+    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)
+        store = AuthStore(self.db_path)
+        store.conn.close()
+        self.audits = []
+        self.worker = Worker(self.db_path, "/fake/host.py",
+                             runner=_completed_runner,
+                             env={"XCM_API_KEY": "test-placeholder"},
+                             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 _merge(self, child_error, audit_rows=()):
+        now = time.time()
+        _insert_task(self.db_path, "P1", status="running", audit_loop=1,
+                     parent=None)
+        _insert_task(self.db_path, "C1", status="failed", parent="P1",
+                     part="a", error=child_error)
+        _set_started_at(self.db_path, "C1", now - 100)
+        for action, detail in audit_rows:
+            _insert_audit(self.db_path, "p1", action, detail,
+                          created_at=now)
+        return self.worker._coord_audit_loop_merge(
+            {"id": "P1", "project_id": "p1", "audit_loop": 1},
+            [{"name": "a", "prompt": "do a"}],
+            [{"id": "C1", "name": "a"}])
+
+    def test_explicit_block_error_only_is_gate_rejected(self):
+        # 显式网关拦的生产方 error 文案（worker.py 网关块），
+        # 无 gate_rejected 行、无 gateway_explicit 行时仍须
+        # 走子串回落归 gate_rejected（旧代码误归 execution
+        # 而推回）。回落审计行须点名新补的 marker。
+        out = self._merge(
+            "default tier explicit model m lacks capability:"
+            " tool_calling; refusing to dispatch a model known to"
+            " lack a required capability")
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+        self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
+        marker_rows = [d for _p, a, d in self.audits
+                       if a == "worker_coord_gate_classified_by_marker"]
+        self.assertTrue(marker_rows)
+        self.assertIn("C1", marker_rows[0])
+        self.assertIn("marker=known to lack a required capability",
+                      marker_rows[0])
+
+    def test_recheck_reject_error_only_is_gate_rejected(self):
+        # 写回复核拒绝的生产方 error 文案（worker.py 复核块），
+        # 同款无结构化行情形。
+        out = self._merge(
+            "default tier persisted model m lacks capability:"
+            " tool_calling; refusing to dispatch a model known to"
+            " lack a required capability")
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+        self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
+        marker_rows = [d for _p, a, d in self.audits
+                       if a == "worker_coord_gate_classified_by_marker"]
+        self.assertTrue(marker_rows)
+        self.assertIn("C1", marker_rows[0])
+        self.assertIn("marker=known to lack a required capability",
+                      marker_rows[0])
+
+
+class _WireCase(unittest.TestCase):
+    """接线级夹具（批 5 _WorkerCase 同款口径，另加 run_one 级
+    测试所需的 CODEX_PUSH_ALLOW_PRIVATE_TARGETS 环境）。"""
+
+    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.audits = []
+
+    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, **kwargs):
+        kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
+        kwargs.setdefault("max_retries", 0)
+        kwargs.setdefault("retry_backoff_seconds", 0)
+        kwargs.setdefault("runner", _completed_runner)
+        worker = Worker(self.db_path, "/fake/host.py", **kwargs)
+        worker._audit = lambda p, a, d="": self.audits.append((p, a, d))
+        return worker
+
+    def _actions(self):
+        return [a for _, a, _ in self.audits]
+
+
+class I2WireTestCase(_WireCase):
+    """I2 + M7：接线 error 返回行实际状态；pass CAS 未命中同款。"""
+
+    def test_pass_cas_miss_returns_row_status_and_keeps_packet(self):
+        # M7：pass 清包 CAS 未命中——行已被并发重排回 queued，
+        # 接线返回 error + 行状态 queued（不再是输入终局
+        # completed），旧包原样不动（事务回滚），状态行不动。
+        _insert_task(self.db_path, "TP2", status="queued", audit_loop=1,
+                     packet='{"round": 1}')
+        _seed_state(self.db_path, "TP2", 1, "pushback")
+        worker = self._worker()
+        task = worker.get_task("TP2")
+        final, _, decision = worker._audit_loop_wire(
+            task, "completed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "error")
+        self.assertEqual(final, "queued")
+        row = _task_row(self.db_path, "TP2")
+        self.assertEqual(row[0], "queued")
+        self.assertEqual(row[1], '{"round": 1}')
+        state = _state_row(self.db_path, "TP2")
+        self.assertEqual(state[0], 1)
+        self.assertEqual(state[2], "pushback")
+        self.assertNotIn("worker_audit_loop_passed", self._actions())
+
+    def test_run_one_suppresses_terminal_on_error_nonterminal(self):
+        # I2 终局块：接线 error 且行已不是终态时，抑制终态审计
+        # 与 failed 通知，返回体状态与行一致（queued）。
+        _insert_task(self.db_path, "TS", status="queued", audit_loop=1)
+        worker = self._worker(runner=_failing_runner)
+        notifies = []
+        worker._notify_task_finished = lambda *a, **k: notifies.append(
+            (a, k))
+        worker._audit_loop_wire = lambda t, f, r, e: (
+            "queued", "boom", "error")
+        out = worker.run_one()
+        self.assertEqual(out["status"], "queued")
+        self.assertNotIn("worker_failed", self._actions())
+        self.assertNotIn("worker_completed", self._actions())
+        self.assertFalse([n for n in notifies if n[0][2] == "failed"])
+
+    def test_run_one_terminal_still_runs_on_error_terminal(self):
+        # 对照：接线 error 但行仍是终态 failed 时，终态审计与
+        # failed 通知照常跑——抑制只针对“行已不是终态”情形，
+        # 不是见 error 就全抑。
+        _insert_task(self.db_path, "TF", status="queued", audit_loop=1)
+        worker = self._worker(runner=_failing_runner)
+        notifies = []
+        worker._notify_task_finished = lambda *a, **k: notifies.append(
+            (a, k))
+        worker._audit_loop_wire = lambda t, f, r, e: (
+            "failed", "boom", "error")
+        out = worker.run_one()
+        self.assertEqual(out["status"], "failed")
+        self.assertIn("worker_failed", self._actions())
+        self.assertTrue([n for n in notifies if n[0][2] == "failed"])
+
+
+class M1MaxRoundsLabelTestCase(unittest.TestCase):
+    """M1：轮次耗尽 + 无 finding 时 reason 为 max_rounds_exceeded。"""
+
+    def test_max_rounds_label_wins_over_no_finding(self):
+        # 上一轮未通过过的门本轮失败、无 finding、无回归、无缺门：
+        # 旧代码 reason 是 gate_failed_without_finding；新代码
+        # max_rounds 分支在前，reason 为 max_rounds_exceeded，
+        # 决策同为 escalate、轮次不变。
+        state = LoopState("T1", round=3, passed_gates=[],
+                          criteria={}, last_decision="pushback")
+        out = evaluate(state, "T1", {"execution": False}, findings=[])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "max_rounds_exceeded")
+        self.assertEqual(out["new_state"].round, 3)
+        self.assertEqual(out["new_state"].last_decision, "escalate")
+
+
+class M2ChosenLastTestCase(_RunOneCase):
+    """M2：两处 worker_routing 写点均以 chosen 结尾（行尾约定）。"""
+
+    def _assert_chosen_last(self, details):
+        self.assertTrue(details)
+        for detail in details:
+            parsed = parse_worker_routing_detail(detail)
+            self.assertIsNotNone(parsed, detail)
+            last_token = detail.strip().rsplit(" ", 1)[-1]
+            self.assertTrue(last_token.startswith("chosen="), detail)
+
+    def test_policy_routing_row_ends_with_chosen(self):
+        # 策略/shadow 决策路径的 worker_routing 行。
+        _insert_task(self.db_path, "M1", audit_loop=0)
+        worker = _worker(self.db_path, {
+            "mode": "shadow",
+            "tiers": {"default": "policy-model"},
+        }, gateway=(None, {}), runner=_failing_runner)
+        worker.run_one()
+        details = _routing_details(self.db_path)
+        self.assertTrue([d for d in details if "source=shadow" in d])
+        self._assert_chosen_last(
+            [d for d in details if "source=shadow" in d])
+
+    def test_explicit_routing_row_ends_with_chosen(self):
+        # 显式任务路径的 worker_routing 行（source=explicit）。
+        _insert_task(self.db_path, "M2", model="explicit-model",
+                     audit_loop=0)
+        worker = _worker(self.db_path, gateway=(None, {}),
+                         runner=_failing_runner)
+        worker.run_one()
+        details = _routing_details(self.db_path)
+        self.assertTrue([d for d in details if "source=explicit" in d])
+        self._assert_chosen_last(
+            [d for d in details if "source=explicit" in d])
+
+
+class M5EscalateRetryTestCase(_WireCase):
+    """M5：升级结果写回有界重试一次（仍带 failed 守卫）。"""
+
+    def _escalating_worker(self):
+        _insert_task(self.db_path, "TE", status="failed", audit_loop=1)
+        _seed_state(self.db_path, "TE", 3, "pushback")
+        return self._worker()
+
+    def test_escalate_result_write_retried_once(self):
+        # 第一次写失败、第二次成功：接线仍返回 escalate，
+        # 写回恰好调两次（重试生效，且不是无界重试）。
+        worker = self._escalating_worker()
+        calls = []
+        real = worker._write_execution_result
+
+        def flaky(*a, **k):
+            calls.append(1)
+            if len(calls) == 1:
+                raise RuntimeError("transient write failure")
+            return real(*a, **k)
+
+        worker._write_execution_result = flaky
+        task = worker.get_task("TE")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "escalate")
+        self.assertEqual(final, "failed")
+        self.assertEqual(len(calls), 2)
+
+    def test_escalate_result_write_bounded_on_repeated_failure(self):
+        # 两次都失败：仍返回 escalate（不谎报成 error），
+        # 写回恰好调两次，不会无界重试。
+        worker = self._escalating_worker()
+        worker._write_execution_result = mock.Mock(
+            side_effect=RuntimeError("persistent write failure"))
+        task = worker.get_task("TE")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "escalate")
+        self.assertEqual(final, "failed")
+        self.assertEqual(worker._write_execution_result.call_count, 2)
+
+
+class M6ClearPacketsEarlyFailureTestCase(_WireCase):
+    """M6：清包前即失败时包列原样不动（旧测试只在清干净后注错）。"""
+
+    def test_clear_packets_failure_before_clear_keeps_packets(self):
+        _insert_task(self.db_path, "PP", status="failed", audit_loop=1,
+                     packet='{"round": 1}')
+        _insert_task(self.db_path, "CC", status="failed", audit_loop=0,
+                     packet='{"round": 1}')
+        worker = self._worker()
+
+        def _boom(path):
+            raise RuntimeError("injected connect failure")
+
+        with mock.patch.object(worker_module, "_db_connect", _boom):
+            # 不抛错，且包列原样不动——失败不改已定结局。
+            worker._coord_clear_packets(
+                {"id": "PP", "project_id": "p1"}, ["CC"])
+        self.assertEqual(_task_row(self.db_path, "PP")[1],
+                         '{"round": 1}')
+        self.assertEqual(_task_row(self.db_path, "CC")[1],
+                         '{"round": 1}')

cloud/worker.py

@@ -508,6 +508,9 @@ def _mark_gate_checked(task, gate):
     无论哪份拷贝打标记，终局块读到的都是同一份集合。
     终局时 _audit_loop_gate_results 只列入被标记过的门，
     不再把没跑过的门硬编码成通过。
+    审计批 6（B1）：run_one 入口已初始化该列表为空——零门
+    任务的 [] 与旧直接调用的 None 是两个不同哨兵，终局块
+    传 [] 时零门不列入，传 None（旧直接调用）时才走旧行为。
     """
     if not isinstance(task, dict):
         return
@@ -533,6 +536,14 @@ _GATE_ERROR_MARKERS = (
     "工具审批声明矛盾",
     "refusing to dispatch an incapable or unverifiable model",
     "fails capability",
+    # 审计批 6（I1）：显式网关拦与写回复核拒绝共用的措辞
+    # "…refusing to dispatch a model known to lack a required
+    # capability"。选全长措辞而不用较短的 "known to lack"：
+    # 全长措辞在全库只出现在上述两处生产方 error 文案里，
+    # 误命中面最小；较短形式可能被执行失败的 error_text
+    # （例如模型输出复述）偶然命中。误命中方向是 escalate
+    # （安全侧、丢一次重试），不是放行，但精度仍优先。
+    "known to lack a required capability",
 )
 
 
@@ -2047,8 +2058,10 @@ class Worker:
           passed_gates 并持久化——“通过”一词在状态表里失真。
           现在：调用方传 gates_checked（run_one 派发点用
           _mark_gate_checked 在 task["_gates_checked"] 里记的
-          本轮实际跑过的门）时，只列入集合内的门；未传
-          （直接调用的旧调用方）时保持旧行为、三门全记 True。
+          本轮实际跑过的门；审计批 6（B1）起 run_one 入口
+          已初始化该列表为空，零门任务传 []、不再传 None）
+          时，只列入集合内的门；未传（直接调用的旧调用方）
+          时保持旧行为、三门全记 True。
           gate_missing 语义核查：引擎只对“上一轮 passed_gates
           里有、本轮缺失”的门升级；同一任务的跑过门集合在
           两轮间是稳定的（显式任务无策略时两轮都不记
@@ -2100,6 +2113,27 @@ class Worker:
                     gate_results[gate] = True
         return gate_results, findings
 
+    def _wire_error_final(self, conn, task_id, final):
+        """CAS 未命中/存库失败后读回行实际状态（审计批 6，I2）。
+
+        接线 error 时调用方（run_one 终局块）不能再拿输入终局
+        当行状态：CAS 未命中意味着行已被第三方改动（例如重排
+        回 queued），继续按输入终局记终态审计、发 failed 通知
+        会与行状态自相矛盾。同一连接在回滚后仍可读，读到即
+        以行为准；读不到（连接已坏、行已删）时回落输入终局，
+        不比旧行为更差。连接都连不上时调用方无连接可用，
+        不走本方法（行未被接线改动，输入终局即行状态）。
+        """
+        try:
+            row = conn.execute(
+                "SELECT status FROM tasks WHERE id = ?",
+                (task_id,)).fetchone()
+        except Exception:
+            return final
+        if row and row[0]:
+            return row[0]
+        return final
+
     def _audit_loop_wire(self, task, final, result, error):
         """审计循环调用方接线：把引擎决策变成真实的任务结局。
 
@@ -2142,6 +2176,15 @@ class Worker:
         result，自带连接）仍是提交后的独立写：它失败时只记
         警告、不再把已 failed 的任务谎报成 error（旧版写回
         失败会返回 error，尽管状态已是 failed）。
+        审计批 6（I2）：CAS 未命中/存库失败返回 error 时，
+        final 不再是输入终局，而是回滚后读回的行实际状态
+        （例如行已被重排回 queued 时返回 queued）；run_one
+        终局块据此抑制终态审计与通知，不再按输入终局谎报。
+        连接都连不上时无连接可读，仍返回输入终局（行未被
+        接线改动，输入终局即行状态）。
+        审计批 6（M5）：升级成功后的结果写回有界重试一次
+        （仍带 only_if_status="failed" 守卫，幂等），两次
+        都失败时仍只记警告、不谎报成 error。
         goal 任务（审计批 5，N-12 定性，只文档化、不改行为）：
         推回只改 tasks.status 与包列（+ 轮次状态表），不碰
         goal / verify_command / goal_max_rounds 列，也不碰宿主
@@ -2221,7 +2264,10 @@ class Worker:
                             message="task status changed before packet clear")
                     except Exception:
                         pass
-                    return final, error, "error"
+                    # 审计批 6（I2）：回滚后读回行实际状态，不再
+                    # 拿输入终局当行状态返回（行可能已被重排）。
+                    return self._wire_error_final(conn, task_id, final), \
+                        error, "error"
                 conn.commit()
                 self._audit(project_id, "worker_audit_loop_passed",
                             "%s round=%s" % (task_id, decision.get("round")))
@@ -2253,7 +2299,9 @@ class Worker:
                             message="task status changed before pushback")
                     except Exception:
                         pass
-                    return final, error, "error"
+                    # 审计批 6（I2）：回滚后读回行实际状态。
+                    return self._wire_error_final(conn, task_id, final), \
+                        error, "error"
                 conn.commit()
                 self._audit(project_id, "worker_audit_loop_pushback",
                             "%s round=%s findings=%s" % (
@@ -2289,7 +2337,9 @@ class Worker:
                             message="task status changed before escalate")
                     except Exception:
                         pass
-                    return final, error, "error"
+                    # 审计批 6（I2）：回滚后读回行实际状态。
+                    return self._wire_error_final(conn, task_id, final), \
+                        error, "error"
                 conn.commit()
                 # 升级审计在提交后记（与旧接线“引擎已记”的行同款，
                 # 只是时序从评估提交后挪到接线提交后，绝不出现
@@ -2310,13 +2360,27 @@ class Worker:
                         "%s reason=%s round=%s"
                         % (task_id, decision.get("reason"),
                            decision.get("round")))
-                try:
-                    self._write_execution_result(
-                        task_id, result_text=(result or {}).get("stdout", ""),
-                        error_text=esc_error, actual_model=None,
-                        started_at=None, finished_at=time.time(),
-                        only_if_status="failed")
-                except Exception as exc:
+                # 审计批 6（M5）：提交后的结果写回仍是独立写，
+                # 但瞬时失败不再直接认命——有界单次重试（仍带
+                # only_if_status="failed" 守卫，重试是幂等的：
+                # 行被并发重排后守卫挡住，不会覆盖新结果）。
+                # 两次都失败时保持旧行为：只记警告、不把已
+                # failed 的任务谎报成 error，DB error_text
+                # 停留旧文本的残留仍在，见报告。
+                _write_exc = None
+                for _attempt in range(2):
+                    try:
+                        self._write_execution_result(
+                            task_id,
+                            result_text=(result or {}).get("stdout", ""),
+                            error_text=esc_error, actual_model=None,
+                            started_at=None, finished_at=time.time(),
+                            only_if_status="failed")
+                        _write_exc = None
+                        break
+                    except Exception as exc:
+                        _write_exc = exc
+                if _write_exc is not None:
                     # 结果文本写回是提交后的独立写：失败只记警告，
                     # 任务已是 failed，不再谎报成 error。
                     try:
@@ -2325,12 +2389,15 @@ class Worker:
                             "worker_audit_loop_escalate_result_failed",
                             task_id=task_id, project_id=project_id,
                             status="warning",
-                            message="escalate result write failed: %s" % exc)
+                            message="escalate result write failed: %s"
+                            % _write_exc)
                     except Exception:
                         pass
                 return "failed", esc_error, "escalate"
             conn.rollback()
-            return final, error, "error"
+            # 审计批 6（I2）：未知决策同样回滚后读回行实际状态。
+            return self._wire_error_final(conn, task_id, final), \
+                error, "error"
         except Exception as exc:
             try:
                 conn.rollback()
@@ -2343,7 +2410,10 @@ class Worker:
                           message="audit loop wire store failed: %s" % exc)
             except Exception:
                 pass
-            return final, error, "error"
+            # 审计批 6（I2）：异常后回滚，行未被接线改动时读回
+            # 即输入终局本身；行已被第三方改动时以行为准。
+            return self._wire_error_final(conn, task_id, final), \
+                error, "error"
         finally:
             try:
                 conn.close()
@@ -7583,6 +7653,17 @@ class Worker:
         except Exception:
             pass
         task = self.get_task(task_id) or task
+        # 审计批 6（B1，批 4 I-10 的 None 哨兵坍塌）：run_one
+        # 入口立即初始化 _gates_checked 为空列表，使“本轮零
+        # 门跑过”（[]）与“旧直接调用”（None）可区分——否则
+        # 零门任务在终局块传 None 时 _audit_loop_gate_results
+        # 走旧行为分支，把没跑过的三门硬编码成通过（批 4
+        # 实测：显式零门任务两轮后 passed_gates 仍是三门）。
+        # 列表对象经后续 dict(task) 浅拷贝共享，标记点与终局
+        # 块读到同一份集合；协调员父任务不走单任务接线，
+        # 初始化对它无害（内存态，不进库、不进审计）。
+        if not isinstance(task.get("_gates_checked"), list):
+            task["_gates_checked"] = []
         # G10：goal 断点续跑标记——认领后立即清掉库里标记（本轮
         # 已认领为 running，不会重复触发；本轮若再被杀回
         # interrupted，用户需重新调 /goal-resume 才会再置标记，
@@ -8005,6 +8086,9 @@ class Worker:
                 # tier/source/applied，评估时建议模型只能记
                 # unknown；新格式补 suggested/chosen，旧解析
                 # 器仍能读（见 cloud/routing_shadow.py）。
+                # 审计批 6（M2）：chosen 必须是行尾字段——影子
+                # 解析器按行尾锚定 chosen，后续写点不得在其
+                # 后追加字段，否则会被吞进 chosen 而误解析。
                 self._audit(task["project_id"], "worker_routing",
                             "%s tier=%s source=%s applied=%s"
                             " suggested=%s chosen=%s" % (
@@ -8061,6 +8145,8 @@ class Worker:
             # 行：评估函数只计 per-tier explicit_n，不计
             # routed_n、不占 seen（见 routing_shadow.py），
             # 父模型语义不变，档模型绝不覆盖它。
+            # 审计批 6（M2）：同款行尾约定——chosen 必须是
+            # 行尾字段，后续写点不得在其后追加字段。
             self._audit(task["project_id"], "worker_routing",
                         "%s tier=%s source=explicit applied=false"
                         " suggested=- chosen=%s" % (
@@ -8588,6 +8674,7 @@ class Worker:
         # 本轮花过的钱照记，推回不等于没花钱。
         audit_loop_decision = None
         audit_loop_pushback = False
+        audit_loop_suppress = False
         if (final_written and not quota_stop
                 and final in ("completed", "failed")
                 and task.get("audit_loop")
@@ -8596,6 +8683,17 @@ class Worker:
                 final, error, audit_loop_decision = \
                     self._audit_loop_wire(task, final, result, error)
                 audit_loop_pushback = (audit_loop_decision == "pushback")
+                # 审计批 6（I2）：接线 error 且行实际状态已不是
+                # 终态（例如 CAS 未命中后行回 queued）时，终态
+                # 审计/时间线/通知必须随之抑制——按输入终局记
+                # worker_failed 并发 failed 通知会与行状态自相
+                # 矛盾。行仍是终态（回滚情形：failed/completed）
+                # 或已被取消（interrupted，按 interrupted 终局
+                # 处理、不发 failed 通知）时不抑制。
+                audit_loop_suppress = (
+                    audit_loop_decision == "error"
+                    and final not in ("completed", "failed",
+                                      "interrupted"))
             except Exception as exc:
                 try:
                     log_event(self.logger, "worker_audit_loop_wire_failed",
@@ -8615,7 +8713,8 @@ class Worker:
             # 持锁写，否则两个线程共用同一 sqlite 连接会直接报错。
             cost_id = self._record_usage_cost(
                 task, actual_model, usage_total, stdout, fallback_count)
-        if final_written and not audit_loop_pushback:
+        if final_written and not audit_loop_pushback \
+                and not audit_loop_suppress:
             # 审计名须与终态一致：interrupted 任务记 worker_failed
             # 会把“可续跑的中断”误读成失败，误导后续排查。
             # 终态审计细节（2026-10-29 任务 A 教训）：interrupted /
@@ -8668,11 +8767,14 @@ class Worker:
             elif final in ("completed", "failed"):
                 self._notify_task_finished(
                     task_id, task["project_id"], final, error=error)
-        elif audit_loop_pushback:
+        elif audit_loop_pushback or audit_loop_suppress:
             # 推回已由接线处理（状态 queued、包列已写、推回审计
             # 已记）：这里不跑终态审计、不改 final、不发通知。
-            # 注意不能落入下面的 else——else 会把 final 改成
-            # interrupted，与库里的 queued 自相矛盾。
+            # 审计批 6（I2）：接线 error 且行已不是终态时同款
+            # 抑制——这里同样不跑终态审计、不改 final、不发
+            # 通知，final 已是接线读回的行实际状态，返回体
+            # 与行一致。注意不能落入下面的 else——else 会把
+            # final 改成 interrupted，与库里的 queued 自相矛盾。
             pass
         else:
             if goal_converted: