Audit batch 5: I-9, I-12, I-14, N-1..N-12, single-task wire single-transaction

cloud/agent_def.py

@@ -216,6 +216,17 @@ def parse_agent_tools_column(raw):
     parse logic therefore changes no observable behaviour on
     either side; the old server/worker defaults (None vs []) were
     only different spellings of the same dispatch outcome.
+    Audit batch 5 (N-9, deliberate choice: document, do not filter):
+    the parser guarantees list shape only, NOT item types — a dirty
+    row like ["a", 1] is returned as-is. Downstream tolerates
+    non-string items, which is why filtering here would be churn,
+    not a fix: the worker required-tools gate only formats items
+    with "%s" (missing_required_tools) and compares them with ==
+    (deny check), the server only JSON-dumps the list, and no use
+    site calls a str method on an item. Filtering at parse time
+    would instead change serialization of old dirty rows (the
+    server today echoes them exactly), so the loose contract stays
+    and is locked by test.
     """
     if raw is None:
         return None

cloud/audit_loop.py

@@ -224,6 +224,28 @@ def _validate_gate_results(gate_results):
             raise ValueError("gate_results must map gate name to bool")
 
 
+def _escalate(task_id, prev_round, passed_gates, criteria, reason,
+              regression=(), missing_gates=(), weakened=()):
+    """升级决策 dict（审计批 5，N-8：纯结构整理，行为不变）。
+
+    evaluate() 内四处 escalate 返回原来各写一遍同样的 dict
+    骨架（new_state + decision/round/packet/reason + 三个清单），
+    只有 passed_gates / criteria / reason / 清单内容不同；抽成
+    助手后各处只传不同项。new_state 的 round 恒为 prev_round
+    （升级不推进轮次）、last_decision 恒为 escalate。
+    """
+    new_state = LoopState(
+        task_id, round=prev_round, passed_gates=sorted(passed_gates),
+        criteria=criteria, last_decision="escalate")
+    return {
+        "decision": "escalate", "round": prev_round, "packet": None,
+        "new_state": new_state, "reason": reason,
+        "regression": list(regression),
+        "missing_gates": list(missing_gates),
+        "weakened": list(weakened),
+    }
+
+
 def evaluate(state, task_id, gate_results, findings=None, fixed=None):
     """纯函数：一轮审计的决策。
 
@@ -269,17 +291,10 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
     # 全量回归：上一轮通过的门本轮必须重跑，缺跑即升级。
     missing_gates = sorted(g for g in prev_passed if g not in gate_results)
     if missing_gates:
-        new_state = LoopState(
-            task_id, round=prev_round,
-            passed_gates=sorted(
-                g for g, v in gate_results.items() if v),
-            criteria=prev_criteria, last_decision="escalate")
-        return {
-            "decision": "escalate", "round": prev_round, "packet": None,
-            "new_state": new_state, "reason": "gate_missing",
-            "regression": [], "missing_gates": missing_gates,
-            "weakened": [],
-        }
+        return _escalate(
+            task_id, prev_round,
+            (g for g, v in gate_results.items() if v),
+            prev_criteria, "gate_missing", missing_gates=missing_gates)
 
     passed_now = {g for g, v in gate_results.items() if v}
     failed_now = {g for g, v in gate_results.items() if not v}
@@ -310,15 +325,11 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
         else:
             weakened.append(fid)
     if weakened:
-        new_state = LoopState(
-            task_id, round=prev_round, passed_gates=sorted(passed_now),
-            criteria=prev_criteria, last_decision="escalate")
-        return {
-            "decision": "escalate", "round": prev_round, "packet": None,
-            "new_state": new_state, "reason": "standards_weakened",
-            "regression": [f.to_dict() for f in regression],
-            "missing_gates": [], "weakened": sorted(weakened),
-        }
+        return _escalate(
+            task_id, prev_round, passed_now, prev_criteria,
+            "standards_weakened",
+            regression=[f.to_dict() for f in regression],
+            weakened=sorted(weakened))
 
     # 新标准 = 未解决的旧标准 + 本轮新 finding（含回归）。
     for finding in all_findings:
@@ -348,26 +359,16 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
     # 转 escalate），新行为后看到的是 gate_failed_without_
     # finding 升级——原因更精确，不是行为倒退。
     if failed_now and not all_findings:
