diff --git a/cloud/models.py b/cloud/models.py
index 6820b02..095048a 100644
--- a/cloud/models.py
+++ b/cloud/models.py
@@ -289,6 +289,115 @@ def split_model_capability_missing(capabilities, model, requires):
     return sorted(known), sorted(unknown)
 
 
+class CapabilityVerdict:
+    """Structured result of one capability check (audit batch 3, I-3).
+
+    One call replaces the previous check+split pair at the five
+    capability sites (decide_routing, worker gateway check, worker
+    persisted-model recheck, dry-run tier/gateway checks, server
+    shadow endpoint). The fields are exactly what those sites used
+    before, so no site changes behaviour by adopting it:
+
+    - ``ok`` / ``missing`` / ``unknown`` are the total-check fields
+      from check_model_capabilities (decide/dry-run/shadow wording);
+    - ``known_missing`` / ``unknown_missing`` are the split fields
+      from split_model_capability_missing (gateway explicit split,
+      persisted recheck);
+    - ``status`` is "ok", or "unknown" when not ok and any part is
+      unverifiable, else "fail" (the dry-run/shadow status rule);
+    - ``verb`` is "cannot verify" when unknown else "lacks" (the
+      dry-run/gateway note wording); it is only meaningful when
+      not ok.
+
+    Edge preserved, not fixed: for a model absent from the
+    directory, ``missing`` lists every requires key (the total
+    check's rule) while ``unknown_missing`` lists only recognised
+    requirement keys (the split's rule). Callers that previously
+    used one function keep getting that function's values; the
+    blocking rules themselves stay at the call sites (gateway has
+    no worker-default fallback, recheck blocks only on known,
+    dry-run/shadow never block), they are not folded into this
+    verdict.
+    """
+
+    def __init__(self, ok, missing, unknown, known_missing,
+                 unknown_missing):
+        self.ok = bool(ok)
+        self.missing = list(missing)
+        self.unknown = bool(unknown)
+        self.known_missing = list(known_missing)
+        self.unknown_missing = list(unknown_missing)
+
+    @property
+    def status(self):
+        if self.ok:
+            return "ok"
+        return "unknown" if self.unknown else "fail"
+
+    @property
+    def verb(self):
+        return "cannot verify" if self.unknown else "lacks"
+
+
+def evaluate_capability(capabilities, model, requires):
+    """Check one model against requires, returning a CapabilityVerdict.
+
+    Pure: it only calls check_model_capabilities and
+    split_model_capability_missing, so its fields are definitionally
+    identical to what the five call sites computed before with two
+    calls. ``requires`` is expected to be already cleaned by
+    parse_tier_requires (all five sites pass tier_requirements);
+    uncleaned keys behave exactly as the two underlying functions
+    behave (see CapabilityVerdict).
+    """
+    ok, missing, unknown = check_model_capabilities(
+        capabilities, model, requires)
+    known, unknown_missing = split_model_capability_missing(
+        capabilities, model, requires)
+    return CapabilityVerdict(ok, missing, unknown, known,
+                             unknown_missing)
+
+
+def merge_policy_documents(base, override):
+    """Merge a file policy document with a JSON override document.
+
+    This is the merge load_routing_policy and dry_run_from_env
+    always shared line-for-line, now in one place (audit batch 3,
+    I-4). ``base`` is the file document (or None when unread/absent),
+    ``override`` is the JSON document (or None when absent/unparsed).
+    Rules, exactly as before:
+
+    - override not a dict: base unchanged;
+    - base not a dict: override;
+    - otherwise: a shallow copy of base updated with override, so
+      JSON replaces whole keys — including the whole "tiers" map.
+      Per-tier merging does NOT happen: the old tiers-update block
+      after update() copied override's tiers back onto themselves
+      (merged["tiers"] already WAS override's tiers), i.e. it was
+      a no-op, and it is kept verbatim below so this function stays
+      a line-for-line move, not a semantic "fix". File tiers absent
+      from the JSON document are therefore dropped, not preserved;
+      changing that would be a behaviour change for another batch.
+
+    Source-level failure records (dropped_tiers/partial_failures
+    from batch 2) are NOT a product of this merge: they are filled
+    in by parse/load afterwards, and this function neither reads
+    nor writes them.
+    """
+    if not isinstance(override, dict):
+        return base
+    if not isinstance(base, dict):
+        return override
+    merged = dict(base)
+    merged.update(override)
+    if isinstance(merged.get("tiers"), dict) and \
+            isinstance(override.get("tiers"), dict):
+        tiers = dict(merged["tiers"])
+        tiers.update(override["tiers"])
+        merged["tiers"] = tiers
+    return merged
+
+
 def prompt_len_bucket(prompt) -> str:
     """Bucket a prompt by length for logging only, never for routing.
 
@@ -578,16 +687,7 @@ def load_routing_policy(env=None):
             failures.append("CLOUD_ROUTING_POLICY_JSON: %s" % exc)
             override = None
         if isinstance(override, dict):
-            if not isinstance(merged, dict):
-                merged = override
-            else:
-                merged = dict(merged)
-                merged.update(override)
-                if isinstance(merged.get("tiers"), dict) and \
-                        isinstance(override.get("tiers"), dict):
-                    tiers = dict(merged["tiers"])
-                    tiers.update(override["tiers"])
-                    merged["tiers"] = tiers
+            merged = merge_policy_documents(merged, override)
     policy = parse_routing_policy(merged)
     if policy is None:
         if configured:
@@ -741,13 +841,13 @@ def decide_routing(task, worker_model, policy=None,
             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(
+    cap = evaluate_capability(
         policy.model_capabilities, tier_model, requires)
     cap_note = ""
-    if requires and not cap_ok:
+    if requires and not cap.ok:
         cap_note = " (capability %s: %s)" % (
-            "unknown" if cap_unknown else "lacks",
-            ", ".join(cap_missing))
+            "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,
@@ -756,7 +856,7 @@ def decide_routing(task, worker_model, policy=None,
             reason="%s tier suggests %s (shadow, not applied)%s"
             % (tier, tier_model, cap_note), signals=signals)
     # effective_mode == "active"
-    if cap_ok:
+    if cap.ok:
         return RoutingDecision(
             task_id=task.get("id"), tier=tier, chosen_model=tier_model,
             suggested_model=None, source="policy",
@@ -767,9 +867,9 @@ def decide_routing(task, worker_model, policy=None,
     # 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(
+    worker_cap = evaluate_capability(
         policy.model_capabilities, worker_model, requires)
-    if worker_ok:
+    if worker_cap.ok:
         return RoutingDecision(
             task_id=task.get("id"), tier=tier, chosen_model=worker_model,
             suggested_model=None, source="worker_default",
diff --git a/cloud/routing_dryrun.py b/cloud/routing_dryrun.py
index 14a247b..cc792b8 100644
--- a/cloud/routing_dryrun.py
+++ b/cloud/routing_dryrun.py
@@ -72,14 +72,16 @@ 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
+    from cloud.models import evaluate_capability
+    from cloud.models import merge_policy_documents
 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
+    from cloud.models import evaluate_capability
+    from cloud.models import merge_policy_documents
 
 
 def _tier_dropped_reason(raw_entry):
@@ -210,12 +212,11 @@ def dry_run_policy_document(data, env_mode_override=None,
             cap = {"required": None, "missing": [], "unknown": False,
                    "status": "not_applicable"}
         else:
-            cap_ok, cap_missing, cap_unknown = check_model_capabilities(
+            cap_verdict = evaluate_capability(
                 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")}
+            cap = {"required": requires, "missing": cap_verdict.missing,
+                   "unknown": cap_verdict.unknown,
+                   "status": cap_verdict.status}
         tiers_report[tier] = {
             "model": policy.tiers[tier], "mode": mode,
             "mode_source": mode_source, "status": status,
@@ -268,17 +269,18 @@ def dry_run_policy_document(data, env_mode_override=None,
                      " -- cannot verify it against the capability"
                      " directory" % (tier, sorted(_req)))
         else:
-            _ok, _missing, _unknown = check_model_capabilities(
+            _gw_verdict = evaluate_capability(
                 policy.model_capabilities, gateway_model, _req)
             _gw["tiers"][tier] = {
-                "status": "ok" if _ok else (
-                    "unknown" if _unknown else "fail"),
-                "missing": _missing, "unknown": _unknown,
+                "status": _gw_verdict.status,
+                "missing": _gw_verdict.missing,
+                "unknown": _gw_verdict.unknown,
                 "tier_mode": _tstatus}
             _note = ("tier %r gateway model %r %s capability:"
                      " %s" % (tier, gateway_model,
-                              "cannot verify" if _unknown else "lacks",
-                              ", ".join(_missing)))
+                              "cannot verify"
+                              if _gw_verdict.unknown else "lacks",
+                              ", ".join(_gw_verdict.missing)))
         if _gw["tiers"][tier]["status"] in ("fail", "unknown"):
             if _tstatus == "active":
                 _gw_blocked = True
@@ -326,7 +328,9 @@ def dry_run_policy_document(data, env_mode_override=None,
 
 
 def dry_run_from_env(env=None):
-    """Dry-run from an env mapping, mirroring load_routing_policy.
+    """Dry-run from an env mapping, sharing merge_policy_documents
+    with load_routing_policy (audit batch 3, I-4) — the merge is
+    one function now, not a mirrored copy.
 
     Returns the report dict. File/JSON failures that the loader
     silently ignores are reported as warnings here.
@@ -354,16 +358,7 @@ def dry_run_from_env(env=None):
                 "(%s); the loader would ignore it" % exc)
             override = None
         if isinstance(override, dict):
-            if not isinstance(merged, dict):
-                merged = override
-            else:
-                merged = dict(merged)
-                merged.update(override)
-                if isinstance(merged.get("tiers"), dict) and \
-                        isinstance(override.get("tiers"), dict):
-                    tiers = dict(merged["tiers"])
-                    tiers.update(override["tiers"])
-                    merged["tiers"] = tiers
+            merged = merge_policy_documents(merged, override)
     _gw_model = (env.get("CLOUD_SUB2API_DEFAULT_MODEL") or "").strip()
     report = dry_run_policy_document(
         merged, env_mode_override=env.get("CLOUD_ROUTING_MODE") or None,
diff --git a/cloud/server.py b/cloud/server.py
index 6996b3c..b38e007 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -70,7 +70,7 @@ from cloud.auth import AuthStore
 from cloud.coordinator import parse_parts as _coord_parse_parts
 from cloud.coordinator import validate_parts as _coord_validate_parts
 from cloud.cost import CostStore, unpriced_token_cap
-from cloud.models import check_model_capabilities as _check_caps
+from cloud.models import evaluate_capability as _evaluate_capability
 from cloud.models import load_routing_policy as _load_routing_policy
 from cloud.routing_shadow import evaluate_shadow as _evaluate_shadow
 from cloud.db import connect as _db_connect
@@ -4348,12 +4348,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                             "model": None, "missing": [],
                             "unknown": False}
                 else:
-                    _ok, _missing, _unknown = _check_caps(
+                    _verdict = _evaluate_capability(
                         _policy.model_capabilities, _model, _req)
-                    _cap = {"status": "ok" if _ok else (
-                        "unknown" if _unknown else "fail"),
-                        "model": _model, "missing": _missing,
-                        "unknown": _unknown}
+                    _cap = {"status": _verdict.status,
+                            "model": _model,
+                            "missing": _verdict.missing,
+                            "unknown": _verdict.unknown}
             _bucket["capability"] = _cap
             if _cap["status"] in ("fail", "unknown"):
                 _cap_flagged.append(
diff --git a/cloud/tests/test_audit_batch3.py b/cloud/tests/test_audit_batch3.py
new file mode 100644
index 0000000..1b2bffb
--- /dev/null
+++ b/cloud/tests/test_audit_batch3.py
@@ -0,0 +1,326 @@
+"""Audit batch 3 (I-3, I-4): capability verdict unification and policy
+merge unification. Pure refactor: no behaviour change.
+
+Run: python3 -m unittest discover -s cloud/tests -p "test_audit_batch3.py" -v
+"""
+
+import copy
+import json
+import os
+import random
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from cloud.models import (
+    check_model_capabilities,
+    decide_routing,
+    evaluate_capability,
+    load_routing_policy,
+    merge_policy_documents,
+    parse_routing_policy,
+    split_model_capability_missing,
+)
+from cloud.routing_dryrun import dry_run_from_env, dry_run_policy_document
+
+REQUIRES = {"tool_calling": True, "vision": True,
+            "min_context_tokens": 1000}
+
+CAPS = {
+    "good": {"tool_calling": True, "vision": True,
+             "context_tokens": 200000},
+    "no-tool": {"tool_calling": False, "vision": True,
+                "context_tokens": 200000},
+    "no-vision-field": {"tool_calling": True,
+                        "context_tokens": 200000},
+    "small-ctx": {"tool_calling": True, "vision": True,
+                  "context_tokens": 100},
+    "no-ctx": {"tool_calling": True, "vision": True},
+    "mixed": {"tool_calling": False, "context_tokens": 200000},
+}
+
+
+def _policy(mode, tier_model, requires, caps, worker_model=None):
+    doc = {
+        "version": "batch3",
+        "mode": mode,
+        "tiers": {"default": {"model": tier_model,
+                               "requires": requires}},
+        "model_capabilities": caps,
+    }
+    return parse_routing_policy(doc)
+
+
+class EvaluateCapabilityTestCase(unittest.TestCase):
+    def test_matrix_known_unknown_mixed_absent_empty(self):
+        cases = [
+            # (model, ok, missing, unknown, known, unknown_missing,
+            #  status, verb)
+            ("good", True, [], False, [], [], "ok", None),
+            ("no-tool", False, ["tool_calling"], False,
+             ["tool_calling"], [], "fail", "lacks"),
+            ("no-vision-field", False, ["vision"], True,
+             [], ["vision"], "unknown", "cannot verify"),
+            ("small-ctx", False, ["min_context_tokens"], False,
+             ["min_context_tokens"], [], "fail", "lacks"),
+            ("no-ctx", False, ["min_context_tokens"], True,
+             [], ["min_context_tokens"], "unknown", "cannot verify"),
+            ("mixed", False, ["tool_calling", "vision"], True,
+             ["tool_calling"], ["vision"], "unknown", "cannot verify"),
+            ("absent-model", False,
+             ["min_context_tokens", "tool_calling", "vision"], True,
+             [], ["min_context_tokens", "tool_calling", "vision"],
+             "unknown", "cannot verify"),
+        ]
+        for (model, ok, missing, unknown, known, unknown_missing,
+             status, verb) in cases:
+            with self.subTest(model=model):
+                verdict = evaluate_capability(CAPS, model, REQUIRES)
+                self.assertEqual(verdict.ok, ok)
+                self.assertEqual(verdict.missing, missing)
+                self.assertEqual(verdict.unknown, unknown)
+                self.assertEqual(verdict.known_missing, known)
+                self.assertEqual(verdict.unknown_missing, unknown_missing)
+                self.assertEqual(verdict.status, status)
+                if verb is not None:
+                    self.assertEqual(verdict.verb, verb)
+
+    def test_empty_requires_always_ok_even_for_absent_model(self):
+        verdict = evaluate_capability(CAPS, "absent-model", {})
+        self.assertTrue(verdict.ok)
+        self.assertEqual(verdict.missing, [])
+        self.assertFalse(verdict.unknown)
+        self.assertEqual(verdict.status, "ok")
+
+    def test_parity_with_old_check_and_split(self):
+        # The refactor contract: every verdict field equals what the
+        # two old functions returned for the same inputs, across a
+        # matrix of directories, models and requires — including the
+        # preserved edge where an absent model plus an uncleaned
+        # requires key makes the total check list the invalid key
+        # while the split lists only recognised keys. That edge is
+        # documented here as preserved parity, not as desired
+        # semantics; changing it would be a behaviour change.
+        requires_sets = [
+            REQUIRES,
+            {"tool_calling": True},
+            {"min_context_tokens": 500},
+            {},
+            {"tool_calling": True, "bogus_key": True},
+        ]
+        models = list(CAPS) + ["absent-model", "", None]
+        for requires in requires_sets:
+            for model in models:
+                with self.subTest(model=model, requires=requires):
+                    verdict = evaluate_capability(
+                        CAPS, model, requires)
+                    ok, missing, unknown = check_model_capabilities(
+                        CAPS, model, requires)
+                    known, unknown_missing = \
+                        split_model_capability_missing(
+                            CAPS, model, requires)
+                    self.assertEqual(verdict.ok, ok)
+                    self.assertEqual(verdict.missing, missing)
+                    self.assertEqual(verdict.unknown, unknown)
+                    self.assertEqual(verdict.known_missing, known)
+                    self.assertEqual(
+                        verdict.unknown_missing, unknown_missing)
+
+
+class DecideRoutingWordingTestCase(unittest.TestCase):
+    def test_active_known_missing_falls_back_to_worker_default(self):
+        policy = _policy("active", "no-tool", REQUIRES, CAPS)
+        decision = decide_routing({}, "good", policy=policy)
+        self.assertEqual(decision.source, "worker_default")
+        self.assertTrue(decision.capability_fallback)
+        self.assertEqual(decision.chosen_model, "good")
+        self.assertIn("capability lacks: tool_calling",
+                      decision.reason)
+
+    def test_active_both_fail_blocks_with_exact_reason_shape(self):
+        policy = _policy("active", "no-tool", REQUIRES, CAPS)
+        decision = decide_routing({}, "no-tool", policy=policy)
+        self.assertEqual(decision.source, "capability_blocked")
+        self.assertIsNone(decision.chosen_model)
+        self.assertIn("refusing to dispatch an incapable or"
+                      " unverifiable model", decision.reason)
+
+    def test_shadow_unknown_and_known_notes(self):
+        policy = _policy("shadow", "no-vision-field", REQUIRES, CAPS)
+        decision = decide_routing({}, "good", policy=policy)
+        self.assertEqual(decision.source, "shadow")
+        self.assertFalse(decision.applied)
+        self.assertIn("capability unknown: vision", decision.reason)
+        policy = _policy("shadow", "no-tool", REQUIRES, CAPS)
+        decision = decide_routing({}, "good", policy=policy)
+        self.assertIn("capability lacks: tool_calling",
+                      decision.reason)
+
+
+class DryRunWordingTestCase(unittest.TestCase):
+    def _doc(self, mode, tier_model):
+        return {
+            "version": "batch3",
+            "mode": mode,
+            "tiers": {"default": {"model": tier_model,
+                                  "requires": REQUIRES}},
+            "model_capabilities": CAPS,
+        }
+
+    def test_tier_check_notes_and_block_flags(self):
+        report = dry_run_policy_document(
+            self._doc("active", "no-tool"))
+        tier = report["tiers"]["default"]
+        self.assertEqual(tier["capabilities"]["status"], "fail")
+        self.assertTrue(report["blocked"])
+        self.assertTrue(any(
+            "lacks capability: tool_calling" in w
+            and "would block 'default' tasks at dispatch" in w
+            for w in report["warnings"]))
+        report = dry_run_policy_document(
+            self._doc("shadow", "no-vision-field"))
+        tier = report["tiers"]["default"]
+        self.assertEqual(tier["capabilities"]["status"], "unknown")
+        self.assertFalse(report["blocked"])
+        self.assertTrue(any(
+            "cannot verify capability: vision" in w
+            and "shadow mode does not block dispatch" in w
+            for w in report["warnings"]))
+
+    def test_gateway_check_notes_and_block_flags(self):
+        report = dry_run_policy_document(
+            self._doc("active", "good"), gateway_model="no-tool")
+        gw = report["gateway"]["tiers"]["default"]
+        self.assertEqual(gw["status"], "fail")
+        self.assertTrue(report["blocked"])
+        self.assertTrue(any(
+            "gateway model 'no-tool' lacks capability: tool_calling"
+            in w
+            and "would block 'default' tasks at the gateway gate" in w
+            for w in report["warnings"]))
+        report = dry_run_policy_document(
+            self._doc("shadow", "good"), gateway_model=None)
+        gw = report["gateway"]["tiers"]["default"]
+        self.assertEqual(gw["status"], "unknown")
+        self.assertTrue(gw["unknown"])
+        self.assertTrue(any(
+            "gateway model is not set in CLOUD_SUB2API_DEFAULT_MODEL"
+            in w for w in report["warnings"]))
+
+
+class MergePolicyDocumentsTestCase(unittest.TestCase):
+    def test_boundary_cases(self):
+        self.assertIsNone(merge_policy_documents(None, None))
+        self.assertEqual(merge_policy_documents(None, {"a": 1}),
+                         {"a": 1})
+        self.assertEqual(merge_policy_documents({"a": 1}, None),
+                         {"a": 1})
+        self.assertEqual(merge_policy_documents({"a": 1}, ["x"]),
+                         {"a": 1})
+        self.assertEqual(merge_policy_documents(["x"], {"a": 1}),
+                         {"a": 1})
+        # No tiers anywhere: plain shallow update.
+        self.assertEqual(
+            merge_policy_documents({"mode": "shadow", "version": "v1"},
+                                   {"mode": "active"}),
+            {"mode": "active", "version": "v1"})
+
+    def test_json_replaces_whole_tiers_map_not_per_tier(self):
+        # Preserved (not fixed) semantics, documented in
+        # merge_policy_documents: JSON tiers replace the whole map.
+        # File tiers absent from JSON are dropped, not preserved.
+        base = {"mode": "shadow",
+                "tiers": {"plan": "m-file-plan",
+                          "default": "m-file-default"}}
+        override = {"tiers": {"plan": "m-json-plan"}}
+        self.assertEqual(merge_policy_documents(base, override),
+                         {"mode": "shadow",
+                          "tiers": {"plan": "m-json-plan"}})
+
+    def test_inputs_not_mutated(self):
+        base = {"mode": "shadow",
+                "tiers": {"plan": "m-file-plan",
+                          "default": "m-file-default"}}
+        override = {"tiers": {"plan": "m-json-plan"}}
+        base_before, override_before = copy.deepcopy(base), \
+            copy.deepcopy(override)
+        merge_policy_documents(base, override)
+        self.assertEqual(base, base_before)
+        self.assertEqual(override, override_before)
+
+    def test_property_random_pairs_match_shallow_update(self):
+        rng = random.Random(20261102)
+        tier_values = ["m1", "m2", {"model": "m3"},
+                       {"model": "m4", "mode": "shadow"},
+                       {"model": "m5", "mode": "bogus"},
+                       {"model": ""}, "  ", 7, None]
+        top_values = ["shadow", "active", "v9", 3, None,
+                      {"x": 1}, ["a"]]
+
+        def _rand_doc():
+            doc = {}
+            if rng.random() < 0.8:
+                doc["mode"] = rng.choice(["shadow", "active", "off"])
+            if rng.random() < 0.5:
+                doc["version"] = rng.choice(top_values)
+            if rng.random() < 0.8:
+                tiers = {}
+                for tier in rng.sample(
+                        ["plan", "goal", "default", "weird"],
+                        rng.randint(0, 3)):
+                    tiers[tier] = rng.choice(tier_values)
+                if rng.random() < 0.1:
+                    doc["tiers"] = rng.choice(["x", 5, None])
+                else:
+                    doc["tiers"] = tiers
+            return doc
+
+        for _ in range(300):
+            base = _rand_doc()
+            override = _rand_doc()
+            got = merge_policy_documents(base, override)
+            expected = dict(base)
+            expected.update(override)
+            self.assertEqual(got, expected)
+
+    def test_loader_and_dryrun_agree_on_same_env(self):
+        # Loader and dry-run now share merge_policy_documents: for
+        # the same file+JSON env, the parsed policy tiers and the
+        # dry-run report tiers (and fingerprints) must agree.
+        cases = [
+            ({"mode": "shadow",
+              "tiers": {"plan": "m-file-plan",
+                        "default": "m-file-default"}},
+             {"tiers": {"plan": "m-json-plan"}}),
+            ({"mode": "active",
+              "tiers": {"default": {"model": "m1",
+                                    "requires": {"tool_calling": True}}},
+              "model_capabilities": {"m1": {"tool_calling": True}}},
+             {"mode": "shadow"}),
+        ]
+        for base_doc, override_doc in cases:
+            with tempfile.NamedTemporaryFile(
+                    "w", suffix=".json", delete=False) as fh:
+                json.dump(base_doc, fh)
+                path = fh.name
+            try:
+                env = {"CLOUD_ROUTING_POLICY_FILE": path,
+                       "CLOUD_ROUTING_POLICY_JSON":
+                           json.dumps(override_doc)}
+                policy = load_routing_policy(env)
+                report = dry_run_from_env(env)
+                self.assertIsNotNone(policy)
+                self.assertTrue(report["valid"])
+                for tier, model in policy.tiers.items():
+                    self.assertEqual(
+                        report["tiers"][tier]["model"], model)
+                self.assertEqual(report["content_fingerprint"],
+                                 policy.content_fingerprint())
+            finally:
+                os.unlink(path)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/worker.py b/cloud/worker.py
index 5fe3650..7b5d656 100644
--- a/cloud/worker.py
+++ b/cloud/worker.py
@@ -165,10 +165,9 @@ from cloud.db import connect as _db_connect
 from cloud.goal_command import parse_goal_command as _parse_goal_command
 from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, RoutingPolicy, decide_routing, \
-    check_model_capabilities, classify_task, \
+    classify_task, evaluate_capability, \
     is_fallbackable, load_routing_policy, parse_routing_policy, \
-    record_fallback, record_routing, \
-    split_model_capability_missing
+    record_fallback, record_routing
 from cloud.routing_shadow import parse_worker_routing_detail as \
     _parse_routing_detail
 from cloud.auto_memory import AutoMemoryStore
@@ -7032,15 +7031,15 @@ class Worker:
         if mode not in ("active", "shadow"):
             return None
         model = task.get("model")
-        ok, missing, unknown = check_model_capabilities(
+        verdict = evaluate_capability(
             self.routing_policy.model_capabilities, model, requires)
-        if ok:
+        if verdict.ok:
             return None
-        known, _unk = split_model_capability_missing(
-            self.routing_policy.model_capabilities, model, requires)
+        known = verdict.known_missing
         detail = ("%s model=%s tier=%s mode=%s missing=%s unknown=%s"
                   % (task["id"], model, tier, mode,
-                     ",".join(missing) or "-", unknown))
+                     ",".join(verdict.missing) or "-",
+                     verdict.unknown))
         if mode == "shadow":
             self._audit(task["project_id"],
                         "worker_capability_recheck_shadow", detail)
@@ -7444,17 +7443,17 @@ class Worker:
             if _gw_requires:
                 _gw_mode = self.routing_policy.tier_mode(_gw_tier)
                 if _gw_mode in ("active", "shadow"):
-                    _gw_ok, _gw_missing, _gw_unknown = \
-                        check_model_capabilities(
-                            self.routing_policy.model_capabilities,
-                            task.get("model"), _gw_requires)
-                    if not _gw_ok:
+                    _gw_verdict = evaluate_capability(
+                        self.routing_policy.model_capabilities,
+                        task.get("model"), _gw_requires)
+                    if not _gw_verdict.ok:
                         _gw_note = ("gateway model %s %s capability:"
                                     " %s" % (
                                         task.get("model"),
                                         "cannot verify"
-                                        if _gw_unknown else "lacks",
-                                        ", ".join(_gw_missing)))
+                                        if _gw_verdict.unknown
+                                        else "lacks",
+                                        ", ".join(_gw_verdict.missing)))
                         # Audit fix 4（第三遍审计分档建议）：
                         # 显式模型按 known/unknown 分档。known
                         # 是目录明确写了不满足（布尔 False、
@@ -7463,10 +7462,7 @@ class Worker:
                         # 无该模型、或条目缺字段，目录陈旧
                         # 不应惩罚用户的正确选择，只审计不拦。
                         # shadow 档两种情况都只审计。
-                        _gw_known, _gw_unk = \
-                            split_model_capability_missing(
-                                self.routing_policy.model_capabilities,
-                                task.get("model"), _gw_requires)
+                        _gw_known = _gw_verdict.known_missing
                         _gw_block = False
                         _gw_error = None
                         if _explicit_model:
@@ -7483,8 +7479,8 @@ class Worker:
                                 " missing=%s unknown=%s blocked=%s"
                                 % (task_id, task.get("model"),
                                    _gw_tier, _gw_mode,
-                                   ",".join(_gw_missing) or "-",
-                                   _gw_unknown,
+                                   ",".join(_gw_verdict.missing) or "-",
+                                   _gw_verdict.unknown,
                                    _gw_blocked))
                             if _gw_blocked:
                                 _gw_block = True