diff --git a/cloud/models.py b/cloud/models.py
index 1186ad1..20f4b53 100644
--- a/cloud/models.py
+++ b/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:
diff --git a/cloud/routing_dryrun.py b/cloud/routing_dryrun.py
index 893020e..901c826 100644
--- a/cloud/routing_dryrun.py
+++ b/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,
diff --git a/cloud/tests/test_routing_capability.py b/cloud/tests/test_routing_capability.py
new file mode 100644
index 0000000..d111836
--- /dev/null
+++ b/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")
diff --git a/cloud/worker.py b/cloud/worker.py
index c0cc65f..e131869 100644
--- a/cloud/worker.py
+++ b/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 单测