fix: N1 persisted-model recheck (recheck + block)

N1-1: writeback scope verified too broad (any chosen, incl. shadow
applied=False, worker_default and worker_explicit) and narrowed to
source=policy + applied only; writeback now records a
worker_model_persisted audit row as the identity marker.
N1: persisted models are rechecked against current tier requires
before dispatch on later attempts (fixes4 split known/unknown):
active + known-missing fails fast (worker_capability_recheck_
rejected, no runner), unknown-only and shadow only audit.
User-explicit, worker-CLI-explicit and gateway-filled tasks skip
the recheck; legacy models are recognised by a policy applied
worker_routing row. Recheck never rewrites tasks.model and never
re-decides.

cloud/tests/test_audit_fixes5.py

@@ -0,0 +1,313 @@
+"""Audit fix 5: N1 persisted-model recheck (recheck + block).
+
+- N1-1 (writeback scope): only source=policy + applied decisions are
+  persisted. Shadow (applied=False), worker_default (no policy / no
+  tier model / tier off / capability fallback) and worker_explicit
+  (CLI model) chosen models must NOT land in tasks.model, or the
+  next attempt would mistake them for a task-explicit model.
+- N1 (recheck): a model persisted by I5 writeback is rechecked
+  against the CURRENT tier requires before dispatch on later
+  attempts. Active + known-missing fails fast (runner never
+  called); unknown-only audits only; shadow never blocks.
+  User-explicit models are untouched by the recheck, and
+  gateway-filled tasks skip it (the gateway check already
+  re-runs every attempt).
+- Identification: writeback records worker_model_persisted;
+  pre-fix5 models are recognised by a legacy worker_routing
+  source=policy applied=true row with the same chosen model.
+  A source=explicit row alone proves nothing: the elif branch
+  writes one for every task that arrives with a model,
+  including persisted ones.
+
+Run: python3 -m unittest discover -s cloud/tests -p "test_audit_fixes5.py" -v
+No real models, no network, no real gateway: gateway steps are
+stubbed exactly like test_audit_fixes.py / test_audit_fixes4.py.
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from types import SimpleNamespace
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from cloud.server import 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"):
+    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, model)"
+            " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester',"
+            " ?, ?, NULL, ?)",
+            (task_id, project_id, status, now, now, model),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _insert_audit(db_path, project_id, action, detail):
+    conn = sqlite3.connect(db_path)
+    try:
+        # audit_log is created lazily by AuthStore on first audit;
+        # pre-seeding history must create it first.
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS audit_log("
+            "id INTEGER PRIMARY KEY AUTOINCREMENT, "
+            "user_id TEXT, project_id TEXT, action TEXT, "
+            "detail TEXT, created_at REAL)")
+        conn.commit()
+        conn.execute(
+            "INSERT INTO audit_log(user_id, project_id, action, detail,"
+            " created_at) VALUES('system', ?, ?, ?, ?)",
+            (project_id, action, detail, time.time()),
+        )
+        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"):
+    from cloud.auth import AuthStore
+    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,
+            model=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, model=model)
+    if gateway is None:
+        worker._ensure_gateway = lambda task: (None, None)
+    else:
+        worker._ensure_gateway = lambda task: gateway
+        worker._sub2api_default_model = lambda e: "gateway-model"
+    worker._calls = calls
+    return worker
+
+
+def _policy(mode, capabilities, requires=None):
+    return {
+        "mode": mode,
+        "tiers": {"default": {
+            "model": "policy-model",
+            "requires": requires if requires is not None
+            else {"tool_calling": True}}},
+        "model_capabilities": capabilities,
+    }
+
+
+class WritebackScopeTestCase(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_policy_applied_decision_is_persisted(self):
+        _insert_task(self.db_path, "T1")
+        worker = _worker(self.db_path, _policy("active", {
+            "policy-model": {"tool_calling": True},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(_row(self.db_path, "T1")["model"],
+                         "policy-model")
+        persisted = [d for a, d in _audit_actions(self.db_path)
+                     if a == "worker_model_persisted"]
+        self.assertTrue(persisted)
+        self.assertIn("model=policy-model", persisted[0])
+
+    def test_shadow_decision_is_not_persisted(self):
+        _insert_task(self.db_path, "T1")
+        worker = _worker(self.db_path, _policy("shadow", {
+            "policy-model": {"tool_calling": True},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertIsNone(_row(self.db_path, "T1")["model"])
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a == "worker_model_persisted"])
+
+    def test_worker_default_decision_is_not_persisted(self):
+        _insert_task(self.db_path, "T1")
+        worker = _worker(self.db_path, None)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertIsNone(_row(self.db_path, "T1")["model"])
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a == "worker_model_persisted"])
+
+    def test_worker_explicit_decision_is_not_persisted(self):
+        _insert_task(self.db_path, "T1")
+        worker = _worker(self.db_path, None, model="cli-model")
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertIsNone(_row(self.db_path, "T1")["model"])
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a == "worker_model_persisted"])
+
+
+class RecheckTestCase(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 _tightened(self, mode, capabilities):
+        return _policy(mode, capabilities)
+
+    def test_persisted_model_unchanged_requires_dispatches(self):
+        _insert_task(self.db_path, "T1", model="persisted-model")
+        _insert_audit(self.db_path, "p1", "worker_model_persisted",
+                      "T1 tier=default model=persisted-model")
+        worker = _worker(self.db_path, self._tightened("active", {
+            "persisted-model": {"tool_calling": True},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a.startswith("worker_capability_recheck")])
+
+    def test_persisted_model_known_missing_is_blocked(self):
+        _insert_task(self.db_path, "T1", model="persisted-model")
+        _insert_audit(self.db_path, "p1", "worker_model_persisted",
+                      "T1 tier=default model=persisted-model")
+        worker = _worker(self.db_path, self._tightened("active", {
+            "persisted-model": {"tool_calling": False},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(worker._calls)
+        self.assertIn("tool_calling", result["error"])
+        self.assertTrue([a for a, _ in _audit_actions(self.db_path)
+                         if a == "worker_capability_recheck_rejected"])
+        # Recheck never rewrites the task model.
+        self.assertEqual(_row(self.db_path, "T1")["model"],
+                         "persisted-model")
+
+    def test_legacy_policy_row_model_is_rechecked(self):
+        _insert_task(self.db_path, "T1", model="persisted-model")
+        _insert_audit(self.db_path, "p1", "worker_routing",
+                      "T1 tier=default source=policy applied=true"
+                      " suggested=- chosen=persisted-model")
+        worker = _worker(self.db_path, self._tightened("active", {
+            "persisted-model": {"tool_calling": False},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertFalse(worker._calls)
+        self.assertTrue([a for a, _ in _audit_actions(self.db_path)
+                         if a == "worker_capability_recheck_rejected"])
+
+    def test_persisted_model_unknown_is_audited_not_blocked(self):
+        _insert_task(self.db_path, "T1", model="persisted-model")
+        _insert_audit(self.db_path, "p1", "worker_model_persisted",
+                      "T1 tier=default model=persisted-model")
+        worker = _worker(self.db_path, self._tightened("active", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        self.assertTrue([a for a, _ in _audit_actions(self.db_path)
+                         if a == "worker_capability_recheck_unknown"])
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a == "worker_capability_recheck_rejected"])
+
+    def test_persisted_model_shadow_known_missing_is_not_blocked(self):
+        _insert_task(self.db_path, "T1", model="persisted-model")
+        _insert_audit(self.db_path, "p1", "worker_model_persisted",
+                      "T1 tier=default model=persisted-model")
+        worker = _worker(self.db_path, self._tightened("shadow", {
+            "persisted-model": {"tool_calling": False},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        self.assertTrue([a for a, _ in _audit_actions(self.db_path)
+                         if a == "worker_capability_recheck_shadow"])
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a == "worker_capability_recheck_rejected"])
+
+    def test_user_explicit_model_is_not_rechecked(self):
+        # An explicit-only history must not be mistaken for a
+        # persisted model: the elif branch writes a source=explicit
+        # row for every task that arrives with a model, so that
+        # row alone proves nothing about origin.
+        _insert_task(self.db_path, "T1", model="explicit-model")
+        _insert_audit(self.db_path, "p1", "worker_routing",
+                      "T1 tier=default source=explicit applied=false"
+                      " suggested=- chosen=explicit-model")
+        worker = _worker(self.db_path, self._tightened("active", {
+            "explicit-model": {"tool_calling": False},
+        }))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a.startswith("worker_capability_recheck")])
+
+    def test_gateway_filled_task_skips_recheck(self):
+        _insert_task(self.db_path, "T1")
+        worker = _worker(self.db_path, self._tightened("active", {
+            "gateway-model": {"tool_calling": True},
+        }), gateway=("sub2api", {}))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertTrue(worker._calls)
+        self.assertFalse([a for a, _ in _audit_actions(self.db_path)
+                          if a.startswith("worker_capability_recheck")])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -160,6 +160,8 @@ from cloud.models import ModelRouter, RoutingPolicy, decide_routing, \
     is_fallbackable, load_routing_policy, parse_routing_policy, \
     record_fallback, record_routing, \
     split_model_capability_missing
+from cloud.routing_shadow import parse_worker_routing_detail as \
+    _parse_routing_detail
 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
@@ -5812,6 +5814,105 @@ class Worker:
             return []
         return _missing_tools(tools, config_text)
 
+    def _is_persisted_model(self, task_id, project_id, model):
+        """判断任务当前模型是不是 I5 写回的模型（Audit fix 5，N1）。
+
+        区分依据，按优先级：
+        1. worker_model_persisted 审计行（写回时同步记，model 匹配）
+           ——最直接的证据；
+        2. 旧模型兜底：worker_routing 里 source=policy、applied=true、
+           chosen 等于当前模型的行（fixes3 的写回只在 policy 决策
+           上留痕，fixes5 之前没有 persisted 审计行）；
+        3. 其余一律视为非写回（用户显式或来源不明）——来源不明
+           默认按显式处理，绝不因猜错来源而拦住用户的显式选择。
+
+        注意 source=explicit 行不能当用户显式的证据：elif 分支
+        会给每一个带模型进场的任务（包括写回模型）都记一条
+        explicit 行，所以它在新旧两种模型的历史里都会出现，
+        区分力为零，故本函数根本不读它。policy 行则不同：
+        带模型进场的任务永远不再跑 decide_routing，所以一条
+        policy 行只可能来自该任务曾经模型为空的那次派发，
+        也就是写回的来源。
+        查找失败（DB 异常）时返回 False 并记警告日志：复核查
+        不到依据就跳过，而不是拦任务。
+        """
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                rows = conn.execute(
+                    "SELECT action, detail FROM audit_log"
+                    " WHERE project_id = ? AND action IN"
+                    " ('worker_model_persisted', 'worker_routing')"
+                    " AND detail LIKE ?",
+                    (project_id, task_id + " %")).fetchall()
+            finally:
+                conn.close()
+        except Exception as exc:
+            log_event(self.logger,
+                      "worker_persisted_model_lookup_failed",
+                      task_id=task_id, project_id=project_id,
+                      status="warning",
+                      message="persisted model lookup failed: %s" % exc)
+            return False
+        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:
+                return True
+        for row in rows:
+            if row[0] != "worker_routing":
+                continue
+            parsed = _parse_routing_detail(row[1])
+            if parsed and parsed["source"] == "policy" \
+                    and parsed["applied"] \
+                    and parsed["chosen_model"] == model:
+                return True
+        return False
+
+    def _recheck_persisted_model(self, task):
+        """复核写回模型（Audit fix 5，N1，复核+拦）。
+
+        只在调用方确认任务带模型进场、且非网关任务时调用。
+        返回错误串（active 下 known 非空时）或 None（放行）。
+        无策略、无档 requires、档模式 off 时不查；shadow 档
+        两种情况都只审计；unknown-only 只审计不拦。复核不改
+        tasks.model、不重新决策、不调 decide_routing——它只是
+        在派发前对旧合格证做一次当前 requires 的重验。
+        """
+        if self.routing_policy is None:
+            return None
+        tier = classify_task(task)
+        requires = self.routing_policy.tier_requirements.get(tier) or {}
+        if not requires:
+            return None
+        mode = self.routing_policy.tier_mode(tier)
+        if mode not in ("active", "shadow"):
+            return None
+        model = task.get("model")
+        ok, missing, unknown = check_model_capabilities(
+            self.routing_policy.model_capabilities, model, requires)
+        if ok:
+            return None
+        known, _unk = split_model_capability_missing(
+            self.routing_policy.model_capabilities, model, requires)
+        detail = ("%s model=%s tier=%s mode=%s missing=%s unknown=%s"
+                  % (task["id"], model, tier, mode,
+                     ",".join(missing) or "-", unknown))
+        if mode == "shadow":
+            self._audit(task["project_id"],
+                        "worker_capability_recheck_shadow", detail)
+            return None
+        if not known:
+            self._audit(task["project_id"],
+                        "worker_capability_recheck_unknown", detail)
+            return None
+        return ("%s tier persisted model %s lacks capability: %s;"
+                " refusing to dispatch a model known to lack a"
+                " required capability"
+                % (tier, model, ", ".join(known)))
+
     def run_one(self, task=None):
         """认领并执行一个任务，返回结果 dict（见模块 docstring）。
 