-        new_state = LoopState(
-            task_id, round=prev_round, passed_gates=sorted(passed_now),
-            criteria=new_criteria, last_decision="escalate")
-        return {
-            "decision": "escalate", "round": prev_round, "packet": None,
-            "new_state": new_state, "reason": "gate_failed_without_finding",
-            "regression": [f.to_dict() for f in regression],
-            "missing_gates": [], "weakened": [],
-        }
+        return _escalate(
+            task_id, prev_round, passed_now, new_criteria,
+            "gate_failed_without_finding",
+            regression=[f.to_dict() for f in regression])
 
     if prev_round >= MAX_PUSHBACKS:
-        new_state = LoopState(
-            task_id, round=prev_round, passed_gates=sorted(passed_now),
-            criteria=new_criteria, last_decision="escalate")
-        return {
-            "decision": "escalate", "round": prev_round, "packet": None,
-            "new_state": new_state, "reason": "max_rounds_exceeded",
-            "regression": [f.to_dict() for f in regression],
-            "missing_gates": [], "weakened": [],
-        }
+        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)

cloud/routing_dryrun.py

@@ -98,6 +98,25 @@ def _tier_dropped_reason(raw_entry):
     return "unrecognised tier entry (expected string or dict)"
 
 
+def _empty_report(warnings, tiers, dropped, unknown, gateway):
+    """无效策略的空报告（审计批 5，N-10：纯结构整理，行为不变）。
+
+    dry_run_policy_document 的两处早期返回（非对象文档、解析
+    出空策略）原来各写一遍同样的 dict 骨架，只有 warnings 不同；
+    抽成助手后各处只传不同项。传入的列表按引用进报告（与原来
+    直接写字面量时同一语义），不拷贝。
+    """
+    return {
+        "valid": False, "blocked": True, "version": None,
+        "content_fingerprint": None,
+        "global_mode": None, "global_mode_source": None,
+        "tiers": tiers, "dropped_tiers": dropped,
+        "unknown_tiers": unknown, "warnings": warnings,
+        "would_activate": [], "would_shadow": [], "would_off": [],
+        "gateway": gateway,
+    }
+
+
 def dry_run_policy_document(data, env_mode_override=None,
                             gateway_model=None):
     """Dry-run one merged policy document (pure, no IO).
@@ -118,17 +137,10 @@ def dry_run_policy_document(data, env_mode_override=None,
                       if gateway_model else "env_unset"),
            "tiers": {}}
     if not isinstance(data, dict):
-        return {
-            "valid": False, "blocked": True, "version": None,
-            "content_fingerprint": None,
-            "global_mode": None, "global_mode_source": None,
-            "tiers": tiers_report, "dropped_tiers": dropped,
-            "unknown_tiers": unknown,
-            "warnings": ["policy document is not an object; "
-                         "the loader would ignore it entirely"],
-            "would_activate": [], "would_shadow": [], "would_off": [],
-            "gateway": _gw,
-        }
+        return _empty_report(
+            ["policy document is not an object; "
+             "the loader would ignore it entirely"],
+            tiers_report, dropped, unknown, _gw)
     raw_tiers = data.get("tiers")
     if isinstance(raw_tiers, dict):
         for name in sorted(raw_tiers):
@@ -142,15 +154,8 @@ def dry_run_policy_document(data, env_mode_override=None,
             "policy is unusable (no usable tiers, or global mode "
             "outside off/shadow/active); the loader would ignore it "
             "entirely and workers would keep the worker default")
-        return {
-            "valid": False, "blocked": True, "version": None,
-            "content_fingerprint": None,
-            "global_mode": None, "global_mode_source": None,
-            "tiers": tiers_report, "dropped_tiers": dropped,
-            "unknown_tiers": unknown, "warnings": warnings,
-            "would_activate": [], "would_shadow": [], "would_off": [],
-            "gateway": _gw,
-        }
+        return _empty_report(
+            warnings, tiers_report, dropped, unknown, _gw)
     global_mode = policy.mode
     global_source = "policy"
     if env_mode_override:

cloud/server.py

@@ -4200,7 +4200,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         # cost_records 表只在 CostStore 首次初始化时建：新鲜
         # 库上直接查会报 no such table，报告端点不应因此 500。
         # 先建表（幂等），再读；建表失败时成本按空算、报告
-        # 里 token/usd 为 0，不假装成本已计。
+        # 里 token/usd 为 0，不假装成本已计。审计批 5（N-1）：
+        # 下面成本 SELECT 同样加 except 兜底（此前只有 finally、
+        # 没有 except，建表失败会直接抛到 handler 层），失败时
+        # 成本按空算并在报告里标 cost_unavailable。
         try:
             _cost_store = CostStore(self.server.db_path)
             try:
@@ -4238,11 +4241,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 " WHERE project_id = ?",
                 (project_id,),
             ).fetchall()
-            cost_rows = conn.execute(
-                "SELECT task_id, total_tokens, usd_micros"
-                " FROM cost_records WHERE project_id = ?",
-                (project_id,),
-            ).fetchall()
+            cost_unavailable = False
+            try:
+                cost_rows = conn.execute(
+                    "SELECT task_id, total_tokens, usd_micros"
+                    " FROM cost_records WHERE project_id = ?",
+                    (project_id,),
+                ).fetchall()
+            except Exception:
+                # 审计批 5（N-1）：成本查询失败（通常是建表失败后
+                # 表仍不存在）时成本按空算，不抛到 handler 层；
+                # 报告里的 cost_unavailable 标明成本不可用。
+                cost_rows = []
+                cost_unavailable = True
         finally:
             conn.close()
         tasks_by_id = {}
@@ -4275,6 +4286,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             tasks_by_id, costs_by_task, explicit_n=explicit_n)
         report["project_id"] = project_id
         report["truncated"] = truncated
+        report["cost_unavailable"] = cost_unavailable
+        # 审计批 5（N-5）：截断取最早 10000 行（ORDER BY
+        # created_at ASC），与协调员门查询取最新 5000 行方向
+        # 相反——调用方须知样本是哪一段：同时返回样本时间窗
+        # （保留样本的最早/最晚 created_at；无样本时为 None）。
+        if audit_rows:
+            report["sample_window"] = {
+                "earliest_created_at": audit_rows[0]["created_at"],
+                "latest_created_at": audit_rows[-1]["created_at"]}
+        else:
+            report["sample_window"] = {
+                "earliest_created_at": None, "latest_created_at": None}
         # 显式失败清单：解析 worker_capability_gateway_explicit
         # 行（格式：task_id model=.. tier=.. mode=.. missing=..
         # unknown=.. blocked=..；fix4 之前的行无 blocked 项，
@@ -4570,6 +4593,57 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                       for path, size in saved],
         })
 
+    @staticmethod
+    def _resolve_create_agent_definition(body):
+        """建任务的 agent 定义解析（审计批 5，I-9③抽取，行为逐字不变）。
+
+        原 _create_task 内 agent/agent_name 解析段平移：返回
+        (definition, error)——error 非空时调用方按原样发 400 且
+        不建任务；definition 为 None 且 error 为空表示未给定义。
+        副作用：无库写、无响应发送（400 由调用方发，与原地一致）。
+        """
+        if body.get("agent") is not None \
+                and body.get("agent_name") is not None:
+            return None, "agent and agent_name cannot both be used"
+        if body.get("agent") is not None \
+                and body.get("agent_version") is not None:
+            return None, ("agent_version is only for agent_name;"
+                          " an inline agent carries its own version")
+        definition = None
+        if body.get("agent") is not None:
+            if not isinstance(body.get("agent"), dict):
+                return None, "agent must be an object"
+            try:
+                definition = _parse_agent_definition(
+                    body.get("agent"),
+                    known_skills=_agent_known_skills())
+            except ValueError as exc:
+                return None, str(exc)
+        if body.get("agent_name") is not None:
+            try:
+                definition = _resolve_agent(
+                    body.get("agent_name"),
+                    version=body.get("agent_version"),
+                    agents_dir=_agent_default_dir(),
+                    known_skills=_agent_known_skills())
+            except ValueError as exc:
+                return None, str(exc)
+        return definition, None
+
+    @staticmethod
+    def _validate_create_audit_loop(body):
+        """建任务的 audit_loop 开关校验（审计批 5，I-9③抽取，行为逐字不变）。
+
+        原 _create_task 内审计循环接线校验段平移：只认 True/1
+        （其它真值一律错，不猜）。返回 (audit_loop_val, error)——
+        error 非空时调用方按原样发 400 且不建任务。
+        """
+        if body.get("audit_loop") is None:
+            return 0, None
+        if body.get("audit_loop") not in (True, 1):
+            return None, "audit_loop must be true or 1"
+        return 1, None
+
     def _create_task(self, user_id):
         body, err = self._read_json()
         if err:
@@ -4606,39 +4680,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         # 优先于定义；解析出的字段落进已有列，定义身份与
         # 指令落进 agent_name/agent_version/agent_instructions
         # 列（v2），prompt 不再带头。
-        definition = None
-        if body.get("agent") is not None \
-                and body.get("agent_name") is not None:
-            self._send_json(400, {
-                "error": "agent and agent_name cannot both be used"})
+        definition, _agent_err = \
+            self._resolve_create_agent_definition(body)
+        if _agent_err:
+            self._send_json(400, {"error": _agent_err})
             return
-        if body.get("agent") is not None \
-                and body.get("agent_version") is not None:
-            self._send_json(400, {
-                "error": "agent_version is only for agent_name;"
-                         " an inline agent carries its own version"})
-            return
-        if body.get("agent") is not None:
-            if not isinstance(body.get("agent"), dict):
-                self._send_json(400, {"error": "agent must be an object"})
-                return
-            try:
-                definition = _parse_agent_definition(
-                    body.get("agent"),
-                    known_skills=_agent_known_skills())
-            except ValueError as exc:
-                self._send_json(400, {"error": str(exc)})
-                return
-        if body.get("agent_name") is not None:
-            try:
-                definition = _resolve_agent(
-                    body.get("agent_name"),
-                    version=body.get("agent_version"),
-                    agents_dir=_agent_default_dir(),
-                    known_skills=_agent_known_skills())
-            except ValueError as exc:
-                self._send_json(400, {"error": str(exc)})
-                return
         depends_on = body.get("depends_on") or []
         if isinstance(depends_on, str):
             depends_on = [depends_on]
@@ -4976,13 +5022,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         # 就不写 audit_loop 列（默认 0），父任务的循环只记在
         # 父任务上，子任务包只是前言载体。goal 任务允许（goal
         # 校验门是单任务接线要消费的确定性门之一）。
-        audit_loop_val = 0
-        if body.get("audit_loop") is not None:
-            if body.get("audit_loop") not in (True, 1):
-                self._send_json(400, {
-                    "error": "audit_loop must be true or 1"})
-                return
-            audit_loop_val = 1
+        audit_loop_val, _al_err = self._validate_create_audit_loop(body)
+        if _al_err:
+            self._send_json(400, {"error": _al_err})
+            return
         agent_name_val = definition.name if definition else None
         agent_version_val = definition.version if definition else None
         agent_instructions_val = (

cloud/tests/test_audit_batch5.py

@@ -0,0 +1,593 @@
+"""审计批 5（I-9、I-14、N-1～N-12、I-12、事务分离）测试。
+
+只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、
+不联网。按任务书：新测试 + 相关旧套件确认无回归；本文件
+只覆盖本批的新行为与新契约，旧行为由旧套件锁定。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+try:
+    from cloud.agent_def import (
+        missing_required_tools,
+        parse_agent_tools_column,
+    )
+    from cloud.audit_loop import Finding
+    from cloud.auth import AuthStore
+    from cloud.models import parse_routing_policy
+    from cloud.routing_dryrun import dry_run_policy_document
+    from cloud.server import CloudControlPlaneHandler, create_server, init_db
+    from cloud.worker import Worker
+    import cloud.worker as worker_module
+    import cloud.server as server_module
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.agent_def import (
+        missing_required_tools,
+        parse_agent_tools_column,
+    )
+    from cloud.audit_loop import Finding
+    from cloud.auth import AuthStore
+    from cloud.models import parse_routing_policy
+    from cloud.routing_dryrun import dry_run_policy_document
+    from cloud.server import CloudControlPlaneHandler, create_server, init_db
+    from cloud.worker import Worker
+    import cloud.worker as worker_module
+    import cloud.server as server_module
+
+
+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 _insert_task(db_path, task_id, status="queued", audit_loop=0,
+                 goal_mode=0, goal=None, verify_command=None,
+                 goal_max_rounds=None, model=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,"
+            " goal_mode, goal, verify_command, goal_max_rounds, model,"
+            " audit_loop_packet)"
+            " VALUES(?, 'p1', 'do work', ?, NULL, '[]', 0, 'tester',"
+            " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?)",
+            (task_id, status, now, now, audit_loop, goal_mode, goal,
+             verify_command, goal_max_rounds, model, packet))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _set_status(db_path, task_id, status):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                     (status, task_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(
+            "SELECT status, audit_loop_packet, goal, verify_command,"
+            " goal_max_rounds, model 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, 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):
+    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, '["execution"]', '{}', last_decision,
+             time.time()))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class _FailingExecuteConn:
+    """注入失败的连接包装：指定 SQL 片段一律抛错，其余透传。"""
+
+    def __init__(self, real, needle):
+        self._real = real
+        self._needle = needle
+
+    def execute(self, sql, params=()):
+        if self._needle in sql:
+            raise RuntimeError("injected store failure")
+        return self._real.execute(sql, params)
+
+    def __getattr__(self, name):
+        return getattr(self._real, name)
+
+
+class _WorkerCase(unittest.TestCase):
+    def setUp(self):
+        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):
+        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 FingerprintTestCase(unittest.TestCase):
+    def test_fallback_models_change_fingerprint(self):
+        # I-14①：指纹宣称含 fallback_models，变化必须敏感。
+        base = {"mode": "shadow", "tiers": {"default": "m1"},
+                "model_capabilities": {"m1": {"tool_calling": True}}}
+        a = parse_routing_policy(base)
+        b = parse_routing_policy(
+            dict(base, fallback_models=["m2"]))
+        self.assertNotEqual(a.content_fingerprint(),
+                            b.content_fingerprint())
+        c = parse_routing_policy(dict(base, fallback_models=["m2"]))
+        self.assertEqual(b.content_fingerprint(),
+                         c.content_fingerprint())
+
+
+class WireTxTestCase(_WorkerCase):
+    def _patched_worker(self, needle):
+        real_connect = worker_module._db_connect
+
+        def _connect(path):
+            return _FailingExecuteConn(real_connect(path), needle)
+
+        patcher = mock.patch.object(worker_module, "_db_connect",
+                                    _connect)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        return self._worker()
+
+    def test_pushback_store_failure_rolls_back_state(self):
+        # 事务分离：推回 UPDATE 失败时状态保存一并回滚，任务不
+        # 以“已推回状态 + failed 任务行”的半截态落地。
+        _insert_task(self.db_path, "TP", status="failed", audit_loop=1)
+        worker = self._patched_worker("SET status = 'queued'")
+        task = worker.get_task("TP")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "error")
+        self.assertEqual(final, "failed")
+        row = _row(self.db_path, "TP")
+        self.assertEqual(row[0], "failed")
+        self.assertIsNone(row[1])
+        self.assertIsNone(_state_row(self.db_path, "TP"))
+        self.assertNotIn("worker_audit_loop_pushback", self._actions())
+
+    def test_escalate_store_failure_claims_nothing(self):
+        # I-14②（改动后新行为）：升级 UPDATE 失败时接线返回
+        # error、任务保持原终局、状态行不动，且不记“已升级”
+        # 审计——升级审计只在提交后记，绝不谎称已升级。
+        _insert_task(self.db_path, "TE", status="failed", audit_loop=1)
+        _seed_state(self.db_path, "TE", 3, "pushback")
+        worker = self._patched_worker("SET status = 'failed'")
+        task = worker.get_task("TE")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "error")
+        self.assertEqual(final, "failed")
+        self.assertEqual(_row(self.db_path, "TE")[0], "failed")
+        state = _state_row(self.db_path, "TE")
+        self.assertEqual(state[0], 3)
+        self.assertEqual(state[1], "pushback")
+        self.assertNotIn("worker_audit_loop_escalated", self._actions())
+
+    def test_escalate_cas_miss_audits_skipped(self):
+        # N-2：升级 CAS 未命中（状态已被改动）时记
+        # escalate_skipped 审计、不谎称已升级，任务与状态原样不动。
+        _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")
+        worker = self._worker()
+        task = worker.get_task("TC")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "error")
+        self.assertEqual(final, "failed")
+        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",
+                      self._actions())
+        self.assertNotIn("worker_audit_loop_escalated", self._actions())
+
+
+class CoordClearTestCase(_WorkerCase):
+    def test_clear_packets_failure_is_best_effort(self):
+        # I-14③：清包失败只记警告、不炸调用方、不改已定结局。
+        _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()
+        worker._coord_clear_packets(
+            {"id": "PP", "project_id": "p1"}, ["CC"])
+        self.assertIsNone(_row(self.db_path, "PP")[1])
+        self.assertIsNone(_row(self.db_path, "CC")[1])
+
+        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"])
+
+
+class ClassifyAndSplitTestCase(_WorkerCase):
+    def test_merge_failed_is_gate_rejected(self):
+        # N-3：merge-failed 归 gate_rejected，不再是
+        # excluded/non_terminal；容忍声明仍优先。
+        worker = self._worker()
+        kind, reason = worker._coord_classify_child(
+            None, {"id": "C1", "status": "merge-failed"}, set())
+        self.assertEqual((kind, reason), ("gate_rejected", "merge_failed"))
+        kind, _ = worker._coord_classify_child(
+            {"on_failure": "tolerate"},
+            {"id": "C2", "status": "merge-failed"}, set())
+        self.assertEqual(kind, "excluded")
+        kind, _ = worker._coord_classify_child(
+            None, {"id": "C3", "status": "merged"}, set())
+        self.assertEqual(kind, "completed")
+
+    def test_wire_splits_subjective_before_evaluate(self):
+        # N-4：单任务接线先拆分再评估——确定性照常推回，主观
+        # 只记审计，不吞推回。
+        _insert_task(self.db_path, "TS", status="failed", audit_loop=1)
+        worker = self._worker()
+        det = Finding("execution_failed", "high", "boom", "re-run",
+                      "task completes", kind="deterministic",
+                      gate="execution")
+        subj = Finding("design_note", "low", "looks odd", "review",
+                       "human ok", kind="subjective", gate=None)
+        worker._audit_loop_gate_results = (
+            lambda task, final, result, gates_checked=None: (
+                {"execution": False}, [det, subj]))
+        task = worker.get_task("TS")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"stdout": ""}, "boom")
+        self.assertEqual(decision, "pushback")
+        self.assertEqual(final, "queued")
+        needs = [d for _, a, d in self.audits
+                 if a == "worker_audit_loop_needs_review"]
+        self.assertTrue(any("design_note" in d for d in needs))
+
+
+class TerminalSqlTestCase(unittest.TestCase):
+    def test_terminal_sql_matches_old_literal_set(self):
+        # N-6：生成的终态串集合与旧手写字面量逐字等价。
+        from cloud.worker import _terminal_status_sql
+        sql = _terminal_status_sql()
+        inner = sql.strip()[1:-1]
+        got = {part.strip().strip("'") for part in inner.split(",")}
+        self.assertEqual(
+            got,
+            {"completed", "failed", "interrupted", "merged",
+             "merge-failed"})
+
+
+class GoalWireTestCase(_WorkerCase):
+    def test_goal_pushback_leaves_goal_columns_and_bounds_rounds(self):
+        # N-12 定性：推回不改 goal 列；包贴进内存 goal 与 prompt；
+        # 审计轮次 3 轮后必升级（无无界循环）。
+        _insert_task(self.db_path, "TG", status="failed", audit_loop=1,
+                     goal_mode=1, goal="reach the goal",
+                     verify_command="true", goal_max_rounds=3)
+        worker = self._worker()
+        task = worker.get_task("TG")
+        final, _, decision = worker._audit_loop_wire(
+            task, "failed", {"goal_met": False, "goal_error": "nope"},
+            "goal unmet")
+        self.assertEqual(decision, "pushback")
+        self.assertEqual(final, "queued")
+        row = _row(self.db_path, "TG")
+        self.assertEqual(row[2], "reach the goal")
+        self.assertEqual(row[3], "true")
+        self.assertEqual(row[4], 3)
+        packet = row[1]
+        self.assertIsNotNone(packet)
+        injected = worker._apply_audit_loop_packet({
+            "id": "TG", "project_id": "p1", "prompt": "do work",
+            "goal": "reach the goal", "goal_mode": 1,
+            "audit_loop_packet": packet})
+        self.assertTrue(injected["prompt"].startswith("[审计推回包"))
+        self.assertTrue(injected["goal"].startswith("[审计推回包"))
+        # 库里的原 prompt/goal 不动。
+        self.assertEqual(_row(self.db_path, "TG")[2], "reach the goal")
+        # 继续推回直至升级：轮次有界。
+        decisions = []
+        for _ in range(3):
+            _set_status(self.db_path, "TG", "failed")
+            task = worker.get_task("TG")
+            _, _, decision = worker._audit_loop_wire(
+                task, "failed",
+                {"goal_met": False, "goal_error": "nope"}, "goal unmet")
+            decisions.append(decision)
+            if decision == "escalate":
+                break
+        self.assertEqual(decisions[-1], "escalate")
+        self.assertEqual(_row(self.db_path, "TG")[0], "failed")
+        self.assertIsNone(_row(self.db_path, "TG")[1])
+
+
+class ToolApprovalEffectiveTestCase(_WorkerCase):
+    def _insert_approval_task(self, task_id, approvals):
+        _insert_task(self.db_path, task_id, status="queued")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET agent_tools = ?,"
+                " agent_tool_approvals = ? WHERE id = ?",
+                (json.dumps(["websearch"]), json.dumps(approvals),
+                 task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def test_ask_is_audited_as_decline_under_host_deny(self):
+        # I-12：派发时真的调 effective，ask 在行里折算成
+        # decline，ask_honoured 恒为 False。
+        self._insert_approval_task("TA", {"websearch": "ask"})
+        worker = self._worker()
+        out = worker.run_one()
+        self.assertEqual(out["status"], "completed")
+        details = [d for _, a, d in self.audits
+                   if a == "worker_tool_approvals_declared"]
+        self.assertTrue(details)
+        self.assertIn('"websearch"', details[0])
+        self.assertIn('"declared": "ask"', details[0])
+        self.assertIn('"effective": "decline"', details[0])
+        self.assertIn('"ask_honoured": false', details[0])
+
+    def test_auto_accepts_only_when_host_config_is_auto(self):
+        # host approval 取自传给 host 的 --config 文件：auto 时
+        # auto 声明 effective 为 accept；默认 deny 时为 decline。
+        self._insert_approval_task("TB", {"websearch": "auto"})
+        cfg = os.path.join(self.tmp.name, "host-config.json")
+        with open(cfg, "w", encoding="utf-8") as fh:
+            json.dump({"approval": "auto"}, fh)
+        worker = self._worker(config_path=cfg)
+        out = worker.run_one()
+        self.assertEqual(out["status"], "completed")
+        details = [d for _, a, d in self.audits
+                   if a == "worker_tool_approvals_declared"]
+        self.assertIn('"effective": "accept"', details[0])
+
+
+class GatewayExtractTestCase(_WorkerCase):
+    def _policy(self, capable):
+        return {"mode": "active",
+                "tiers": {"default": {
+                    "model": "m1",
+                    "requires": {"tool_calling": True}}},
+                "model_capabilities": {"m1": {"tool_calling": capable}}}
+
+    def test_blocked_and_pass_through(self):
+        # I-9①：抽取后行为逐字不变——被拦时早返回 dict +
+        # 门审计 + 标记点；满足时返回 None 继续派发。
+        _insert_task(self.db_path, "TG1", status="queued")
+        worker = self._worker(routing_policy=self._policy(False))
+        task = {"id": "TG1", "project_id": "p1", "model": "m1"}
+        blocked = worker._gateway_capability_check(
+            task, "sub2api", False, time.time())
+        self.assertIsNotNone(blocked)
+        self.assertEqual(blocked["status"], "failed")
+        self.assertIn("worker_capability_gate_rejected", self._actions())
+        self.assertIn("capability", task.get("_gates_checked", []))
+
+        self.audits.clear()
+        _insert_task(self.db_path, "TG2", status="queued")
+        worker2 = self._worker(routing_policy=self._policy(True))
+        task2 = {"id": "TG2", "project_id": "p1", "model": "m1"}
+        self.assertIsNone(worker2._gateway_capability_check(
+            task2, "sub2api", False, time.time()))
+        self.assertNotIn("worker_capability_gate_rejected",
+                         self._actions())
+
+
+class PersistExtractTestCase(_WorkerCase):
+    def test_writeback_only_for_applied_policy(self):
+        # I-9②：只持久化 source=policy 且 applied 的决策。
+        _insert_task(self.db_path, "TW", status="queued", model=None)
+        worker = self._worker()
+        task = {"id": "TW", "project_id": "p1"}
+        worker._persist_policy_model(
+            task, SimpleNamespace(chosen_model="m9", source="policy",
+                                  applied=True, tier="default"))
+        self.assertEqual(_row(self.db_path, "TW")[5], "m9")
+        self.assertIn("worker_model_persisted", self._actions())
+        self.audits.clear()
+        _insert_task(self.db_path, "TW2", status="queued", model=None)
+        worker._persist_policy_model(
+            {"id": "TW2", "project_id": "p1"},
+            SimpleNamespace(chosen_model="m9", source="shadow",
+                            applied=False, tier="default"))
+        self.assertIsNone(_row(self.db_path, "TW2")[5])
+        self.assertNotIn("worker_model_persisted", self._actions())
+
+
+class ServerHelperTestCase(unittest.TestCase):
+    def test_agent_definition_and_audit_loop_helpers(self):
+        # I-9③：两个小函数与原地行为一致（错误文案逐字）。
+        resolve = CloudControlPlaneHandler._resolve_create_agent_definition
+        _, err = resolve({"agent": {"a": 1}, "agent_name": "x"})
+        self.assertEqual(err, "agent and agent_name cannot both be used")
+        _, err = resolve({"agent": "not-a-dict"})
+        self.assertEqual(err, "agent must be an object")
+        validate = CloudControlPlaneHandler._validate_create_audit_loop
+        self.assertEqual(validate({}), (0, None))
+        self.assertEqual(validate({"audit_loop": True}), (1, None))
+        self.assertEqual(validate({"audit_loop": 1}), (1, None))
+        _, err = validate({"audit_loop": 2})
+        self.assertEqual(err, "audit_loop must be true or 1")
+
+
+class ParseToleranceTestCase(unittest.TestCase):
+    def test_non_string_items_survive_and_are_tolerated(self):
+        # N-9（文档选项）：解析层只保 list 形状，下游容忍非
+        # 字符串项——不过滤、不改旧脏行序列化。
+        self.assertEqual(parse_agent_tools_column('["a", 1]'),
+                         ["a", 1])
+        self.assertEqual(missing_required_tools(["a", 1], ""),
+                         ["a", 1])
+
+
+class DryRunEmptyTestCase(unittest.TestCase):
+    def test_early_returns_share_shape(self):
+        # N-10：两处早期返回经 _empty_report 归一，键集合一致。
+        a = dry_run_policy_document("not-a-dict")
+        b = dry_run_policy_document({"mode": "bogus"})
+        self.assertFalse(a["valid"])
+        self.assertFalse(b["valid"])
+        self.assertEqual(set(a.keys()), set(b.keys()))
+
+
+class ShadowEndpointTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port,
+                                          timeout=10)
+        headers = {"Content-Type": "application/json"}
+        if key:
+            headers["X-API-Key"] = key
+        conn.request(method, path,
+                     json.dumps(body) if body is not None else None,
+                     headers)
+        resp = conn.getresponse()
+        raw = resp.read()
+        conn.close()
+        try:
+            data = json.loads(raw)
+        except ValueError:
+            data = None
+        return resp.status, data
+
+    def _shadow(self):
+        status, report = self._req(
+            "GET", "/routing-shadow?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200, report)
+        return report
+
+    def test_cost_query_failure_flagged_not_500(self):
+        # N-1 + I-14④：成本查询失败时报告仍 200、成本按空算、
+        # cost_unavailable 标明，不抛到 handler 层。
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute("DROP TABLE IF EXISTS cost_records")
+            conn.commit()
+        finally:
+            conn.close()
+
+        class _BoomCostStore:
+            def __init__(self, *args, **kwargs):
+                raise RuntimeError("injected cost store failure")
+
+        with mock.patch.object(server_module, "CostStore",
+                               _BoomCostStore):
+            report = self._shadow()
+        self.assertTrue(report["cost_unavailable"])
+
+    def test_sample_window_and_normal_cost_flag(self):
+        # N-5：截断样本（最早 10000 行）附带最早/最晚时间窗；
+        # 正常路径 cost_unavailable 为 False。
+        conn = sqlite3.connect(self.db_path)
+        try:
+            for task_id, created in (("t1", 10.0), ("t2", 20.0)):
+                conn.execute(
+                    "INSERT INTO audit_log(user_id, project_id, action,"
+                    " detail, created_at) VALUES(?, ?, ?, ?, ?)",
+                    ("system", self.project["id"], "worker_routing",
+                     "%s tier=plan source=shadow applied=False"
+                     " suggested=mx" % task_id, created))
+            conn.commit()
+        finally:
+            conn.close()
+        report = self._shadow()
+        self.assertFalse(report["cost_unavailable"])
+        self.assertEqual(report["sample_window"],
+                         {"earliest_created_at": 10.0,
+                          "latest_created_at": 20.0})
+        self.assertFalse(report["truncated"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_audit_loop_wire.py

@@ -232,9 +232,13 @@ class WireTestCase(unittest.TestCase):
     def test_needs_review_leaves_outcome_untouched(self):
         _insert_task(self.db_path, "T6", status="failed", audit_loop=1)
         worker = self._worker()
-        worker.audit_loop_evaluate = lambda t, g, findings=None, fixed=None: {
-            "enabled": True, "decision": "needs_review", "packet": None,
-            "new_state": None, "reason": "subjective"}
+        # 审计批 5：接线改走 _audit_loop_evaluate_on_conn（同连接
+        # 单事务），桩点随之移到新内核；意图不变——防御分支仍在，
+        # needs_review 时终局、包列一律不动、只记审计。
+        worker._audit_loop_evaluate_on_conn = (
+            lambda conn, t, g, findings=None, fixed=None: {
+                "enabled": True, "decision": "needs_review", "packet": None,
+                "new_state": None, "reason": "subjective"})
         task = worker.get_task("T6")
         final, _, decision = worker._audit_loop_wire(
             task, "failed", {"stdout": ""}, "boom")