feat: capability-based routing (v8)

Policy carries an operator-declared model_capabilities directory
and per-tier requires (tool_calling/reasoning/vision booleans,
min_context_tokens). decide_routing checks the policy tier model
in active mode: capable applies, incapable falls back to a capable
worker default, and if both fail or are unknown it returns
capability_blocked, which the worker turns into a failed task
(v3 tools-gate shape, runner never called). Shadow mode only
annotates the reason, never blocks. Explicit task and worker CLI
models bypass the check by design. Invalid requires drops only
that tier (fail-closed, like invalid tier mode). Dry-run reports
per-tier capability status and blocks on active tiers that fail
or cannot be verified.

cloud/models.py

@@ -117,6 +117,131 @@ class ModelRouter:
 ROUTING_MODES = ("off", "shadow", "active")
 ROUTING_TIERS = ("plan", "goal", "default")
 
+# 能力路由（v8）：能力目录与档需求的字段名。布尔能力只认
+# True 为“具备”，False 为“已知不具备”，缺失为“未知”——
+# 缺失绝不按 False 算（那会把“没声明”说成“没能力”），
+# 也绝不按 True 算（那会把“没声明”说成“有能力”）。
+CAPABILITY_BOOL_KEYS = ("tool_calling", "reasoning", "vision")
+CAPABILITY_INT_KEYS = ("context_tokens",)
+REQUIREMENT_BOOL_KEYS = CAPABILITY_BOOL_KEYS
+REQUIREMENT_INT_KEYS = ("min_context_tokens",)
+
+
+def _is_bool(value):
+    return isinstance(value, bool)
+
+
+def _is_int(value):
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
+def parse_model_capabilities(raw):
+    """清洗能力目录：{model: {context_tokens, tool_calling,
+    reasoning, vision}}，只保留字段全合法的条目。
+
+    目录是运营方手工维护的声明，不是中转 /models 的实测
+    （中转只回 id，不回能力与价格，见路由 v0 的价格口径）：
+    运营方写错一个条目时，整条丢掉比只丢错字段更安全——
+    只丢错字段会让“要求工具调用”悄悄变成“不查工具调用”。
+    非 dict、模型名为空、context_tokens 非正整数、布尔
+    能力非布尔，一律整条丢掉，调用方据此把该模型的能力
+    判为未知。
+    """
+    out = {}
+    if not isinstance(raw, dict):
+        return out
+    for model, entry in raw.items():
+        model = _clean_model_name(model)
+        if not model or not isinstance(entry, dict):
+            continue
+        cleaned = {}
+        valid = True
+        for key in CAPABILITY_INT_KEYS:
+            if key in entry:
+                if not _is_int(entry[key]) or entry[key] <= 0:
+                    valid = False
+                    break
+                cleaned[key] = entry[key]
+        if not valid:
+            continue
+        for key in CAPABILITY_BOOL_KEYS:
+            if key in entry:
+                if not _is_bool(entry[key]):
+                    valid = False
+                    break
+                cleaned[key] = entry[key]
+        if not valid:
+            continue
+        out[model] = cleaned
+    return out
+
+
+def parse_tier_requires(raw):
+    """校验一档的需求：返回清洗后的 dict，或 None 表示非法。
+
+    合法需求：布尔能力键只接受布尔值（False 等于“无此
+    要求”，不是“要求不具备”）；min_context_tokens 只接受
+    正整数；未知键一律非法——运营方拼错需求名时，若静默
+    忽略，他会以为该要求已生效，实际没生效。非法需求的
+    处理在 parse_routing_policy：整档丢掉（与无效档模式
+    同款 fail-closed），不连累其他档。
+    """
+    if raw is None:
+        return {}
+    if not isinstance(raw, dict):
+        return None
+    out = {}
+    for key, value in raw.items():
+        if key in REQUIREMENT_BOOL_KEYS:
+            if not _is_bool(value):
+                return None
+            if value:
+                out[key] = True
+        elif key in REQUIREMENT_INT_KEYS:
+            if not _is_int(value) or value <= 0:
+                return None
+            out[key] = value
+        else:
+            return None
+    return out
+
+
+def check_model_capabilities(capabilities, model, requires):
+    """查一个模型是否满足需求：返回 (ok, missing, unknown)。
+
+    requires 为空时恒为 ok。模型不在目录里时全部需求都
+    无法验证：ok=False，missing 列全部需求名，unknown=True。
+    目录里有模型但某项能力缺失时，该项计 unknown（不是
+    已知不具备），missing 里同样列它——active 模式下
+    fail 与 unknown 都拦任务，区别只在报错文案里写清是
+    “已知缺乏”还是“无法验证”。
+    """
+    requires = requires or {}
+    if not requires:
+        return True, [], False
+    model = _clean_model_name(model)
+    entry = (capabilities or {}).get(model) if model else None
+    if entry is None:
+        return False, sorted(requires.keys()), True
+    missing = []
+    unknown = False
+    for key, value in requires.items():
+        if key in REQUIREMENT_BOOL_KEYS:
+            flag = entry.get(key)
+            if flag is True:
+                continue
+            missing.append(key)
+            if flag is None:
+                unknown = True
+        elif key in REQUIREMENT_INT_KEYS:
+            context = entry.get("context_tokens")
+            if context is None:
+                missing.append(key)
+                unknown = True
+            elif context < value:
+                missing.append(key)
+    return (not missing), sorted(missing), unknown
+
 
 def prompt_len_bucket(prompt) -> str:
     """Bucket a prompt by length for logging only, never for routing.
@@ -194,10 +319,15 @@ class RoutingPolicy:
     remains the only source that can enable a real fallback retry,
     so a policy file alone can never start burning a second
     attempt's quota.
+    ``model_capabilities`` (v8) is the operator-declared
+    capability directory, and ``tier_requirements`` maps a tier
+    to its requires dict; both are filtered to tiers/models
+    that actually exist, exactly like ``tier_modes``.
     """
 
     def __init__(self, version=None, mode="shadow", tiers=None,
-                 fallback_models=None, tier_modes=None):
+                 fallback_models=None, tier_modes=None,
+                 model_capabilities=None, tier_requirements=None):
         self.version = version if isinstance(version, str) \
             and version.strip() else None
         self.mode = mode if mode in ROUTING_MODES else "off"