@@ -6361,17 +6462,27 @@ class Worker:
             # 照常生效，只是下次派发可能再次决策），但
             # 必须记警告日志，不能静默吞掉（Audit fix 4
             # N5：否则 I5 的保证悄悄失效还看不出来）。
-            # Audit fix 4（N1 trade-off，只写注释、不改
+            # Audit fix 5（N1-1，写回范围核实）：旧代码对
-            # 语义）：写回是一把双刃剑——非网关路径下，
+            # 任何 chosen 都写回，包括 shadow（applied=False，
-            # 写回的模型使后续尝试走显式分支，decide_routing
+            # chosen 只是 worker 默认）与 worker_default（无
-            # 的能力检查不再跑；两次尝试之间档 requires
+            # 策略/档无模型/档 off/能力回落）与 worker_explicit
-            # 一收紧，续跑任务将带着旧模型、不经复核直
+            # （CLI 模型）。那些都不是档策略的 applied 决策，
-            # 接派发。网关路径不受影响：网关填模型的任务
+            # 写回会把它们伪装成任务显式模型：下次进场走
-            # 跳过 decide_routing 与写回，每次尝试都重新
+            # 显式分支不再决策，CLI 改了也不生效，shadow
-            # 填网关模型并重新接受网关能力检查。复核方案
+            # 的建议模型还会被当成已应用。本轨起只持久化
-            # （写回时记校验版本、或续跑时重查当前
+            # source=policy 且 applied 的决策。
-            # requires）是下一批的事，本轨不改写回语义。
+            # Audit fix 5（N1）：写回的另一面由派发前复核
-            if decision.chosen_model:
+            # 兜底（见 run_one 末尾的复核块）——写回模型
+            # 在后续尝试里走显式分支，decide_routing 的
+            # 能力检查不再跑；两次尝试之间档 requires 一
+            # 收紧，复核会在派发前用当前 requires 重验，
+            # 已知不满足在 active 下拦、unknown 只审计。
+            # 网关路径不受影响：网关填模型的任务跳过
+            # decide_routing 与写回，每次尝试都重新填网关
+            # 模型并重新接受网关能力检查。
+            _persisted = False
+            if decision.chosen_model and decision.source == "policy" \
+                    and decision.applied:
                 try:
                     _wb_conn = _db_connect(self.db_path)
                     try:
