diff --git a/modelopt/torch/models/_onfly_decompress.py b/modelopt/torch/models/_onfly_decompress.py
new file mode 100644
index 00000000000..1200032a42e
--- /dev/null
+++ b/modelopt/torch/models/_onfly_decompress.py
@@ -0,0 +1,490 @@
+# 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 _SUB_CONFIG_ATTRS, require_transformers
+
+
+def to_native_backbone_config(
+    vlm_config,
+    model_type: str = "deepseek_v3",
+    architectures: "tuple[str, ...]" = ("DeepseekV3ForCausalLM",),
+):
+    """The transformers-native backbone config for a VLM config's bundled ``text_config``.
+
+    Drops the remote ``auto_map`` and the stale ``quantization_config``, and names the native model
+    type and class (hf_ptq's VLM detection reads ``architectures``). The decompression mark carries
+    over. Defaults target the Kimi DeepSeek-V3 backbone.
+    """
+    from transformers.models.auto.configuration_auto import CONFIG_MAPPING
+
+    text_config = getattr(vlm_config, "text_config", None)
+    if text_config is None:
+        raise ValueError("to_native_backbone_config: config has no .text_config (not a VLM?)")
+    if model_type not in CONFIG_MAPPING:
+        raise ValueError(f"to_native_backbone_config: unknown native model_type {model_type!r}")
+    d = text_config.to_dict() if hasattr(text_config, "to_dict") else dict(text_config)
+    for key in ("auto_map", "quantization_config", "_name_or_path"):
+        d.pop(key, None)
+    d["model_type"] = model_type
+    d["architectures"] = list(architectures)
+    native = CONFIG_MAPPING[model_type].from_dict(d)
+    stored_as = getattr(vlm_config, _MARK, None) or getattr(text_config, _MARK, None)
+    if stored_as is not None:
+        # A backbone loaded from this config on its own decompresses as the VLM would.
+        _mark(native, stored_as)
+    return native
+
+
+def _config_levels(hf_config) -> list:
+    """``hf_config`` and the nested language-model configs the expert-parallel load reads."""
+    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 _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
+    try:
+        from compressed_tensors.compressors.base import BaseCompressor
+        from compressed_tensors.quantization import QuantizationConfig
+    except ImportError as e:
+        # No pip extra of ModelOpt installs it: only a compressed checkpoint needs it.
+        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
+    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.
+    """
+    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."""
+    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.
+    """
+    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 (to_native_backbone_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 29282c5f4cf..0bc4d9d313c 100644
--- a/modelopt/torch/utils/moe.py
+++ b/modelopt/torch/utils/moe.py
@@ -55,6 +55,11 @@ def require_transformers(min_version: str, feature: str) -> None:
         )
 
 
+# The nested configs a VLM's expert-parallel plan and expert count are read from. A spec attaching
+# its plan anywhere else would be read as having none, and the MoE planned as a dense model.
+_SUB_CONFIG_ATTRS = ("text_config", "llm_config")
+
+
 def _ep_config(hf_config):
     """Return the config that carries the EP plan / expert count.
 
diff --git a/noxfile.py b/noxfile.py
index 8528f7136e0..427a4d6dbad 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/unit/torch/models/test_onfly_decompress.py b/tests/unit/torch/models/test_onfly_decompress.py
new file mode 100644
index 00000000000..7271a21052c
--- /dev/null
+++ b/tests/unit/torch/models/test_onfly_decompress.py
@@ -0,0 +1,1041 @@
+# 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"])
+@pytest.mark.parametrize("loaded", ["vlm", "backbone"])
+def test_a_vlm_marked_at_either_level_loads_through_the_rewritten_converters(
+    monkeypatch, marked, loaded
+):
+    """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. Whether the load builds the VLM, its text config replaced
+    by the native backbone's, or that backbone on its own, its 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)
+    vlm.text_config = onfly.to_native_backbone_config(vlm)
+    model = SimpleNamespace(config=copy.deepcopy(vlm if loaded == "vlm" else vlm.text_config))
+
+    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"
+
+
+@pytest.mark.parametrize("as_dict", [False, True], ids=["config-object", "raw-dict"])
+def test_the_native_backbone_config_is_a_clean_stock_deepseek_v3(as_dict):
+    """The bundled text config becomes transformers' own ``DeepseekV3Config`` with the same
+    architecture, minus ``auto_map`` (which routes back to the remote code) and the stale
+    ``quantization_config`` (which would build a competing quantizer). ``architectures`` names the
+    native class: hf_ptq's VLM detection iterates it."""
+    fields = {
+        "hidden_size": 64,
+        "num_hidden_layers": 3,
+        "n_routed_experts": 8,
+        "moe_intermediate_size": 16,
+        "auto_map": {"AutoConfig": "configuration_deepseek.DeepseekV3Config"},
+        "quantization_config": {"quant_method": "compressed-tensors"},
+    }
+    if as_dict:
+        text = {**fields, "model_type": "remote_deepseek_v3"}
+    else:
+        text = _RemoteTextConfig(**fields)
+
+    native = onfly.to_native_backbone_config(SimpleNamespace(text_config=text))
+
+    assert type(native) is DeepseekV3Config
+    assert native.model_type == "deepseek_v3"
+    assert native.architectures == ["DeepseekV3ForCausalLM"]
+    assert (native.hidden_size, native.num_hidden_layers) == (64, 3)
+    assert (native.n_routed_experts, native.moe_intermediate_size) == (8, 16)
+    assert getattr(native, "auto_map", None) is None
+    assert getattr(native, "quantization_config", None) is None
+
+
+def test_the_native_backbone_config_needs_a_text_config_and_a_known_model_type():
+    with pytest.raises(ValueError, match="text_config"):
+        onfly.to_native_backbone_config(SimpleNamespace())
+    with pytest.raises(ValueError, match="unknown native model_type"):
+        onfly.to_native_backbone_config(SimpleNamespace(text_config={}), model_type="not_a_model")
+
+
+# -------------------------------------------------------- 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)"
+    )
+    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):
+    """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.
+    """
+    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)
+    )
+    state = load_file(str(path / "model.safetensors"))
+    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
+        dense[key] = values[torch.randint(0, len(values), (rows, cols), generator=gen)]
+        scales = torch.full((rows, cols // 32), scale)
+        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
+            )
+    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("fmt", sorted(_FORMATS))
+def test_a_compressed_checkpoint_loads_exactly_and_saves_under_its_own_names(tmp_path, fmt):
+    """A real ``from_pretrained`` through the installed decompressor: the fused experts hold the
+    checkpoint's weights exactly. 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)
+    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)
+    state = load_file(str(tmp_path / "model.safetensors"))
+    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))