diff --git a/modelopt/torch/models/_onfly_decompress.py b/modelopt/torch/models/_onfly_decompress.py
new file mode 100644
index 000000000000..24da3d1413f8
--- /dev/null
+++ b/modelopt/torch/models/_onfly_decompress.py
@@ -0,0 +1,458 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Decompress a compressed-tensors MoE checkpoint's experts as ``from_pretrained`` loads them.
+
+Each rank decompresses only its own experts, with the ``compressed_tensors`` library, so expert-parallel
+PTQ runs on the original checkpoint. Needs transformers 5.8, whose ``Concatenate`` reads only its keys.
+"""
+
+import functools
+import os
+
+import torch
+from safetensors import safe_open
+
+from modelopt.torch.utils import print_rank_0
+from modelopt.torch.utils.moe import config_levels, require_transformers
+
+
+def _find_quantization_config(hf_config):
+    """Locate the compressed-tensors ``quantization_config`` (top-level, else on the nested
+    language-model config) and return it as a plain dict, or ``None`` for a config without one."""
+    qc = next(
+        (
+            found
+            for c in config_levels(hf_config)
+            if (found := getattr(c, "quantization_config", None)) is not None
+        ),
+        None,
+    )
+    if qc is None:
+        return None
+    if hasattr(qc, "to_dict"):
+        qc = qc.to_dict()
+    elif not isinstance(qc, dict):
+        qc = dict(qc)
+    return qc
+
+
+def _read_compression(hf_config):
+    """``(compressor, scheme, param_names, format)`` for the checkpoint's compressed-tensors config.
+
+    ``param_names`` are the keys one weight is stored under. The one config group must target the
+    experts; several groups are refused. ``None`` for a config with no ``quantization_config``."""
+    # With no config, compressed_tensors need not even be installed.
+    quantization_config = _find_quantization_config(hf_config)
+    if quantization_config is None:
+        return None
+    # Optional: no pip extra of ModelOpt installs compressed-tensors, as only a compressed
+    # checkpoint needs it.
+    try:
+        from compressed_tensors.compressors.base import BaseCompressor
+        from compressed_tensors.quantization import QuantizationConfig
+    except ImportError as e:
+        raise ImportError(
+            "On-the-fly decompression of a compressed-tensors checkpoint needs the "
+            "compressed-tensors package. Install it with: pip install compressed-tensors"
+        ) from e
+    # transformers is optional: this module imports without it.
+    from transformers.utils.versions import require_version
+
+    # From 0.16, a compressor's parameter names and decompress are classmethods taking the scheme.
+    require_version("compressed-tensors>=0.16", "pip install -U 'compressed-tensors>=0.16'")
+
+    qconfig = QuantizationConfig.model_validate(quantization_config)
+    groups = qconfig.config_groups
+    if not groups:
+        raise ValueError(
+            "install_onfly_decompress: the checkpoint's compressed-tensors quantization_config has "
+            "no config groups: nothing to decompress."
+        )
+    if len(groups) > 1:
+        # Only the expert converters are rewritten: which group holds the experts would be a guess.
+        raise NotImplementedError(
+            f"install_onfly_decompress: the checkpoint has {len(groups)} compressed-tensors config "
+            f"groups ({', '.join(groups)}); on-the-fly decompression handles a single group, "
+            "compressing only the experts. Decompress the checkpoint first."
+        )
+    [scheme] = groups.values()
+    # Regex targets alone must name the experts. A class target such as Linear does not say which
+    # modules it compressed: check_compressed_keys checks the names the checkpoint stores instead.
+    if not any(not t.startswith("re:") or "experts" in t for t in scheme.targets):
+        raise NotImplementedError(
+            f"install_onfly_decompress: the checkpoint's compressed-tensors config group targets "
+            f"{scheme.targets}, not the experts; on-the-fly decompression handles a checkpoint "
+            "whose experts are what it compresses. Decompress the checkpoint first."
+        )
+    # A group's own format wins: the config's top level then says "mixed-precision".
+    fmt = getattr(scheme, "format", None) or qconfig.format
+    fmt = fmt.value if hasattr(fmt, "value") else str(fmt)
+    if fmt not in BaseCompressor.registered_names():
+        raise NotImplementedError(
+            f"install_onfly_decompress: compressed-tensors registers no compressor for the format "
+            f"{fmt!r}, so on-the-fly decompression cannot read it. Decompress the checkpoint first."
+        )
+    compressor = BaseCompressor.get_value_from_registry(fmt)
+    param_names = tuple(compressor.compression_param_names(scheme))
+    return compressor, scheme, param_names, fmt
+
+
+def _make_op_class(compressor, scheme, param_names):
+    """Build the per-expert decompress op, bound to a compressor, its scheme and the stored keys.
+
+    It subclasses ``MergeModulelist`` only so the loader shards experts at load, handing it ``None``
+    for each expert another rank owns; the merge later in the chain still does the stacking.
+    """
+    # transformers is optional, and core_model_loading is new in transformers 5.
+    from transformers.core_model_loading import MergeModulelist
+
+    anchor = "." + param_names[0]
+    others = [(pn, "." + pn) for pn in param_names[1:]]
+
+    class DecompressExperts(MergeModulelist):
+        """Decompress each owned expert's stored params to bf16, under the first stored key."""
+
+        def __init__(self):
+            super().__init__(dim=0)  # dim unused: convert is overridden
+
+        @torch.no_grad()
+        def convert(self, input_dict, *args, **kwargs):
+            out = {}
+            for pat in list(input_dict.keys()):
+                marker = "$" if pat.endswith("$") else ""
+                base = pat[:-1] if marker else pat
+                if not base.endswith(anchor):
+                    continue
+                stem = base[: -len(anchor)]
+                lists = {param_names[0]: input_dict.pop(pat)}
+                # A checkpoint may store a subset of the scheme's params (Kimi-K3's mxfp4 stores no
+                # weight_global_scale): only those present go to the compressor.
+                for pn, suf in others:
+                    sib = input_dict.pop(stem + suf + marker, None)
+                    if sib is not None:
+                        lists[pn] = sib
+                anchor_list = lists[param_names[0]]
+                present = [pn for pn in param_names if pn in lists]
+                if any(len(lists[pn]) != len(anchor_list) for pn in present):
+                    counts = ", ".join(f"{len(lists[pn])} {pn}" for pn in present)
+                    raise ValueError(
+                        f"DecompressExperts: {stem} stores {counts}, not one per expert"
+                    )
+                # Experts other ranks own arrive as None: only this rank's are decompressed.
+                out[pat] = [
+                    compressor.decompress({pn: lists[pn][i] for pn in present}, scheme)[
+                        "weight"
+                    ].to(torch.bfloat16)
+                    for i in range(len(anchor_list))
+                    if anchor_list[i] is not None
+                ]
+            if input_dict:
+                # A stored param without its weight: the merge would fuse it into the wrong slot.
+                raise ValueError(
+                    f"DecompressExperts: unconsumed checkpoint keys {sorted(input_dict)}"
+                )
+            return out
+
+    return DecompressExperts
+
+
+@functools.cache
+def _decompressing_converter_class():
+    """The ``WeightConverter`` subclass a rewritten converter is, reversing as the one it rewrote."""
+    # transformers is optional, and core_model_loading is new in transformers 5.
+    from transformers.core_model_loading import WeightConverter
+
+    class DecompressingConverter(WeightConverter):
+        """Reverses as the converter it rewrote: a save names the experts back as ``.weight`` keys."""
+
+        __slots__ = ("_reversible",)
+
+        def reverse_transform(self):
+            return self._reversible.reverse_transform()
+
+    return DecompressingConverter
+
+
+def _compressedify(conv, compressor, scheme, param_names):
+    """Rewrite a per-expert converter to read the stored keys, with ``DecompressExperts`` first.
+
+    Each ``.weight`` source becomes its stored keys, grouped by key with the anchor first, which
+    keeps gate before up for the merge. The rewritten converter reverses as ``conv`` does.
+    """
+    DecompressExperts = _make_op_class(compressor, scheme, param_names)  # noqa: N806 (a class, keeping its own name)
+    by_pname = {pn: [] for pn in param_names}
+    for p in conv.source_patterns:
+        base = p.removesuffix("$")
+        if not base.endswith(".weight") or ".experts." not in base:
+            # Only the experts' weights are stored compressed: rewritten, a dense weight would match
+            # no stored key and load newly initialized. A bias is expected in a converter of its own
+            # (check_compressed_keys lets one through); grouped with the weight, the merge takes it.
+            raise NotImplementedError(
+                f"install_onfly_decompress: the per-expert converter for {conv.target_patterns} "
+                f"reads {p!r} alongside the experts' weights, which is no compressed tensor; "
+                "on-the-fly decompression rewrites a converter that reads the experts' weights only. "
+                "Decompress the checkpoint first."
+            )
+        stem = base[: -len(".weight")]
+        for pn in param_names:
+            by_pname[pn].append(stem + "." + pn + "$")
+    new_sources = [s for pn in param_names for s in by_pname[pn]]
+    tgt = getattr(conv, "_original_target_patterns", conv.target_patterns)
+    new = _decompressing_converter_class()(
+        source_patterns=new_sources,
+        target_patterns=tgt,
+        operations=[DecompressExperts(), *conv.operations],
+    )
+    new._reversible = conv
+    # What else transformers sets on a converter it maps, whichever of them this release has.
+    for attr in ("scope_prefix", "base_model_prefix", "force_cpu"):
+        if hasattr(conv, attr):
+            setattr(new, attr, getattr(conv, attr))
+    return new
+
+
+# Whether the load under way is one whose converters the hooks rewrote (from_pretrained asks for the
+# mapping before it reads a tensor). Module state: the loader materializes in worker threads.
+_LOAD = {"decompressing": False}
+
+
+def _install_keep_int_native():
+    """Patch ``spawn_materialize`` so a decompressing load never casts an integer source to a float.
+
+    Without a quantizer the loader casts each source to its parameter's dtype in the copy, or from
+    transformers 5.14 in ``DtensorShardOperation.shard_tensor``: both start here. Other loads cast.
+    """
+    # transformers is optional, and core_model_loading is new in transformers 5.
+    import transformers.core_model_loading as cml
+
+    if getattr(cml.spawn_materialize, "_onfly_patched", False):
+        return
+    orig = cml.spawn_materialize
+
+    # Any other argument passes through, positionally or by keyword: a release may add some.
+    def spawn_materialize(thread_pool, tensor, device=None, dtype=None, *args, **kwargs):
+        if _LOAD["decompressing"] and dtype is not None and dtype.is_floating_point:
+            # A safetensors slice is judged by its stored dtype, without reading it.
+            stored = tensor.get_dtype() if hasattr(tensor, "get_dtype") else None
+            if not (stored.startswith(("F", "BF")) if stored else tensor.is_floating_point()):
+                dtype = None  # keep packed/shape/zero-point/g_idx integers native
+        return orig(thread_pool, tensor, device, dtype, *args, **kwargs)
+
+    setattr(spawn_materialize, "_onfly_patched", True)
+    cml.spawn_materialize = spawn_materialize
+
+
+# Marks a config whose model's converters the installed hooks rewrite (holds the names each
+# compressed weight is stored under).
+_MARK = "_modelopt_onfly_decompress"
+
+
+def _mark(hf_config, stored_as):
+    """Mark ``hf_config``'s load for the installed hooks; ``from_pretrained``'s copy of it keeps the mark."""
+    setattr(hf_config, _MARK, stored_as)
+
+
+def is_marked(hf_config) -> bool:
+    """Whether :func:`install_onfly_decompress` marked ``hf_config``, or its nested language-model
+    config, for its hooks."""
+    return any(getattr(c, _MARK, None) is not None for c in config_levels(hf_config))
+
+
+def stored_names(hf_config) -> tuple[str, ...]:
+    """The names the installed compressed-tensors stores each weight under for ``hf_config``'s scheme."""
+    for c in config_levels(hf_config):
+        if (stored_as := getattr(c, _MARK, None)) is not None:
+            return stored_as
+    raise ValueError("The config is not marked for on-the-fly decompression: install it first.")
+
+
+def install_onfly_decompress(hf_config):
+    """Make ``from_pretrained`` decompress the experts of the checkpoint ``hf_config`` describes.
+
+    Call before ``from_pretrained``. The hooks are process-wide but rewrite only the converters of a
+    model built from a config they marked: one compressed checkpoint per process, installing again
+    for it marks that config too, and one compressed differently is refused.
+    """
+    if isinstance(hf_config, dict):
+        # The load finds the hooks by a mark on the config object, which a dict cannot carry.
+        raise TypeError("install_onfly_decompress: pass the model's config object, not a dict")
+    # The loader sets each parameter from what the conversion hands it, uncast. Every level counts:
+    # a backbone config derived from the text config keeps its dtype.
+    for level in config_levels(hf_config):
+        dtype = getattr(level, "dtype", None)
+        if dtype is not None and str(dtype).removeprefix("torch.") != "bfloat16":
+            raise NotImplementedError(
+                f"install_onfly_decompress: the experts decompress to bfloat16, but this config "
+                f"loads in {dtype}. Load the checkpoint in bfloat16."
+            )
+    # Before compressed-tensors is needed: a release below the floor is what to fix first.
+    require_transformers("5.8", "on-the-fly compressed-tensors decompression")
+    compression = _read_compression(hf_config)
+    # Imported once the release is known to have them: transformers 4 has no core_model_loading.
+    import transformers.modeling_utils as mu
+    from transformers.core_model_loading import WeightConverter
+
+    installed = getattr(mu.get_model_conversion_mapping, "_onfly_compression", None)
+    if installed is not None:
+        if compression is None:
+            # A marked config its prepare dropped quantization_config from (Kimi-K3's) may be prepared
+            # again; an unmarked one would load its experts newly initialized.
+            if not is_marked(hf_config):
+                raise ValueError(
+                    "install_onfly_decompress: config has no compressed-tensors "
+                    "quantization_config and is not marked: nothing would decompress it"
+                )
+            return
+        if compression != installed:
+            raise RuntimeError(
+                f"install_onfly_decompress: this process already decompresses a '{installed[3]}' "
+                f"checkpoint (stored params {installed[2]}); one compressed differently ('{compression[3]}', "
+                f"stored params {compression[2]}) needs a process of its own"
+            )
+        # The same checkpoint through a config of its own, which its load must find marked too.
+        _mark(hf_config, installed[2])
+        return
+    if compression is None:
+        # Nothing to decompress: refused before anything is patched.
+        raise ValueError(
+            "install_onfly_decompress: config has no compressed-tensors quantization_config"
+        )
+    compressor, scheme, param_names, fmt = compression
+    _install_keep_int_native()
+    orig = mu.get_model_conversion_mapping
+
+    def patched(*args, **kwargs):
+        convs = orig(*args, **kwargs)
+        config = getattr(args[0] if args else kwargs["model"], "config", None)
+        _LOAD["decompressing"] = is_marked(config)
+        if not _LOAD["decompressing"]:
+            # Not the compressed checkpoint: rewritten, an uncompressed MoE's experts would match no
+            # stored key and load newly initialized without an error.
+            return convs
+        out, rewritten = [], 0
+        for c in convs:
+            # only the per-expert MoE converters (have `.weight` sources under `.experts.`, which
+            # leaves a shared expert's `shared_experts.` out, as check_compressed_keys does)
+            if isinstance(c, WeightConverter) and any(
+                ".experts." in p and p.endswith((".weight", ".weight$")) for p in c.source_patterns
+            ):
+                out.append(_compressedify(c, compressor, scheme, param_names))
+                rewritten += 1
+            else:
+                out.append(c)
+        if not rewritten:
+            # The stock converters would read .weight keys the checkpoint does not store.
+            raise RuntimeError(
+                "install_onfly_decompress: the model's conversion mapping has no per-expert "
+                "converter to rewrite, so its compressed experts would load newly initialized."
+            )
+        # from_pretrained loads from a copy of the marked config; the mark must not reach the config
+        # a later save writes, at any level. Cleared once the experts' converters are rewritten.
+        for c in config_levels(config):
+            if hasattr(c, _MARK):
+                delattr(c, _MARK)
+        return out
+
+    setattr(patched, "_onfly_compression", compression)
+    mu.get_model_conversion_mapping = patched
+    _mark(hf_config, param_names)
+    print_rank_0(
+        f"[onfly_decompress] installed: compressed_tensors '{fmt}' fused-MoE experts decompress "
+        f"(stored params={param_names}) -> bf16 during from_pretrained"
+    )
+
+
+def check_compressed_keys(hf_config, ckpt_dir) -> dict[str, str]:
+    """Refuse a checkpoint the decompression would load wrong, by the tensors ``ckpt_dir`` stores.
+
+    Only the experts decompress, and the load drops the config any other compressed weight would need:
+    the experts alone must be compressed, every one of them. The load may cast their float tensors to
+    bf16 before decompressing, so one bf16 cannot hold exactly (an fp32 scale) is refused too. Reads
+    only the tensors' names and dtypes, and returns the ``tensor name -> file`` map it read.
+    """
+    # A plugin: it imports huggingface_hub, which this module needs nowhere else.
+    from modelopt.torch.utils.plugins.hf_checkpoint_utils import checkpoint_tensor_files
+
+    stored_as = stored_names(hf_config)
+    tensor_files = checkpoint_tensor_files(ckpt_dir)
+    compressed: set[str] = set()
+    expert_weights: set[str] = set()
+    stems_storing: dict[str, set[str]] = {}
+    for name in tensor_files:
+        stem, _, stored = name.rpartition(".")
+        if stored in stored_as:
+            stems_storing.setdefault(stored, set()).add(stem)
+        if stored in stored_as and stored != "weight":
+            compressed.add(stem)
+        elif stored == "weight" and ".experts." in name:
+            expert_weights.add(stem)
+    outside = sorted(stem for stem in compressed if ".experts." not in stem)
+    plain = sorted(expert_weights - compressed)
+    if outside or plain or not compressed:
+        raise NotImplementedError(
+            f"The checkpoint compresses {len(compressed)} weights, {len(outside)} of them outside the "
+            f"experts {outside[:1]}, and stores {len(plain)} expert weights uncompressed {plain[:1]}: "
+            "on-the-fly decompression loads the experts alone, every one compressed, so the rest "
+            "would load newly initialized. Decompress the checkpoint first."
+        )
+    # The names are what the installed compressed-tensors gives the scheme: a weight's name a newer
+    # release added would match no rewritten source, and the experts would decompress without it.
+    # Only weight* names are checked (every compressor's today), so an expert's bias, which loads
+    # through its own converter, is not refused.
+    unknown = [n for n in sorted(tensor_files) if n.rpartition(".")[0] in compressed]
+    unknown = [n for n in unknown if n.rpartition(".")[2] not in stored_as]
+    unknown = [n for n in unknown if n.rpartition(".")[2].startswith("weight")]
+    if unknown:
+        raise NotImplementedError(
+            f"The checkpoint stores {len(unknown)} tensors under its compressed weights that this "
+            f"compressed-tensors release does not name for the scheme {unknown[:1]}: the decompression "
+            f"reads only {list(stored_as)}. Upgrade compressed-tensors, or decompress it first."
+        )
+    # The load pairs each expert's stored tensors by the order it collects them, not by expert id: one
+    # missing a scale would pair a later expert's weight with an earlier one's scale. A name no weight
+    # stores (Kimi-K3's weight_global_scale) is simply not read.
+    unpaired = sorted(
+        (stored, sorted(compressed - stems))
+        for stored, stems in stems_storing.items()
+        if stems & compressed and compressed - stems
+    )
+    if unpaired:
+        stored, missing = unpaired[0]
+        raise NotImplementedError(
+            f"The checkpoint stores {stored} for some compressed weights but not for {len(missing)} "
+            f"others {missing[:1]}: the decompression pairs each expert's stored tensors by position, "
+            "so every compressed weight must store the same names. Decompress the checkpoint first."
+        )
+    by_file: dict[str, list[str]] = {}
+    for name, file in tensor_files.items():
+        stem, _, stored = name.rpartition(".")
+        if stem in compressed and stored in stored_as:
+            by_file.setdefault(file, []).append(name)
+    inexact: set[str] = set()
+    for file, names in by_file.items():
+        with safe_open(os.path.join(ckpt_dir, file), framework="pt") as f:
+            for name in names:
+                dtype = f.get_slice(name).get_dtype()
+                if dtype in ("F16", "F32", "F64"):
+                    inexact.add(f"{name.rpartition('.')[2]} ({dtype})")
+    if inexact:
+        raise NotImplementedError(
+            f"The checkpoint's compressed experts store {', '.join(sorted(inexact))}, which bf16 "
+            "cannot hold exactly: the load may cast them to bf16, the dtype the experts decompress "
+            "to, before decompressing, shifting every weight they scale. Decompress the checkpoint "
+            "first."
+        )
+    return tensor_files
diff --git a/modelopt/torch/utils/moe.py b/modelopt/torch/utils/moe.py
index 15de325fd9f7..72e3579c4c1c 100644
--- a/modelopt/torch/utils/moe.py
+++ b/modelopt/torch/utils/moe.py
@@ -64,6 +64,12 @@ def require_transformers(min_version: str, feature: str) -> None:
 _SUB_CONFIG_ATTRS = ("text_config", "llm_config")
 
 