@@ -6382,12 +6493,23 @@ class Worker:
                         _wb_conn.commit()
                     finally:
                         _wb_conn.close()
+                    _persisted = True
                 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)
+                if _persisted:
+                    # 写回模型的身份标记：后续尝试据此区分
+                    # 写回模型与用户显式模型（见
+                    # _is_persisted_model）。审计本身 best-effort，
+                    # 丢了也不影响写回，旧模型靠 policy 行兜底。
+                    self._audit(task["project_id"],
+                                "worker_model_persisted",
+                                "%s tier=%s model=%s" % (
+                                    task_id, decision.tier,
+                                    decision.chosen_model))
             task["model"] = decision.chosen_model or self.model
         elif not _gateway_filled:
             # Audit fix 3（I8）：显式模型任务（含协调员子任务
@@ -6401,6 +6523,45 @@ class Worker:
                         " suggested=- chosen=%s" % (
                             task_id, classify_task(task),
                             task.get("model")))
+        # Audit fix 5（N1，复核+拦）：写回模型在后续尝试里
+        # 走显式分支、能力检查不再跑，档 requires 收紧会
+        # 带着过期合格证派发，故在 runner 调用之前复核一次。
+        # 只在任务带模型进场、且非网关任务时复核：网关任务
+        # 每尝试都走网关检查（sub2api 时模式 off 也不查，
+        # 那是运营方选的 off，不能由复核越过）；用户显式与
+        # worker CLI 显式跳过（语义不变，见 _is_persisted_model
+        # 的来源不明默认显式规则）；复核不改库模型、不重新
+        # 决策。首轮刚写回的任务不在此列：本轮 decide_routing
+        # 已按当前 requires 查过，复核只针对跨尝试的旧模型。
+        if _explicit_model and not _gateway_filled \
+                and gateway_provider != "sub2api" \
+                and self._is_persisted_model(task_id,
+                                             task["project_id"],
+                                             task.get("model")):
+            _recheck_error = self._recheck_persisted_model(task)
+            if _recheck_error:
+                self._set_status(task_id, "failed")
+                self._write_execution_result(
+                    task_id, error_text=_recheck_error,
+                    actual_model=self.model,
+                    started_at=started_at, finished_at=time.time(),
+                    only_if_status="failed")
+                self._audit(task["project_id"],
+                            "worker_capability_recheck_rejected",
+                            "%s: %s" % (task_id, _recheck_error))
+                log_event(self.logger,
+                          "worker_capability_recheck_rejected",
+                          task_id=task_id, project_id=task["project_id"],
+                          status="failed")
+                self._cleanup_task_seed(task)
+                self._running_clear(task_id)
+                return {"dispatched": True, "task_id": task_id,
+                        "host_name": self.executor._host_name(task)
+                        if hasattr(self.executor, "_host_name")
+                        else task_id,
+                        "status": "failed", "executor": self._executor_name(),
+                        "stdout": "", "error": _recheck_error, "events": [],
+                        "cost_record_id": None}
         # agent 指令通道（v2）：local/docker 由 sandbox 的
         # --developer-instructions 派发，prompt 保持干净；通用
         # cloud 执行器没有这个通道，只能在这里把 v1 的带标记