@@ -206,6 +336,14 @@ class RoutingPolicy:
         for tier, tier_mode in (tier_modes or {}).items():
             if tier in self.tiers and tier_mode in ROUTING_MODES:
                 self.tier_modes[tier] = tier_mode
+        self.model_capabilities = parse_model_capabilities(
+            model_capabilities)
+        self.tier_requirements = {}
+        for tier, requires in (tier_requirements or {}).items():
+            if tier in self.tiers and requires:
+                cleaned = parse_tier_requires(requires)
+                if cleaned is not None:
+                    self.tier_requirements[tier] = cleaned
 
     def tier_mode(self, tier):
         """Effective mode for one tier: explicit tier mode, else global."""
@@ -252,13 +390,32 @@ def parse_routing_policy(data):
             tiers.pop(tier, None)
     if not tiers:
         return None
+    # Per-tier requirements (v8): a tier entry dict may carry
+    # "requires". An invalid requires (unknown key, bad value)
+    # drops the whole tier entry, same fail-closed rule as an
+    # invalid mode: a typo'd requirement must never be silently
+    # ignored while the operator believes it is enforced.
+    tier_requirements = {}
+    for tier, entry in raw_tiers.items():
+        if not isinstance(entry, dict) or "requires" not in entry:
+            continue
+        cleaned = parse_tier_requires(entry.get("requires"))
+        if cleaned is None:
+            tiers.pop(tier, None)
+        elif cleaned:
+            tier_requirements[tier] = cleaned
+    if not tiers:
+        return None
     fallback_models = data.get("fallback_models")
     if fallback_models is not None and not isinstance(fallback_models,
                                                      list):
         return None
     return RoutingPolicy(version=data.get("version"), mode=mode,
                          tiers=tiers, fallback_models=fallback_models,
-                         tier_modes=tier_modes)
+                         tier_modes=tier_modes,
+                         model_capabilities=data.get(
+                             "model_capabilities"),
+                         tier_requirements=tier_requirements)
 
 
 def load_routing_policy(env=None):