+def config_levels(hf_config) -> list:
+    """``hf_config`` and its nested language-model configs, those it has, which may carry its settings."""
+    nested = (getattr(hf_config, attr, None) for attr in _SUB_CONFIG_ATTRS)
+    return [c for c in (hf_config, *nested) if c is not None]
+
+
 def _ep_config(hf_config):
     """Return the config that carries the EP plan / expert count.
 
diff --git a/modelopt/torch/utils/plugins/model_load_utils.py b/modelopt/torch/utils/plugins/model_load_utils.py
index d33f87d5ba33..ee23676b4c39 100644
--- a/modelopt/torch/utils/plugins/model_load_utils.py
+++ b/modelopt/torch/utils/plugins/model_load_utils.py
@@ -37,6 +37,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
+from safetensors.torch import safe_open
 from torch.distributed import device_mesh
 from torch.distributed.checkpoint import HuggingFaceStorageReader
 from torch.distributed.checkpoint.state_dict import StateDictOptions, set_model_state_dict
@@ -82,12 +83,14 @@
     _spec_refuses_ep,
     apply_distributed_support,
     check_distributed_support,
+    config_levels,
     register_torch_a2a,
 )
 from modelopt.torch.utils.plugins.hf_checkpoint_utils import (
     checkpoint_tensor_files,
     ensure_local_checkpoint,
     indexed_weight_map,
+    off_index_safetensors_files,
     read_safetensors_subset,
 )
 
@@ -398,7 +401,9 @@ def _group_sources_by_layer(
     return layer_sources, non_layer_sources, unplaced
 
 
-def unplaced_source_keys(model: nn.Module, ckpt_path: str) -> list[str]:
+def unplaced_source_keys(
+    model: nn.Module, ckpt_path: str, tensor_files: "dict[str, str] | None" = None
+) -> list[str]:
     """Checkpoint tensors the loaded ``model`` has no parameter for, e.g. an MTP head.
 
     Every tensor in the weight files the loader opened
@@ -421,8 +426,10 @@ def unplaced_source_keys(model: nn.Module, ckpt_path: str) -> list[str]:
     and a model our own loader built from its config has none, so the stock mapping answers there.
 
     A checkpoint without safetensors (``pytorch_model.bin``) has nothing the export can carry: ``[]``.
+    Pass ``tensor_files``, the checkpoint's ``tensor name -> file`` map, when the load already read it.
     """
-    tensor_files = checkpoint_tensor_files(ckpt_path)
+    if tensor_files is None:
+        tensor_files = checkpoint_tensor_files(ckpt_path)
     if not tensor_files:
         return []
     # Every name of a tied parameter: a checkpoint can store the one named_parameters() drops by
@@ -445,7 +452,10 @@ def unplaced_source_keys(model: nn.Module, ckpt_path: str) -> list[str]:
 
 
 def record_unplaced_source_keys(
-    model: nn.Module, ckpt_path: str, keys: "Iterable[str] | None" = None
+    model: nn.Module,
+    ckpt_path: str,
+    keys: "Iterable[str] | None" = None,
+    tensor_files: "dict[str, str] | None" = None,
 ) -> list[str]:
     """Record on ``model`` the checkpoint tensors it has no parameter for, for the export to carry.
 
