Audit batch 4: I-6 whitespace audit format, I-7 explicit non-sub2api capability gate, I-8 coord gate attribution, I-10 checked gates, I-11 gate-without-finding escalate, I-13 structured-first classification

cloud/audit_loop.py

@@ -231,9 +231,16 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
     本轮全部门的 {门名: 是否通过}，必须含全部已跑过的门；findings
     为本轮确定性/主观发现；fixed 为 {finding_id: 已修复证据}，调用方
     以此证明上一轮 finding 已修复（无证据不得丢标准）。
+    注意：fixed 仅供测试/未来调用方——生产两条接线（单任务
+    _audit_loop_wire、协调员 _coord_audit_loop_merge）均不传
+    fixed，只认重报与门通过；不要顺手给生产接线加 fixed
+    语义，那是另一项功能。
     返回决策 dict：decision 为 pass / pushback / escalate /
     needs_review 之一；packet 只在 pushback 时有值；new_state 只在
     pass / pushback / escalate 时有值（needs_review 不写状态）。
+    有失败门但无 finding 时返回 escalate（reason 为
+    gate_failed_without_finding），不再走到 PushbackPacket 抛
+    ValueError——调用方拿到的是决策，不是异常。
     """
     findings = list(findings or [])
     fixed = dict(fixed or {})
@@ -329,6 +336,28 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None):
 
     # 有 finding 但门全过、finding 为空的矛盾态：门说通过、发现说有
     # 问题，以 finding 为准（finding 是审计的直接证据）。
+    # 审计批 4（I-11）：有失败门但无 finding 时，旧代码走到
+    # 下面 PushbackPacket 抛 ValueError（findings 为空）——
+    # 调用方拿到的是异常而不是决策。显式升级：门说失败、
+    # 发现说没问题，无法构成推回包，reason 点名缺 finding；
+    # 状态正常保存为 escalate（不再像异常路径那样回滚）。
+    # 生产两条接线实际走不到这支（单任务接线在终局非
+    # completed 时必有 execution_failed finding，协调员合流
+    # 为失败子任务必有 finding），它是直接调用方的契约修复。
+    # 注意 wire 层原来看到的是 evaluate_failed 升级（异常
+    # 转 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": [],
+        }
+
     if prev_round >= MAX_PUSHBACKS:
         new_state = LoopState(
             task_id, round=prev_round, passed_gates=sorted(passed_now),

cloud/models.py

@@ -155,7 +155,11 @@ def parse_model_capabilities(raw):
         return out
     for model, entry in raw.items():
         model = _clean_model_name(model)
-        if not model or not isinstance(entry, dict):
+        # 审计批 4（I-6）：能力目录键含空白时整条丢弃——
+        # 含空白的键在空白分隔的审计 detail 里无法被消费点
+        # 稳健识别，保留它只会让能力检查静默走 unknown。
+        if not model or _model_name_has_whitespace(model) \
+                or not isinstance(entry, dict):
             continue
         cleaned = {}
         valid = True
@@ -432,17 +436,28 @@ def classify_task(task) -> str:
 
 
 def _clean_model_name(value):
-    # Audit fix 3（M7，文档级）：只 strip、不校验字符集。
-    # 含空白的模型名仍会原样进审计 detail，而影子审计
-    # 格式是空白分隔，整行解析失败即无声丢样本（不报错、
-    # 不计数）。这里不加字符集校验——那会改变模型名接受
-    # 语义、可能拦住合法模型名；风险以文档记录，修复
-    # 需另起一轨（审计格式改分隔或模型名白名单）。
+    """把模型名规范成可比较的字符串：仅 strip。
+
+    注意：这里刻意不做空白白名单。decide_routing 里清洗后
+    的 None 会被当成“无显式模型”而改走策略/worker 默认路由，
+    全局白名单会把“显式模型含空白（目前原样派发、provider
+    侧 loud fail）”悄悄改成“静默替换用户选型”，比现状更差。
+    空白模型名的处理因此分两层（审计批 4，I-6）：
+    (a) 策略写入路径白名单——能力目录键、档模型、
+    fallback_models 含空白的一律丢弃；(b) 运行时消费点
+    稳健解析——影子 chosen 贪到行尾、_is_persisted_model
+    按键提取 model 值，不再依赖空白分隔的位置。
+    """
     if isinstance(value, str) and value.strip():
         return value.strip()
     return None
 
 
+def _model_name_has_whitespace(name):
+    """清洗后的模型名是否含空白（审计批 4，I-6 白名单判定）。"""
+    return isinstance(name, str) and any(ch.isspace() for ch in name)
+
+
 def _clean_tiers(raw):
     """Return {tier: model} keeping only valid non-empty entries."""
     tiers = {}
@@ -450,13 +465,17 @@ def _clean_tiers(raw):
         return tiers
     for tier in ROUTING_TIERS:
         model = _clean_model_name(raw.get(tier))
-        if isinstance(model, str):
+        # 审计批 4（I-6）：档模型含空白时整档丢弃（parse 层
+        # 另行记入 dropped_tiers + 警告），不让含空白的模型
+        # 名进空白分隔的审计 detail。
+        if isinstance(model, str) \
+                and not _model_name_has_whitespace(model):
             # A tier entry may itself be a dict {"model": ...}; accept
             # that shape too so policy authors are not punished for it.
             tiers[tier] = model
         elif isinstance(raw.get(tier), dict):
             model = _clean_model_name(raw[tier].get("model"))
-            if model:
+            if model and not _model_name_has_whitespace(model):
                 tiers[tier] = model
     return tiers
 
@@ -501,7 +520,12 @@ class RoutingPolicy:
         self.version = version if isinstance(version, str) \
             and version.strip() else None
         self.mode = mode if mode in ROUTING_MODES else "off"
-        self.tiers = dict(tiers or {})
+        # 审计批 4（I-6）：直接构造的策略同样过白名单——含
+        # 空白的档模型进不了空白分隔的审计 detail，整档丢弃
+        # （与 parse 层口径一致；parse 层另行记 dropped_tiers）。
+        self.tiers = {tier: model for tier, model in (tiers or {}).items()
+                      if not (isinstance(model, str)
+                              and _model_name_has_whitespace(model))}
         self.tier_modes = {}
         for tier, tier_mode in (tier_modes or {}).items():
             if tier in self.tiers and tier_mode in ROUTING_MODES:
@@ -517,7 +541,9 @@ class RoutingPolicy:
         self.fallback_models = []
         for item in fallback_models or []:
             model = _clean_model_name(item)
-            if model and model not in self.fallback_models:
+            # 审计批 4（I-6）：fallback 模型名同款白名单。
+            if model and not _model_name_has_whitespace(model) \
+                    and model not in self.fallback_models:
                 self.fallback_models.append(model)
         self.model_capabilities = parse_model_capabilities(
             model_capabilities)
@@ -598,6 +624,23 @@ def parse_routing_policy(data):
     # for an empty model) is not a second drop and is not logged:
     # the warning must name a tier this pop really removed.
     dropped_tiers = []
+    # 审计批 4（I-6）：_clean_tiers 已把含空白模型名的档
+    # 丢掉，这里只记名——与无效模式/无效 requires 的 pop
+    # 同款可见性。只认 ROUTING_TIERS 内的档（未知档名本
+    # 来就被 _clean_tiers 忽略，不是“被丢弃”）；档已被
+    # _clean_tiers 滤掉时，后面的 mode/requires pop 因
+    # `tier in tiers` 为假不会重复记，不会双计。
+    for tier, entry in raw_tiers.items():
+        if tier not in ROUTING_TIERS or tier in tiers:
+            continue
+        raw_model = entry if isinstance(entry, str) else \
+            entry.get("model") if isinstance(entry, dict) else None
+        cleaned = _clean_model_name(raw_model)
+        if cleaned and _model_name_has_whitespace(cleaned):
+            dropped_tiers.append(tier)
+            _logger.warning(
+                "routing policy tier %r dropped: model name contains"
+                " whitespace %r", tier, cleaned)
     tier_modes = {}
     for tier, entry in raw_tiers.items():
         if not isinstance(entry, dict) or "mode" not in entry:

cloud/routing_shadow.py

@@ -60,7 +60,16 @@ _DETAIL_RE = re.compile(
     r"^(?P<task_id>\S+)\s+tier=(?P<tier>\S+)\s+source=(?P<source>\S+)"
     r"\s+applied=(?P<applied>\S+)"
     r"(?:\s+suggested=(?P<suggested>\S+))?"
-    r"(?:\s+chosen=(?P<chosen>\S+))?\s*$")
+    # Audit batch 4 (I-6): chosen is the LAST field, so it may
+    # greedily run to end of line — a legacy chosen model name
+    # containing whitespace ("my model") now parses instead of
+    # dropping the whole row. suggested stays \S+ (it is a middle
+    # field); new policy models cannot contain whitespace (the
+    # batch-4 policy whitelist drops them), so new rows always
+    # parse, and legacy suggested-with-whitespace rows still land
+    # in the unparseable bucket (counted, batch 2) rather than
+    # being silently lost.
+    r"(?:\s+chosen=(?P<chosen>.+?))?\s*$")
 
 
 def parse_worker_routing_detail(detail):
@@ -70,9 +79,11 @@ def parse_worker_routing_detail(detail):
     A ``-`` placeholder also means None, never a model named "-".
     Audit batch 2 (I-1): 解析语义本身不变（失败仍返回 None），
     但 evaluate_shadow 不再静默跳过：不可解析的行计入
-    dropped_rows["unparseable"]，报告里可见（例如模型名含
-    空白会使整行匹配失败，见 M7——行不再无声消失，只是
-    不计为样本）。
+    dropped_rows["unparseable"]，报告里可见。
+    Audit batch 4 (I-6): chosen 是行尾字段，含空白的
+    legacy chosen 模型名现在能解析（见 _DETAIL_RE 注释）；
+    suggested 仍要求无空白，legacy suggested 含空白的行
+    仍进 unparseable 桶计数，不静默消失。
     """
     if not isinstance(detail, str):
         return None