@@ -347,6 +504,28 @@ def decide_routing(task, worker_model, policy=None,
        suggests it, ``off`` falls back to the worker default;
     4. worker default chain (config file / ``DEFAULT_MODEL``).
 
+    Capability filter (v8): when the tier carries ``requires``,
+    the policy tier model is checked against the policy's
+    ``model_capabilities`` directory. Explicit task models and
+    worker CLI models are never checked -- an explicit choice
+    is never overridden by a policy. In ``shadow`` mode a
+    mismatch only annotates the reason (shadow never changes
+    behaviour, never blocks). In ``active`` mode a mismatch
+    first tries the worker default: if the worker default
+    satisfies the same requirements it is used instead (source
+    worker_default, reason names the tier model's shortfall);
+    if the worker default also fails or is unknown, the
+    decision is ``source="capability_blocked"`` with
+    ``chosen_model=None`` -- the worker turns that into a
+    failed task with a clear error (same fail-fast shape as
+    the v3 required-tools gate), never silently dispatching
+    an incapable or unverifiable model. A tier with no
+    requirements, and a policy with no directory, behave
+    exactly as before. The directory is operator-declared,
+    never fetched from a provider, so a stale directory is
+    the operator's responsibility to fix; the dry-run tool
+    reports capability mismatches before activation.
+
     Shadow mode returns the worker default as ``chosen_model`` and the
     policy tier model as ``suggested_model`` with ``applied=False``:
     it observes without changing behaviour. This function is pure: it
@@ -399,19 +578,51 @@ def decide_routing(task, worker_model, policy=None,
             suggested_model=None, source="worker_default",
             policy_version=policy.version, applied=True,
             reason="%s tier mode is off" % tier, signals=signals)
+    requires = policy.tier_requirements.get(tier) or {}
+    cap_ok, cap_missing, cap_unknown = check_model_capabilities(
+        policy.model_capabilities, tier_model, requires)
+    cap_note = ""
+    if requires and not cap_ok:
+        cap_note = " (capability %s: %s)" % (
+            "unknown" if cap_unknown else "lacks",
+            ", ".join(cap_missing))
     if effective_mode == "shadow":
         return RoutingDecision(
             task_id=task.get("id"), tier=tier, chosen_model=worker_model,
             suggested_model=tier_model, source="shadow",
             policy_version=policy.version, applied=False,
-            reason="%s tier suggests %s (shadow, not applied)"
-            % (tier, tier_model), signals=signals)
+            reason="%s tier suggests %s (shadow, not applied)%s"
+            % (tier, tier_model, cap_note), signals=signals)
     # effective_mode == "active"
+    if cap_ok:
+        return RoutingDecision(
+            task_id=task.get("id"), tier=tier, chosen_model=tier_model,
+            suggested_model=None, source="policy",
+            policy_version=policy.version, applied=True,
+            reason="%s tier -> %s" % (tier, tier_model),
+            signals=signals)
+    # Tier model cannot be verified as capable. Try the worker
+    # default against the same requirements before blocking:
+    # a capable worker default keeps the task moving without
+    # violating the requirement.
+    worker_ok, _w_missing, _w_unknown = check_model_capabilities(
+        policy.model_capabilities, worker_model, requires)
+    if worker_ok:
+        return RoutingDecision(
+            task_id=task.get("id"), tier=tier, chosen_model=worker_model,
+            suggested_model=None, source="worker_default",
+            policy_version=policy.version, applied=True,
+            reason="%s tier model %s%s, worker default satisfies"
+            " requirements, falling back" % (tier, tier_model, cap_note),
+            signals=signals)
     return RoutingDecision(
-        task_id=task.get("id"), tier=tier, chosen_model=tier_model,
-        suggested_model=None, source="policy",
-        policy_version=policy.version, applied=True,
-        reason="%s tier -> %s" % (tier, tier_model), signals=signals)
+        task_id=task.get("id"), tier=tier, chosen_model=None,
+        suggested_model=tier_model, source="capability_blocked",
+        policy_version=policy.version, applied=False,
+        reason="%s tier model %s%s and worker default also cannot"
+        " satisfy requirements; refusing to dispatch an incapable"
+        " or unverifiable model" % (tier, tier_model, cap_note),
+        signals=signals)
 
 
 def record_routing(task_record, decision) -> dict:

cloud/routing_dryrun.py