@@ -453,10 +463,19 @@ def record_unplaced_source_keys(
     at load time, and the export reads it back (``_modelopt_unplaced_source_keys``, beside the
     checkpoint in ``_modelopt_source_checkpoint``) rather than deriving it again. Pass ``keys`` when
     the loader already found them while placing the weights; otherwise :func:`unplaced_source_keys`
-    derives them from the loaded model. Every loader of a model that may be exported calls this:
-    :func:`load_pretrained` and :func:`parallel_load_and_prepare_fsdp2` do.
+    derives them from the loaded model, from ``tensor_files`` when the load already read them. Every
+    loader of a model that may be exported calls this: :func:`load_pretrained` and
+    :func:`parallel_load_and_prepare_fsdp2` do.
     """
-    keys = sorted(unplaced_source_keys(model, ckpt_path) if keys is None else keys)
+    if keys is None:
+        # The map by keyword, and only when the load read it: a caller that replaced
+        # unplaced_source_keys with its two-argument form gets its keys as before.
+        keys = (
+            unplaced_source_keys(model, ckpt_path)
+            if tensor_files is None
+            else unplaced_source_keys(model, ckpt_path, tensor_files=tensor_files)
+        )
+    keys = sorted(keys)
     model._modelopt_unplaced_source_keys = keys
     model._modelopt_source_checkpoint = str(ckpt_path)
     return keys
@@ -1998,6 +2017,14 @@ def _ep_load_refusal(hf_config, ep_plan: dict[str, str]) -> Exception | None:
     return ValueError(refusal) if refusal is not None else None
 
 
+def _decompressed_as_loaded(cfg) -> bool:
+    """Whether a model spec installed the on-the-fly decompressor for this config's checkpoint."""
+    # Importing modelopt.torch.models registers every model spec; only expert-parallel loads need them.
+    from modelopt.torch.models import _onfly_decompress
+
+    return _onfly_decompress.is_marked(cfg)
+
+
 def _ckpt_quant_method(cfg):
     """Return the checkpoint's native quantization method, or ``None``.
 
@@ -2005,7 +2032,7 @@ def _ckpt_quant_method(cfg):
     or ``None`` if the checkpoint isn't natively quantized. Checks the top config, then a VLM's
     ``text_config`` or ``llm_config``.
     """
-    for c in (cfg, getattr(cfg, "text_config", None), getattr(cfg, "llm_config", None)):
+    for c in config_levels(cfg):
         qc = getattr(c, "quantization_config", None)
         m = qc.get("quant_method") if isinstance(qc, dict) else getattr(qc, "quant_method", None)
         if m:
@@ -2153,7 +2180,7 @@ def load_and_prepare_ep(
         # checkpoint of the model, nor a plan of its own, would load.
         check_distributed_support(hf_config.model_type)
         quant_method = _ckpt_quant_method(hf_config)
-        if quant_method not in _EP_SOURCE_FORMATS:
+        if quant_method not in _EP_SOURCE_FORMATS and not _decompressed_as_loaded(hf_config):
             # transformers' own quantizer would load it as stored, for PTQ to treat as plain weights.
             dequantized = ", ".join(sorted(f for f in _EP_SOURCE_FORMATS if f))
             loads = "unquantized checkpoints" + (
@@ -2165,6 +2192,24 @@ def load_and_prepare_ep(
             )
         # The swapped plan is this load's own: the caller's config keeps the plan its checkpoint declares.
         hf_config = copy.deepcopy(hf_config)
+        if _decompressed_as_loaded(hf_config):
+            # The decompressor hands the experts over dequantized. Left on the config (Kimi-K2.5's
+            # config class copies its text config's to the top level), the checkpoint's
+            # quantization_config would build transformers' compressed-tensors quantizer, which from
+            # 5.13 claims the same expert converters first. Deleted rather than set to None: before
+            # 5.11 transformers takes any quantization_config attribute, None too, for a quantized
+            # checkpoint. Every level _ckpt_quant_method reads is cleared.
+            levels = config_levels(hf_config)
+            for cfg in levels:
+                vars(cfg).pop("quantization_config", None)
+            # Checked once every level is cleared: a property forwarding to a nested level's is gone
+            # then, while a class attribute, or a property of its own, is not.
+            if kept := [cfg for cfg in levels if hasattr(cfg, "quantization_config")]:
+                raise NotImplementedError(
+                    f"{type(kept[0]).__name__} keeps quantization_config off its instance, so this "
+                    "load cannot clear it, and transformers' own quantizer would compete with the "
+                    "decompressor. Decompress the checkpoint first."
+                )
         ep_cfg = _ep_config(hf_config)  # for VLMs the plan lives on .text_config
         ep_plan = dict(getattr(ep_cfg, "base_model_ep_plan", None) or {})
         if not ep_plan:
@@ -2222,8 +2267,61 @@ def gpu() -> tuple[bool, bool, str | None]:
     dtype = getattr(hf_config, "dtype", None) or torch.bfloat16
     # The load and the export both read the local snapshot, made only once the checks pass, so a
     # refused load downloads nothing.
+    compressed_names: set[str] = set()
+    tensor_files: dict[str, str] | None = None  # read once, by the decompressing load's checks
     with _refused_together("the local checkpoint", alone=False):
         _, ckpt_path = ensure_local_checkpoint(ckpt_path, **hub_options)
+        if _decompressed_as_loaded(hf_config):
+            # Only the experts decompress as they load, and the config cannot say what else the
+            # checkpoint compressed, or in which dtypes: what it stores is checked before anything
+            # loads. Imported here for the reason _decompressed_as_loaded gives.
+            from modelopt.torch.models import _onfly_decompress
+
+            tensor_files = _onfly_decompress.check_compressed_keys(hf_config, ckpt_path)
+            # Minus "weight": a float format's anchor is .weight itself, and an unplaced plain weight
+            # is what the export has always carried as stored.
+            compressed_names = set(_onfly_decompress.stored_names(hf_config)) - {"weight"}
+            # An MTP layer is stored past the decoder layers the model builds, so its compressed
+            # experts would be refused only after the whole load, by the check that follows it.
+            # Refused here instead, before anything loads, from the names just read; that check
+            # stays for any other shape. transformers 5.18 builds no decoder layer past
+            # num_hidden_layers, so this refuses nothing that check would let through there.
+            built = getattr(ep_cfg, "num_hidden_layers", None)
+            past = (
+                []
+                if built is None
+                else sorted(
+                    name
+                    for name in tensor_files
+                    if name.rpartition(".")[2] in compressed_names
+                    and (layer := re.search(r"(?:^|\.)layers\.(\d+)\.", name))
+                    and int(layer[1]) >= built
+                )
+            )
+            if past:
+                raise NotImplementedError(
+                    f"The checkpoint stores {len(past)} compressed tensors past the model's {built} "
+                    f"decoder layers, such as {past[0]} (an MTP layer's, typically): the model builds "
+                    "no parameter for them, and the export would copy them as stored, with no "
+                    "quantization_config to read them. Decompress the checkpoint first."
+                )
+            # The files the load never opens (GLM-4.7's standalone mtp.safetensors, say) the export
+            # copies as stored, and this load drops the config a compressed tensor there needs:
+            # refused here too. A plain one is carried as before.
+            sidecar = []
+            for file_name in off_index_safetensors_files(ckpt_path):
+                with safe_open(os.path.join(ckpt_path, file_name), framework="pt") as f:
+                    sidecar += [
+                        f"{file_name}'s {name}"
+                        for name in f.keys()  # noqa: SIM118 - safe_open is not iterable
+                        if name.rpartition(".")[2] in compressed_names
+                    ]
+            if sidecar:
+                raise NotImplementedError(
+                    f"The checkpoint stores {len(sidecar)} compressed tensors in files the load "
+                    f"never opens, such as {sidecar[0]}: the export would copy them as stored, with "
+                    "no quantization_config to read them. Decompress the checkpoint first."
+                )
 
     # Experts are sharded over ``ep`` and replicated over ``edp``; the dense part is replicated over
     # the whole world (``dp``).
@@ -2300,7 +2398,16 @@ def gpu() -> tuple[bool, bool, str | None]:
 
         # The checkpoint tensors the model has no parameter for (an MTP head), worked out once, at load,
         # for the export to carry.
-        record_unplaced_source_keys(model, ckpt_path)
+        unplaced = record_unplaced_source_keys(model, ckpt_path, tensor_files=tensor_files)
+        # The export would copy them as stored, and this load dropped the quantization_config that
+        # reads a compressed one, such as an MTP layer's experts. A suffix is enough to tell them:
+        # check_compressed_keys refused any compressed tensor outside the experts.
+        if carried := [key for key in unplaced if key.rpartition(".")[2] in compressed_names]:
+            raise NotImplementedError(
+                f"The checkpoint stores {len(carried)} compressed tensors the model has no parameter "
+                f"for, such as {carried[0]}: the export would copy them as stored, with no "
+                "quantization_config to read them. Decompress the checkpoint first."
+            )
 
         # The model carries the ep submesh, for calibration to re-install the all-to-all forwards
         # over (expert_parallel_mesh); transformers sets it differently by release.
diff --git a/noxfile.py b/noxfile.py
index 4fa9fc107afa..f4a8ecb726a8 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -66,7 +66,11 @@ def _cov_args():
 @nox.parametrize("torch_ver", [nox.param(k, id=k) for k in TORCH_VERSIONS])
 def unit(session, torch_ver, tf_ver):
     """Unit tests — parametrized over torch and transformers versions."""
-    session.install(TORCH_VERSIONS[torch_ver], "-e", ".[all,dev-test]")
+    # compressed-tensors runs the real-library tests of on-the-fly expert decompression; it goes in
+    # with the torchvision pin so the resolver keeps the leg's torch. Not in dev-test: the
+    # transformers-free partial_unit(torch) lane installs that extra, and compressed-tensors pulls in
+    # transformers.
+    session.install(TORCH_VERSIONS[torch_ver], "-e", ".[all,dev-test]", "compressed-tensors")
     tf_pins = TRANSFORMERS_VERSIONS[tf_ver]
     if tf_pins:
         session.install(*tf_pins)
diff --git a/tests/gpu/torch/utils/test_expert_parallel_load.py b/tests/gpu/torch/utils/test_expert_parallel_load.py
index 849305f4f6a8..f26bb6506a81 100644
--- a/tests/gpu/torch/utils/test_expert_parallel_load.py
+++ b/tests/gpu/torch/utils/test_expert_parallel_load.py
@@ -17,20 +17,30 @@
 
 ``load_and_prepare_ep`` shards an MoE's routed experts over the ranks, and the ``torch_a2a`` experts
 forward sends each token to the rank that holds its expert. Every rank runs its own batch through the
-sharded model, and the result must match the whole model run on that batch alone.
+sharded model, and the result must match the whole model run on that batch alone. A checkpoint whose
+experts are compressed-tensors compressed loads the same way, each rank decompressing its own.
 """
 
+import json
+from types import SimpleNamespace
+
 import pytest
 import torch
+import torch.distributed as dist
+from _test_utils.torch.distributed.utils import bounded_collectives
 from packaging.version import Version
+from safetensors.torch import load_file, save_file
 
 transformers = pytest.importorskip("transformers")
 if Version(transformers.__version__) < Version("5.7"):
     pytest.skip("qwen3_moe's base_model_ep_plan ships in transformers 5.7", allow_module_level=True)
 
+import transformers.core_model_loading as cml
+import transformers.modeling_utils as mu
 from torch.distributed.tensor import DTensor
 from transformers import AutoConfig, AutoModelForCausalLM, Qwen3MoeConfig
 
+from modelopt.torch.models import _onfly_decompress
 from modelopt.torch.utils.distributed import (
     DistributionPlan,
     distribution_plan,
@@ -101,6 +111,130 @@ def _sharded_model_matches_the_whole_one(rank, size, ckpt_dir, ep_degree):
     torch.testing.assert_close(sharded.float(), reference.float(), atol=2e-2, rtol=2e-2)
 
 
+# The compressions the published Kimi checkpoints use, INT4 (Kimi-K2.5) and MXFP4 (Kimi-K3), through
+# compressed-tensors (quant_method "compressed-tensors"), which the on-the-fly decompressor reads; not
+# transformers' own MXFP4 (quant_method "mxfp4"). The experts draw values each format holds exactly,
+# at a power-of-two scale bf16 holds too, so they decompress to the dense model's weights exactly.
+_INT4 = {"num_bits": 4, "type": "int", "symmetric": True, "strategy": "group", "group_size": 32}
+_E2M1 = torch.tensor([0, 0.5, 1, 1.5, 2, 3, 4, 6])
+_COMPRESSIONS = {
+    "int4": ("pack-quantized", _INT4, torch.arange(-8, 8) / 64, 1 / 64),
+    "mxfp4": (
+        "mxfp4-pack-quantized",
+        {**_INT4, "type": "float", "scale_dtype": "torch.uint8"},
+        torch.cat([_E2M1, -_E2M1[1:]]) / 16,
+        1 / 16,
+    ),
+}
+
+
+def _save_compressed_tiny_qwen3_moe(dense_dir, packed_dir, compression):
+    """The tiny Qwen3-MoE twice: whole in bf16, and with its experts compressed by ``compression``."""
+    fmt, weights, values, scale = _COMPRESSIONS[compression]
+    _save_tiny_qwen3_moe(dense_dir)
+    # Copied out of the file, which is written over below.
+    state = {k: v.clone() for k, v in load_file(str(dense_dir / "model.safetensors")).items()}
+    gen = torch.Generator().manual_seed(0)
+    for key in [k for k in state if ".experts." in k]:  # ...mlp.experts.<e>.gate_proj.weight
+        draw = torch.randint(0, len(values), state[key].shape, generator=gen)
+        state[key] = values[draw].to(torch.bfloat16)
+    save_file(state, str(dense_dir / "model.safetensors"), metadata={"format": "pt"})
+
+    quantization_config = {
+        "quant_method": "compressed-tensors",
+        "format": fmt,
+        "config_groups": {"experts": {"targets": ["Linear"], "weights": weights, "format": fmt}},
+    }
+    compressor, scheme, _, _ = _onfly_decompress._read_compression(
+        SimpleNamespace(quantization_config=quantization_config)
+    )
+    packed = {}
+    for key, tensor in state.items():
+        if ".experts." not in key:
+            packed[key] = tensor
+            continue
+        rows, cols = tensor.shape
+        scales = torch.full((rows, cols // 32), scale)
+        stored = compressor.compress({"weight": tensor.float(), "weight_scale": scales}, scheme)
+        for name, value in stored.items():  # floats as bf16, as the Kimi checkpoints store them
+            packed[key.removesuffix("weight") + name] = (
+                value.bfloat16() if value.is_floating_point() else value
+            )
+    packed_dir.mkdir()
+    save_file(packed, str(packed_dir / "model.safetensors"), metadata={"format": "pt"})
+    config = json.loads((dense_dir / "config.json").read_text())
+    config["quantization_config"] = quantization_config
+    (packed_dir / "config.json").write_text(json.dumps(config))
+
+
+def _compressed_experts_load_on_their_own_ranks(rank, size, dense_dir, packed_dir):
+    device = torch.device(f"cuda:{rank}")
+    problems = []
+    # The decompressor's hooks are process-wide, and the workers outlive the job.
+    hooks = (mu.get_model_conversion_mapping, cml.spawn_materialize)
+    try:
+        with bounded_collectives(seconds=120):
+            config = AutoConfig.from_pretrained(packed_dir)
+            _onfly_decompress.install_onfly_decompress(config)  # as a model spec's prepare does
+            model = load_and_prepare_ep(
+                packed_dir, DistributionPlan(size, ep_size=size), hf_config=config
+            ).eval()
+            whole = AutoModelForCausalLM.from_pretrained(dense_dir, dtype=torch.bfloat16)
+            whole = whole.to(device).eval()
+
+            # This rank's experts, decompressed as they loaded, are the dense model's, exactly.
+            per_rank = NUM_EXPERTS // size
+            owned = slice(rank * per_rank, (rank + 1) * per_rank)
+            for i, (layer, whole_layer) in enumerate(zip(model.model.layers, whole.model.layers)):
+                for name in ("gate_up_proj", "down_proj"):
+                    local = _local(getattr(layer.mlp.experts, name))
+                    if local.shape[0] != per_rank:
+                        problems.append(f"layer {i} {name}: {local.shape[0]} experts held")
+                    elif not torch.equal(local, getattr(whole_layer.mlp.experts, name)[owned]):
+                        problems.append(f"layer {i} {name}: rank {rank}'s experts differ")
+            # Every stored tensor of the experts it built went into them.
+            if model._modelopt_unplaced_source_keys:
+                problems.append(f"unplaced: {model._modelopt_unplaced_source_keys[:3]}")
+            set_distribution_plan(model, DistributionPlan(size, ep_size=size))
+
+            generator = torch.Generator().manual_seed(rank)
+            input_ids = torch.randint(0, 128, (2, 16), generator=generator).to(device)
+            with torch.no_grad():
+                sharded = model(input_ids).logits.float()
+                reference = whole(input_ids).logits.float()
+            if not torch.allclose(sharded, reference, atol=2e-2, rtol=2e-2):
+                problems.append(f"logits differ by {(sharded - reference).abs().max().item():.3g}")
+
+            gathered = [None] * size
+            dist.all_gather_object(gathered, problems)
+    finally:
+        mu.get_model_conversion_mapping, cml.spawn_materialize = hooks
+    found = [problem for rank_problems in gathered for problem in rank_problems]
+    assert not found, "\n".join(found[:20])
+
+
+@pytest.mark.parametrize("compression", sorted(_COMPRESSIONS))
+def test_compressed_experts_load_expert_parallel_each_rank_decompressing_its_own(
+    need_2_gpus, dist_workers, tmp_path, compression
+):
+    """A checkpoint whose experts are compressed (INT4, MXFP4) loads expert-parallel through the
+    on-the-fly decompressor: each rank holds exactly the dense model's weights for its own experts,
+    every stored tensor is placed, and the sharded model matches the dense one loaded whole. Each
+    rank's checks are gathered before any rank fails, so a regression fails instead of hanging."""
+    pytest.importorskip(
+        "compressed_tensors",
+        minversion="0.16",
+        reason="compressed-tensors is needed: the checkpoint's experts are compressed",
+    )
+    if Version(transformers.__version__) < Version("5.8"):
+        pytest.skip("the on-the-fly decompression needs transformers 5.8")
+    if NUM_EXPERTS % dist_workers.world_size:
+        pytest.skip(f"{NUM_EXPERTS} experts do not shard over {dist_workers.world_size} GPUs")
+    dense_dir, packed_dir = tmp_path / "dense", tmp_path / "packed"
+    _save_compressed_tiny_qwen3_moe(dense_dir, packed_dir, compression)
+    dist_workers.run(_compressed_experts_load_on_their_own_ranks, str(dense_dir), str(packed_dir))
+
+
 @pytest.mark.parametrize("ep_share", [1, 2], ids=["ep-world", "ep-half"])
 def test_expert_parallel_model_matches_the_whole_model(
     need_2_gpus, dist_workers, tmp_path, ep_share
diff --git a/tests/unit/torch/export/test_unified_export_hf.py b/tests/unit/torch/export/test_unified_export_hf.py
index 1b8d55373fae..12126c61a277 100644
--- a/tests/unit/torch/export/test_unified_export_hf.py
+++ b/tests/unit/torch/export/test_unified_export_hf.py
@@ -1080,7 +1080,9 @@ def test_an_unrecorded_model_derives_its_unplaced_keys_once(tmp_path, monkeypatc
     )
     derived_from = []
     monkeypatch.setattr(
-        model_load_utils, "unplaced_source_keys", lambda m, c: derived_from.append(c) or [mtp]
+        model_load_utils,
+        "unplaced_source_keys",
+        lambda m, c, tensor_files=None: derived_from.append(c) or [mtp],
     )
     model = _ProvenanceModel(name_or_path=tmp_path)
 
diff --git a/tests/unit/torch/models/test_onfly_decompress.py b/tests/unit/torch/models/test_onfly_decompress.py
new file mode 100644
index 000000000000..3c68410d6fcd
--- /dev/null
+++ b/tests/unit/torch/models/test_onfly_decompress.py
@@ -0,0 +1,1024 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""On-the-fly decompression of compressed-tensors MoE experts while ``from_pretrained`` loads.
+
+The tests drive transformers' own weight-loading objects the way its loader does --
+``rename_source_key`` routes a checkpoint key, ``add_tensor`` collects it, ``convert`` fuses it --
+with a stand-in compressor, so they need neither a checkpoint nor ``compressed_tensors``. The tests
+that use the real ``compressed_tensors`` registry, among them real ``from_pretrained`` loads of tiny
+compressed checkpoints, skip when it cannot be imported.
+"""
+
+import copy
+import importlib.metadata
+import json
+import sys
+from concurrent.futures import Future
+from types import SimpleNamespace
+
+import pytest
+import torch
+from safetensors import safe_open
+from safetensors.torch import load_file, save_file
+from torch.distributed.tensor import Shard, distribute_tensor
+
+transformers = pytest.importorskip("transformers")
+
+from packaging.version import Version
+
+if Version(transformers.__version__) < Version("5.8"):
+    pytest.skip("the on-the-fly decompression needs transformers >= 5.8", allow_module_level=True)
+
+import transformers.core_model_loading as cml
+import transformers.modeling_utils as mu
+from transformers import (
+    AutoConfig,
+    AutoModelForCausalLM,
+    DeepseekV3Config,
+    LlamaConfig,
+    MixtralConfig,
+    PretrainedConfig,
+)
+from transformers.conversion_mapping import get_checkpoint_conversion_mapping
+from transformers.core_model_loading import (
+    Chunk,
+    Concatenate,
+    MergeModulelist,
+    WeightConverter,
+    WeightRenaming,
+)
+
+from modelopt.torch.export.quant_aware_conversion import build_reverse_name_mapper
+from modelopt.torch.models import _onfly_decompress as onfly
+
+PACKED = ("weight_packed", "weight_scale", "weight_shape")  # what pack-quantized stores per weight
+
+
+@pytest.fixture
+def restore_loader_hooks(monkeypatch):
+    """Undo the process-wide transformers hooks the decompressor installs, and start the test with
+    no decompressing load under way, whatever an earlier test's load left."""
+    for owner, name in ((cml, "spawn_materialize"), (mu, "get_model_conversion_mapping")):
+        monkeypatch.setattr(owner, name, getattr(owner, name))
+    monkeypatch.setattr(onfly, "_LOAD", {"decompressing": False})
+
+
+class _ScaledCompressor:
+    """Stands in for a compressed_tensors compressor: ``weight = stored weight * weight_scale``,
+    the stored weight being ``weight_packed`` (pack-quantized) or ``weight`` (float formats)."""
+
+    @classmethod
+    def decompress(cls, state_dict, scheme):
+        stored = state_dict.get("weight_packed", state_dict.get("weight"))
+        return {"weight": stored.float() * state_dict["weight_scale"]}
+
+
+def _stock_expert_converters():
+    """transformers' own per-expert -> fused converters for DeepSeek-V3: ``(gate_up, down)``."""
+    by_target = {
+        conv.target_patterns[0]: conv
+        for conv in get_checkpoint_conversion_mapping("deepseek_v3")
+        if isinstance(conv, WeightConverter)
+    }
+    return by_target["mlp.experts.gate_up_proj"], by_target["mlp.experts.down_proj"]
+
+
+def _load(converters, checkpoint):
+    """Route, collect and fuse ``checkpoint`` the way transformers' loader does.
+
+    The first converter whose sources match a key claims it; every fused target gets its own deep
+    copy of that converter, fed in checkpoint order with completed futures (async loading).
+    """
+    per_target = {}
+    for key, tensor in checkpoint.items():
+        for conv in converters:
+            target, pattern = conv.rename_source_key(key)
+            if pattern is not None:
+                break
+        else:
+            raise AssertionError(f"no converter claims {key}")
+        future = Future()
+        future.set_result(tensor)
+        per_target.setdefault(target, copy.deepcopy(conv)).add_tensor(target, key, pattern, future)
+    fused = {}
+    for target, conv in per_target.items():
+        fused.update(conv.convert(target))
+    return fused
+
+
+# ------------------------------------------------------------- rewriting the per-expert converters
+
+
+def test_only_expert_weight_sources_are_rewritten_and_stored_params_need_their_weight():
+    """Only the experts' ``.weight`` sources name a compressed tensor: a converter reading another --
+    an expert's bias, or a dense weight fused with the experts' -- is refused where it is rewritten,
+    before any tensor is read. A stored param without its weight still reaches the decompress op,
+    which refuses it rather than hand the merge a tensor it would fuse into the wrong slot."""
+    for other in ("mlp.experts.*.gate_proj.bias", "mlp.gate_proj.weight"):
+        conv = WeightConverter(
+            source_patterns=["mlp.experts.*.gate_proj.weight", other],
+            target_patterns="mlp.experts.gate_proj",
+            operations=[MergeModulelist(dim=0)],
+        )
+        with pytest.raises(NotImplementedError, match=r"alongside the experts. weights"):
+            onfly._compressedify(conv, _ScaledCompressor, None, PACKED)
+
+    gate_up, _ = _stock_expert_converters()
+    scales_alone = {"model.layers.0.mlp.experts.0.gate_proj.weight_scale": torch.ones(2, 2)}
+    with pytest.raises(ValueError, match=r"unconsumed checkpoint keys .*gate_proj\.weight_scale"):
+        _load([onfly._compressedify(gate_up, _ScaledCompressor, None, PACKED)], scales_alone)
+
+
+def test_a_rewritten_converter_keeps_the_scope_transformers_set_on_it():
+    """transformers scopes a sub-model's converters (a VLM's language model) once it has built them;
+    the rewrite keeps whichever of those fields this release has."""
+    _, down = _stock_expert_converters()
+    down = copy.deepcopy(down)
+    fields = {"scope_prefix": "language_model", "base_model_prefix": "model", "force_cpu": True}
+    fields = {name: value for name, value in fields.items() if hasattr(type(down), name)}
+    for name, value in fields.items():
+        setattr(down, name, value)
+
+    rewritten = onfly._compressedify(down, _ScaledCompressor, None, PACKED)
+
+    assert {name: getattr(rewritten, name) for name in fields} == fields
+
+
+# ----------------------------------------------------------------- decompressing during the load
+
+
+@pytest.mark.parametrize(
+    "stored", [PACKED, ("weight", "weight_scale")], ids=["pack-quantized", "weight-and-scale"]
+)
+def test_packed_experts_decompress_then_fuse_like_an_uncompressed_checkpoint(stored):
+    """Through transformers' loader objects, per-expert compressed weights and scales come out as
+    the fused bf16 ``gate_up_proj`` ([gate; up] per expert) and the stacked ``down_proj``.
+
+    The decompress op runs first, as a ``MergeModulelist``: loaders shard experts at load only when
+    the first op is one. A float format stores ``weight`` next to ``weight_scale``, and transformers
+    regex-*searches* source patterns, so only anchored patterns keep a weight from claiming its
+    scale.
+    """
+    n_experts, inter, hidden = 3, 2, 4
+    shapes = {"gate": (inter, hidden), "up": (inter, hidden), "down": (hidden, inter)}
+    gen = torch.Generator().manual_seed(0)
+    dense, checkpoint = {}, {}
+    for e in range(n_experts):
+        for proj, shape in shapes.items():
+            packed = torch.randint(-8, 8, shape, generator=gen, dtype=torch.int32)
+            scale = torch.rand(shape[0], 1, generator=gen) + 0.5
+            dense[proj, e] = (packed.float() * scale).bfloat16()
+            key = f"model.layers.0.mlp.experts.{e}.{proj}_proj."
+            values = {"weight_packed": packed, "weight": packed, "weight_scale": scale}
+            values["weight_shape"] = torch.tensor(shape)
+            checkpoint |= {key + name: values[name] for name in stored}
+    converters = [
+        onfly._compressedify(conv, _ScaledCompressor, None, stored)
+        for conv in _stock_expert_converters()
+    ]
+    assert all(isinstance(conv.operations[0], MergeModulelist) for conv in converters)
+
+    fused = _load(converters, checkpoint)
+    gate_up = fused.pop("model.layers.0.mlp.experts.gate_up_proj")
+    down = fused.pop("model.layers.0.mlp.experts.down_proj")
+
+    assert not fused
+    assert gate_up.dtype == down.dtype == torch.bfloat16
+    expected = [torch.cat([dense["gate", e], dense["up", e]]) for e in range(n_experts)]
+    assert torch.equal(gate_up, torch.stack(expected))
+    assert torch.equal(down, torch.stack([dense["down", e] for e in range(n_experts)]))
+
+
+def test_only_owned_experts_and_stored_siblings_reach_the_compressor():
+    """Under EP the loader hands experts this rank does not own over as ``None``; they are skipped,
+    so a rank decompresses only its own. A checkpoint may store fewer params than the scheme
+    declares (no ``weight_global_scale`` in Kimi-K3's mxfp4); missing ones are not passed, but
+    one stored for some experts and not others is refused."""
+    seen = []
+
+    class RecordingCompressor(_ScaledCompressor):
+        @classmethod
+        def decompress(cls, state_dict, scheme):
+            seen.append((sorted(state_dict), scheme))
+            return super().decompress(state_dict, scheme)
+
+    _, down = _stock_expert_converters()
+    names = ("weight_packed", "weight_scale", "weight_global_scale")
+    decompress = onfly._compressedify(down, RecordingCompressor, "scheme", names).operations[0]
+    packed = [torch.full((2, 2), e + 1, dtype=torch.int32) for e in range(3)]
+    scale = torch.full((2, 1), 0.5)
+
+    out = decompress.convert(
+        {
+            "mlp.experts.*.down_proj.weight_packed$": [packed[0], None, packed[2]],
+            "mlp.experts.*.down_proj.weight_scale$": [scale, None, scale],
+        }
+    )
+
+    [(pattern, experts)] = out.items()
+    assert pattern == "mlp.experts.*.down_proj.weight_packed$"  # the anchor, for the stock merge
+    assert [t.dtype for t in experts] == [torch.bfloat16] * 2
+    assert [t[0, 0].item() for t in experts] == [0.5, 1.5]
+    assert seen == [(["weight_packed", "weight_scale"], "scheme")] * 2
+    # Paired by position, an expert missing one of them would shift every later expert's pairing.
+    short = {
+        "mlp.experts.*.down_proj.weight_packed$": packed,
+        "mlp.experts.*.down_proj.weight_scale$": [scale, scale],
+    }
+    with pytest.raises(ValueError, match="3 weight_packed, 2 weight_scale, not one per expert"):
+        decompress.convert(short)
+
+
+PACKED_INT32 = torch.tensor([[2**31 - 15, -(2**31) + 3]], dtype=torch.int32)  # not exact in bf16
+
+
+def _read(path, sharding_op=None, dtype=torch.bfloat16):
+    """Read ``path``'s one tensor the way the loader does: a safetensors slice, through spawn_materialize."""
+    with safe_open(str(path), framework="pt") as stored:
+        [key] = stored.keys()
+        kwargs = {"sharding_op": sharding_op, "tensor_idx": 0} if sharding_op else {}
+        return cml.spawn_materialize(None, stored.get_slice(key), "cpu", dtype, **kwargs)()
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_integer_checkpoint_tensors_keep_their_dtype_on_load(tmp_path):
+    """Without a quantizer the loader casts every tensor to the model's float dtype as it reads it,
+    which destroys packed weights, shapes and zero-points. In a decompressing load integers stay
+    integers; floats are still cast, and integer-to-integer casts still happen. Any other load casts
+    as before."""
+    save_file({"packed": PACKED_INT32}, str(tmp_path / "packed.safetensors"))
+    onfly._install_keep_int_native()
+    assert _read(tmp_path / "packed.safetensors").dtype == torch.bfloat16
+    # Outside a decompressing load the hook looks at nothing it is handed.
+    cml.spawn_materialize(None, object(), "cpu", torch.bfloat16)
+    onfly._LOAD["decompressing"] = True
+
+    kept = _read(tmp_path / "packed.safetensors")
+
+    assert kept.dtype == torch.int32
+    assert torch.equal(kept, PACKED_INT32)
+    assert _read(tmp_path / "packed.safetensors", dtype=torch.int64).dtype == torch.int64
+    assert (
+        cml.spawn_materialize(None, torch.ones(2), "cpu", torch.bfloat16)().dtype == torch.bfloat16
+    )
+    assert cml.spawn_materialize(None, PACKED_INT32, "cpu", torch.bfloat16)().dtype == torch.int32
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_an_integer_sharded_onto_the_expert_mesh_keeps_its_dtype(tmp_path, gloo_world_of_one):
+    """From transformers 5.14 an expert-parallel load shards each expert's tensor onto the mesh as it
+    reads it, and the shard op casts it to the parameter's dtype, not the copy: a decompressing load
+    keeps the packed integers there too."""
+    sharding_utils = pytest.importorskip("transformers.distributed.sharding_utils")
+    experts = torch.zeros(2, 4, 8, dtype=torch.bfloat16)  # a fused expert param, sharded by expert
+    shard = sharding_utils.DtensorShardOperation(
+        distribute_tensor(experts, gloo_world_of_one, [Shard(0)])
+    )
+    save_file({"packed": PACKED_INT32}, str(tmp_path / "packed.safetensors"))
+    onfly._install_keep_int_native()
+    onfly._LOAD["decompressing"] = True
+
+    kept = _read(tmp_path / "packed.safetensors", sharding_op=shard)
+
+    assert kept.dtype == torch.int32
+    assert torch.equal(kept, PACKED_INT32)
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_a_read_parameter_a_release_adds_passes_through(monkeypatch):
+    """The patch sits on every later load's reads, so a parameter past the dtype (5.14's sharding_op
+    and tensor_idx, or one a release adds) reaches the loader's own, positionally or by keyword."""
+    seen = []
+
+    def release_spawn(
+        thread_pool, tensor, device=None, dtype=None, sharding_op=None, tensor_idx=None
+    ):
+        seen.append((sharding_op, tensor_idx))
+        return tensor
+
+    monkeypatch.setattr(cml, "spawn_materialize", release_spawn)
+    onfly._install_keep_int_native()
+
+    cml.spawn_materialize(None, torch.ones(2), "cpu", torch.bfloat16, "op", 3)
+    cml.spawn_materialize(
+        None, torch.ones(2), "cpu", torch.bfloat16, sharding_op="op", tensor_idx=3
+    )
+
+    assert seen == [("op", 3), ("op", 3)]
+
+
+# ------------------------------------------------------------------------ installing the rewrite
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_install_rewrites_only_the_per_expert_weight_converters(monkeypatch):
+    """The loader's conversion list also holds renamings and other converters; only converters that
+    fuse per-expert ``.weight`` tensors are rewritten, not a shared expert's, and from_pretrained's
+    arguments reach the stock mapping unchanged."""
+    renaming = WeightRenaming(r"experts\.(\d+)\.w1\.weight", r"experts.\1.gate_proj.weight")
+    qkv = WeightConverter(
+        "self_attn.qkv_proj.weight",
+        ["self_attn.q_proj.weight", "self_attn.k_proj.weight", "self_attn.v_proj.weight"],
+        [Chunk(dim=0)],
+    )
+    expert_bias = WeightConverter(
+        "mlp.experts.*.down_proj.bias", "mlp.experts.down_proj_bias", [MergeModulelist(dim=0)]
+    )
+    shared_expert = WeightConverter(
+        ["mlp.shared_experts.gate_proj.weight", "mlp.shared_experts.up_proj.weight"],
+        "mlp.shared_experts.gate_up_proj",
+        [Concatenate(dim=0)],
+    )
+    gate_up, down = _stock_expert_converters()
+    stock = [renaming, qkv, expert_bias, shared_expert, gate_up, down]
+    calls = []
+
+    def stock_mapping(model, key_mapping=None, hf_quantizer=None, add_legacy=True):
+        calls.append((model, key_mapping, hf_quantizer, add_legacy))
+        return list(stock)
+
+    compression = (_ScaledCompressor, None, PACKED, "stub-format")
+    monkeypatch.setattr(mu, "get_model_conversion_mapping", stock_mapping)
+    monkeypatch.setattr(onfly, "_read_compression", lambda cfg: compression)
+    config = SimpleNamespace()
+    onfly.install_onfly_decompress(config)
+    model = SimpleNamespace(config=config)
+
+    # positionally, the way from_pretrained calls it, and a later parameter too
+    out = mu.get_model_conversion_mapping(model, {"old": "new"}, "quantizer", False)
+
+    assert calls == [(model, {"old": "new"}, "quantizer", False)]
+    assert len(out) == len(stock)
+    assert all(kept is original for kept, original in zip(out[:4], stock[:4]))
+    assert [conv.source_patterns[0] for conv in out[4:]] == [
+        "mlp.experts.*.gate_proj.weight_packed$",
+        "mlp.experts.*.down_proj.weight_packed$",
+    ]
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_a_marked_model_with_no_expert_converter_to_rewrite_is_refused(monkeypatch):
+    """With nothing rewritten, the stock converters would look for plain ``.weight`` keys the
+    checkpoint does not store, and the experts would load newly initialized. The load is refused,
+    and the config, whose converters were not rewritten, keeps its mark."""
+    renaming = WeightRenaming(r"experts\.(\d+)\.w1\.weight", r"experts.\1.gate_proj.weight")
+    monkeypatch.setattr(mu, "get_model_conversion_mapping", lambda model, **kw: [renaming])
+    compression = (_ScaledCompressor, None, PACKED, "stub-format")
+    monkeypatch.setattr(onfly, "_read_compression", lambda cfg: compression)
+    config = SimpleNamespace()
+    onfly.install_onfly_decompress(config)
+    model = SimpleNamespace(config=copy.deepcopy(config))
+
+    with pytest.raises(RuntimeError, match="no per-expert converter to rewrite"):
+        mu.get_model_conversion_mapping(model)
+    assert onfly.is_marked(model.config)
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_installing_twice_wraps_the_loader_once(monkeypatch):
+    """``prepare`` may run more than once per process. Installing again for the same checkpoint must
+    not wrap the hooks again: with a ``weight`` anchor the outer wrap would rewrite the rewritten
+    converter and decompress twice."""
+    _, down = _stock_expert_converters()
+    reads = []
+
+    def read(cfg):
+        reads.append(cfg)
+        return _ScaledCompressor, None, ("weight", "weight_scale"), "stub-format"
+
+    def stock_mapping(model, **kwargs):
+        return [copy.deepcopy(down)]
+
+    monkeypatch.setattr(mu, "get_model_conversion_mapping", stock_mapping)
+    monkeypatch.setattr(onfly, "_read_compression", read)
+
+    def hooks():
+        return mu.get_model_conversion_mapping, cml.spawn_materialize
+
+    onfly.install_onfly_decompress(SimpleNamespace())
+    installed = hooks()
+    again = SimpleNamespace()
+    onfly.install_onfly_decompress(again)
+
+    assert hooks() == installed
+    assert len(reads) == 2
+    [conv] = mu.get_model_conversion_mapping(SimpleNamespace(config=again))
+    assert len(conv.operations) == len(down.operations) + 1
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_a_checkpoint_compressed_differently_needs_a_process_of_its_own(monkeypatch):
+    """The hooks are process-wide: decompressing a second checkpoint with the first one's scheme and
+    stored params would load it wrong without a word, so its install is refused and the hooks stay as
+    they are. A config the installing prepare already stripped of its quantization_config is the
+    first checkpoint again; one with none that was never marked is refused, since nothing would
+    decompress its experts; one whose quantization_config fails to parse is not, and its error
+    goes through."""
+    _, down = _stock_expert_converters()
+    compressions = {
+        "a": (_ScaledCompressor, None, PACKED, "stub-format"),
+        "b": (_ScaledCompressor, None, (*PACKED, "weight_zero_point"), "stub-format"),
+        None: None,
+    }
+
+    def read(cfg):
+        if cfg.checkpoint == "malformed":
+            # compressed_tensors' parse raises pydantic's ValidationError, a ValueError.
+            raise ValueError("1 validation error for QuantizationConfig")
+        return compressions[cfg.checkpoint]
+
+    monkeypatch.setattr(
+        mu, "get_model_conversion_mapping", lambda model, **kw: [copy.deepcopy(down)]
+    )
+    monkeypatch.setattr(onfly, "_read_compression", read)
+
+    first = SimpleNamespace(checkpoint="a")
+    onfly.install_onfly_decompress(first)
+    installed = mu.get_model_conversion_mapping
+    first.checkpoint = None  # its prepare dropped the quantization_config after installing
+    onfly.install_onfly_decompress(first)
+    with pytest.raises(ValueError, match="is not marked"):
+        onfly.install_onfly_decompress(SimpleNamespace(checkpoint=None))
+    with pytest.raises(RuntimeError, match="needs a process of its own"):
+        onfly.install_onfly_decompress(SimpleNamespace(checkpoint="b"))
+    with pytest.raises(ValueError, match="validation error"):
+        onfly.install_onfly_decompress(SimpleNamespace(checkpoint="malformed"))
+
+    assert mu.get_model_conversion_mapping is installed
+
+
+def _install_for_deepseek_v3(monkeypatch):
+    """Install for a compressed DeepSeek-V3 config; a config without quantization_config reads as none."""
+    compression = (_ScaledCompressor, None, PACKED, "stub-format")
+
+    def read(cfg):
+        return None if getattr(cfg, "quantization_config", None) is None else compression
+
+    monkeypatch.setattr(
+        mu, "get_model_conversion_mapping", lambda model, **kw: list(_stock_expert_converters())
+    )
+    monkeypatch.setattr(onfly, "_read_compression", read)
+    config = DeepseekV3Config(num_hidden_layers=1)
+    config.quantization_config = {"format": "stub-format"}
+    onfly.install_onfly_decompress(config)
+    return config
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+@pytest.mark.parametrize("loaded", ["another-model", "stripped-copy", "fresh-config"])
+def test_only_the_compressed_checkpoint_loads_through_the_rewritten_converters(monkeypatch, loaded):
+    """The hooks outlive the compressed checkpoint's load. Rewritten, another (uncompressed) MoE's
+    experts would match no stored key, load newly initialized, and from_pretrained would not
+    complain. from_pretrained builds the model from a deep copy of the config ``prepare`` handed
+    over; that copy must still select the rewrite, whether ``prepare`` ran again on the config it
+    stripped of its quantization_config (Kimi-K3) or on a fresh config of the same checkpoint, and
+    the mark must not reach the config a later save writes."""
+    config = _install_for_deepseek_v3(monkeypatch)
+    if loaded == "another-model":
+        config = DeepseekV3Config(num_hidden_layers=1)
+    else:
+        if loaded == "stripped-copy":
+            config.quantization_config = None
+        else:
+            config = DeepseekV3Config(num_hidden_layers=1)
+            config.quantization_config = {"format": "stub-format"}
+        onfly.install_onfly_decompress(config)
+    model = SimpleNamespace(config=copy.deepcopy(config))
+
+    out = mu.get_model_conversion_mapping(model)
+
+    stored = "weight" if loaded == "another-model" else "weight_packed$"
+    assert [conv.source_patterns[0] for conv in out] == [
+        f"mlp.experts.*.gate_proj.{stored}",
+        f"mlp.experts.*.down_proj.{stored}",
+    ]
+    assert onfly._MARK not in model.config.to_dict()
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+@pytest.mark.parametrize("marked", ["vlm", "text-config"])
+def test_a_vlm_marked_at_either_level_loads_through_the_rewritten_converters(monkeypatch, marked):
+    """A VLM carries its compressed-tensors config at the top level or on its text config, and the
+    install marks the config it is handed. Either way the VLM's expert converters are rewritten, and
+    neither level keeps the mark for a later save."""
+    monkeypatch.setattr(
+        mu, "get_model_conversion_mapping", lambda model, **kw: list(_stock_expert_converters())
+    )
+    compression = (_ScaledCompressor, None, PACKED, "stub-format")
+    monkeypatch.setattr(onfly, "_read_compression", lambda cfg: compression)
+    text = _RemoteTextConfig(hidden_size=64, quantization_config={"format": "stub-format"})
+    vlm = SimpleNamespace(text_config=text)
+    onfly.install_onfly_decompress(vlm if marked == "vlm" else text)
+    model = SimpleNamespace(config=copy.deepcopy(vlm))
+
+    out = mu.get_model_conversion_mapping(model)
+
+    assert [conv.source_patterns[0] for conv in out] == [
+        "mlp.experts.*.gate_proj.weight_packed$",
+        "mlp.experts.*.down_proj.weight_packed$",
+    ]
+    assert not onfly.is_marked(model.config)
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_a_language_model_config_named_llm_config_is_read_as_text_config_is(monkeypatch):
+    """Some VLMs name their language-model config llm_config, which the expert-parallel load reads
+    as it reads text_config: marked there, the VLM loads through the rewritten converters, and the
+    mark is cleared."""
+    monkeypatch.setattr(
+        mu, "get_model_conversion_mapping", lambda model, **kw: list(_stock_expert_converters())
+    )
+    compression = (_ScaledCompressor, None, PACKED, "stub-format")
+    monkeypatch.setattr(onfly, "_read_compression", lambda cfg: compression)
+    llm = SimpleNamespace()
+    onfly.install_onfly_decompress(llm)
+    model = SimpleNamespace(config=SimpleNamespace(llm_config=copy.deepcopy(llm)))
+
+    out = mu.get_model_conversion_mapping(model)
+
+    assert [conv.source_patterns[0] for conv in out] == [
+        "mlp.experts.*.gate_proj.weight_packed$",
+        "mlp.experts.*.down_proj.weight_packed$",
+    ]
+    assert not onfly.is_marked(model.config)
+
+
+# ------------------------------------------------------------------------ reading the checkpoint
+
+
+@pytest.mark.parametrize(
+    ("config", "found"),
+    [
+        (
+            SimpleNamespace(
+                quantization_config={"at": "top"},
+                text_config=SimpleNamespace(quantization_config={"at": "text"}),
+            ),
+            {"at": "top"},
+        ),
+        (
+            SimpleNamespace(text_config=SimpleNamespace(quantization_config={"at": "text"})),
+            {"at": "text"},
+        ),
+        (
+            SimpleNamespace(llm_config=SimpleNamespace(quantization_config={"at": "llm"})),
+            {"at": "llm"},
+        ),
+        (
+            SimpleNamespace(quantization_config=SimpleNamespace(to_dict=lambda: {"at": "obj"})),
+            {"at": "obj"},
+        ),
+    ],
+    ids=["top-level-first", "text-config", "llm-config", "config-object"],
+)
+def test_the_quantization_config_is_found_at_the_top_level_then_on_the_text_config(config, found):
+    """A VLM may carry its compressed-tensors config at the top level or on its language-model
+    config (``text_config`` or ``llm_config``), as a config object or a dict; it is always returned
+    as a plain dict."""
+    assert onfly._find_quantization_config(config) == found
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+@pytest.mark.parametrize(
+    ("config", "error", "match"),
+    [
+        (SimpleNamespace(text_config=SimpleNamespace()), ValueError, "quantization_config"),
+        ({"quantization_config": {"format": "pack-quantized"}}, TypeError, "not a dict"),
+        (
+            SimpleNamespace(dtype=torch.float16, quantization_config={"format": "pack-quantized"}),
+            NotImplementedError,
+            "loads in torch.float16",
+        ),
+        (
+            SimpleNamespace(
+                text_config=SimpleNamespace(dtype=torch.float16),
+                quantization_config={"format": "pack-quantized"},
+            ),
+            NotImplementedError,
+            "loads in torch.float16",
+        ),
+    ],
+    ids=["nothing-to-decompress", "raw-dict", "not-bf16", "not-bf16-text-config"],
+)
+def test_a_config_with_nothing_to_decompress_leaves_the_loader_alone(
+    monkeypatch, config, error, match
+):
+    """By the time a bf16 checkpoint's caller sees that error, nothing may be patched yet -- and
+    compressed_tensors, which such a checkpoint never needs, need not be installed. Nor for a raw
+    dict, which cannot carry the mark its load would be found by, or for a load in a dtype other
+    than the bf16 the experts decompress to, at any config level (a backbone config derived from the
+    text config keeps its dtype): the loader would set them beside the rest uncast."""
+    monkeypatch.setitem(sys.modules, "compressed_tensors", None)
+    hooks = (
+        mu.get_model_conversion_mapping,
+        WeightConverter.add_tensor,
+        WeightConverter.materialize_tensors,
+        cml.spawn_materialize,
+    )
+
+    with pytest.raises(error, match=match):
+        onfly.install_onfly_decompress(config)
+
+    assert (
+        mu.get_model_conversion_mapping,
+        WeightConverter.add_tensor,
+        WeightConverter.materialize_tensors,
+        cml.spawn_materialize,
+    ) == hooks
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_a_transformers_below_the_floor_is_named_before_compressed_tensors(monkeypatch):
+    """On a transformers the decompression cannot run on, the first error names the release, not
+    the compressed-tensors package a user would install only to hit the release next."""
+    monkeypatch.setitem(sys.modules, "compressed_tensors", None)
+    monkeypatch.setattr(transformers, "__version__", "5.5.4")
+
+    with pytest.raises(RuntimeError, match=r"transformers >= 5\.8"):
+        onfly.install_onfly_decompress(
+            SimpleNamespace(quantization_config={"format": "pack-quantized"})
+        )
+
+
+EXPERTS = [f"model.layers.1.mlp.experts.{e}.gate_proj" for e in range(3)]
+EXPERT = EXPERTS[0]
+ATTENTION = "model.layers.1.self_attn.q_proj"
+SHARED_EXPERT = "model.layers.1.mlp.shared_experts.up_proj"  # a dense MLP, not the routed experts
+STORED_DTYPES = {  # as compressed-tensors stores them; every other tensor is bf16 here
+    "weight_packed": torch.int32,
+    "weight_shape": torch.int32,
+    "weight_global_scale": torch.float32,
+}
+
+
+@pytest.mark.parametrize(
+    ("stored_as", "names", "match"),
+    [
+        (
+            PACKED,
+            [f"{EXPERT}.{n}" for n in PACKED] + [f"{ATTENTION}.weight", f"{SHARED_EXPERT}.weight"],
+            None,
+        ),
+        (
+            ("weight", "weight_scale"),
+            [f"{EXPERT}.weight", f"{EXPERT}.weight_scale", f"{ATTENTION}.weight"],
+            None,
+        ),
+        (PACKED, [f"{p}.{n}" for p in (EXPERT, ATTENTION) for n in PACKED], "1 of them outside"),
+        (PACKED, [f"{EXPERT}.weight"] + [f"{ATTENTION}.{n}" for n in PACKED], "1 of them outside"),
+        (
+            PACKED,
+            [f"{EXPERT}.{n}" for n in PACKED] + ["model.layers.1.mlp.experts.1.gate_proj.weight"],
+            "1 expert weights uncompressed",
+        ),
+        (PACKED, [f"{ATTENTION}.weight", f"{SHARED_EXPERT}.weight"], "compresses 0 weights"),
+        (
+            (*PACKED, "weight_global_scale"),
+            [f"{EXPERT}.{n}" for n in (*PACKED, "weight_global_scale")],
+            r"store weight_global_scale \(F32\), which bf16 cannot hold exactly",
+        ),
+        (
+            PACKED,
+            [f"{EXPERT}.{n}" for n in PACKED] + [f"{EXPERTS[1]}.weight_scale"],
+            "stores weight_packed for some compressed weights but not for 1 others",
+        ),
+        (
+            PACKED,
+            [f"{EXPERT}.{n}" for n in (*PACKED, "weight_zero_point")],
+            "stores 1 tensors under its compressed weights that this compressed-tensors release",
+        ),
+        (PACKED, [f"{EXPERT}.{n}" for n in (*PACKED, "bias")], None),
+        (
+            PACKED,
+            [f"{EXPERTS[e]}.{n}" for e in (0, 1) for n in ("weight_packed", "weight_shape")]
+            + [f"{EXPERTS[e]}.weight_scale" for e in (1, 2)],
+            "stores weight_packed for some compressed weights but not for 1 others",
+        ),
+    ],
+    ids=[
+        "experts-alone",
+        "experts-alone-float-format",
+        "every-linear",
+        "all-but-the-experts",
+        "an-expert-left-uncompressed",
+        "nothing-compressed",
+        "an-fp32-scale",
+        "a-scale-alone",
+        "a-name-this-release-does-not-know",
+        "an-expert-bias-loads-apart",
+        "equal-counts-other-experts",
+    ],
+)
+def test_a_checkpoint_must_compress_its_experts_and_nothing_else(tmp_path, stored_as, names, match):
+    """The config cannot say which modules a class target such as Linear compressed, so the names
+    the checkpoint stores decide: anything else compressed, or an expert stored uncompressed, would
+    load newly initialized, without a quantizer to load it or a converter that reads it. A float
+    stored for the experts must be one bf16 holds exactly: the load may cast it to bf16 first. And
+    every compressed weight stores the same names, all of them names the installed compressed-tensors
+    gives the scheme: the load pairs them by position, not by expert, so a scale stored for other
+    experts in equal number would pair with the wrong weights, and a name it does not know would be
+    left out of the decompression."""
+    tensors = {
+        name: torch.zeros(1, dtype=STORED_DTYPES.get(name.rpartition(".")[2], torch.bfloat16))
+        for name in names
+    }
+    save_file(tensors, str(tmp_path / "model.safetensors"))
+    config = SimpleNamespace(text_config=SimpleNamespace())
+    onfly._mark(config.text_config, stored_as)
+
+    if match is None:
+        onfly.check_compressed_keys(config, tmp_path)
+    else:
+        with pytest.raises(NotImplementedError, match=match):
+            onfly.check_compressed_keys(config, tmp_path)
+
+
+def test_an_unmarked_config_has_no_stored_names_to_check(tmp_path):
+    """The check reads what the checkpoint stores by the names the mark holds, so a config the
+    decompressor never marked is refused, by name, rather than with a bare StopIteration."""
+    save_file({f"{EXPERT}.weight_packed": torch.zeros(1)}, str(tmp_path / "model.safetensors"))
+
+    with pytest.raises(ValueError, match="not marked for on-the-fly decompression"):
+        onfly.check_compressed_keys(SimpleNamespace(text_config=SimpleNamespace()), tmp_path)
+
+
+class _RemoteTextConfig(PretrainedConfig):
+    """Stands in for the remote-code DeepSeek-V3 config a VLM bundles as its ``text_config``."""
+
+    model_type = "remote_deepseek_v3"
+
+
+# -------------------------------------------------------- with the real compressed_tensors library
+
+
+def _int4(symmetric, targets):
+    """A compressed-tensors INT4, group-32 config group."""
+    weights = {"num_bits": 4, "type": "int", "symmetric": symmetric, "strategy": "group"}
+    return {"targets": targets, "weights": {**weights, "group_size": 32}}
+
+
+def _pack_quantized(**groups):
+    """A compressed-tensors ``quantization_config`` in the ``pack-quantized`` format."""
+    return {
+        "quant_method": "compressed-tensors",
+        "format": "pack-quantized",
+        "config_groups": groups,
+    }
+
+
+ATTN = ["re:.*self_attn.*"]
+EXPERTS = ["re:.*experts.*"]
+
+
+@pytest.mark.parametrize(
+    ("groups", "error", "match"),
+    [
+        (
+            {"attn": _int4(True, ATTN), "moe": _int4(False, EXPERTS)},
+            NotImplementedError,
+            "2 compressed-tensors config groups",
+        ),
+        (
+            {"attn": _int4(False, ATTN), "dense": _int4(True, ["Linear"])},
+            NotImplementedError,
+            "2 compressed-tensors config groups",
+        ),
+        ({"attn": _int4(True, ATTN)}, NotImplementedError, "not the experts"),
+        ({}, ValueError, "nothing to decompress"),
+    ],
+    ids=["an-expert-group-among-others", "no-expert-group", "one-group-not-the-experts", "none"],
+)
+def test_a_checkpoint_whose_experts_are_not_its_one_compressed_group_is_refused(
+    groups, error, match
+):
+    """Only the expert converters are rewritten: another group's weights would load newly
+    initialized, a group guessed for the experts would decompress them with the wrong scheme, and
+    experts no group targets are stored as plain weights the rewritten converters would not find."""
+    pytest.importorskip(
+        "compressed_tensors",
+        minversion="0.16",
+        reason="needs compressed-tensors (CI's unit session installs it)",
+    )
+    config = SimpleNamespace(quantization_config=_pack_quantized(**groups))
+
+    with pytest.raises(error, match=match):
+        onfly._read_compression(config)
+
+
+@pytest.mark.parametrize(
+    ("top", "group", "match"),
+    [("mixed-precision", "pack-quantized", None), ("fakequant", None, "no compressor")],
+    ids=["the-group-names-it", "none-registered"],
+)
+def test_the_format_is_the_one_group_s_own_and_must_have_a_compressor(top, group, match):
+    """compressed-tensors writes "mixed-precision" at the top level once a group names its own
+    format; the one group's format is the one to decompress with. A format nothing registers is
+    refused by this module, rather than failing inside the library's registry."""
+    pytest.importorskip(
+        "compressed_tensors",
+        minversion="0.16",
+        reason="needs compressed-tensors (CI's unit session installs it)",
+    )
+    experts = _int4(True, EXPERTS)
+    if group is not None:
+        experts["format"] = group
+    config = SimpleNamespace(quantization_config={**_pack_quantized(moe=experts), "format": top})
+
+    if match is None:
+        assert onfly._read_compression(config)[3] == group
+    else:
+        with pytest.raises(NotImplementedError, match=match):
+            onfly._read_compression(config)
+
+
+def test_a_compressed_tensors_before_per_scheme_compressors_is_refused_by_its_release(monkeypatch):
+    """compressed-tensors 0.13 states a compressor's parameter names as a property, which a call
+    by scheme fails on deep in the setup; such a release is refused first, naming the one needed."""
+    pytest.importorskip(
+        "compressed_tensors", reason="needs compressed-tensors (CI's unit session installs it)"
+    )
+    # Optional dependency: imported after importorskip so the test skips without it.
+    from compressed_tensors.compressors.base import BaseCompressor
+
+    class _PropertyForm:
+        """A compressor as compressed-tensors 0.13 registers one."""
+
+        @property
+        def compression_param_names(self):
+            return ("weight_packed", "weight_scale")
+
+    installed = importlib.metadata.version
+    monkeypatch.setattr(
+        importlib.metadata,
+        "version",
+        lambda name: "0.13.0" if name == "compressed-tensors" else installed(name),
+    )
+    monkeypatch.setattr(
+        BaseCompressor, "get_value_from_registry", staticmethod(lambda name: _PropertyForm)
+    )
+    config = SimpleNamespace(quantization_config=_pack_quantized(moe=_int4(True, EXPERTS)))
+
+    with pytest.raises(ImportError, match=r"compressed-tensors>=0\.16 is required.*==0\.13\.0"):
+        onfly._read_compression(config)
+
+
+# The compressions the published Kimi checkpoints use: INT4 (Kimi-K2.5) and MXFP4 (Kimi-K3), each
+# with the values its format holds exactly, and a scale it holds exactly.
+_INT4 = {"num_bits": 4, "type": "int", "symmetric": True, "strategy": "group", "group_size": 32}
+_FP4_E2M1 = torch.tensor([0, 0.5, 1, 1.5, 2, 3, 4, 6])
+_FORMATS = {
+    "pack-quantized": (_INT4, torch.arange(-8, 8) * 0.5, 0.5),
+    "mxfp4-pack-quantized": (
+        {**_INT4, "type": "float", "scale_dtype": "torch.uint8"},
+        torch.cat([_FP4_E2M1, -_FP4_E2M1[1:]]),
+        1.0,
+    ),
+}
+
+
+def _compressed_mixtral(path, fmt, sharded=False):
+    """Save a tiny Mixtral whose experts ``fmt`` compresses; return its config and dense experts.
+
+    Mixtral's checkpoint names its experts ``w1``/``w3``/``w2``, which the model fuses and the export
+    names back: names a save gets wrong would show. Each expert has a scale of its own, so experts
+    paired with another's scales would show too. ``sharded`` stores the experts' packed weights and
+    their scales in two shards in opposite expert order.
+    """
+    weights, values, scale = _FORMATS[fmt]
+    config = MixtralConfig(
+        vocab_size=64,
+        hidden_size=64,
+        intermediate_size=64,
+        num_hidden_layers=1,
+        num_attention_heads=4,
+        num_key_value_heads=2,
+        num_local_experts=2,
+        num_experts_per_tok=1,
+        tie_word_embeddings=False,
+    )
+    AutoModelForCausalLM.from_config(config).to(torch.bfloat16).save_pretrained(path)
+    quantization_config = {
+        "quant_method": "compressed-tensors",
+        "format": fmt,
+        "config_groups": {"experts": {"targets": ["Linear"], "weights": weights, "format": fmt}},
+    }
+    compressor, scheme, _, _ = onfly._read_compression(
+        SimpleNamespace(quantization_config=quantization_config)
+    )
+    # Copied out of the file, which is written over below.
+    state = {k: v.clone() for k, v in load_file(str(path / "model.safetensors")).items()}
+    gen = torch.Generator().manual_seed(0)
+    dense = {}
+    for key in sorted(k for k in list(state) if ".experts." in k):  # ...experts.<e>.w1.weight
+        rows, cols = state.pop(key).shape
+        factor = 2 ** int(key.split(".experts.")[1].split(".")[0])  # per expert, exact in both
+        dense[key] = values[torch.randint(0, len(values), (rows, cols), generator=gen)] * factor
+        scales = torch.full((rows, cols // 32), scale * factor)
+        stored = compressor.compress({"weight": dense[key], "weight_scale": scales}, scheme)
+        for name, tensor in stored.items():  # floats as bf16, as the Kimi checkpoints store them
+            state[key.removesuffix("weight") + name] = (
+                tensor.bfloat16() if tensor.is_floating_point() else tensor
+            )
+    if sharded:
+        # Expert 0's packed weights with expert 1's scales, and the other way around.
+        shards = {"model-00001-of-00002.safetensors": {}, "model-00002-of-00002.safetensors": {}}
+        for key, tensor in state.items():
+            expert = key.split(".experts.")[1][0] if ".experts." in key else "0"
+            first = (expert == "0") == ("weight_scale" not in key)
+            shards[list(shards)[0 if first else 1]][key] = tensor
+        (path / "model.safetensors").unlink()
+        for name, shard in shards.items():
+            save_file(shard, str(path / name), metadata={"format": "pt"})
+        weight_map = {key: name for name, shard in shards.items() for key in shard}
+        index = {"metadata": {}, "weight_map": weight_map}
+        (path / "model.safetensors.index.json").write_text(json.dumps(index))
+    else:
+        save_file(state, str(path / "model.safetensors"), metadata={"format": "pt"})
+    config_json = json.loads((path / "config.json").read_text())
+    config_json["quantization_config"] = quantization_config
+    (path / "config.json").write_text(json.dumps(config_json))
+    return AutoConfig.from_pretrained(path), dense
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+@pytest.mark.parametrize("sharded", [False, True], ids=["one-file", "sharded"])
+@pytest.mark.parametrize("fmt", sorted(_FORMATS))
+def test_a_compressed_checkpoint_loads_exactly_and_saves_under_its_own_names(
+    tmp_path, fmt, sharded
+):
+    """A real ``from_pretrained`` through the installed decompressor: the fused experts hold the
+    checkpoint's weights exactly, each expert's scaled by its own scales, also when the packed
+    weights and the scales sit in different shards in opposite expert order. The converters the load
+    kept reverse as the original fusion, so a save writes the experts back as the checkpoint's own
+    ``w1``/``w3``/``w2`` weights, decompressed, and the export names its expanded experts after
+    them."""
+    pytest.importorskip(
+        "compressed_tensors",
+        minversion="0.16",
+        reason="needs compressed-tensors (CI's unit session installs it)",
+    )
+    config, dense = _compressed_mixtral(tmp_path, fmt, sharded)
+    onfly.install_onfly_decompress(config)
+    del config.quantization_config  # as the expert-parallel load does: no quantizer competes
+
+    model = AutoModelForCausalLM.from_pretrained(tmp_path, config=config, dtype=torch.bfloat16)
+
+    experts, stored = model.model.layers[0].mlp.experts, "model.layers.0.block_sparse_moe.experts"
+    for e in range(2):
+        gate_up = torch.cat([dense[f"{stored}.{e}.w1.weight"], dense[f"{stored}.{e}.w3.weight"]])
+        assert torch.equal(experts.gate_up_proj[e], gate_up.bfloat16())
+        assert torch.equal(experts.down_proj[e], dense[f"{stored}.{e}.w2.weight"].bfloat16())
+    saved = cml.revert_weight_conversion(model, model.state_dict())
+    for key, weight in dense.items():
+        assert torch.equal(saved[key], weight.bfloat16())
+    export_name = build_reverse_name_mapper(model, tensor_keys=True)
+    assert [
+        export_name(f"model.layers.0.mlp.experts.1.{part}.weight")
+        for part in ("gate_proj", "up_proj", "down_proj")
+    ] == [f"{stored}.1.{hub}.weight" for hub in ("w1", "w3", "w2")]
+
+
+@pytest.mark.usefixtures("restore_loader_hooks")
+def test_a_load_the_decompressor_did_not_mark_casts_as_before(tmp_path):
+    """The hooks are process-wide, but only a decompressing load keeps an integer source integer:
+    a later load of a model no config of which was marked casts it to its float parameter."""
+    pytest.importorskip(
+        "compressed_tensors",
+        minversion="0.16",
+        reason="needs compressed-tensors (CI's unit session installs it)",
+    )
+    onfly.install_onfly_decompress(
+        SimpleNamespace(quantization_config=_pack_quantized(experts=_int4(True, ["Linear"])))
+    )
+    config = LlamaConfig(
+        vocab_size=64,
+        hidden_size=32,
+        intermediate_size=64,
+        num_hidden_layers=1,
+        num_attention_heads=4,
+        num_key_value_heads=2,
+        tie_word_embeddings=False,
+    )
+    AutoModelForCausalLM.from_config(config).save_pretrained(tmp_path)
+    # Copied out of the file, which is written over below.
+    state = {k: v.clone() for k, v in load_file(str(tmp_path / "model.safetensors")).items()}
+    state["model.norm.weight"] = torch.full_like(state["model.norm.weight"], 3, dtype=torch.int8)
+    save_file(state, str(tmp_path / "model.safetensors"), metadata={"format": "pt"})
+
+    model = AutoModelForCausalLM.from_pretrained(tmp_path, dtype=torch.bfloat16)
+
+    assert model.model.norm.weight.dtype == torch.bfloat16
+    assert torch.equal(model.model.norm.weight, torch.full_like(model.model.norm.weight, 3))
diff --git a/tests/unit/torch/utils/test_model_load_utils.py b/tests/unit/torch/utils/test_model_load_utils.py
index 70da8ef18ea3..3669a1ccf33d 100644
--- a/tests/unit/torch/utils/test_model_load_utils.py
+++ b/tests/unit/torch/utils/test_model_load_utils.py
@@ -75,7 +75,7 @@
 import modelopt.torch.quantization as mtq
 import modelopt.torch.utils.distributed as dist
 import modelopt.torch.utils.moe as moe
-from modelopt.torch.models import get_spec
+from modelopt.torch.models import _onfly_decompress, get_spec
 from modelopt.torch.quantization.plugins import accelerate as accelerate_plugin
 from modelopt.torch.quantization.plugins.accelerate import init_quantized_weights
 from modelopt.torch.quantization.utils.layerwise_calib import LayerActivationCollector
@@ -2778,7 +2778,7 @@ def from_pretrained(path, **kwargs):
         "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
     )
     monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
-    monkeypatch.setattr(model_load_utils, "record_unplaced_source_keys", lambda *args: [])
+    monkeypatch.setattr(model_load_utils, "record_unplaced_source_keys", lambda *args, **kwargs: [])
     config = SimpleNamespace(
         model_type=_NO_SPEC,
         base_model_ep_plan={"layers.*.mlp.experts.*.gate_up_proj": "grouped_gemm"},
@@ -2827,7 +2827,7 @@ def next_check(*args, **kwargs):
         "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
     )
     monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
-    monkeypatch.setattr(model_load_utils, "record_unplaced_source_keys", lambda *args: None)
+    monkeypatch.setattr(model_load_utils, "record_unplaced_source_keys", lambda *args, **kwargs: [])
     monkeypatch.setattr(model_load_utils, "_seed_parallel_states", next_check)
 
     class _Config(transformers.PretrainedConfig):
@@ -2889,7 +2889,7 @@ def next_check(*args, **kwargs):
         "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: mesh
     )
     monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
-    monkeypatch.setattr(model_load_utils, "record_unplaced_source_keys", lambda *args: None)
+    monkeypatch.setattr(model_load_utils, "record_unplaced_source_keys", lambda *args, **kwargs: [])
     monkeypatch.setattr(model_load_utils, "_seed_parallel_states", next_check)
     config = Qwen3MoeConfig(
         hidden_size=16,
@@ -2964,6 +2964,219 @@ def auto_class(hf_config, require_lm_head=True, trust_remote_code=False):
     assert classes == [True]
 
 
+def test_a_compressed_checkpoint_a_spec_decompresses_as_it_loads_passes(
+    ep_world_of_two, monkeypatch, tmp_path
+):
+    """A model spec that installs the on-the-fly decompressor (Kimi-K2.5's) marks the config; that
+    checkpoint passes the format check and loads without its quantization_config, at every level,
+    which would build transformers' own compressed-tensors quantizer to compete with the
+    decompressor. The same checkpoint unmarked is refused, and the caller's config keeps its own.
+    So is one that compresses more than its experts, by the names it stores, before it loads, and a
+    config whose class keeps quantization_config, which the load cannot clear; one whose class only
+    forwards its text config's loads."""
+
+    class _LoadedError(Exception):
+        """Stops the load at from_pretrained."""
+
+    loaded = {}
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, **kwargs):
+            loaded.update(kwargs)
+            raise _LoadedError
+
+    monkeypatch.setattr(
+        "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
+    )
+    monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
+    stored_as = {"weight_packed": torch.int32, "weight_scale": torch.bfloat16}
+    expert, attention = "model.layers.1.mlp.experts.0.up_proj", "model.layers.1.self_attn.o_proj"
+    for name, compressed in (("experts", [expert]), ("more", [expert, attention])):
+        stored = {
+            f"{m}.{n}": torch.ones(1, dtype=d) for m in compressed for n, d in stored_as.items()
+        }
+        (tmp_path / name).mkdir()
+        save_file(stored, str(tmp_path / name / "model.safetensors"))
+    quantization_config = {"quant_method": "compressed-tensors"}
+    text = SimpleNamespace(quantization_config=quantization_config, base_model_ep_plan=_EP_PLAN)
+    llm = SimpleNamespace(quantization_config=quantization_config)
+    config = SimpleNamespace(
+        model_type=_NO_SPEC,
+        quantization_config=quantization_config,
+        text_config=text,
+        llm_config=llm,
+    )
+    plan = DistributionPlan(2, ep_size=2)
+
+    with pytest.raises(NotImplementedError, match="cannot dequantize"):
+        load_and_prepare_ep(str(tmp_path / "experts"), plan, hf_config=config)
+    _onfly_decompress._mark(config, tuple(stored_as))
+    with pytest.raises(NotImplementedError, match="1 of them outside the experts"):
+        load_and_prepare_ep(str(tmp_path / "more"), plan, hf_config=config)
+    assert not loaded
+    with pytest.raises(_LoadedError):
+        load_and_prepare_ep(str(tmp_path / "experts"), plan, hf_config=config)
+
+    for level in (loaded["config"], loaded["config"].text_config, loaded["config"].llm_config):
+        assert not hasattr(level, "quantization_config")
+    assert config.quantization_config is text.quantization_config is quantization_config
+    assert llm.quantization_config is quantization_config
+
+    # A config class that keeps quantization_config as a class attribute cannot have it cleared.
+    typed_config = type("_Typed", (SimpleNamespace,), {"quantization_config": quantization_config})
+    typed = typed_config(
+        model_type=_NO_SPEC, text_config=SimpleNamespace(base_model_ep_plan=_EP_PLAN)
+    )
+    _onfly_decompress._mark(typed, tuple(stored_as))
+    loaded.clear()
+    with pytest.raises(NotImplementedError, match="keeps quantization_config off its instance"):
+        load_and_prepare_ep(str(tmp_path / "experts"), plan, hf_config=typed)
+    assert not loaded
+
+    # One whose class forwards the text config's, as Kimi-K2.5's copies it, loads once both are gone.
+    class _Forwarding(SimpleNamespace):
+        @property
+        def quantization_config(self):
+            return self.text_config.quantization_config
+
+    text = SimpleNamespace(quantization_config=quantization_config, base_model_ep_plan=_EP_PLAN)
+    forwarding = _Forwarding(model_type=_NO_SPEC, text_config=text)
+    _onfly_decompress._mark(forwarding, tuple(stored_as))
+    with pytest.raises(_LoadedError):
+        load_and_prepare_ep(str(tmp_path / "experts"), plan, hf_config=forwarding)
+
+
+@pytest.mark.parametrize(
+    ("compressed", "layers"),
+    [(False, 3), (True, 3), (True, 2)],
+    ids=["uncompressed", "compressed", "compressed-past-the-layers"],
+)
+def test_a_decompressed_load_refuses_compressed_tensors_its_model_does_not_place(
+    ep_world_of_two, monkeypatch, tmp_path, compressed, layers
+):
+    """The export copies the tensors the model has no parameter for as stored, and this load dropped
+    the quantization_config that reads a compressed one: an unplaced tensor under a stored name the
+    mark holds (an MTP layer's compressed experts) is refused, one under the plain ``weight`` name is
+    not. Compressed experts stored past the config's decoder layers are refused before anything
+    loads. Which keys go unplaced is stood in for here; a real decompressed load's are pinned in
+    test_unplaced_source_keys.py."""
+
+    class _PastError(Exception):
+        """Raised by the load's next check."""
+
+    def next_check(*args, **kwargs):
+        raise _PastError
+
+    model = model_with_decoder_at("model.")
+    loads = []
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, config, **kwargs):
+            loads.append(path)
+            model.config = copy.deepcopy(config)
+            return model
+
+    mtp = "model.layers.2.mlp.experts.0.up_proj"  # unplaced, as an MTP layer's are
+    stored_as = {"weight_packed": torch.int32, "weight_scale": torch.bfloat16}
+    names = list(stored_as) if compressed else ["weight"]
+    stored = {f"{m}.{n}": torch.ones(1, dtype=d) for m in (mtp,) for n, d in stored_as.items()}
+    save_file(stored, str(tmp_path / "model.safetensors"))
+    monkeypatch.setattr(
+        "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
+    )
+    monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
+    monkeypatch.setattr(
+        model_load_utils,
+        "unplaced_source_keys",
+        lambda *args, **kwargs: [f"{mtp}.{n}" for n in names],
+    )
+    monkeypatch.setattr(model_load_utils, "_seed_parallel_states", next_check)
+    config = SimpleNamespace(
+        model_type=_NO_SPEC,
+        num_experts=8,
+        num_hidden_layers=layers,
+        base_model_ep_plan=dict(_EP_PLAN),
+        quantization_config={"quant_method": "compressed-tensors"},
+    )
+    _onfly_decompress._mark(config, tuple(stored_as))
+
+    with pytest.raises(NotImplementedError if compressed else _PastError) as refused:
+        load_and_prepare_ep(str(tmp_path), DistributionPlan(2, ep_size=2), hf_config=config)
+    if layers == 2:
+        assert not loads
+        assert f"2 compressed tensors past the model's 2 decoder layers, such as {mtp}." in str(
+            refused.value
+        )
+    elif compressed:
+        assert f"2 compressed tensors the model has no parameter for, such as {mtp}" in str(
+            refused.value
+        )
+
+
+@pytest.mark.parametrize("compressed", [False, True], ids=["plain", "compressed"])
+def test_a_decompressed_load_refuses_compressed_tensors_in_a_file_it_never_opens(
+    ep_world_of_two, monkeypatch, tmp_path, compressed
+):
+    """The export copies the safetensors files the load never opens as stored (GLM-4.7 keeps its MTP
+    head in a standalone ``mtp.safetensors``), and this load drops the quantization_config a
+    compressed tensor there needs: such a file is refused before anything loads. A plain one is
+    carried, and the load goes on."""
+
+    class _NextCheckError(Exception):
+        """Raised by the load's next check."""
+
+    def next_check(*args, **kwargs):
+        raise _NextCheckError
+
+    model = model_with_decoder_at("model.")
+    loads = []
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, config, **kwargs):
+            loads.append(path)
+            model.config = copy.deepcopy(config)
+            return model
+
+    stored_as = {"weight_packed": torch.int32, "weight_scale": torch.bfloat16}
+    experts = "model.layers.0.mlp.experts.0.up_proj"
+    save_file(
+        {f"{experts}.{n}": torch.ones(1, dtype=d) for n, d in stored_as.items()},
+        str(tmp_path / "model.safetensors"),
+    )
+    mtp = "model.layers.1.mlp.experts.0.up_proj"
+    off_index = stored_as if compressed else {"weight": torch.bfloat16}
+    save_file(
+        {f"{mtp}.{n}": torch.ones(1, dtype=d) for n, d in off_index.items()},
+        str(tmp_path / "mtp.safetensors"),
+    )
+    monkeypatch.setattr(
+        "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
+    )
+    monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
+    monkeypatch.setattr(model_load_utils, "unplaced_source_keys", lambda *args, **kwargs: [])
+    monkeypatch.setattr(model_load_utils, "_seed_parallel_states", next_check)
+    config = SimpleNamespace(
+        model_type=_NO_SPEC,
+        num_experts=8,
+        num_hidden_layers=2,
+        base_model_ep_plan=dict(_EP_PLAN),
+        quantization_config={"quant_method": "compressed-tensors"},
+    )
+    _onfly_decompress._mark(config, tuple(stored_as))
+
+    with pytest.raises(NotImplementedError if compressed else _NextCheckError) as refused:
+        load_and_prepare_ep(str(tmp_path), DistributionPlan(2, ep_size=2), hf_config=config)
+    assert bool(loads) != compressed
+    if compressed:
+        assert (
+            f"2 compressed tensors in files the load never opens, such as mtp.safetensors's {mtp}."
+            in str(refused.value)
+        )
+
+
 # ----------------------------------------------------------------------- _seed_parallel_states
 
 
diff --git a/tests/unit/torch/utils/test_unplaced_source_keys.py b/tests/unit/torch/utils/test_unplaced_source_keys.py
index 46a5e7ef6347..1882b7c5cc88 100644
--- a/tests/unit/torch/utils/test_unplaced_source_keys.py
+++ b/tests/unit/torch/utils/test_unplaced_source_keys.py
@@ -25,6 +25,7 @@
 
 import pytest
 import torch
+from packaging.version import Version
 from safetensors.torch import load_file, save_file
 
 pytest.importorskip("accelerate")
@@ -40,7 +41,9 @@
     LlamaForCausalLM,
     Mxfp4Config,
 )
+from transformers import __version__ as transformers_version
 
+from modelopt.torch.models import _onfly_decompress
 from modelopt.torch.utils.distributed import DistributionPlan, distribution_plan
 from modelopt.torch.utils.plugins import model_load_utils
 from modelopt.torch.utils.plugins.model_load_utils import (
@@ -48,6 +51,32 @@
     parallel_load_and_prepare_fsdp2,
 )
 
+
+def test_a_two_argument_unplaced_source_keys_still_records_the_keys(monkeypatch, tmp_path):
+    """The map a load already read reaches ``unplaced_source_keys`` by keyword, and only when there is
+    one, so a caller that replaced the function with its two-argument form still records keys."""
+    seen = []
+
+    def two_arguments(model, ckpt_path):
+        seen.append(ckpt_path)
+        return ["mtp.fc.weight"]
+
+    monkeypatch.setattr(model_load_utils, "unplaced_source_keys", two_arguments)
+    model = LlamaForCausalLM(_tiny_llama_config())
+
+    assert model_load_utils.record_unplaced_source_keys(model, str(tmp_path)) == ["mtp.fc.weight"]
+    assert seen == [str(tmp_path)]
+    read = {"mtp.fc.weight": "model.safetensors"}
+    monkeypatch.setattr(
+        model_load_utils,
+        "unplaced_source_keys",
+        lambda model, ckpt_path, tensor_files=None: sorted(tensor_files),
+    )
+    assert model_load_utils.record_unplaced_source_keys(
+        model, str(tmp_path), tensor_files=read
+    ) == ["mtp.fc.weight"]
+
+
 # A layer past ``num_hidden_layers``: genuinely stray, so it must always be recorded.
 STRAY = "model.layers.1.input_layernorm.weight"
 
@@ -391,3 +420,80 @@ def test_mxfp4_expert_blocks_folded_in_at_load_are_placed(tmp_path):
     model = _load(tmp_path, quantization_config=Mxfp4Config(dequantize=True), dtype=torch.bfloat16)
 
     assert model._modelopt_unplaced_source_keys == [STRAY]
+
+
+def test_compressed_experts_decompressed_at_load_are_placed(tmp_path, monkeypatch):
+    """The on-the-fly decompressor feeds each expert's stored ``weight_packed``/``weight_scale``/
+    ``weight_shape`` into the fused weights. A built layer's are placed, through the converters
+    ``from_pretrained`` recorded (the stock mapping places none of them), so the expert-parallel load
+    passes them; only a layer the model does not build (an MTP layer) keeps its own, which that load
+    refuses to carry."""
+    pytest.importorskip(
+        "compressed_tensors",
+        minversion="0.16",
+        reason="needs compressed-tensors (CI's unit session installs it)",
+    )
+    if Version(transformers_version) < Version("5.8"):
+        pytest.skip("the on-the-fly decompression needs transformers >= 5.8")
+    # Imported once the release is known to have them: transformers 4 has no core_model_loading.
+    import transformers.core_model_loading as cml
+    import transformers.modeling_utils as mu
+
+    # The decompressor's hooks are process-wide: undone after the test.
+    for owner, name in ((cml, "spawn_materialize"), (mu, "get_model_conversion_mapping")):
+        monkeypatch.setattr(owner, name, getattr(owner, name))
+    monkeypatch.setattr(_onfly_decompress, "_LOAD", {"decompressing": False})
+    hidden, inter, experts = 64, 32, 2
+    config = AutoConfig.for_model(
+        "qwen3_moe",
+        vocab_size=64,
+        hidden_size=hidden,
+        intermediate_size=64,
+        moe_intermediate_size=inter,
+        num_hidden_layers=1,
+        num_attention_heads=4,
+        num_key_value_heads=2,
+        head_dim=16,
+        num_experts=experts,
+        num_experts_per_tok=1,
+        tie_word_embeddings=False,
+        architectures=["Qwen3MoeForCausalLM"],
+    )
+    group = {"num_bits": 4, "type": "int", "symmetric": True, "strategy": "group", "group_size": 32}
+    config.quantization_config = {
+        "quant_method": "compressed-tensors",
+        "format": "pack-quantized",
+        "config_groups": {"experts": {"targets": ["re:.*experts.*"], "weights": group}},
+    }
+    compressor, scheme, stored_as, _ = _onfly_decompress._read_compression(config)
+    shapes = {
+        "gate_proj": (inter, hidden),
+        "up_proj": (inter, hidden),
+        "down_proj": (hidden, inter),
+    }
+    state, mtp = {}, []
+    for layer in (0, 1):  # layer 1 is past num_hidden_layers: an MTP layer, say
+        for e in range(experts):
+            for proj, (rows, cols) in shapes.items():
+                weight = {
+                    "weight": torch.ones(rows, cols),
+                    "weight_scale": torch.ones(rows, cols // 32),
+                }
+                stem = f"model.layers.{layer}.mlp.experts.{e}.{proj}."
+                for name, tensor in compressor.compress(weight, scheme).items():
+                    state[stem + name] = tensor
+                    if layer:
+                        mtp.append(stem + name)
+    for name, tensor in AutoModelForCausalLM.from_config(config).state_dict().items():
+        if ".experts." not in name:
+            state[name] = tensor.to(torch.bfloat16)
+    _save(tmp_path, config, state, config.quantization_config)
+    _onfly_decompress.install_onfly_decompress(config)
+    del config.quantization_config  # as the expert-parallel load does once the hooks are in
+
+    model = AutoModelForCausalLM.from_pretrained(str(tmp_path), config=config, dtype=torch.bfloat16)
+    if getattr(model, "_weight_conversions", None) is None:
+        pytest.skip("this transformers does not record the conversions from_pretrained ran")
+
+    assert sorted(stored_as) == ["weight_packed", "weight_scale", "weight_shape"]
+    assert model_load_utils.unplaced_source_keys(model, str(tmp_path)) == sorted(mtp)