@@ -89,7 +100,7 @@ def parse_worker_routing_detail(detail):
         "source": match.group("source"),
         "applied": applied,
         "suggested_model": None if suggested in (None, "-") else suggested,
-        "chosen_model": None if chosen in (None, "-") else chosen,
+        "chosen_model": None if chosen in (None, "-") else chosen.strip(),
     }
 
 

cloud/tests/test_audit_batch4.py

@@ -0,0 +1,673 @@
+"""审计修复批 4 测试：I-6、I-7、I-8、I-10、I-11、I-13。
+
+只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、
+不联网、不登录任何服务；env 里的值只是占位用的假名。
+
+行为变更锁定（本批有意改语义，测试即契约）：
+- I-6：策略写入路径（能力目录键、档模型、fallback_models）
+  含空白的一律丢弃；影子行 chosen（行尾字段）含空白可解析；
+  persisted 行模型含空白按鍵提取、复核不再默认显式 bypass。
+- I-7：显式 + 非 sub2api 网关在 active 下已知不满足时从
+  “直接派发”变为“拦”（同款审计 action、error 文案、
+  blocked 标志）；unknown-only 与 shadow 仍只审计。
+- I-8：缺 key / 仓库门审计 detail 带任务 id 前缀，补进门拒绝
+  查询集合；仓库解析失败子任务归 gate_rejected、父走升级。
+- I-10：gate_results 只列本轮实际跑过的门；显式与政策任务
+  两轮往返均不误触发 gate_missing。
+- I-11：失败门 + 空 findings 时 evaluate 返回 escalate
+  （gate_failed_without_finding），不再抛 ValueError。
+- I-13：分类优先读结构化审计行，子串只作最后回落且回落
+  命中时记 worker_coord_gate_classified_by_marker 审计行。
+"""
+
+import json
+import logging
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.audit_loop import Finding, evaluate
+    from cloud.auth import AuthStore
+    from cloud.models import (
+        RoutingPolicy, evaluate_capability, parse_model_capabilities,
+        parse_routing_policy)
+    from cloud.routing_shadow import (
+        evaluate_shadow, parse_worker_routing_detail)
+    from cloud.server import init_db
+    from cloud.worker import Worker, _GATE_ERROR_MARKERS
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.audit_loop import Finding, evaluate
+    from cloud.auth import AuthStore
+    from cloud.models import (
+        RoutingPolicy, evaluate_capability, parse_model_capabilities,
+        parse_routing_policy)
+    from cloud.routing_shadow import (
+        evaluate_shadow, parse_worker_routing_detail)
+    from cloud.server import init_db
+    from cloud.worker import Worker, _GATE_ERROR_MARKERS
+
+
+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):
+    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)"
+            " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester',"
+            " ?, ?, NULL, ?, ?, ?, ?, ?, ?)",
+            (task_id, project_id, status, now, now, audit_loop,
+             parent, part, result, error, model),
+        )
+        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 _audit_actions(db_path, project_id="p1"):
+    store = AuthStore(db_path)
+    try:
+        return [(e["action"], e["detail"]) for e in store.list_audit(
+            project_id)]
+    finally:
+        store.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
+
+
+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 _task_row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(
+            "SELECT status, error_text FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+
+
+class WhitespacePolicyTestCase(unittest.TestCase):
+    """I-6：策略写入路径白名单。"""
+
+    def test_whitespace_tier_model_dropped_and_named(self):
+        with self.assertLogs("cloud.models", level="WARNING") as logs:
+            policy = parse_routing_policy({
+                "mode": "active",
+                "tiers": {"plan": "my model", "default": "ok-model"},
+            })
+        self.assertIsNotNone(policy)
+        self.assertEqual(policy.tiers, {"default": "ok-model"})
+        self.assertEqual(policy.dropped_tiers, ["plan"])
+        self.assertTrue(any("plan" in line for line in logs.output))
+
+    def test_whitespace_capability_key_dropped_is_unknown(self):
+        caps = parse_model_capabilities({
+            "my model": {"tool_calling": True},
+            "ok": {"tool_calling": True},
+        })
+        self.assertEqual(caps, {"ok": {"tool_calling": True}})
+        verdict = evaluate_capability(caps, "my model",
+                                      {"tool_calling": True})
+        self.assertFalse(verdict.ok)
+        self.assertTrue(verdict.unknown)
+        self.assertEqual(verdict.unknown_missing, ["tool_calling"])
+        self.assertEqual(verdict.known_missing, [])
+
+    def test_whitespace_fallback_and_direct_tiers_dropped(self):
+        policy = RoutingPolicy(
+            tiers={"default": "my model", "plan": "ok"},
+            fallback_models=["my model", "ok"])
+        self.assertEqual(policy.tiers, {"plan": "ok"})
+        self.assertEqual(policy.fallback_models, ["ok"])
+
+    def test_shadow_row_with_whitespace_chosen_parses(self):
+        parsed = parse_worker_routing_detail(
+            "T9 tier=default source=policy applied=true"
+            " suggested=- chosen=my model")
+        self.assertIsNotNone(parsed)
+        self.assertEqual(parsed["chosen_model"], "my model")
+        # 进样本计数，不再整行丢弃。
+        report = evaluate_shadow([
+            {"detail": "T9 tier=default source=policy applied=true"
+                       " suggested=- chosen=my model",
+             "created_at": 1000.0}], {}, {})
+        self.assertEqual(report["routed_task_n"], 1)
+        self.assertEqual(report["dropped_rows"]["unparseable"], 0)
+
+    def test_legacy_suggested_whitespace_still_counted_not_lost(self):
+        # suggested 是中间字段，仍要求无空白：legacy 含空白
+        # suggested 的行解析失败，但按批 2 计入丢行桶，不静默消失。
+        parsed = parse_worker_routing_detail(
+            "T9 tier=default source=policy applied=true"
+            " suggested=my model chosen=ok")
+        self.assertIsNone(parsed)
+        report = evaluate_shadow([
+            {"detail": "T9 tier=default source=policy applied=true"
+                       " suggested=my model chosen=ok",
+             "created_at": 1000.0}], {}, {})
+        self.assertEqual(report["routed_task_n"], 0)
+        self.assertEqual(report["dropped_rows"]["unparseable"], 1)
+
+
+class PersistedWhitespaceModelTestCase(unittest.TestCase):
+    """I-6：含空白写回模型的复核不再被 bypass。"""
+
+    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()
+
+    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 _policy(self):
+        return {
+            "mode": "active",
+            "tiers": {"default": {"model": "policy-model",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "my model": {"tool_calling": False},
+            },
+        }
+
+    def test_persisted_whitespace_model_matches_and_is_rechecked(self):
+        _insert_audit(self.db_path, "p1", "worker_model_persisted",
+                      "T1 tier=default model=my model")
+        worker = _worker(self.db_path, self._policy())
+        audits = []
+        worker._audit = lambda p, a, d="": audits.append((p, a, d))
+        # 旧 split()[2] 定位置取在这里永远匹配不上（model
+        # 含空白被切成两段）；新按键提取必须认出写回模型。
+        self.assertTrue(worker._is_persisted_model(
+            "T1", "p1", "my model"))
+        # 复核必须真正跑起来、不再默认显式 bypass：含空白的
+        # 模型名进不了能力目录（I-6 白名单已丢弃该键），故
+        # 复核结果是 unknown——按 fix5 分档只审计不拦；断言
+        # 点是复核确实跑了（有 recheck_unknown 审计行），
+        # 而不是悄悄按显式放行（那时连审计行都没有）。
+        # active 下 known 非空才拦，空白模型永远到不了
+        # known（目录里没有它的条目），这是白名单的直接
+        # 推论，不是复核 bypass。
+        error = worker._recheck_persisted_model({
+            "id": "T1", "project_id": "p1", "model": "my model"})
+        self.assertIsNone(error)
+        self.assertTrue([a for _p, a, _d in audits
+                         if a == "worker_capability_recheck_unknown"])
+        self.assertTrue(any("my model" in d for _p, a, d in audits
+                            if a == "worker_capability_recheck_unknown"))
+
+    def test_persisted_whitespace_model_mismatch_is_not_persisted(self):
+        _insert_audit(self.db_path, "p1", "worker_model_persisted",
+                      "T1 tier=default model=my model")
+        worker = _worker(self.db_path, self._policy())
+        self.assertFalse(worker._is_persisted_model(
+            "T1", "p1", "other model"))
+        self.assertFalse(worker._is_persisted_model(
+            "T2", "p1", "my model"))
+
+
+class ExplicitNonSub2apiGateTestCase(unittest.TestCase):
+    """I-7：显式 + 非 sub2api 网关的能力门。"""
+
+    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()
+
+    def _policy(self, mode, capabilities):
+        return {
+            "mode": mode,
+            "tiers": {"default": {"model": "policy-model",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": capabilities,
+        }
+
+    def test_known_missing_explicit_non_sub2api_is_blocked(self):
+        # 行为变更：旧行为是直接派发（能力门只在 sub2api
+        # 时跑），新行为是 active 下已知不满足即拦。
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        worker = _worker(self.db_path, self._policy("active", {
+            "explicit-model": {"tool_calling": False},
+        }), gateway=("other-gw", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(worker._calls)
+        self.assertIn("lacks capability", result["error"])
+        actions = _audit_actions(self.db_path)
+        explicit = [d for a, d in actions
+                    if a == "worker_capability_gateway_explicit"]
+        self.assertTrue(explicit)
+        self.assertIn("blocked=True", explicit[0])
+        self.assertTrue([a for a, _ in actions
+                         if a == "worker_capability_gate_rejected"])
+
+    def test_unknown_explicit_non_sub2api_is_audited_not_blocked(self):
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        worker = _worker(self.db_path, self._policy("active", {}),
+                         gateway=("other-gw", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        actions = _audit_actions(self.db_path)
+        explicit = [d for a, d in actions
+                    if a == "worker_capability_gateway_explicit"]
+        self.assertTrue(explicit)
+        self.assertIn("blocked=False", explicit[0])
+        self.assertFalse([a for a, _ in actions
+                          if a == "worker_capability_gate_rejected"])
+
+    def test_shadow_explicit_non_sub2api_is_audited_not_blocked(self):
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        worker = _worker(self.db_path, self._policy("shadow", {
+            "explicit-model": {"tool_calling": False},
+        }), gateway=("other-gw", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        actions = _audit_actions(self.db_path)
+        explicit = [d for a, d in actions
+                    if a == "worker_capability_gateway_explicit"]
+        self.assertTrue(explicit)
+        self.assertIn("blocked=False", explicit[0])
+
+    def test_nonexplicit_non_sub2api_has_no_gateway_gate(self):
+        # 非显式 + 非 sub2api 根本不进网关能力块：模型为空，
+        # 后续走 decide_routing，不存在独立门误拦。
+        _insert_task(self.db_path, "E1")
+        worker = _worker(self.db_path, self._policy("active", {
+            "policy-model": {"tool_calling": True},
+        }), gateway=("other-gw", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        actions = _audit_actions(self.db_path)
+        self.assertFalse([a for a, _ in actions
+                          if a == "worker_capability_gateway_explicit"])
+
+    def test_explicit_no_gateway_keeps_explicit_semantics(self):
+        # 边界：显式 + 无网关（provider 为 None）不进网关
+        # 能力块——它根本没走网关，显式语义不变：已知不满足
+        # 也不在这里拦（写回模型仍走后续复核，用户显式仍
+        # 不被复核，fixes5 契约）。无网关显式任务被网关块
+        # 拦住会是语义倒退，不是 I-7 的意图。
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        worker = _worker(self.db_path, self._policy("active", {
+            "explicit-model": {"tool_calling": False},
+        }), gateway=(None, {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        actions = _audit_actions(self.db_path)
+        self.assertFalse([a for a, _ in actions
+                          if a == "worker_capability_gateway_explicit"])
+        self.assertFalse([a for a, _ in actions
+                          if a == "worker_capability_gate_rejected"])
+
+    def test_explicit_sub2api_path_unchanged(self):
+        # sub2api 显式路径与原来逐字一致：同一审计 action、
+        # 同一 error 文案、同一 blocked 标志。
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        worker = _worker(self.db_path, self._policy("active", {
+            "explicit-model": {"tool_calling": False},
+        }), gateway=("sub2api", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(worker._calls)
+        self.assertIn("lacks capability", result["error"])
+        actions = _audit_actions(self.db_path)
+        explicit = [d for a, d in actions
+                    if a == "worker_capability_gateway_explicit"]
+        self.assertTrue(explicit)
+        self.assertIn("blocked=True", explicit[0])
+
+
+class CoordGateAttributionTestCase(unittest.TestCase):
+    """I-8 + I-13：协调员门拒绝归因与分类回落可见性。"""
+
+    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)
+        out = self.worker._coord_audit_loop_merge(
+            {"id": "P1", "project_id": "p1", "audit_loop": 1},
+            [{"name": "a", "prompt": "do a"}],
+            [{"id": "C1", "name": "a"}])
+        return out
+
+    def test_repo_failure_child_is_gate_rejected_not_execution(self):
+        out = self._merge(
+            "仓库解析失败: bad repo",
+            audit_rows=[("worker_repo_failed",
+                         "C1: 仓库解析失败: bad repo")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+        self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
+        # 结构化证据已命中，不走子串回落、无回落审计行。
+        self.assertFalse([a for _p, a, _d in self.audits
+                          if a == "worker_coord_gate_classified_by_marker"])
+
+    def test_missing_key_child_is_gate_rejected_not_execution(self):
+        out = self._merge(
+            "Missing environment variable: OPENKUN_API_KEY",
+            audit_rows=[("worker_missing_key",
+                         "C1: Missing environment variable:"
+                         " OPENKUN_API_KEY")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+        self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
+
+    def test_recheck_rejected_child_is_gate_rejected(self):
+        out = self._merge(
+            "default tier persisted model m lacks capability",
+            audit_rows=[("worker_capability_recheck_rejected",
+                         "C1: default tier persisted model m lacks"
+                         " capability")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+
+    def test_structured_evidence_wins_over_reworded_error(self):
+        # error_text 措辞改掉（不含任何 marker 子串），结构化
+        # 审计行不变时分类仍为 gate_rejected：分类不再依赖文案。
+        out = self._merge(
+            "repo exploded unexpectedly",
+            audit_rows=[("worker_repo_failed",
+                         "C1: 仓库解析失败: bad repo")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+        self.assertFalse([a for _p, a, _d in self.audits
+                          if a == "worker_coord_gate_classified_by_marker"])
+
+    def test_marker_fallback_is_audited(self):
+        # 删结构化审计行、仅留 error_text 时分类走回落，
+        # 且合流里有回落审计行点名 marker。
+        out = self._merge("必需工具缺失: playwright")
+        self.assertEqual(out["decision"], "escalate")
+        self.assertIn("a", out.get("gate_rejected", []))
+        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=必需工具缺失", marker_rows[0])
+
+    def test_gate_lookup_failure_still_escalates(self):
+        # I-5 回归：审计表缺失导致查询失败时合流走 escalate，
+        # 不改回空集推回（与批 2 测试同款口径，本夹具先删表）。
+        _insert_task(self.db_path, "P1", status="running", audit_loop=1)
+        _insert_task(self.db_path, "C1", status="failed", parent="P1",
+                     part="a", error="boom")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute("DROP TABLE IF EXISTS audit_log")
+            conn.commit()
+        finally:
+            conn.close()
+        out = self.worker._coord_audit_loop_merge(
+            {"id": "P1", "project_id": "p1", "audit_loop": 1},
+            [{"name": "a", "prompt": "do a"}],
+            [{"id": "C1", "name": "a"}])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertTrue(out.get("gate_lookup_failed"))
+        self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
+
+
+class GateResultsCheckedTestCase(unittest.TestCase):
+    """I-10：gate_results 只列实际跑过的门 + 两轮不误升级。"""
+
+    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()
+
+    def test_direct_call_checked_subset_and_legacy(self):
+        worker = _worker(self.db_path)
+        task = {"id": "T1", "project_id": "p1"}
+        gates, _ = worker._audit_loop_gate_results(
+            task, "failed", {"error": "boom"}, gates_checked=["gateway"])
+        self.assertEqual(gates, {"execution": False, "gateway": True})
+        gates, _ = worker._audit_loop_gate_results(
+            task, "failed", {"error": "boom"}, gates_checked=[])
+        self.assertEqual(gates, {"execution": False})
+        # 未传 gates_checked 的旧直接调用保持旧行为。
+        gates, _ = worker._audit_loop_gate_results(
+            task, "failed", {"error": "boom"})
+        self.assertEqual(gates["capability"], True)
+        self.assertEqual(gates["required_tools"], True)
+        self.assertEqual(gates["gateway"], True)
+
+    def test_policy_task_two_rounds_pushback_without_gate_missing(self):
+        _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)
+        # 政策任务本轮实际跑过的门只有 capability（网关桩
+        # 回 None、无 agent_tools 故工具门未跑），状态里
+        # passed_gates 不含没跑过的门。
+        self.assertEqual(json.loads(row[1]), ["capability"])
+        self.assertEqual(row[2], "pushback")
+
+    def test_explicit_task_two_rounds_pushback_without_gate_missing(self):
+        _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(row[2], "pushback")
+
+
+class EvaluateNoFindingTestCase(unittest.TestCase):
+    """I-11：失败门 + 空 findings 返回 escalate，不抛异常。"""
+
+    def test_failed_gate_without_finding_escalates(self):
+        out = evaluate(None, "T1", {"execution": False}, findings=[])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "gate_failed_without_finding")
+        self.assertEqual(out["round"], 0)
+        self.assertIsNone(out["packet"])
+        self.assertEqual(out["new_state"].round, 0)
+        self.assertEqual(out["new_state"].last_decision, "escalate")
+
+    def test_failed_gate_without_finding_keeps_round(self):
+        from cloud.audit_loop import LoopState
+        state = LoopState("T1", round=2, passed_gates=["execution"],
+                          criteria={}, last_decision="pushback")
+        # 注意：上一轮通过的门本轮变失败会生成 regression
+        # finding（all_findings 非空），走不到本支；所以
+        # 本例让失败的是上一轮未通过过的门（capability），
+        # 且上一轮通过的 execution 本轮仍通过——无回归、
+        # 无缺门、无弱化、无 finding，只有失败门。
+        out = evaluate(state, "T1",
+                       {"execution": True, "capability": False},
+                       findings=[])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "gate_failed_without_finding")
+        self.assertEqual(out["new_state"].round, 2)
+
+    def test_finding_still_pushbacks_and_max_rounds_unchanged(self):
+        finding = Finding("execution_failed", "high", "boom", "re-run",
+                          "task completes", kind="deterministic",
+                          gate="execution")
+        out = evaluate(None, "T1", {"execution": False},
+                       findings=[finding])
+        self.assertEqual(out["decision"], "pushback")
+        self.assertEqual(out["round"], 1)
+        from cloud.audit_loop import LoopState
+        state = LoopState("T1", round=3, passed_gates=[],
+                          criteria={}, last_decision="pushback")
+        out = evaluate(state, "T1", {"execution": False},
+                       findings=[finding])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "max_rounds_exceeded")
+
+
+class GateErrorMarkersTestCase(unittest.TestCase):
+    """I-13：分类子串表抽成模块常量。"""
+
+    def test_marker_constant_covers_known_markers(self):
+        self.assertIn("Missing environment variable:",
+                      _GATE_ERROR_MARKERS)
+        self.assertIn("网关准备失败", _GATE_ERROR_MARKERS)
+        self.assertIn("必需工具缺失", _GATE_ERROR_MARKERS)
+        self.assertIn("工具审批声明矛盾", _GATE_ERROR_MARKERS)
+        self.assertIn("refusing to dispatch an incapable or"
+                      " unverifiable model", _GATE_ERROR_MARKERS)
+        self.assertIn("fails capability", _GATE_ERROR_MARKERS)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -482,6 +482,45 @@ class _CoordGateLookupFailed(Exception):
     """
 
 
+def _mark_gate_checked(task, gate):
+    """在任务 dict 上记一笔“某门本轮确实跑过”（审计批 4，I-10）。
+
+    用 task["_gates_checked"] 列表承载，键名带下划线、私有、
+    内存态，与 _gateway_provider/_gateway_env 同款约定：
+    不进库、不进审计 detail、不随 prompt 落库。列表而非
+    集合，保证 task dict 仍是 JSON 安全的；run_one 内 task
+    经多次 dict(task) 浅拷贝，拷贝共享同一列表对象，因此
+    无论哪份拷贝打标记，终局块读到的都是同一份集合。
+    终局时 _audit_loop_gate_results 只列入被标记过的门，
+    不再把没跑过的门硬编码成通过。
+    """
+    if not isinstance(task, dict):
+        return
+    checked = task.get("_gates_checked")
+    if not isinstance(checked, list):
+        checked = []
+        task["_gates_checked"] = checked
+    if gate not in checked:
+        checked.append(gate)
+
+
+# 审计批 4（I-13）：合流分类用的门拒绝子串标记表。
+# 分类优先读结构化证据（_coord_gate_rejected_child_ids 的
+# 审计行集合），子串只作最后回落——回落命中时合流记一条
+# worker_coord_gate_classified_by_marker 审计行点名 marker，
+# 让回落可见。生产方 error 文案本身不改措辞、不做全仓
+# 常量化重构（那是另一批的事）；分类侧的这份拷贝抽成
+# 常量，文案改动时分类侧只有这一处要改。
+_GATE_ERROR_MARKERS = (
+    "Missing environment variable:",
+    "网关准备失败",
+    "必需工具缺失",
+    "工具审批声明矛盾",
+    "refusing to dispatch an incapable or unverifiable model",
+    "fails capability",
+)
+
+
 class Worker:
     """单机 worker：一轮认领一个 queued 任务并执行到底。
 
@@ -1924,7 +1963,8 @@ class Worker:
             result["new_state"] = result["new_state"].to_dict()
         return result
 
-    def _audit_loop_gate_results(self, task, final, result):
+    def _audit_loop_gate_results(self, task, final, result,
+                                 gates_checked=None):
         """本轮已产生的确定性门结果（只取已有结果，不捏造判定）。
 
         门的取值口径（核实结论）：
@@ -1938,8 +1978,25 @@ class Worker:
         - capability / required_tools / gateway：本方法只在终局
           块被调用，而终局块的前提是尝试跑到了 runner——此前
           的前置门若拒绝，run_one 早已在 runner 之前提前返回，
-          根本到不了这里；所以能到这里的尝试，这三门要么已
-          通过、要么不适用，记 True 是事实陈述，不是复查。
+          根本到不了这里；所以能到这里的尝试，已跑过的门
+          要么已通过、要么不适用。审计批 4（I-10）之前这里
+          把三门无条件记 True，显式任务（尤其 I-7 修复前的
+          非 sub2api 组合）能力门根本没跑过却同样记入
+          passed_gates 并持久化——“通过”一词在状态表里失真。
+          现在：调用方传 gates_checked（run_one 派发点用
+          _mark_gate_checked 在 task["_gates_checked"] 里记的
+          本轮实际跑过的门）时，只列入集合内的门；未传
+          （直接调用的旧调用方）时保持旧行为、三门全记 True。
+          gate_missing 语义核查：引擎只对“上一轮 passed_gates
+          里有、本轮缺失”的门升级；同一任务的跑过门集合在
+          两轮间是稳定的（显式任务无策略时两轮都不记
+          capability，政策任务两轮都记；门集合由任务/策略/
+          执行器决定，不由轮次决定），所以显式任务两轮不
+          会因“not_checked”误触发升级——两轮往返测试锁定。
+          唯一可预见的边界是部署时已在循环中的任务：旧状态
+          passed_gates 里含硬编码 True 的三门，新代码首轮
+          可能缺省其中几门 → gate_missing 升级一次，方向是
+          fail-closed 不是放行。
           前置门拒绝的路径不接审计循环（见 _audit_loop_wire
           的文档）：那类任务在 agent 跑任何一轮之前就已失败，
           推回包没有可投递的 agent 回合，且重排会确定性地
@@ -1970,9 +2027,15 @@ class Worker:
                     "re-run goal verification",
                     "goal verification passes",
                     kind="deterministic", gate="goal_verify"))
-        gate_results["capability"] = True
-        gate_results["required_tools"] = True
-        gate_results["gateway"] = True
+        if gates_checked is None:
+            gate_results["capability"] = True
+            gate_results["required_tools"] = True
+            gate_results["gateway"] = True
+        else:
+            checked = set(gates_checked)
+            for gate in ("capability", "required_tools", "gateway"):
+                if gate in checked:
+                    gate_results[gate] = True
         return gate_results, findings
 
     def _audit_loop_wire(self, task, final, result, error):
@@ -2007,7 +2070,11 @@ class Worker:
         if task.get("coordinator_parts"):
             return final, error, "skipped"
         gate_results, findings = self._audit_loop_gate_results(
-            task, final, result)
+            # 审计批 4（I-10）：把派发点记的本轮实际跑过
+            # 的门透传进去；没记过的门不列入 gate_results，
+            # 不再硬编码成通过。未传（旧直接调用）时保持
+            # 旧行为，见 _audit_loop_gate_results 文档。
+            task, final, result, task.get("_gates_checked"))
         try:
             decision = self.audit_loop_evaluate(
                 task, gate_results, findings=findings)
@@ -6129,11 +6196,19 @@ class Worker:
                                        attempt_started_at=None):
         """前置门拒绝的子任务 id 集（合流分类用）。
 
-        依据只有审计行：能力门、工具门、审批门的拒绝审计 detail
-        是 "<子任务 id>: <错误>" 格式，可按 id 前缀归属；缺 key
-        门（worker_missing_key）与仓库门（worker_repo_failed）
-        的 detail 只有错误文本、没有任务 id，审计行无法归属，
-        故不在此列——那两类只走下面的 error_text 强标记回落。
+        依据只有审计行：下列 6 个拒绝审计的 detail 均是
+        "<子任务 id>: <错误>" 格式，可按 id 前缀归属——
+        能力门、工具门、审批门（旧有 3 门），以及审计批 4
+        （I-8）补进的缺 key 门（worker_missing_key）、仓库门
+        （worker_repo_failed）、写回复核拒绝门
+        （worker_capability_recheck_rejected，其 detail 本就
+        是 "<id>: <错误>" 格式）。不补进集合的同类审计行：
+        worker_capability_gateway_explicit（含 blocked=false
+        的纯审计行，补进会把未被拦的显式任务误判为门拒绝）、
+        worker_capability_gateway_shadow（shadow 永不拦）、
+        worker_capability_fallback（回落不是拒绝）；
+        worker_gateway_failed（网关准备失败）已有 error_text
+        标记覆盖，不补进集合。
         审计是 best-effort，查不到不等于没拒绝，调用方不得把
         “无审计行”当成“没被门拒绝”的证据（回落标记见分类）。
 
@@ -6175,7 +6250,10 @@ class Worker:
         """
         actions = ("worker_capability_gate_rejected",
                    "worker_tool_gate_rejected",
-                   "worker_tool_approval_gate_rejected")
+                   "worker_tool_approval_gate_rejected",
+                   "worker_missing_key",
+                   "worker_repo_failed",
+                   "worker_capability_recheck_rejected")
         rows = None
         try:
             conn = _db_connect(self.db_path)
@@ -6183,17 +6261,21 @@ class Worker:
                 try:
                     rows = conn.execute(
                         "SELECT detail, created_at FROM audit_log"
-                        " WHERE project_id = ? AND action IN (?, ?, ?)"
+                        " WHERE project_id = ? AND action IN"
+                        " (?, ?, ?, ?, ?, ?)"
                         " ORDER BY created_at DESC, id DESC LIMIT 5000",
                         (project_id, actions[0], actions[1],
-                         actions[2])).fetchall()
+                         actions[2], actions[3], actions[4],
+                         actions[5])).fetchall()
                 except Exception:
                     rows = conn.execute(
                         "SELECT detail, created_at FROM audit_log"
-                        " WHERE project_id = ? AND action IN (?, ?, ?)"
+                        " WHERE project_id = ? AND action IN"
+                        " (?, ?, ?, ?, ?, ?)"
                         " ORDER BY created_at DESC LIMIT 5000",
                         (project_id, actions[0], actions[1],
-                         actions[2])).fetchall()
+                         actions[2], actions[3], actions[4],
+                         actions[5])).fetchall()
             finally:
                 conn.close()
         except Exception as exc:
@@ -6259,13 +6341,7 @@ class Worker:
         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"):
+        for marker in _GATE_ERROR_MARKERS:
             if marker in err:
                 return "gate_rejected", "gate_error_marker"
         if status == "failed" or status == "interrupted":
@@ -6392,6 +6468,19 @@ class Worker:
                 excluded_notes.append("%s:%s" % (name, reason))
             elif kind == "gate_rejected":
                 gate_rejected.append(name)
+                if reason == "gate_error_marker":
+                    # 审计批 4（I-13）：子串回落命中必须可见，
+                    # 不能静默——记一条点名 marker 的审计行，
+                    # 让“分类依赖 error_text 回落”在审计里
+                    # 可查；文案改动导致回落失效时，这里
+                    # 的缺失本身就是信号。
+                    fired = next(
+                        (m for m in _GATE_ERROR_MARKERS
+                         if m in (row.get("error_text") or "")),
+                        "?")
+                    self._audit(project_id,
+                                "worker_coord_gate_classified_by_marker",
+                                "%s marker=%s" % (cid, fired))
             elif kind == "execution":
                 gate_results["child:" + name] = False
                 findings.append(_AuditFinding(
@@ -6910,6 +6999,11 @@ class Worker:
             return []
         if not isinstance(self.executor, DockerExecutor):
             return []
+        # 审计批 4（I-10）：走到这里说明工具门真正查了租户
+        # 配置（读失败也算跑过：返回 [] 的放行按 I-5 记审计，
+        # 且该任务随后在租户准备失败处提前返回、到不了
+        # 终局块，标记本身无实际影响）。
+        _mark_gate_checked(task, "required_tools")
         try:
             home = self._tenant_dirs(
                 task.get("project_id"))["codex_home"]
@@ -6994,12 +7088,22 @@ class Worker:
                         "worker_persisted_model_lookup_failed",
                         "%s error=%s" % (task_id, exc))
             return False
+        # 审计批 4（I-6）：persisted 行格式固定为
+        # "%s tier=%s model=%s"，model 是最后一个字段，
+        # 按键提取而不是 split()[2] 定位置取。旧代码在
+        # 模型名含空白时永远匹配不上，于是写回模型被
+        # 默认当成“用户显式”而跳过复核——两处同时失效
+        # 的另一半（影子解析是 routing_shadow 侧）。
+        # 格式完全不符的行仍视为无证据：worker 写出的
+        # persisted 行格式固定，格式不符不可能是写回行，
+        # fail-closed 备选（按写回对待）在此不适用。
+        persisted_re = re.compile(
+            r"^%s\s+tier=\S+\s+model=(.*)$" % re.escape(task_id))
         for row in rows:
             if row[0] != "worker_model_persisted":
                 continue
-            parts = (row[1] or "").split()
-            if len(parts) >= 3 and parts[0] == task_id \
-                    and parts[2] == "model=%s" % model:
+            match = persisted_re.match((row[1] or "").strip())
+            if match and match.group(1).strip() == model:
                 return True
         for row in rows:
             if row[0] != "worker_routing":
@@ -7033,6 +7137,9 @@ class Worker:
         model = task.get("model")
         verdict = evaluate_capability(
             self.routing_policy.model_capabilities, model, requires)
+        # 审计批 4（I-10）：走到这里说明复核真正算了
+        # verdict（无策略/无 requires/off 档已在上面返回）。
+        _mark_gate_checked(task, "capability")
         if verdict.ok:
             return None
         known = verdict.known_missing
@@ -7206,7 +7313,12 @@ class Worker:
             # 缺 key 门提前返回：同样要清暂存，否则与仓库解析
             # 失败一路同款残留（见 _cleanup_task_seed）。
             self._cleanup_task_seed(task)
-            self._audit(task["project_id"], "worker_missing_key", error)
+            # 审计批 4（I-8）：缺 key 门审计 detail 改为
+            # "<任务 id>: <错误>"，与能力/工具/审批门同款
+            # 可归属格式，协调员合流才能按 id 前缀认领；
+            # error_text 子串标记仍能命中（子串匹配）。
+            self._audit(task["project_id"], "worker_missing_key",
+                        "%s: %s" % (task_id, error))
             log_event(self.logger, "worker_missing_key", task_id=task_id,
                       project_id=task["project_id"], status="failed")
             self._notify_task_finished(
@@ -7234,7 +7346,11 @@ class Worker:
             # 仓库解析失败提前返回：够不到正常终局后的清理，
             # 这里必须先清暂存，否则 bundle 目录永久残留。
             self._cleanup_task_seed(task)
-            self._audit(task["project_id"], "worker_repo_failed", repo_error)
+            # 审计批 4（I-8）：仓库门审计 detail 改为
+            # "<任务 id>: <错误>"，与 3 门同款可归属格式；
+            # error_text 子串标记仍能命中（子串匹配）。
+            self._audit(task["project_id"], "worker_repo_failed",
+                        "%s: %s" % (task_id, repo_error))
             log_event(self.logger, "worker_repo_failed", task_id=task_id,
                       project_id=task["project_id"], status="failed")
             self._notify_task_finished(
@@ -7419,6 +7535,12 @@ class Worker:
                     "stdout": "", "error": error, "events": [],
                     "cost_record_id": None}
 
+        if gateway_provider:
+            # 审计批 4（I-10）：网关准备实际跑过且成功
+            # （失败已在上面提前返回），终局时 gateway
+            # 门可列入 gate_results。
+            _mark_gate_checked(task, "gateway")
+
         # Audit fix：网关能力检查——此前路由决策只在模型
         # 为空时跑，sub2api 网关先填默认模型后决策直接跳
         # 过，能力门在生产路径上完全不生效。补齐：网关填
@@ -7435,8 +7557,29 @@ class Worker:
         # 不满足才拦，unknown 只审计（见下），因为显式
         # 优先保护的是用户选了哪个模型，不是选的模型
         # 一定能跑；目录陈旧不应惩罚正确的显式选择。
-        if gateway_provider == "sub2api" \
-                and self.routing_policy is not None:
+        # 审计批 4（I-7）：显式模型的 known/unknown 分档检查
+        # 与网关无关——网关分支只决定“待查模型是哪一个”
+        # （网关填充模型或显式模型），分档判定走第 3 批统一
+        # 的 evaluate_capability。进入条件因此是“有策略且
+        # （走了 sub2api 网关，或显式任务走了任一真实网关）”：
+        # 非显式 + 非 sub2api 的任务根本不进块（它们模型
+        # 为空、后续走 decide_routing，不存在误拦）；显式
+        # + 无网关（provider 为 None）的任务也不进块——它
+        # 根本没走网关，显式语义不变（写回模型仍走后续
+        # 复核，用户显式仍不被复核，见 fixes5 契约）；
+        # 显式 + sub2api 路径与原来逐字一致（同一审计
+        # action、同一 error 文案、同一 blocked 标志），
+        # 不会重复审计或重复拦；显式 + 非 sub2api 网关
+        # 在 active 下已知不满足时从“直接派发”变为“拦”，
+        # 这是有意的行为变更。
+        # 附带：已持久化的显式任务在非 sub2api 网关下会
+        # 同时走本门与后续写回复核（本门不拦时复核再按
+        # 分档审计/拦），两者审计 action 不同
+        # （gateway_explicit 与 recheck_*），不是重复拦。
+        if self.routing_policy is not None \
+                and (gateway_provider == "sub2api"
+                     or (_explicit_model
+                         and gateway_provider is not None)):
             _gw_tier = classify_task(task)
             _gw_requires = self.routing_policy.tier_requirements.get(
                 _gw_tier) or {}
@@ -7446,6 +7589,9 @@ class Worker:
                     _gw_verdict = evaluate_capability(
                         self.routing_policy.model_capabilities,
                         task.get("model"), _gw_requires)
+                    # 审计批 4（I-10）：走到这里说明网关
+                    # 能力检查真正算了 verdict。
+                    _mark_gate_checked(task, "capability")
                     if not _gw_verdict.ok:
                         _gw_note = ("gateway model %s %s capability:"
                                     " %s" % (
@@ -7584,6 +7730,16 @@ class Worker:
             # 断言 events[0] 就是 fallback 事件、无事件时为空），
             # 而旧行为本身已由 task["model"] / actual_model 体现，
             # 不需要再用一条事件重复说明。
+            # 审计批 4（I-10）：decide_routing 对档模型算过
+            # capability 的决策才记 capability——policy/shadow
+            # 决策算过、capability_blocked 决策算过、
+            # capability_fallback 决策算过；worker_default 经
+            # 无档模型/off/无策略、explicit/worker_explicit
+            # 提前返回时均未算，不记。
+            if decision.source in ("policy", "shadow",
+                                   "capability_blocked") \
+                    or getattr(decision, "capability_fallback", False):
+                _mark_gate_checked(task, "capability")
             if decision.source in ("policy", "shadow"):
                 record_routing({"id": task_id, "events": events}, decision)
                 # 影子评估（v0b）需要建议模型：旧格式只有