@@ -19,7 +19,13 @@ What it reports, per tier (plan / goal / default):
   that tier,
 * tiers absent from the document (worker default, not routed),
 * unknown tier names in the document (ignored by the parser),
-* whether any tier would actually activate on next dispatch.
+* whether any tier would actually activate on next dispatch,
+* capability checks (v8): per tier, whether its model meets
+  its requires against the policy's model_capabilities
+  directory -- an active tier that fails or cannot be verified
+  is a BLOCKER (it would block dispatch at the worker
+  capability gate); a shadow/off tier with the same problem
+  is a warning only.
 
 What it does NOT do, honestly:
 * It does NOT validate model names against a provider or relay
@@ -56,11 +62,15 @@ import sys
 try:
     from cloud.models import ROUTING_MODES, ROUTING_TIERS
     from cloud.models import parse_routing_policy
+    from cloud.models import parse_tier_requires
+    from cloud.models import check_model_capabilities
 except ImportError:  # run as a script from the repo root
     sys.path.insert(0, os.path.join(
         os.path.dirname(os.path.abspath(__file__)), ".."))
     from cloud.models import ROUTING_MODES, ROUTING_TIERS
     from cloud.models import parse_routing_policy
+    from cloud.models import parse_tier_requires
+    from cloud.models import check_model_capabilities
 
 
 def _tier_dropped_reason(raw_entry):
@@ -68,6 +78,9 @@ def _tier_dropped_reason(raw_entry):
         if "mode" in raw_entry and \
                 raw_entry.get("mode") not in ROUTING_MODES:
             return "invalid tier mode %r" % (raw_entry.get("mode"),)
+        if "requires" in raw_entry and \
+                parse_tier_requires(raw_entry.get("requires")) is None:
+            return "invalid tier requires %r" % (raw_entry.get("requires"),)
         return "no usable model"
     if isinstance(raw_entry, str):
         return "empty model" if not raw_entry.strip() else "unknown"
@@ -135,6 +148,7 @@ def dry_run_policy_document(data, env_mode_override=None):
     # without mutating the parsed policy: a tier with an explicit
     # mode keeps it, a tier without one follows the overridden global.
     effective_policy_mode = global_mode
+    cap_blocked = False
     for tier in ROUTING_TIERS:
         raw_entry = raw_tiers.get(tier) if isinstance(raw_tiers,
                                                       dict) else None
@@ -166,11 +180,43 @@ def dry_run_policy_document(data, env_mode_override=None):
             mode_source = "global_mode"
         status = "active" if mode == "active" else (
             "shadow" if mode == "shadow" else "off")
+        # Capability check (v8): the same check decide_routing
+        # runs, reported before activation. "not_applicable"
+        # when the tier carries no requirements; "unknown" when
+        # requirements exist but the model (or a needed field)
+        # is absent from the directory.
+        requires = policy.tier_requirements.get(tier) or {}
+        if not requires:
+            cap = {"required": None, "missing": [], "unknown": False,
+                   "status": "not_applicable"}
+        else:
+            cap_ok, cap_missing, cap_unknown = check_model_capabilities(
+                policy.model_capabilities, policy.tiers[tier], requires)
+            cap = {"required": requires, "missing": cap_missing,
+                   "unknown": cap_unknown,
+                   "status": "ok" if cap_ok else (
+                       "unknown" if cap_unknown else "fail")}
         tiers_report[tier] = {
             "model": policy.tiers[tier], "mode": mode,
             "mode_source": mode_source, "status": status,
-            "dropped_reason": None,
+            "dropped_reason": None, "capabilities": cap,
         }
+        if cap["status"] in ("fail", "unknown"):
+            note = ("tier %r model %r %s capability: %s" % (
+                tier, policy.tiers[tier],
+                "cannot verify" if cap["status"] == "unknown"
+                else "lacks",
+                ", ".join(cap["missing"])))
+            if status == "active":
+                # An active tier that fails/unknowns its capability
+                # check would block dispatch (worker capability
+                # gate), so it blocks the dry-run too.
+                cap_blocked = True
+                warnings.append(note + "; applying this policy "
+                                "would block %r tasks at dispatch" % tier)
+            else:
+                warnings.append(note + "; %s mode does not block "
+                                "dispatch" % status)
     would_activate = [t for t in ROUTING_TIERS
                       if tiers_report[t]["status"] == "active"]
     would_shadow = [t for t in ROUTING_TIERS
@@ -193,7 +239,7 @@ def dry_run_policy_document(data, env_mode_override=None):
             "tasks on next worker dispatch if this policy were "
             "applied; this dry-run does not activate anything" %
             ", ".join(would_activate))
