fix: third-audit gate findings (explicit split, shadow explicit failures, N2-N5, N1 note)

Explicit gateway capability now splits known/unknown: known-missing
explicit models fail-fast under active (same shape as the gateway
default gate), unknown explicit models stay audit-only, shadow never
blocks; explicit priority now protects which model the user chose,
not that the chosen model must run. Shadow report reads
worker_capability_gateway_explicit rows as explicit_capability_failures
with a field note distinguishing the three explicit counts.
N2: coordinator requeue CAS rowcount=0 branch covered via a proxy
connection test (no refactor needed). N3: shadow endpoint test on
the create_server + real HTTP skeleton (capability, truncated, and
explicit-failure fields). N4: creation-layer rejection tests for
tools/instructions/tool_approvals on goal/coordinator tasks.
N5: model writeback failure logs a warning (best-effort kept).
N1: writeback capability-recheck trade-off documented in comment
only; recheck itself is a later batch.

cloud/models.py

@@ -246,6 +246,49 @@ def check_model_capabilities(capabilities, model, requires):
     return (not missing), sorted(missing), unknown
 
 
+def split_model_capability_missing(capabilities, model, requires):
+    """把 check_model_capabilities 的缺项拆成已知 / 未知两档。
+
+    Audit fix 4（第三遍审计分档建议）：显式模型的网关检查不能
+    只用 (missing, unknown) 的总口径——unknown 为真时可能是
+    “目录里根本没有该模型”，也可能是“条目里某项能力没填”，
+    两者决策不同。返回 (known_missing, unknown_missing)：
+
+    - 目录里没有该模型：全部需求都算 unknown，known 为空；
+    - 布尔能力明确为 False、或 context_tokens 明确低于需求：
+      known（目录明确写了不满足）；
+    - 布尔能力未填（None）、或 context_tokens 未填：unknown。
+
+    与 check_model_capabilities 的口径一致：requires 里非布尔/
+    非整数键被忽略，不计入任何一档。
+    """
+    requires = requires or {}
+    model = _clean_model_name(model)
+    entry = (capabilities or {}).get(model) if model else None
+    if entry is None:
+        return [], sorted(
+            k for k in requires
+            if k in REQUIREMENT_BOOL_KEYS or k in REQUIREMENT_INT_KEYS)
+    known = []
+    unknown = []
+    for key, value in requires.items():
+        if key in REQUIREMENT_BOOL_KEYS:
+            flag = entry.get(key)
+            if flag is True:
+                continue
+            if flag is None:
+                unknown.append(key)
+            else:
+                known.append(key)
+        elif key in REQUIREMENT_INT_KEYS:
+            context = entry.get("context_tokens")
+            if context is None:
+                unknown.append(key)
+            elif context < value:
+                known.append(key)
+    return sorted(known), sorted(unknown)
+
+
 def prompt_len_bucket(prompt) -> str:
     """Bucket a prompt by length for logging only, never for routing.
 

cloud/server.py

@@ -4195,6 +4195,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             ).fetchall()
             truncated = len(audit_rows) > 10000
             audit_rows = audit_rows[:10000]
+            # Audit fix 4（第三遍审计显式闭环）：显式能力检查
+            # 结果只记 worker_capability_gateway_explicit，影子
+            # 端点此前只读 worker_routing，运营方根本看不到它。
+            # 在此之外加读显式行，附为独立字段，不与已有两个
+            # explicit_n 口径混在一起（见报告里的字段说明）。
+            explicit_rows = conn.execute(
+                "SELECT detail FROM audit_log"
+                " WHERE project_id = ?"
+                " AND action = 'worker_capability_gateway_explicit'"
+                " ORDER BY created_at, id LIMIT 10001",
+                (project_id,),
+            ).fetchall()
+            explicit_truncated = len(explicit_rows) > 10000
+            explicit_rows = explicit_rows[:10000]
             task_rows = conn.execute(
                 "SELECT id, status, model FROM tasks"
                 " WHERE project_id = ?",
@@ -4229,6 +4243,51 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             tasks_by_id, costs_by_task, explicit_n=explicit_n)
         report["project_id"] = project_id
         report["truncated"] = truncated
+        # 显式失败清单：解析 worker_capability_gateway_explicit
+        # 行（格式：task_id model=.. tier=.. mode=.. missing=..
+        # unknown=.. blocked=..；fix4 之前的行无 blocked 项，
+        # 按 False 计）。这与报告里两个 explicit_n 不同义：
+        # 顶层 explicit_n 是任务表里带模型字段的任务数；
+        # per-tier explicit_n 是 worker_routing source=explicit
+        # 审计行数；本字段只计网关显式能力检查的结果行，
+        # 含被拦的已知不满足与只审计的 unknown，blocked
+        # 字段区分两者。解析失败的行直接跳过、不谎报。
+        _explicit_failures = []
+        for _row in explicit_rows:
+            _detail = _row["detail"] or ""
+            _tokens = _detail.split()
+            if not _tokens:
+                continue
+            _kv = {}
+            for _tok in _tokens[1:]:
+                if "=" in _tok:
+                    _k, _v = _tok.split("=", 1)
+                    _kv[_k] = _v
+            if "model" not in _kv or "tier" not in _kv:
+                continue
+            _missing = _kv.get("missing") or "-"
+            _explicit_failures.append({
+                "task_id": _tokens[0],
+                "model": _kv["model"],
+                "tier": _kv["tier"],
+                "mode": _kv.get("mode") or "-",
+                "missing": [] if _missing == "-" else _missing.split(","),
+                "unknown": _kv.get("unknown") == "True",
+                "blocked": _kv.get("blocked") == "True",
+            })
+        report["explicit_capability_failures"] = _explicit_failures
+        report["explicit_capability_failures_truncated"] = \
+            explicit_truncated
+        report["explicit_counts_note"] = (
+            "explicit_n (top level) counts tasks whose model field is"
+            " set in the tasks table; per-tier explicit_n counts"
+            " worker_routing rows with source=explicit; "
+            "explicit_capability_failures lists"
+            " worker_capability_gateway_explicit rows only, i.e."
+            " gateway capability check outcomes for explicit models,"
+            " with blocked=True for known-missing models that were"
+            " refused and blocked=False for unknown models that were"
+            " audited but not blocked.")
         # Audit fix 3（I2，端点侧）：只读加载当前策略，
         # 给每档附能力状态（档模型 vs requires）。加载
         # 失败/未配置时状态为 policy_unavailable，不拦、

cloud/tests/test_audit_fixes3.py

@@ -113,7 +113,11 @@ class B1ExplicitGatewayTestCase(unittest.TestCase):
     def tearDown(self):
         self.tmp.cleanup()
 
-    def test_explicit_model_is_audited_not_blocked(self):
+    def test_explicit_model_known_missing_is_blocked(self):
+        # Audit fix 4: known-missing explicit models are now blocked
+        # (split from unknown, which stays audit-only). The old
+        # audit-only expectation for tool_calling=False moved to
+        # test_audit_fixes4's unknown case.
         _insert_task(self.db_path, "E1", model="explicit-model")
         worker = _worker(self.db_path, {
             "mode": "active",
@@ -124,16 +128,16 @@ class B1ExplicitGatewayTestCase(unittest.TestCase):
             },
         }, gateway=("sub2api", {}))
         result = worker.run_one()
-        self.assertEqual(result["status"], "completed")
-        self.assertTrue(worker._calls)
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(worker._calls)
+        self.assertIn("tool_calling", 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("explicit-model", explicit[0])
-        self.assertIn("tool_calling", explicit[0])
-        self.assertFalse([a for a, _ in actions
-                          if a == "worker_capability_gate_rejected"])
+        self.assertIn("blocked=True", explicit[0])
+        self.assertTrue([a for a, _ in actions
+                         if a == "worker_capability_gate_rejected"])
 
     def test_gateway_default_active_still_blocks(self):
         _insert_task(self.db_path, "G1")

cloud/tests/test_audit_fixes4.py

@@ -0,0 +1,481 @@
+"""Audit fix 4: third-audit gate findings.
+
+- Explicit gateway capability split: known-missing blocks (active),
+  unknown audits only; shadow never blocks.
+- Explicit failures surfaced in the shadow report as an independent
+  field, distinct from the two explicit_n counts.
+- N2: coordinator requeue CAS rowcount=0 branch, via a proxy
+  connection (the auditor's low-cost approach, no refactor).
+- N3: shadow endpoint test on the create_server + real HTTP skeleton.
+- N4: creation-layer rejections for tools / instructions /
+  tool_approvals on goal/coordinator tasks, in one pass.
+- N5: model writeback failure logs a warning (best-effort kept).
+
+Run: python3 -m unittest discover -s cloud/tests -p "test_audit_fixes4.py" -v
+No real models, no network, no real gateway: gateway steps are
+stubbed exactly like test_audit_fixes.py / test_audit_fixes3.py.
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+import cloud.worker as worker_module
+from cloud.auth import AuthStore
+from cloud.models import split_model_capability_missing
+from cloud.server import create_server, init_db
+from cloud.worker import Worker
+
+
+def _completed_runner(cmd):
+    if "dispatch" in cmd:
+        name = cmd[cmd.index("--name") + 1]
+    else:
+        name = cmd[cmd.index("--task") + 1]
+    return SimpleNamespace(
+        returncode=0,
+        stdout='{"status": "completed", "result": "done",'
+               ' "task_id": "%s", "thread_id": "th1"}' % name,
+        stderr="")
+
+
+def _insert_task(db_path, task_id, model=None, status="queued",
+                 project_id="p1", coordinator_parts=None,
+                 parent_task_id=None, coordinator_part=None,
+                 result_text=None, error_text=None, paused=0,
+                 interrupted_reason=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, coordinator_parts,"
+            " parent_task_id, coordinator_part, result_text, error_text,"
+            " model, paused_by_user, interrupted_reason)"
+            " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester',"
+            " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (task_id, project_id, status, now, now,
+             json.dumps(coordinator_parts) if coordinator_parts else None,
+             parent_task_id, coordinator_part, result_text, error_text,
+             model, paused, interrupted_reason),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    try:
+        row = conn.execute("SELECT * FROM tasks WHERE id = ?",
+                           (task_id,)).fetchone()
+        return dict(row) if row else None
+    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 _worker(db_path, policy_doc=None, env=None, gateway=None):
+    calls = []
+
+    def recording(cmd):
+        calls.append(list(cmd))
+        return _completed_runner(cmd)
+
+    worker = Worker(db_path, "/fake/host.py", runner=recording,
+                    env=env or {"XCM_API_KEY": "test-placeholder"},
+                    routing_policy=policy_doc)
+    if gateway is not None:
+        worker._ensure_gateway = lambda task: gateway
+        worker._sub2api_default_model = lambda e: "gateway-model"
+    worker._calls = calls
+    return worker
+
+
+class SplitMissingTestCase(unittest.TestCase):
+    def test_absent_model_is_all_unknown(self):
+        known, unknown = split_model_capability_missing(
+            {}, "no-such-model", {"tool_calling": True,
+                                   "min_context_tokens": 1000})
+        self.assertEqual(known, [])
+        self.assertEqual(unknown, ["min_context_tokens", "tool_calling"])
+
+    def test_explicit_false_is_known(self):
+        known, unknown = split_model_capability_missing(
+            {"m": {"tool_calling": False}},
+            "m", {"tool_calling": True})
+        self.assertEqual(known, ["tool_calling"])
+        self.assertEqual(unknown, [])
+
+    def test_unfilled_field_is_unknown(self):
+        known, unknown = split_model_capability_missing(
+            {"m": {}}, "m", {"tool_calling": True})
+        self.assertEqual(known, [])
+        self.assertEqual(unknown, ["tool_calling"])
+
+    def test_short_context_is_known_and_missing_context_is_unknown(self):
+        known, unknown = split_model_capability_missing(
+            {"m": {"context_tokens": 100}},
+            "m", {"min_context_tokens": 1000})
+        self.assertEqual(known, ["min_context_tokens"])
+        known2, unknown2 = split_model_capability_missing(
+            {"m": {}}, "m", {"min_context_tokens": 1000})
+        self.assertEqual(known2, [])
+        self.assertEqual(unknown2, ["min_context_tokens"])
+
+    def test_mixed_known_and_unknown_keeps_known(self):
+        known, unknown = split_model_capability_missing(
+            {"m": {"tool_calling": False}},
+            "m", {"tool_calling": True, "vision": True})
+        self.assertEqual(known, ["tool_calling"])
+        self.assertEqual(unknown, ["vision"])
+
+
+class ExplicitGatewaySplitTestCase(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)
+
+    def tearDown(self):
+        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_is_blocked(self):
+        _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("tool_calling", 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_is_audited_not_blocked(self):
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        worker = _worker(self.db_path, self._policy("active", {}),
+                         gateway=("sub2api", {}))
+        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_mixed_known_and_unknown_explicit_is_blocked(self):
+        # One known-missing capability is enough to block: the task
+        # cannot succeed regardless of the unknown field.
+        _insert_task(self.db_path, "E1", model="explicit-model")
+        policy = self._policy("active", {
+            "explicit-model": {"tool_calling": False},
+        })
+        policy["tiers"]["default"]["requires"] = {
+            "tool_calling": True, "vision": True}
+        worker = _worker(self.db_path, policy, gateway=("sub2api", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(worker._calls)
+
+    def test_shadow_known_missing_explicit_is_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=("sub2api", {}))
+        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"])
+
+
+class I7RowcountZeroTestCase(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)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_rowcount_zero_skips_requeue_and_audit(self):
+        parts = [{"name": "后端", "prompt": "做后端"}]
+        _insert_task(self.db_path, "coord", coordinator_parts=parts)
+        _insert_task(self.db_path, "c-old", status="failed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="旧输出", error_text="旧错")
+        worker = _worker(self.db_path, gateway=(None, None))
+        task = worker.get_task("coord")
+        orig_connect = worker_module._db_connect
+        db_path = self.db_path
+
+        class _FlippingConn:
+            """Proxy: flip the child to completed just before the
+            requeue UPDATE executes, so its rowcount is 0. The
+            snapshot connection is wrapped too, but only the
+            requeue UPDATE triggers the flip."""
+
+            def __init__(self, real):
+                self._real = real
+                self._flipped = False
+
+            def execute(self, sql, *args):
+                if not self._flipped and \
+                        "UPDATE tasks SET status = 'queued'" in sql:
+                    flip = sqlite3.connect(db_path)
+                    try:
+                        flip.execute(
+                            "UPDATE tasks SET status = 'completed'"
+                            " WHERE id = 'c-old'")
+                        flip.commit()
+                    finally:
+                        flip.close()
+                    self._flipped = True
+                return self._real.execute(sql, *args)
+
+            def commit(self):
+                return self._real.commit()
+
+            def close(self):
+                return self._real.close()
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+
+        with mock.patch.object(
+                worker_module, "_db_connect",
+                side_effect=lambda p: _FlippingConn(orig_connect(p))):
+            children = worker._coord_sync_children(
+                task, task["coordinator_parts"])
+        self.assertEqual(children, [{"name": "后端", "id": "c-old"}])
+        old = _row(self.db_path, "c-old")
+        self.assertEqual(old["status"], "completed")
+        self.assertEqual(old["result_text"], "旧输出")
+        actions = _audit_actions(self.db_path)
+        self.assertFalse([a for a, _ in actions
+                          if a == "coordinator_child_requeued"])
+
+
+class _HttpServerCase(unittest.TestCase):
+    def setUp(self):
+        self._old_policy_json = os.environ.get(
+            "CLOUD_ROUTING_POLICY_JSON")
+        self._old_policy_file = os.environ.get(
+            "CLOUD_ROUTING_POLICY_FILE")
+        os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None)
+        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()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        if self._old_policy_json is None:
+            os.environ.pop("CLOUD_ROUTING_POLICY_JSON", None)
+        else:
+            os.environ["CLOUD_ROUTING_POLICY_JSON"] = \
+                self._old_policy_json
+        if self._old_policy_file is None:
+            os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None)
+        else:
+            os.environ["CLOUD_ROUTING_POLICY_FILE"] = \
+                self._old_policy_file
+        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)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _project(self, name="p"):
+        status, project = self._req(
+            "POST", "/projects", {"name": name, "auto_pr": 1},
+            self.owner_key)
+        self.assertEqual(status, 201, project)
+        return project["id"]
+
+    def _insert_audit(self, project_id, action, detail):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO audit_log(user_id, project_id, action,"
+                " detail, created_at) VALUES(?, ?, ?, ?, ?)",
+                ("tester", project_id, action, detail, time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+
+
+class ShadowEndpointTestCase(_HttpServerCase):
+    def test_report_carries_capability_and_explicit_failures(self):
+        os.environ["CLOUD_ROUTING_POLICY_JSON"] = json.dumps({
+            "mode": "active",
+            "tiers": {"default": {"model": "policy-model",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "policy-model": {"tool_calling": False},
+            },
+        })
+        pid = self._project()
+        _insert_task(self.db_path, "T1", model="policy-model",
+                     status="completed", project_id=pid)
+        self._insert_audit(
+            pid, "worker_routing",
+            "T1 tier=default source=policy applied=true"
+            " suggested=policy-model chosen=policy-model")
+        self._insert_audit(
+            pid, "worker_capability_gateway_explicit",
+            "E1 model=explicit-model tier=default mode=active"
+            " missing=tool_calling unknown=False blocked=True")
+        status, report = self._req(
+            "GET", "/routing-shadow?project_id=" + pid,
+            key=self.owner_key)
+        self.assertEqual(status, 200, report)
+        self.assertEqual(
+            report["tiers"]["default"]["capability"]["status"], "fail")
+        self.assertEqual(
+            report["tiers"]["default"]["capability"]["missing"],
+            ["tool_calling"])
+        self.assertFalse(report["truncated"])
+        failures = report["explicit_capability_failures"]
+        self.assertEqual(len(failures), 1)
+        self.assertEqual(failures[0], {
+            "task_id": "E1", "model": "explicit-model",
+            "tier": "default", "mode": "active",
+            "missing": ["tool_calling"], "unknown": False,
+            "blocked": True})
+        self.assertFalse(report["explicit_capability_failures_truncated"])
+        self.assertIn("top level", report["explicit_counts_note"])
+        self.assertIn("per-tier", report["explicit_counts_note"])
+        self.assertIn("worker_capability_gateway_explicit",
+                      report["explicit_counts_note"])
+
+    def test_report_truncated_flag(self):
+        pid = self._project()
+        _insert_task(self.db_path, "T1", model="policy-model",
+                     status="completed", project_id=pid)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.executemany(
+                "INSERT INTO audit_log(user_id, project_id, action,"
+                " detail, created_at) VALUES(?, ?, ?, ?, ?)",
+                [("tester", pid, "worker_routing",
+                  "T1 tier=default source=policy applied=true"
+                  " suggested=policy-model chosen=policy-model",
+                  time.time())] * 10001)
+            conn.commit()
+        finally:
+            conn.close()
+        status, report = self._req(
+            "GET", "/routing-shadow?project_id=" + pid,
+            key=self.owner_key)
+        self.assertEqual(status, 200, report)
+        self.assertTrue(report["truncated"])
+        self.assertIn("truncated", report["recommendation"])
+
+
+class CreateRejectionTestCase(_HttpServerCase):
+    def _create(self, extra):
+        pid = self._project()
+        body = {"project_id": pid, "prompt": "协调干活"}
+        body.update(extra)
+        return self._req("POST", "/tasks", body, self.owner_key)
+
+    def test_coordinator_rejects_tools_instructions_approvals(self):
+        parts = [{"name": "后端", "prompt": "做后端"}]
+        agents = [
+            {"name": "x-agent", "tools": ["websearch"]},
+            {"name": "x-agent", "instructions": "Do it."},
+            {"name": "x-agent", "tool_approvals": {"websearch": "ask"}},
+        ]
+        for agent in agents:
+            status, data = self._create({
+                "coordinator_parts": parts, "agent": agent})
+            self.assertEqual(status, 400, (agent, data))
+            self.assertIn("cannot be used with", data["error"])
+
+    def test_goal_rejects_tools_instructions_approvals(self):
+        agents = [
+            {"name": "x-agent", "tools": ["websearch"]},
+            {"name": "x-agent", "instructions": "Do it."},
+            {"name": "x-agent", "tool_approvals": {"websearch": "ask"}},
+        ]
+        for agent in agents:
+            status, data = self._create({
+                "goal_mode": 1, "agent": agent})
+            self.assertEqual(status, 400, (agent, data))
+            self.assertIn("cannot be used with", data["error"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -158,7 +158,8 @@ from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, RoutingPolicy, decide_routing, \
     check_model_capabilities, classify_task, \
     is_fallbackable, load_routing_policy, parse_routing_policy, \
-    record_fallback, record_routing
+    record_fallback, record_routing, \
+    split_model_capability_missing
 from cloud.auto_memory import AutoMemoryStore
 from cloud.auto_memory import build_injection_prefix as _auto_prefix
 from cloud.auto_memory import process_due_tasks as _auto_process_due
@@ -6121,8 +6122,10 @@ class Worker:
         # Audit fix 3（B1）：触发条件是“走了 sub2api 网关”，
         # 不是“网关填了模型”。任务级显式模型此前因
         # _gateway_filled=False 全程跳过能力检查，是生产
-        # 覆盖的一半缺口；显式优先原则不变——显式模型
-        # 不满足/unknown 时只审计、不拦任务（见下）。
+        # 覆盖的一半缺口。Audit fix 4（第三遍审计分档建议）：
+        # 显式优先保护的是“用户选了哪个模型”，不是“选的
+        # 模型一定能跑”——显式模型已知不满足时同样拦，
+        # 只有无法验证时才只审计不拦（见下）。
         _explicit_model = bool(task.get("model"))
         _gateway_filled = False
         gateway_provider = None
@@ -6167,6 +6170,10 @@ class Worker:
         # active 下不满足即拦，不存在“先试 worker 默认”
         # 的中间态；运营方须先把网关模型写进能力目录，
         # 否则 unknown 同样拦（fail-closed，与 v8 一致）。
+        # 例外：任务级显式模型按 Audit fix 4 分档——已知
+        # 不满足才拦，unknown 只审计（见下），因为显式
+        # 优先保护的是用户选了哪个模型，不是选的模型
+        # 一定能跑；目录陈旧不应惩罚正确的显式选择。
         if gateway_provider == "sub2api" \
                 and self.routing_policy is not None:
             _gw_tier = classify_task(task)
@@ -6186,19 +6193,46 @@ class Worker:
                                         "cannot verify"
                                         if _gw_unknown else "lacks",
                                         ", ".join(_gw_missing)))
+                        # Audit fix 4（第三遍审计分档建议）：
+                        # 显式模型按 known/unknown 分档。known
+                        # 是目录明确写了不满足（布尔 False、
+                        # 或 context 低于需求），与网关默认
+                        # 模型同款 fail-fast；unknown 是目录
+                        # 无该模型、或条目缺字段，目录陈旧
+                        # 不应惩罚用户的正确选择，只审计不拦。
+                        # shadow 档两种情况都只审计。
+                        _gw_known, _gw_unk = \
+                            split_model_capability_missing(
+                                self.routing_policy.model_capabilities,
+                                task.get("model"), _gw_requires)
+                        _gw_block = False
+                        _gw_error = None
                         if _explicit_model:
-                            # 显式优先：不拦任务，但绝不能无声
-                            # 豁免——记独立审计，含缺项/unknown，
-                            # 影子解析器不认此 action，不污染样本。
+                            _gw_blocked = (
+                                _gw_mode == "active" and bool(_gw_known))
+                            # 显式检查结果独立审计，含 blocked
+                            # 标志：被拦的已知不满足也要进影子
+                            # 报告的显式失败清单，否则报告会
+                            # 漏掉恰恰最该被看到的一类。
                             self._audit(
                                 task["project_id"],
                                 "worker_capability_gateway_explicit",
                                 "%s model=%s tier=%s mode=%s"
-                                " missing=%s unknown=%s" % (
-                                    task_id, task.get("model"),
-                                    _gw_tier, _gw_mode,
-                                    ",".join(_gw_missing) or "-",
-                                    _gw_unknown))
+                                " missing=%s unknown=%s blocked=%s"
+                                % (task_id, task.get("model"),
+                                   _gw_tier, _gw_mode,
+                                   ",".join(_gw_missing) or "-",
+                                   _gw_unknown,
+                                   _gw_blocked))
+                            if _gw_blocked:
+                                _gw_block = True
+                                _gw_error = (
+                                    "%s tier explicit model %s lacks"
+                                    " capability: %s; refusing to"
+                                    " dispatch a model known to lack"
+                                    " a required capability"
+                                    % (_gw_tier, task.get("model"),
+                                       ", ".join(_gw_known)))
                         elif _gw_mode == "shadow":
                             self._audit(
                                 task["project_id"],
@@ -6206,11 +6240,13 @@ class Worker:
                                 "%s tier=%s %s (shadow, not blocked)"
                                 % (task_id, _gw_tier, _gw_note))
                         else:
+                            _gw_block = True
                             _gw_error = (
                                 "%s tier gateway model %s; refusing"
                                 " to dispatch an incapable or"
                                 " unverifiable model"
                                 % (_gw_tier, _gw_note))
+                        if _gw_block:
                             self._set_status(task_id, "failed")
                             self._write_execution_result(
                                 task_id, error_text=_gw_error,
@@ -6322,7 +6358,19 @@ class Worker:
             # 策略一改同一任务中途就换模型。CAS 式写回：
             # 只在库模型仍为空时写，绝不覆盖显式模型；
             # 写失败 best-effort，不炸派发（本轮内存模型
-            # 照常生效，只是下次派发可能再次决策）。
+            # 照常生效，只是下次派发可能再次决策），但
+            # 必须记警告日志，不能静默吞掉（Audit fix 4
+            # N5：否则 I5 的保证悄悄失效还看不出来）。
+            # Audit fix 4（N1 trade-off，只写注释、不改
+            # 语义）：写回是一把双刃剑——非网关路径下，
+            # 写回的模型使后续尝试走显式分支，decide_routing
+            # 的能力检查不再跑；两次尝试之间档 requires
+            # 一收紧，续跑任务将带着旧模型、不经复核直
+            # 接派发。网关路径不受影响：网关填模型的任务
+            # 跳过 decide_routing 与写回，每次尝试都重新
+            # 填网关模型并重新接受网关能力检查。复核方案
+            # （写回时记校验版本、或续跑时重查当前
+            # requires）是下一批的事，本轨不改写回语义。
             if decision.chosen_model:
                 try:
                     _wb_conn = _db_connect(self.db_path)
@@ -6334,8 +6382,12 @@ class Worker:
                         _wb_conn.commit()
                     finally:
                         _wb_conn.close()
-                except Exception:
-                    pass
+                except Exception as exc:
+                    log_event(self.logger, "worker_model_writeback_failed",
+                              task_id=task_id,
+                              project_id=task["project_id"],
+                              status="warning",
+                              message="model writeback failed: %s" % exc)
             task["model"] = decision.chosen_model or self.model
         elif not _gateway_filled:
             # Audit fix 3（I8）：显式模型任务（含协调员子任务