-    blocked = bool(dropped or unknown)
+    blocked = bool(dropped or unknown or cap_blocked)
     return {
         "valid": True, "blocked": blocked,
         "version": policy.version, "global_mode": global_mode,

cloud/tests/test_routing_capability.py

@@ -0,0 +1,318 @@
+"""Capability routing (v8): directory, requires, decision filter,
+worker capability gate, and dry-run reporting.
+
+Run: python3 -m unittest discover -s cloud/tests -p "test_routing_capability.py" -v
+"""
+
+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.models import (check_model_capabilities, decide_routing,
+                          parse_model_capabilities, parse_routing_policy,
+                          parse_tier_requires)
+from cloud.routing_dryrun import dry_run_policy_document
+from cloud.server import init_db
+from cloud.worker import Worker
+
+
+def _policy(**overrides):
+    doc = {
+        "version": "cap-v8",
+        "mode": "active",
+        "tiers": {
+            "plan": {"model": "plan-model",
+                     "requires": {"tool_calling": True,
+                                  "min_context_tokens": 128000}},
+        },
+        "model_capabilities": {
+            "plan-model": {"tool_calling": True,
+                           "context_tokens": 200000},
+            "worker-model": {"tool_calling": True,
+                             "context_tokens": 200000},
+            "no-tools-model": {"tool_calling": False,
+                               "context_tokens": 200000},
+            "small-model": {"tool_calling": True,
+                            "context_tokens": 32000},
+        },
+    }
+    doc.update(overrides)
+    return parse_routing_policy(doc)
+
+
+class CapabilityParseTestCase(unittest.TestCase):
+    def test_clean_capabilities_drops_bad_entries(self):
+        caps = parse_model_capabilities({
+            "good": {"tool_calling": True, "context_tokens": 1000},
+            "bad-bool": {"tool_calling": "yes"},
+            "bad-int": {"context_tokens": -5},
+            "": {"tool_calling": True},
+            "not-dict": "x",
+        })
+        self.assertEqual(list(caps.keys()), ["good"])
+
+    def test_requires_rejects_unknown_key_and_bad_value(self):
+        self.assertIsNone(parse_tier_requires({"fuzzy": True}))
+        self.assertIsNone(parse_tier_requires({"tool_calling": "yes"}))
+        self.assertIsNone(parse_tier_requires({"min_context_tokens": 0}))
+        self.assertEqual(parse_tier_requires({"tool_calling": False}), {})
+        self.assertEqual(
+            parse_tier_requires({"tool_calling": True,
+                                 "min_context_tokens": 100}),
+            {"tool_calling": True, "min_context_tokens": 100})
+
+    def test_invalid_requires_drops_only_that_tier(self):
+        doc = {
+            "mode": "active",
+            "tiers": {
+                "plan": {"model": "plan-model",
+                         "requires": {"fuzzy": True}},
+                "default": "default-model",
+            },
+        }
+        policy = parse_routing_policy(doc)
+        self.assertNotIn("plan", policy.tiers)
+        self.assertIn("default", policy.tiers)
+
+    def test_check_matrix(self):
+        caps = {"m": {"tool_calling": True, "context_tokens": 1000}}
+        self.assertEqual(check_model_capabilities(caps, "m", {}),
+                         (True, [], False))
+        self.assertEqual(check_model_capabilities(
+            caps, "m", {"tool_calling": True}), (True, [], False))
+        self.assertEqual(check_model_capabilities(
+            caps, "m", {"vision": True}), (False, ["vision"], True))
+        self.assertEqual(check_model_capabilities(
+            caps, "m", {"min_context_tokens": 2000}),
+            (False, ["min_context_tokens"], False))
+        self.assertEqual(check_model_capabilities(
+            caps, "absent", {"tool_calling": True}),
+            (False, ["tool_calling"], True))
+
+
+class CapabilityDecisionTestCase(unittest.TestCase):
+    def test_active_capable_applies_policy_model(self):
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "worker-model",
+            policy=_policy())
+        self.assertEqual(decision.source, "policy")
+        self.assertEqual(decision.chosen_model, "plan-model")
+        self.assertTrue(decision.applied)
+
+    def test_active_incapable_falls_back_to_capable_worker_default(self):
+        policy = _policy()
+        policy.tiers["plan"] = "no-tools-model"
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "worker-model", policy=policy)
+        self.assertEqual(decision.source, "worker_default")
+        self.assertEqual(decision.chosen_model, "worker-model")
+        self.assertIn("lacks", decision.reason)
+
+    def test_active_incapable_and_worker_incapable_blocks(self):
+        policy = _policy()
+        policy.tiers["plan"] = "no-tools-model"
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "small-model", policy=policy)
+        self.assertEqual(decision.source, "capability_blocked")
+        self.assertIsNone(decision.chosen_model)
+        self.assertFalse(decision.applied)
+        # The reason names the tier model's shortfall (tool
+        # calling); the worker default's own shortfall (context)
+        # is what made the fallback fail, not the reason text.
+        self.assertIn("tool_calling", decision.reason)
+
+    def test_active_unknown_model_blocks(self):
+        policy = _policy()
+        policy.tiers["plan"] = "unlisted-model"
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "worker-model", policy=policy)
+        # worker default is capable, so it is used instead of blocking
+        self.assertEqual(decision.source, "worker_default")
+        self.assertEqual(decision.chosen_model, "worker-model")
+        self.assertIn("unknown", decision.reason)
+
+    def test_active_unknown_model_and_worker_unknown_blocks(self):
+        policy = _policy()
+        policy.tiers["plan"] = "unlisted-model"
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "also-unlisted", policy=policy)
+        self.assertEqual(decision.source, "capability_blocked")
+        self.assertIsNone(decision.chosen_model)
+
+    def test_shadow_incapable_never_blocks(self):
+        policy = parse_routing_policy({
+            "version": "cap-v8", "mode": "shadow",
+            "tiers": {"plan": {"model": "no-tools-model",
+                               "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "no-tools-model": {"tool_calling": False},
+            },
+        })
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "worker-model", policy=policy)
+        self.assertEqual(decision.source, "shadow")
+        self.assertEqual(decision.chosen_model, "worker-model")
+        self.assertEqual(decision.suggested_model, "no-tools-model")
+        self.assertIn("lacks", decision.reason)
+
+    def test_explicit_models_bypass_capability_check(self):
+        policy = _policy()
+        policy.tiers["plan"] = "no-tools-model"
+        decision = decide_routing(
+            {"id": "t1", "plan": True, "model": "no-tools-model"},
+            "small-model", policy=policy)
+        self.assertEqual(decision.source, "explicit")
+        self.assertEqual(decision.chosen_model, "no-tools-model")
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "no-tools-model",
+            policy=policy, worker_model_explicit=True)
+        self.assertEqual(decision.source, "worker_explicit")
+        self.assertEqual(decision.chosen_model, "no-tools-model")
+
+    def test_no_requirements_behaves_as_before(self):
+        policy = parse_routing_policy({
+            "mode": "active", "tiers": {"plan": "plan-model"}})
+        decision = decide_routing(
+            {"id": "t1", "plan": True}, "worker-model", policy=policy)
+        self.assertEqual(decision.source, "policy")
+        self.assertEqual(decision.chosen_model, "plan-model")
+
+
+class CapabilityDryRunTestCase(unittest.TestCase):
+    def test_active_incapable_blocks_dry_run(self):
+        report = dry_run_policy_document({
+            "mode": "active",
+            "tiers": {"plan": {"model": "no-tools-model",
+                               "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "no-tools-model": {"tool_calling": False},
+            },
+        })
+        self.assertTrue(report["blocked"])
+        self.assertEqual(
+            report["tiers"]["plan"]["capabilities"]["status"], "fail")
+
+    def test_shadow_incapable_warns_only(self):
+        report = dry_run_policy_document({
+            "mode": "shadow",
+            "tiers": {"plan": {"model": "no-tools-model",
+                               "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "no-tools-model": {"tool_calling": False},
+            },
+        })
+        self.assertFalse(report["blocked"])
+        self.assertTrue(any("lacks" in w for w in report["warnings"]))
+
+    def test_active_capable_does_not_block(self):
+        report = dry_run_policy_document({
+            "mode": "active",
+            "tiers": {"plan": {"model": "plan-model",
+                               "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "plan-model": {"tool_calling": True},
+            },
+        })
+        self.assertFalse(report["blocked"])
+        self.assertEqual(
+            report["tiers"]["plan"]["capabilities"]["status"], "ok")
+
+
+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='{"task": "%s", "status": "completed", "text": "done"}' % name,
+        stderr="")
+
+
+def _insert_task(db_path, task_id):
+    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)"
+            " VALUES(?, 'p1', 'do work', 'queued', NULL, '[]', 0,"
+            " 'tester', ?, ?, NULL)", (task_id, now, now))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class CapabilityWorkerGateTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.calls = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self, policy_doc, model):
+        def recording(cmd):
+            self.calls.append(list(cmd))
+            return _completed_runner(cmd)
+
+        # model_explicit=False: the worker model is a plain
+        # default, not a CLI pin -- a CLI pin would bypass the
+        # policy (and the capability check) by design, so the
+        # gate test must not use one.
+        return Worker(self.db_path, "/fake/host.py", runner=recording,
+                      env={"XCM_API_KEY": "test-placeholder"},
+                      model=model, model_explicit=False,
+                      routing_policy=policy_doc)
+
+    def test_blocked_decision_fails_task_without_running(self):
+        _insert_task(self.db_path, "A")
+        worker = self._worker({
+            "mode": "active",
+            "tiers": {"default": {"model": "no-tools-model",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "no-tools-model": {"tool_calling": False},
+                "small-model": {"tool_calling": True,
+                                "context_tokens": 32000},
+            },
+        }, model="small-model")
+        # small-model is capable on tool calling but the task is
+        # default-tier: the tier model fails tool calling, and the
+        # worker default must be checked against the same
+        # requires -- here it passes, so this is NOT a block.
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+
+    def test_blocked_decision_fails_task_when_worker_also_incapable(self):
+        _insert_task(self.db_path, "B")
+        worker = self._worker({
+            "mode": "active",
+            "tiers": {"default": {"model": "no-tools-model",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": {
+                "no-tools-model": {"tool_calling": False},
+                "no-tools-worker": {"tool_calling": False},
+            },
+        }, model="no-tools-worker")
+        result = worker.run_one()
+        self.assertTrue(result["dispatched"])
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("tool_calling", result["error"])
+        self.assertEqual(self.calls, [])  # gate fires before runner
+        conn = sqlite3.connect(self.db_path)
+        try:
+            status = conn.execute(
+                "SELECT status FROM tasks WHERE id = 'B'").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(status, "failed")

cloud/worker.py

@@ -6036,6 +6036,36 @@ class Worker:
             decision = decide_routing(
                 task, self.model, policy=self.routing_policy,
                 worker_model_explicit=self.model_explicit)
+            # 能力门（v8）：策略档模型与 worker 默认都满足不了
+            # 档需求时，绝不静默派给一个不具备或无法验证的模型，
+            # 按 v3 必需工具门同款 fail-fast：任务直接 failed、
+            # 审计点名原因、清占位、不调 runner。显式任务模型
+            # 与 worker CLI 模型在 decide_routing 内已优先，
+            # 永远不会走到这里被能力门拦。
+            if decision.source == "capability_blocked":
+                _cap_error = decision.reason
+                self._set_status(task_id, "failed")
+                self._write_execution_result(
+                    task_id, error_text=_cap_error,
+                    actual_model=self.model,
+                    started_at=started_at, finished_at=time.time(),
+                    only_if_status="failed")
+                self._audit(task["project_id"],
+                            "worker_capability_gate_rejected",
+                            "%s: %s" % (task_id, _cap_error))
+                log_event(self.logger,
+                          "worker_capability_gate_rejected",
+                          task_id=task_id, project_id=task["project_id"],
+                          status="failed")
+                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": _cap_error, "events": [],
+                        "cost_record_id": None}
             # 只在策略真正介入时记事件与审计：worker_default /
             # worker_explicit 就是旧行为，记一条 model_routing 会
             # 打破现有调用方对 events 的假设（例如 fallback 单测