diff --git a/modelopt/torch/quantization/plugins/huggingface.py b/modelopt/torch/quantization/plugins/huggingface.py
index 2223608c283..83cd2ba117c 100644
--- a/modelopt/torch/quantization/plugins/huggingface.py
+++ b/modelopt/torch/quantization/plugins/huggingface.py
@@ -41,7 +41,7 @@
 from modelopt.torch.kernels.quantization.gemm import IS_AVAILABLE as IS_TRITON_AVAILABLE
 from modelopt.torch.opt.dynamic import DynamicModule
 from modelopt.torch.utils import warn_rank_0
-from modelopt.torch.utils.distributed import ParallelState
+from modelopt.torch.utils.distributed import ParallelState, recorded_distribution_plan
 
 from ..algorithms import AutoQuantizeGradientSearcher
 from ..conversion import register
@@ -596,7 +596,9 @@ def convert_hf_parallel_linears_on_the_fly(model):
         elif HFRowParallelLinear.is_compatible(module, tp_mesh):
             HFRowParallelLinear.convert(module)
             converted = True
-    if tp_mesh is not None and not converted:
+    # An expert-parallel load keeps its experts' mesh there too, and shards no Linear by design.
+    plan = recorded_distribution_plan(model)
+    if tp_mesh is not None and not converted and not (plan is not None and plan.is_expert_parallel):
         warnings.warn(
             "Found a transformers tensor-parallel mesh but no TP-sharded linear layer; they will be"
             " quantized as unsharded layers."
diff --git a/modelopt/torch/utils/moe.py b/modelopt/torch/utils/moe.py
index 2aff08e6891..29282c5f4cf 100644
--- a/modelopt/torch/utils/moe.py
+++ b/modelopt/torch/utils/moe.py
@@ -15,25 +15,25 @@
 
 """HuggingFace-coupled MoE expert-parallel (EP) helpers.
 
-Finding a model's routed experts, choosing the EP degree they are sharded over, and the DP=EP
-all-to-all expert transport (``torch_a2a``).
+Finding a model's routed experts, choosing the EP degree they are sharded over, the DP=EP
+all-to-all expert transport (``torch_a2a``). The expert-parallel model loader is
+:func:`~modelopt.torch.utils.plugins.model_load_utils.load_and_prepare_ep`.
 
 The all-to-all transport is plain ``torch.distributed.all_to_all_single`` (NCCL) wrapped around
-transformers' quantizable ``_grouped_linear``. ``register_torch_a2a()`` registers it into
-transformers' ``ExpertsInterface`` lazily -- never at import time, so importing this module has no
-transformers-``ExpertsInterface`` side effect.
+transformers' quantizable ``_grouped_linear``. ``register_torch_a2a()`` registers
+it into transformers' ``ExpertsInterface`` (lazily, from ``load_and_prepare_ep``'s a2a path -- never at
+import time, so importing this module has no transformers-``ExpertsInterface`` side effect).
 """
 
 import warnings
-from typing import TYPE_CHECKING
 
 import torch
 import torch.distributed as dist
+import torch.nn as nn
 from packaging.version import parse
 from torch.distributed.nn.functional import all_to_all_single as autograd_all_to_all_single
 
-if TYPE_CHECKING:
-    from torch import nn
+from .distributed import distribution_plan
 
 
 def require_transformers(min_version: str, feature: str) -> None:
@@ -406,7 +406,9 @@ def register_torch_a2a():
     These two styles are functionally transformers' ``megamoe_router`` / ``megamoe_experts`` (added
     in 5.10.4 for DeepGEMM Mega-MoE), reimplemented here so the transport works below 5.10.4 too.
     The base classes (``RouterParallel``, ``TensorParallelLayer``) are stock wherever the EP infra
-    exists (transformers >= 5.6); ``grouped_gemm`` (expert weight sharding) is stock too.
+    exists (transformers >= 5.6); ``grouped_gemm`` (expert weight sharding) is stock too. A model
+    builds with ``torch_a2a`` only from 5.7, the floor ``load_and_prepare_ep`` enforces: 5.6 accepts
+    only the experts implementations it ships.
     """
     global _TORCH_A2A_REGISTERED
     if _TORCH_A2A_REGISTERED:
@@ -473,3 +475,23 @@ def shard_tensor(self, param, tensor_idx=None, device=None, dtype=None):
     ALL_PARALLEL_STYLES.register("torch_a2a_router", TorchA2ARouter())
     ALL_PARALLEL_STYLES.register("torch_a2a_experts", TorchA2AExperts())
     _TORCH_A2A_REGISTERED = True
+
+
+def expert_parallel_mesh(model: nn.Module):
+    """The mesh ``model``'s routed experts are sharded over, or ``None`` if its plan shards none.
+
+    :func:`~modelopt.torch.utils.plugins.model_load_utils.load_and_prepare_ep` loads the model over
+    its expert-parallel submesh, and keeps that submesh on the model as ``_device_mesh``.
+    Calibration takes the mesh to re-install the all-to-all forwards
+    (:func:`~modelopt.torch.utils.distributed.distributed_forward_loop`).
+    """
+    if not distribution_plan(model).is_expert_parallel:
+        return None
+    mesh = getattr(model, "_device_mesh", None)
+    if mesh is None:
+        raise RuntimeError(
+            f"{type(model).__name__} is planned expert-parallel but carries no device mesh. Pass "
+            "the model load_and_prepare_ep returned, not a submodule of it: the loader keeps the "
+            "mesh it shards the experts over on that model alone."
+        )
+    return mesh
diff --git a/modelopt/torch/utils/plugins/model_load_utils.py b/modelopt/torch/utils/plugins/model_load_utils.py
index 094a2d79504..705e21d5a93 100644
--- a/modelopt/torch/utils/plugins/model_load_utils.py
+++ b/modelopt/torch/utils/plugins/model_load_utils.py
@@ -15,7 +15,9 @@
 
 """HuggingFace model loading: :func:`load_pretrained`, by a plan to distribute the model."""
 
+import contextlib
 import copy
+import dataclasses
 import logging
 import os
 import re
@@ -28,10 +30,12 @@
 import torch
 import torch.distributed.checkpoint as dcp
 import torch.nn as nn
+import transformers.distributed as hf_distributed
 from huggingface_hub import snapshot_download
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
+from torch.distributed import device_mesh
 from torch.distributed.checkpoint import HuggingFaceStorageReader
 from torch.distributed.checkpoint.state_dict import StateDictOptions, set_model_state_dict
 from torch.distributed.fsdp import FSDPModule
@@ -49,9 +53,10 @@
 from transformers.core_model_loading import WeightConverter, dot_natural_key, rename_source_key
 
 from modelopt.torch.utils import distributed as dist
-from modelopt.torch.utils import env
+from modelopt.torch.utils import env, moe
 from modelopt.torch.utils.distributed import (
     DistributionPlan,
+    ParallelState,
     barrier,
     broadcast_state_dict,
     is_initialized,
@@ -60,11 +65,13 @@
 from modelopt.torch.utils.fsdp2_utils import fsdp2_wrap
 from modelopt.torch.utils.logging import print_rank_0, warn_rank_0
 from modelopt.torch.utils.moe import (
+    _EXPERTS_STYLES,
     _choose_ep_degree,
     _ep_config,
     _num_experts_from_config,
     _routed_expert_keys,
     _routed_expert_modules,
+    register_torch_a2a,
 )
 from modelopt.torch.utils.plugins.hf_checkpoint_utils import (
     checkpoint_tensor_files,
@@ -73,6 +80,18 @@
     read_safetensors_subset,
 )
 
+__all__ = [
+    "build_meta_causal_lm",
+    "fsdp2_wrap",
+    "is_fsdp2_text_only_vlm",
+    "load_and_prepare_ep",
+    "load_pretrained",
+    "parallel_load_and_prepare_fsdp2",
+    "record_unplaced_source_keys",
+    "resolve_distribution_plan",
+    "unplaced_source_keys",
+]
+
 logger = logging.getLogger(__name__)
 
 
@@ -1583,3 +1602,368 @@ def checkpoint_read() -> bool:
     if plan != DistributionPlan():
         print_rank_0(f"Distribution: {plan!r}")
     return plan, hf_config
+
+
+def _seed_parallel_states(model: nn.Module, expert_names, dense_state, expert_state) -> None:
+    """Seed every module's ``_parallel_state`` for calibration's amax sync.
+
+    The routed-expert containers named in ``expert_names`` and everything under them get
+    ``expert_state``; every other module gets ``dense_state``. Seeding the plain modules works because
+    quantization keeps a ``_parallel_state`` that is already present when it converts a module -- the same
+    way Megatron's expert MLPs seed theirs.
+    """
+    prefixes = tuple(name + "." for name in expert_names)
+    for name, module in model.named_modules():
+        module._parallel_state = (
+            expert_state if prefixes and (name + ".").startswith(prefixes) else dense_state
+        )
+
+
+# The checkpoint formats load_and_prepare_ep dequantizes as it loads, beside unquantized ones (None).
+# Not MXFP4: transformers dequantizes it into GPT-OSS's transposed expert layout, whatever the model's.
+_EP_SOURCE_FORMATS: frozenset[str | None] = frozenset({None})
+
+
+def _ckpt_quant_method(cfg):
+    """Return the checkpoint's native quantization method, or ``None``.
+
+    Return the checkpoint's native quantization method (e.g. ``"fp8"`` for block-fp8 DeepSeek-V3),
+    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)):
+        qc = getattr(c, "quantization_config", None)
+        m = qc.get("quant_method") if isinstance(qc, dict) else getattr(qc, "quant_method", None)
+        if m:
+            # transformers' config objects carry a QuantizationMethod enum, whose str() is not its value.
+            return str(getattr(m, "value", m))
+    return None
+
+
+def _transformers_mesh_builder():
+    """The module whose ``initialize_distributed_mesh`` ``from_pretrained`` calls, or ``None``.
+
+    Newer transformers drops ``from_pretrained``'s ``device_mesh`` once ``distributed_config`` is
+    given: it builds its own world-sized ``tp`` mesh from the config's sizes. That shards the experts
+    over every rank instead of this rank's EP group whenever the EP degree is below the world size.
+    Older releases honor ``device_mesh`` and have no such builder.
+    """
+    try:
+        import transformers.distributed.mixin as mixin
+    except ImportError:
+        return None
+    return mixin if hasattr(mixin, "initialize_distributed_mesh") else None
+
+
+@contextlib.contextmanager
+def _experts_sharded_over(ep_mesh):
+    """Make ``from_pretrained`` shard the experts over ``ep_mesh`` on releases that build their own mesh."""
+    mixin = _transformers_mesh_builder()
+    if mixin is None:
+        yield
+        return
+
+    def _ep_mesh(*args, **kwargs):
+        index = torch.cuda.current_device() if ep_mesh.device_type == "cuda" else None
+        return torch.device(ep_mesh.device_type, index), ep_mesh
+
+    original = mixin.initialize_distributed_mesh
+    mixin.initialize_distributed_mesh = _ep_mesh
+    try:
+        yield
+    finally:
+        mixin.initialize_distributed_mesh = original
+
+
+def load_and_prepare_ep(
+    ckpt_path: str,
+    plan: DistributionPlan,
+    trust_remote_code: bool = False,
+    attn_implementation: str | None = None,
+    freeze: bool = True,
+    hf_config=None,
+    require_lm_head: bool = True,
+) -> nn.Module:
+    """Load a MoE causal LM expert-parallel for PTQ: each rank holds its share of the routed experts.
+
+    Collective: every rank of the process group calls it with the same ``plan``, an expert-parallel
+    :class:`~modelopt.torch.utils.distributed.DistributionPlan` for the whole group. The caller sets
+    the group up and binds each rank to its own GPU first, as
+    :func:`~modelopt.torch.utils.distributed.distributed_run` does. Each rank holds
+    ``num_experts / ep_size`` routed experts and a replica of the dense part, and tokens reach their
+    experts' ranks by all-to-all, so every rank must calibrate on data of its own.
+
+    It loads unquantized checkpoints, on transformers 5.7 or later. A refusal on any rank -- of its
+    transformers, the config, the checkpoint, its GPU, or in the load -- raises on every rank. In
+    the load that rests on ``from_pretrained`` running no collective, as transformers'
+    expert-parallel load does not on 5.7 through 5.18.
+
+    The model builds with its LM head where an auto class builds one; ``require_lm_head=False``
+    builds a VLM without it. Without ``hf_config`` every rank reads the checkpoint's config itself;
+    on many ranks, pass it. The caller records the plan on the model
+    (:func:`~modelopt.torch.utils.distributed.set_distribution_plan`), as :func:`load_pretrained`
+    does; :func:`~modelopt.torch.utils.moe.expert_parallel_mesh` then finds its mesh.
+    """
+    if not plan.is_expert_parallel:
+        raise ValueError(f"{plan!r} shards no routed experts, which is what this load is for.")
+    if plan.is_fsdp_dense:
+        raise NotImplementedError(f"{plan!r}: this load replicates the dense part on every rank.")
+    # Bringing the group up is the launcher's job; this load needs it live and device-bound.
+    if not dist.is_initialized():
+        raise RuntimeError(
+            "load_and_prepare_ep requires a live, device-bound process group; set it up first, e.g. "
+            "with modelopt.torch.utils.distributed.distributed_run() under torchrun."
+        )
+    if plan.world_size != dist.size():
+        raise ValueError(
+            f"{plan!r} is for {plan.world_size} ranks, but this run has {dist.size()}."
+        )
+    world_size, ep_degree, edp = plan.world_size, plan.ep_size, plan.edp
+    # The ep_plan styles this load keeps: grouped_gemm, and those it swaps for its torch_a2a styles.
+    # The checks and the swapped plan both read it.
+    swap = {
+        "ep_router": "torch_a2a_router",
+        "moe_tp_experts": "torch_a2a_experts",
+        "nemotron_experts_ep": "torch_a2a_experts",
+    }
+    kept_styles = {"grouped_gemm", *swap}
+    # Each rank checks its own transformers, the config it reads or was handed, and the plan against
+    # it: one refusing fails every rank here, before the checkpoint's fetch and the meshes.
+    with _refused_together("the expert-parallel load's checks", alone=False):
+        # 5.7 is the first release that builds a model with the torch_a2a experts.
+        moe.require_transformers(
+            "5.7", "ModelOpt DP=EP expert-parallel PTQ (transformers MoE EP framework)"
+        )
+        if hf_config is None:
+            config_kwargs: dict[str, Any] = {"trust_remote_code": trust_remote_code}
+            if attn_implementation is not None:
+                config_kwargs["attn_implementation"] = attn_implementation
+            hf_config = AutoConfig.from_pretrained(ckpt_path, **config_kwargs)
+        quant_method = _ckpt_quant_method(hf_config)
+        if quant_method not in _EP_SOURCE_FORMATS:
+            # 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" + (
+                f" and dequantizes {dequantized}" if dequantized else ""
+            )
+            raise NotImplementedError(
+                f"The expert-parallel load cannot dequantize a {quant_method!r} checkpoint: it loads "
+                f"{loads}. Start from an unquantized copy of the model."
+            )
+        # The swapped plan is this load's own: the caller's config keeps the plan its checkpoint declares.
+        hf_config = copy.deepcopy(hf_config)
+        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:
+            raise ValueError("expert-parallel PTQ requires the model to declare base_model_ep_plan")
+        if not _routed_expert_keys(ep_plan):
+            raise ValueError(
+                f"base_model_ep_plan={ep_plan} styles no routed experts ({', '.join(_EXPERTS_STYLES)} "
+                "or grouped_gemm), so this load would leave them whole on every rank."
+            )
+        # Only grouped_gemm entries shard an expert's weights here: the torch_a2a styles the
+        # containers take for the load shard nothing themselves (any other style is refused below).
+        unsharded = sorted(
+            key
+            for key in _routed_expert_keys(ep_plan)
+            if not any(
+                style == "grouped_gemm" and name.startswith(key + ".")
+                for name, style in ep_plan.items()
+            )
+        )
+        if unsharded:
+            raise ValueError(
+                f"base_model_ep_plan={ep_plan} shards no weights of {', '.join(unsharded)} through "
+                "grouped_gemm, the only expert-weight style this load shards by, so every rank would "
+                "hold every expert."
+            )
+        # The load drops any other style an expert weight takes, so that weight would load whole
+        # beside its sharded siblings, and the grouped forward would pair slices of different experts.
+        containers = _routed_expert_keys(ep_plan)
+        dropped = sorted(
+            name
+            for name, style in ep_plan.items()
+            if style not in kept_styles
+            and name not in containers
+            and any(name.startswith(key + ".") for key in containers)
+        )
+        if dropped:
+            raise ValueError(
+                f"base_model_ep_plan={ep_plan} styles {', '.join(dropped)} by a style this load "
+                "drops, so those expert weights would load whole on every rank beside their sharded "
+                "siblings."
+            )
+        # The all-to-all transport needs every rank to hold as many experts, counted as the plan's
+        # decision counts them; a count the config does not name is checked once the experts load.
+        num_experts = _num_experts_from_config(ep_cfg)
+        if num_experts % ep_degree:
+            raise ValueError(
+                f"{plan!r}: ep_size={ep_degree} must divide the {num_experts} routed experts, which the "
+                "all-to-all transport shares out equally over the expert-parallel ranks."
+            )
+
+    # Each rank builds the meshes on the GPU it is bound to, and DeviceMesh binds one not bound yet
+    # by its local rank. A rank that would so end up on no GPU, or on one another rank is on (as an
+    # import that initializes CUDA leaves every rank of a launch that binds none), fails alone in the
+    # meshes while the others wait: every rank decides from one gather, and all refuse alike.
+    def gpu() -> tuple[bool, bool, str | None]:
+        """Whether this rank has a GPU, whether it waits to be bound to one, and the bound one's UUID."""
+        if not torch.cuda.is_initialized():
+            return _has_own_gpu(), torch.cuda.is_available(), None
+        props = torch.cuda.get_device_properties(torch.cuda.current_device())
+        return _has_own_gpu(), False, str(props.uuid)
+
+    gpus = dist.all_reduce(
+        "each rank's GPU",
+        lambda: [gpu()],
+        combine=lambda ranks: [g for facts in ranks for g in facts],
+    )
+    unbound = [rank for rank, (has, waits, _) in enumerate(gpus) if waits and not has]
+    if unbound:
+        raise RuntimeError(
+            "The expert-parallel load builds its meshes on the GPU each rank is bound to, and rank(s) "
+            f"{unbound} are bound to none yet, nor does their local rank name one they see. Bind each "
+            "rank to its GPU before the load (torch.cuda.set_device), as distributed_run() does."
+        )
+    without = [rank for rank, (has, _, _) in enumerate(gpus) if not has]
+    if without and len(without) < len(gpus):
+        raise RuntimeError(
+            "The expert-parallel load builds its meshes on each rank's own GPU, and rank(s) "
+            f"{without} have none while the others have theirs. Run one process per GPU, as "
+            "torchrun --nproc_per_node=<GPUs per node> does."
+        )
+    uuids = [uuid for _, _, uuid in gpus]
+    shared = [rank for rank, uuid in enumerate(uuids) if uuid is not None and uuids.count(uuid) > 1]
+    if shared:
+        raise RuntimeError(
+            "The expert-parallel load builds its meshes on the GPU each rank is bound to, and rank(s) "
+            f"{shared} are bound to the same one. Bind each rank to its own GPU before the load "
+            "(torch.cuda.set_device), as distributed_run() does."
+        )
+    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.
+    with _refused_together("the local checkpoint", alone=False):
+        _, ckpt_path = ensure_local_checkpoint(ckpt_path)
+
+    # Experts are sharded over ``ep`` and replicated over ``edp``; the dense part is replicated over
+    # the whole world (``dp``).
+    sparse_moe_mesh = device_mesh.init_device_mesh(
+        "cuda", (edp, ep_degree), mesh_dim_names=("edp", "ep")
+    )
+    dense_mesh = device_mesh.init_device_mesh("cuda", (world_size,), mesh_dim_names=("dp",))
+    ep_submesh = sparse_moe_mesh["ep"]
+
+    # No tensor parallelism: tp_plan="auto" only switches on the path through which transformers
+    # applies the expert plan, and the swapped plan holds expert entries only. Each argument goes on
+    # DistributedConfig where this release has a field for it, else to from_pretrained.
+    _dc_fields = {f.name for f in dataclasses.fields(hf_distributed.DistributedConfig)}
+    _dc_has_tp_plan = "tp_plan" in _dc_fields
+    dc_kwargs: dict[str, Any] = {"enable_expert_parallel": True}
+    if _dc_has_tp_plan:
+        dc_kwargs["tp_plan"] = "auto"
+    if _transformers_mesh_builder() is not None and "tp_size" in _dc_fields:
+        # This release sizes its own mesh from tp_size; _experts_sharded_over swaps in the EP mesh.
+        dc_kwargs["tp_size"] = ep_degree
+    distributed_config = hf_distributed.DistributedConfig(**dc_kwargs)
+    from_pretrained_kwargs: dict[str, Any] = {
+        "dtype": dtype,
+        "device_mesh": ep_submesh,
+        "distributed_config": distributed_config,
+        "trust_remote_code": trust_remote_code,
+    }
+    if not _dc_has_tp_plan:
+        from_pretrained_kwargs["tp_plan"] = "auto"
+    if attn_implementation is not None:
+        from_pretrained_kwargs["attn_implementation"] = attn_implementation
+    # Swap the plan to the torch_a2a styles and select the torch_a2a experts forward: tokens reach
+    # their experts' ranks by all-to-all around the quantizable grouped GEMM.
+    register_torch_a2a()  # idempotent; register the experts forward + the two styles, right before use
+    # Keep expert-related entries (grouped_gemm expert params + the router/experts styles); DROP the TP
+    # styles (a2a builds no TP group for them -> would shard attention across the wrong mesh).
+    swapped_plan = {k: swap.get(v, v) for k, v in ep_plan.items() if v in kept_styles}
+    # Tag every routed-experts container torch_a2a_experts, found as the plan's decision finds them:
+    # some plans style only their parameters, grouped_gemm.
+    swapped_plan.update(dict.fromkeys(_routed_expert_keys(ep_plan), "torch_a2a_experts"))
+    from_pretrained_kwargs["experts_implementation"] = "torch_a2a"
+    # The checkpoint's own plan, for the config the model keeps (below); None: a class default.
+    declared_plan = vars(ep_cfg).get("base_model_ep_plan")
+    ep_cfg.base_model_ep_plan = swapped_plan
+    from_pretrained_kwargs["config"] = hf_config
+    print_rank_0(f"[ep] ep_plan={swapped_plan} experts_implementation=torch_a2a")
+    # A rank failing in the load fails every rank here, before the warmup's collectives:
+    # from_pretrained runs no collective, so the others finish it and meet the failure in the gather.
+    with _refused_together("the expert-parallel load", alone=False):
+        # The class the plan's decision chose, with its LM head where an auto class builds one.
+        loader = _auto_class(hf_config, require_lm_head, trust_remote_code=trust_remote_code)
+        print_rank_0(f"[ep] loader={loader.__name__} (require_lm_head={require_lm_head})")
+        with _experts_sharded_over(ep_submesh):
+            model = loader.from_pretrained(ckpt_path, **from_pretrained_kwargs)
+        model.eval()
+        # The config the model keeps goes back to the checkpoint's plan, as an export writes it out
+        # and nothing without ModelOpt knows the torch_a2a styles.
+        kept_config = getattr(model, "config", None)
+        if kept_config is not None:
+            kept_ep_config = _ep_config(kept_config)
+            if declared_plan is None:
+                vars(kept_ep_config).pop("base_model_ep_plan", None)
+            else:
+                kept_ep_config.base_model_ep_plan = declared_plan
+        # The experts forward reads the implementation from the config when it runs, which
+        # from_pretrained left on grouped_mm: torch_a2a on every (sub)config. An export writes none.
+        for _m in model.modules():
+            _cfg = getattr(_m, "config", None)
+            if _cfg is not None and hasattr(_cfg, "_experts_implementation"):
+                _cfg._experts_implementation = "torch_a2a"
+
+        # 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)
+
+        # 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.
+        model._device_mesh = ep_submesh
+        if hasattr(model, "tie_weights"):
+            model.tie_weights()
+        if freeze:
+            model.requires_grad_(False)
+
+        # The routed-expert containers, from the plan that drove the sharding.
+        expert_modules = _routed_expert_modules(model, ep_plan)
+        if not expert_modules:
+            raise ValueError(
+                f"base_model_ep_plan={ep_plan} names routed-expert containers that no module of "
+                f"{type(model).__name__} matches, so calibration would reduce the experts' amax over "
+                "ranks holding different experts. Quantize the model without expert parallelism."
+            )
+
+    # Each module's ParallelState for calibration's amax sync: dense modules reduce over the whole world;
+    # routed experts reduce over their expert-data-parallel replicas, and share one input amax across the
+    # layer's experts over the expert-parallel group. A size-1 axis is left unused (-1): no collective.
+    def _axis(mesh):
+        return mesh.get_group() if mesh.size() > 1 else -1
+
+    _seed_parallel_states(
+        model,
+        expert_modules,
+        dense_state=ParallelState(data_parallel_group=_axis(dense_mesh["dp"])),
+        expert_state=ParallelState(
+            data_parallel_group=_axis(sparse_moe_mesh["edp"]),
+            expert_model_parallel_group=_axis(sparse_moe_mesh["ep"]),
+        ),
+    )
+    print_rank_0(
+        f"[ep] seeded the expert-parallel state of {len(expert_modules)} routed-expert "
+        "module(s) from base_model_ep_plan"
+    )
+    # Build each amax collective's NCCL comm now, in the same order on every rank, rather than lazily
+    # mid-calibration, where subgroups phase-skewed at DP>1 deadlock. Size-1 axes run none.
+    warm_device = next(model.parameters()).device
+    for _sub in (dense_mesh["dp"], sparse_moe_mesh["ep"], sparse_moe_mesh["edp"]):
+        if _sub.size() <= 1:
+            continue
+        torch.distributed.barrier()
+        torch.distributed.all_reduce(torch.ones(1, device=warm_device), group=_sub.get_group())
+        torch.cuda.synchronize()
+    torch.distributed.barrier()
+    return model
diff --git a/tests/gpu/torch/utils/test_expert_parallel_load.py b/tests/gpu/torch/utils/test_expert_parallel_load.py
new file mode 100644
index 00000000000..849305f4f6a
--- /dev/null
+++ b/tests/gpu/torch/utils/test_expert_parallel_load.py
@@ -0,0 +1,113 @@
+# 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.
+
+"""Expert-parallel loading and the all-to-all expert transport on real GPUs.
+
+``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.
+"""
+
+import pytest
+import torch
+from packaging.version import Version
+
+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)
+
+from torch.distributed.tensor import DTensor
+from transformers import AutoConfig, AutoModelForCausalLM, Qwen3MoeConfig
+
+from modelopt.torch.utils.distributed import (
+    DistributionPlan,
+    distribution_plan,
+    set_distribution_plan,
+)
+from modelopt.torch.utils.moe import expert_parallel_mesh
+from modelopt.torch.utils.plugins.model_load_utils import load_and_prepare_ep
+
+NUM_EXPERTS = 8
+
+
+def _save_tiny_qwen3_moe(ckpt_dir):
+    config = Qwen3MoeConfig(
+        hidden_size=64,
+        intermediate_size=128,
+        moe_intermediate_size=32,
+        num_hidden_layers=2,
+        num_attention_heads=4,
+        num_key_value_heads=2,
+        num_experts=NUM_EXPERTS,
+        num_experts_per_tok=2,
+        vocab_size=128,
+        max_position_embeddings=64,
+    )
+    torch.manual_seed(0)
+    AutoModelForCausalLM.from_config(config).to(torch.bfloat16).save_pretrained(ckpt_dir)
+
+
+def _local(tensor):
+    return tensor.to_local() if isinstance(tensor, DTensor) else tensor
+
+
+def _sharded_model_matches_the_whole_one(rank, size, ckpt_dir, ep_degree):
+    device = torch.device(f"cuda:{rank}")
+    config = AutoConfig.from_pretrained(ckpt_dir)  # carries qwen3_moe's own base_model_ep_plan
+    plan = DistributionPlan(size, ep_size=ep_degree)
+    model = load_and_prepare_ep(ckpt_dir, plan, hf_config=config).eval()
+    # The loader swaps the plan on its own copy: the caller's config keeps the checkpoint's.
+    assert config.base_model_ep_plan == AutoConfig.from_pretrained(ckpt_dir).base_model_ep_plan
+    # The loader records no plan; loading through it directly, the caller records one.
+    with pytest.raises(RuntimeError, match="no distribution plan recorded"):
+        distribution_plan(model)
+    set_distribution_plan(model, plan)
+    # Calibration re-installs the all-to-all over the mesh the loader keeps on the model.
+    assert expert_parallel_mesh(model).size() == ep_degree
+    whole = AutoModelForCausalLM.from_pretrained(ckpt_dir, dtype=torch.bfloat16).to(device).eval()
+
+    # Each rank holds its contiguous share of every layer's experts within its EP group, whose
+    # ranks are consecutive.
+    per_rank = NUM_EXPERTS // ep_degree
+    ep_rank = rank % ep_degree
+    owned = slice(ep_rank * per_rank, (ep_rank + 1) * per_rank)
+    for layer, whole_layer in zip(model.model.layers, whole.model.layers):
+        for name in ("gate_up_proj", "down_proj"):
+            local = _local(getattr(layer.mlp.experts, name))
+            assert local.shape[0] == per_rank
+            torch.testing.assert_close(local, getattr(whole_layer.mlp.experts, name)[owned])
+    # The load recorded what it could not place, for the export to carry: here, nothing.
+    assert model._modelopt_unplaced_source_keys == []
+    assert model._modelopt_source_checkpoint == ckpt_dir
+
+    # Every rank calibrates on different tokens, so each sends its own batch through the experts.
+    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
+        reference = whole(input_ids).logits
+    torch.testing.assert_close(sharded.float(), reference.float(), atol=2e-2, rtol=2e-2)
+
+
+@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
+):
+    """With the experts sharded over every rank, or over half of them and replicated across the halves."""
+    ep_degree = dist_workers.world_size // ep_share
+    if ep_degree < 2 or dist_workers.world_size % ep_share or NUM_EXPERTS % ep_degree:
+        pytest.skip(f"{NUM_EXPERTS} experts do not shard over EP {ep_degree} on these GPUs")
+    _save_tiny_qwen3_moe(tmp_path)
+    dist_workers.run(_sharded_model_matches_the_whole_one, str(tmp_path), ep_degree)
diff --git a/tests/unit/torch/quantization/plugins/test_huggingface.py b/tests/unit/torch/quantization/plugins/test_huggingface.py
index 078f5dacd8d..136de332d7b 100644
--- a/tests/unit/torch/quantization/plugins/test_huggingface.py
+++ b/tests/unit/torch/quantization/plugins/test_huggingface.py
@@ -36,10 +36,12 @@
 from modelopt.torch.quantization.plugins.huggingface import (
     _QuantHFParallelLinear,
     _TransposedExpertsCalibMixin,
+    convert_hf_parallel_linears_on_the_fly,
     get_homogeneous_hf_decoder_layers,
     is_homogeneous_hf_model,
 )
 from modelopt.torch.quantization.utils.layerwise_calib import LayerActivationCollector
+from modelopt.torch.utils.distributed import DistributionPlan, set_distribution_plan
 
 pytest.importorskip("transformers")
 
@@ -101,6 +103,25 @@ class ParallelLinear:
     assert linear.weight is original_weight
 
 
+@pytest.mark.parametrize("expert_parallel", [False, True], ids=["no-plan", "expert-parallel"])
+def test_a_mesh_without_tp_sharded_linears_warns_unless_the_model_is_expert_parallel(
+    expert_parallel,
+):
+    """A transformers mesh with no TP-sharded Linear under it is worth a warning, except on a model
+    loaded expert-parallel: its mesh is the experts', and no Linear is meant to be sharded."""
+    model = nn.Sequential(nn.Linear(4, 4))
+    model._device_mesh = object()
+    if expert_parallel:
+        set_distribution_plan(model, DistributionPlan(2, ep_size=2))
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        convert_hf_parallel_linears_on_the_fly(model)
+
+    warned = any("no TP-sharded linear layer" in str(w.message) for w in caught)
+    assert warned is not expert_parallel
+
+
 def test_convert_conv1d():
     set_seed()
     assert transformers.pytorch_utils.Conv1D in QuantModuleRegistry
diff --git a/tests/unit/torch/utils/test_model_load_utils.py b/tests/unit/torch/utils/test_model_load_utils.py
index 9266e601180..5f07787ddee 100644
--- a/tests/unit/torch/utils/test_model_load_utils.py
+++ b/tests/unit/torch/utils/test_model_load_utils.py
@@ -22,10 +22,12 @@
 tests of each part use a process group, two gloo ranks.
 """
 
+import copy
+import dataclasses
 import inspect
 import math
 import os
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
 from types import SimpleNamespace
 from unittest import mock
 
@@ -34,9 +36,11 @@
 from _test_utils.torch.distributed.utils import bounded_collectives
 from _test_utils.torch.moe_models import model_with_decoder_at
 from safetensors.torch import save_file
+from torch import nn
 
 pytest.importorskip("accelerate")
 transformers = pytest.importorskip("transformers")
+import transformers.distributed as hf_distributed
 from _test_utils.torch.transformers_models import (
     _TinyQwen35Tokenizer,
     get_tiny_qwen3_5_moe_vl,
@@ -50,29 +54,39 @@
     AutoModelForImageTextToText,
     AutoModelForSeq2SeqLM,
     AutoModelForSpeechSeq2Seq,
+    FineGrainedFP8Config,
     LlamaConfig,
     LlamaForCausalLM,
     Phi4MultimodalConfig,
+    PreTrainedModel,
     Qwen2_5_VLConfig,
     Qwen3MoeConfig,
+    Qwen3MoeForCausalLM,
     T5Config,
     WhisperConfig,
 )
+from transformers.distributed import DistributedConfig
 
 import modelopt.torch.models.hf as models_hf
 import modelopt.torch.quantization as mtq
 import modelopt.torch.utils.distributed as dist
+import modelopt.torch.utils.moe as moe
 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
 from modelopt.torch.utils.distributed import DistributionPlan, distribution_plan
-from modelopt.torch.utils.plugins import model_load_utils
+from modelopt.torch.utils.plugins import hf_checkpoint_utils, model_load_utils
 from modelopt.torch.utils.plugins.model_load_utils import (
     _auto_class,
+    _ckpt_quant_method,
     _conversion_plan,
     _convert_keys,
+    _experts_sharded_over,
     _resolve_target,
+    _seed_parallel_states,
+    _transformers_mesh_builder,
     build_meta_causal_lm,
+    load_and_prepare_ep,
     load_pretrained,
     record_unplaced_source_keys,
     resolve_distribution_plan,
@@ -1830,3 +1844,717 @@ def test_what_needs_the_group_is_refused_in_a_launch_that_sets_none_up(
     with pytest.raises(error, match=match):
         load_pretrained(str(tmp_path), distribution=distribution)
     assert loaders == {}
+
+
+# ======================================================================== the expert-parallel load
+# What the expert-parallel load reads off the loaded model and its config: it takes an expert-parallel
+# plan, ``_seed_parallel_states`` gives every module the parallel state its amax syncs over, and
+# ``_ckpt_quant_method`` reads the checkpoint's native quantization. Everything runs in one CPU process
+# but the bounded two-rank tests at the end.
+
+# ------------------------------------------------------------------------------- the plan
+
+_EP_PLAN = {
+    "layers.*.mlp.gate": "ep_router",
+    "layers.*.mlp.experts.gate_up_proj": "grouped_gemm",
+    "layers.*.mlp.experts.down_proj": "grouped_gemm",
+    "layers.*.mlp.experts": "moe_tp_experts",
+}
+
+
+class _Mesh:
+    """A device mesh, as far as the loader reads one; its groups run no collective (``-1``)."""
+
+    device_type = "cpu"
+
+    def __getitem__(self, name):
+        return self
+
+    def size(self):
+        return 2
+
+    def get_group(self):
+        return -1
+
+
+def _cuda(gpu, bound=True, count=1, uuid="GPU-0"):
+    """Patches pinning the CUDA state the GPU check reads: ``count`` visible GPUs and this rank on
+    cuda:0, the one named ``uuid``, bound to it or not yet; or, without ``gpu``, none. The check then
+    reads the case, not the host the test runs on, whose visible GPU no test binds."""
+    state = {
+        "is_available": gpu,
+        "is_initialized": gpu and bound,
+        "current_device": 0,
+        "device_count": count if gpu else 0,
+        "get_device_properties": SimpleNamespace(uuid=uuid),
+    }
+    return [mock.patch(f"torch.cuda.{name}", return_value=value) for name, value in state.items()]
+
+
+@pytest.fixture
+def gloo_world_of_one(gloo_world_of_one):
+    """The one-rank world, with no GPU whatever the host has (:func:`_cuda`)."""
+    with ExitStack() as stack:
+        for patch in _cuda(False):
+            stack.enter_context(patch)
+        yield gloo_world_of_one
+
+
+@pytest.fixture
+def ep_world_of_two(monkeypatch, gloo_world_of_one):
+    """A two-rank expert-parallel world as far as the loader's checks see, and the checkpoint paths
+    it resolves; nothing else is set up."""
+    resolved = []
+    monkeypatch.setattr(dist, "is_initialized", lambda: True)
+    monkeypatch.setattr(dist, "size", lambda group=None: 2)
+    monkeypatch.setattr(torch.distributed, "get_rank", lambda group=None: 0)
+    monkeypatch.setattr(moe, "require_transformers", lambda *args: None)
+    monkeypatch.setattr(
+        model_load_utils, "ensure_local_checkpoint", lambda p, **k: resolved.append(p) or (None, p)
+    )
+    return resolved
+
+
+@pytest.mark.parametrize(
+    ("plan", "error", "match"),
+    [
+        (DistributionPlan(2), ValueError, "shards no routed experts"),
+        (DistributionPlan(2, ep_size=2, dense="fsdp"), NotImplementedError, "replicates the dense"),
+    ],
+    ids=["no-expert-parallelism", "sharded-dense"],
+)
+def test_the_loader_refuses_a_plan_it_cannot_load_by(plan, error, match):
+    """Before anything else: no process group, transformers release or checkpoint is looked at."""
+    with pytest.raises(error, match=match):
+        load_and_prepare_ep("unused", plan)
+
+
+def test_the_loader_refuses_a_plan_for_another_world(gloo_world_of_one):
+    with pytest.raises(ValueError, match="is for 2 ranks, but this run has 1"):
+        load_and_prepare_ep("unused", DistributionPlan(2, ep_size=2))
+
+
+def test_the_loader_refuses_a_transformers_that_builds_no_model_with_torch_a2a(
+    monkeypatch, gloo_world_of_one
+):
+    """transformers 5.6 checks a model's experts implementation against the ones it ships, so its
+    from_pretrained would refuse torch_a2a once the load had set everything up: refused first."""
+    monkeypatch.setattr(dist, "is_initialized", lambda: True)
+    monkeypatch.setattr(dist, "size", lambda group=None: 2)
+    monkeypatch.setattr("transformers.__version__", "5.6.2")
+
+    with pytest.raises(RuntimeError, match=r"requires transformers >= 5\.7"):
+        load_and_prepare_ep("unused", DistributionPlan(2, ep_size=2), hf_config=SimpleNamespace())
+
+
+def test_a_hub_id_is_made_local_before_the_weights_load(monkeypatch, gloo_world_of_one):
+    """The export reads the tensors the load could not place back from the checkpoint it records,
+    which must be a local directory: a Hub ID there leaves the export nothing to read. The load
+    resolves it once the plan and the config pass, before anything reads a weight."""
+
+    class _ResolvedError(Exception):
+        """Stops the load where it resolves the checkpoint."""
+
+    resolved = []
+
+    def resolve(path, **hub_options):
+        resolved.append(path)
+        raise _ResolvedError
+
+    monkeypatch.setattr(dist, "is_initialized", lambda: True)
+    monkeypatch.setattr(dist, "size", lambda group=None: 2)
+    monkeypatch.setattr(torch.distributed, "get_rank", lambda group=None: 0)
+    monkeypatch.setattr(moe, "require_transformers", lambda *args: None)
+    monkeypatch.setattr(model_load_utils, "ensure_local_checkpoint", resolve)
+
+    with pytest.raises(_ResolvedError):
+        load_and_prepare_ep(
+            "org/model",
+            DistributionPlan(2, ep_size=2),
+            hf_config=SimpleNamespace(base_model_ep_plan=_EP_PLAN),
+        )
+
+    assert resolved == ["org/model"]
+
+
+def test_the_experts_are_counted_in_the_config_holding_the_plan(monkeypatch, gloo_world_of_one):
+    """As the distribution plan's decision counts them: a VLM whose nested config holds the MoE and
+    its plan has its experts counted there, not by an expert-ish field on its root config."""
+
+    class _CheckedError(Exception):
+        """Stops the load once its checks pass."""
+
+    def resolve(path, **hub_options):
+        raise _CheckedError
+
+    monkeypatch.setattr(dist, "is_initialized", lambda: True)
+    monkeypatch.setattr(dist, "size", lambda group=None: 4)
+    monkeypatch.setattr(torch.distributed, "get_rank", lambda group=None: 0)
+    monkeypatch.setattr(moe, "require_transformers", lambda *args: None)
+    monkeypatch.setattr(model_load_utils, "ensure_local_checkpoint", resolve)
+    text = SimpleNamespace(base_model_ep_plan=_EP_PLAN, num_experts=8)
+
+    with pytest.raises(_CheckedError):
+        load_and_prepare_ep(
+            "org/model",
+            DistributionPlan(4, ep_size=4),
+            hf_config=SimpleNamespace(text_config=text, num_experts=3),
+        )
+
+
+@pytest.mark.parametrize(
+    ("config", "error", "match"),
+    [
+        *(
+            (
+                SimpleNamespace(quantization_config={"quant_method": method}),
+                NotImplementedError,
+                f"cannot dequantize a '{method}'",
+            )
+            for method in ("gptq", "modelopt", "compressed-tensors", "mxfp4")
+        ),
+        (SimpleNamespace(), ValueError, "declare base_model_ep_plan"),
+        (
+            SimpleNamespace(base_model_ep_plan={"layers.*.self_attn.q_proj": "colwise"}),
+            ValueError,
+            "styles no routed experts",
+        ),
+        *(
+            (
+                SimpleNamespace(
+                    base_model_ep_plan={"layers.*.mlp.experts": "moe_tp_experts", **params}
+                ),
+                ValueError,
+                r"shards no weights of layers\.\*\.mlp\.experts through grouped_gemm",
+            )
+            for params in ({}, {"layers.*.mlp.experts.gate_up_proj": "packed_colwise"})
+        ),
+        (
+            SimpleNamespace(
+                base_model_ep_plan={**_EP_PLAN, "layers.*.mlp.experts.down_proj": "packed_rowwise"}
+            ),
+            ValueError,
+            r"styles layers\.\*\.mlp\.experts\.down_proj by a style this load drops",
+        ),
+    ],
+    ids=[
+        "gptq",
+        "modelopt",
+        "compressed-tensors",
+        "mxfp4",
+        "no-ep-plan",
+        "no-routed-experts",
+        "experts-container-only",
+        "experts-sharded-otherwise",
+        "experts-sharded-partly-otherwise",
+    ],
+)
+def test_a_config_the_load_cannot_take_is_refused_before_anything_loads(
+    ep_world_of_two, config, error, match
+):
+    """A format transformers' own quantizer would load as stored, for PTQ to treat as plain weights,
+    or, MXFP4, dequantize into GPT-OSS's transposed expert layout whatever the model's, or a plan
+    that leaves the experts whole on every rank: one naming no routed experts, or none of
+    whose entries shards their weights by grouped_gemm, the only expert-weight style the load
+    shards by, or one that styles some of their weights otherwise. The load refuses each before it
+    reads anything."""
+    with pytest.raises(error, match=match):
+        load_and_prepare_ep("org/model", DistributionPlan(2, ep_size=2), hf_config=config)
+    assert ep_world_of_two == []
+
+
+def test_an_ep_size_that_does_not_divide_the_experts_is_refused_before_anything_loads(
+    ep_world_of_two,
+):
+    """The all-to-all transport finds an expert's rank by dividing its id by this rank's share of
+    the experts: with 3 experts over 2 ranks the ranks would disagree on who holds which, and send
+    tokens to the wrong experts."""
+    config = SimpleNamespace(base_model_ep_plan=_EP_PLAN, num_experts=3)
+
+    with pytest.raises(ValueError, match="ep_size=2 must divide the 3 routed experts"):
+        load_and_prepare_ep("org/model", DistributionPlan(2, ep_size=2), hf_config=config)
+    assert ep_world_of_two == []
+
+
+def test_a_plan_whose_expert_containers_match_no_module_is_refused(ep_world_of_two, monkeypatch):
+    """A plan that styles the grouped-GEMM parameters a level below the experts names a container no
+    module is. Loaded anyway, every module would get the dense state, and the experts' amax would
+    reduce over ranks holding different experts."""
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, **kwargs):
+            return model_with_decoder_at("model.")
+
+    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, "record_unplaced_source_keys", lambda *args: [])
+    config = SimpleNamespace(
+        base_model_ep_plan={"layers.*.mlp.experts.*.gate_up_proj": "grouped_gemm"}
+    )
+
+    with pytest.raises(ValueError, match="that no module of"):
+        load_and_prepare_ep("org/model", DistributionPlan(2, ep_size=2), hf_config=config)
+
+
+@pytest.mark.parametrize("declared", [True, False], ids=["on-the-config", "a-class-default"])
+def test_the_model_keeps_the_plan_its_checkpoint_declares(
+    ep_world_of_two, gloo_world_of_one, monkeypatch, declared
+):
+    """The load shards by the torch_a2a styles, but the config the model keeps goes back to the
+    checkpoint's plan: an export writes it out, and nothing without ModelOpt knows those styles. A
+    plan the config class ships (transformers' own, e.g. Qwen3-MoE's) stays off the instance."""
+
+    class _PastError(Exception):
+        """Raised by the load's next check."""
+
+    model = model_with_decoder_at("model.")
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, config, **kwargs):
+            model.config = copy.deepcopy(config)  # transformers keeps a copy of the config
+            return model
+
+    def next_check(*args, **kwargs):
+        raise _PastError
+
+    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, "record_unplaced_source_keys", lambda *args: None)
+    monkeypatch.setattr(model_load_utils, "_seed_parallel_states", next_check)
+
+    class _Config(SimpleNamespace):
+        base_model_ep_plan = _EP_PLAN
+
+    # 8 experts over 2 ranks: the stand-in model holds this rank's 4.
+    config = _Config(num_experts=8, **({"base_model_ep_plan": dict(_EP_PLAN)} if declared else {}))
+
+    with pytest.raises(_PastError):
+        load_and_prepare_ep("org/model", DistributionPlan(2, ep_size=2), hf_config=config)
+
+    assert model.config.base_model_ep_plan == _EP_PLAN
+    assert ("base_model_ep_plan" in vars(model.config)) is declared
+
+
+def test_a_loaded_model_keeps_its_sharding_styles_out_of_its_saved_config(
+    ep_world_of_two, gloo_world_of_one, monkeypatch
+):
+    """What the loader leaves on a transformers model, as transformers treats it. Built with the
+    torch_a2a styles, the model keeps them as ``model.tp_plan``, which calibration re-installs the
+    all-to-all forwards by, and the expert-parallel mesh it re-installs them over, which this stand-in
+    from_pretrained does not set; while its config goes back to the checkpoint's plan; and the experts
+    implementation its forwards dispatch on, torch_a2a, stays out of the config it serializes,
+    which an export writes."""
+    if not Qwen3MoeConfig.base_model_ep_plan:
+        pytest.skip("this transformers ships no expert-parallel plan for Qwen3-MoE")
+
+    class _PastError(Exception):
+        """Raised by the load's next check."""
+
+    built = []
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, config, **kwargs):
+            # As transformers' expert-parallel load builds it: this rank's 2 of the 4 experts.
+            config.distributed_config = DistributedConfig(enable_expert_parallel=True)
+            share = copy.deepcopy(config)
+            share.num_experts //= 2
+            with torch.device("meta"):
+                model = Qwen3MoeForCausalLM(share)
+            model.config = config
+            built.append(model)
+            return model
+
+    def next_check(*args, **kwargs):
+        raise _PastError
+
+    mesh = _Mesh()
+    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, "record_unplaced_source_keys", lambda *args: None)
+    monkeypatch.setattr(model_load_utils, "_seed_parallel_states", next_check)
+    config = Qwen3MoeConfig(
+        hidden_size=16,
+        intermediate_size=32,
+        moe_intermediate_size=8,
+        num_hidden_layers=1,
+        num_attention_heads=2,
+        num_key_value_heads=1,
+        vocab_size=32,
+        num_experts=4,
+        num_experts_per_tok=1,
+    )
+
+    with pytest.raises(_PastError):
+        load_and_prepare_ep("org/model", DistributionPlan(2, ep_size=2), hf_config=config)
+
+    (model,) = built
+    assert model.config.base_model_ep_plan == Qwen3MoeConfig.base_model_ep_plan
+    assert "torch_a2a_experts" in set(model.tp_plan.values())
+    assert model._device_mesh is mesh
+    assert model.config._experts_implementation == "torch_a2a"
+    assert not any("experts_implementation" in key for key in model.config.to_dict())
+
+
+def test_the_weights_load_in_the_checkpoints_dtype_through_its_class(ep_world_of_two, monkeypatch):
+    """The checkpoint's ``dtype`` reaches ``from_pretrained``, and the class that builds it is chosen
+    with the trust the caller gave its code."""
+
+    class _LoadedError(Exception):
+        """Stops the load at from_pretrained."""
+
+    loaded, classes = {}, []
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, **kwargs):
+            loaded.update(kwargs)
+            raise _LoadedError
+
+    def auto_class(hf_config, require_lm_head=True, trust_remote_code=False):
+        classes.append(trust_remote_code)
+        return _Loader
+
+    monkeypatch.setattr(
+        "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
+    )
+    monkeypatch.setattr(model_load_utils, "_auto_class", auto_class)
+    config = SimpleNamespace(dtype=torch.float16, base_model_ep_plan=_EP_PLAN)
+
+    with pytest.raises(_LoadedError):
+        load_and_prepare_ep(
+            "org/model", DistributionPlan(2, ep_size=2), trust_remote_code=True, hf_config=config
+        )
+    assert loaded["dtype"] is torch.float16
+    assert classes == [True]
+
+
+# ----------------------------------------------------------------------- _seed_parallel_states
+
+
+def test_expert_containers_and_their_children_get_the_expert_state():
+    """Everything else, the root and the containers' own parents included, gets the dense state; so
+    does a sibling whose name merely starts with a container's (Qwen2-MoE keeps
+    ``shared_expert_gate`` beside ``shared_expert``)."""
+    model = model_with_decoder_at("model.")
+    for layer in model.model.layers:
+        layer.mlp.experts.weight_quantizer = nn.Identity()  # a child, as quantizer insertion adds
+    model.model.layers[0].mlp.experts_gate = nn.Linear(2, 1)
+    experts = {f"model.layers.{i}.mlp.experts": model.model.layers[i].mlp.experts for i in (0, 1)}
+    dense, expert = object(), object()
+
+    _seed_parallel_states(model, experts, dense_state=dense, expert_state=expert)
+
+    states = {name: module._parallel_state for name, module in model.named_modules()}
+    assert {name for name, state in states.items() if state is expert} == {
+        "model.layers.0.mlp.experts",
+        "model.layers.0.mlp.experts.weight_quantizer",
+        "model.layers.1.mlp.experts",
+        "model.layers.1.mlp.experts.weight_quantizer",
+    }
+    assert all(state is dense for state in states.values() if state is not expert)
+
+
+# --------------------------------------------------------------------------- _ckpt_quant_method
+
+
+@pytest.mark.parametrize(
+    ("config", "expected"),
+    [
+        (SimpleNamespace(quantization_config={"quant_method": "fp8"}), "fp8"),
+        (SimpleNamespace(quantization_config=SimpleNamespace(quant_method="mxfp4")), "mxfp4"),
+        (
+            SimpleNamespace(
+                text_config=SimpleNamespace(quantization_config={"quant_method": "fp8"})
+            ),
+            "fp8",
+        ),
+        (
+            SimpleNamespace(
+                llm_config=SimpleNamespace(quantization_config={"quant_method": "fp8"})
+            ),
+            "fp8",
+        ),
+        (
+            SimpleNamespace(
+                quantization_config={"quant_method": "mxfp4"},
+                text_config=SimpleNamespace(quantization_config={"quant_method": "fp8"}),
+            ),
+            "mxfp4",
+        ),
+        (SimpleNamespace(quantization_config=FineGrainedFP8Config()), "fp8"),
+        (SimpleNamespace(quantization_config={"weight_block_size": [128, 128]}), None),
+        (SimpleNamespace(text_config=None), None),
+    ],
+    ids=[
+        "dict",
+        "object",
+        "vlm-text-config",
+        "vlm-llm-config",
+        "top-level-wins",
+        "transformers-config-object",
+        "no-method",
+        "unquantized",
+    ],
+)
+def test_quant_method_comes_from_the_config_or_a_vlms_text_config(config, expected):
+    """The top-level config first, else a VLM's text or LLM config. A loaded model's config holds
+    transformers' config object, not the JSON dict, and callers compare against ``"fp8"`` /
+    ``"mxfp4"``: it reports the plain method name, not its enum."""
+    assert _ckpt_quant_method(config) == expected
+
+
+def test_from_pretrained_builds_its_mesh_from_the_ep_mesh_it_is_given():
+    """Newer transformers ignores ``from_pretrained``'s ``device_mesh`` and builds a world-sized one,
+    which shards the experts over every rank when EP is below the world size. During the load it gets
+    the EP mesh instead, and its own builder is back afterwards, even when the load raises."""
+    mixin = _transformers_mesh_builder()
+    if mixin is None:
+        pytest.skip("this transformers honors from_pretrained's device_mesh")
+    builder = mixin.initialize_distributed_mesh
+    ep_mesh = SimpleNamespace(device_type="cpu")
+    config = DistributedConfig(enable_expert_parallel=True, tp_plan="auto", tp_size=2)
+    with _experts_sharded_over(ep_mesh):
+        _, device_map, device_mesh = PreTrainedModel.prepare_distribute_model(config)
+
+    assert device_mesh is ep_mesh
+    assert device_map == torch.device("cpu")
+    assert mixin.initialize_distributed_mesh is builder
+    with pytest.raises(RuntimeError), _experts_sharded_over(ep_mesh):
+        raise RuntimeError("the load failed")
+    assert mixin.initialize_distributed_mesh is builder
+
+
+@pytest.mark.parametrize(
+    ("fields", "builds_its_mesh", "config", "tp_plan"),
+    [
+        # transformers 5.6-5.9, 5.11, 5.12
+        (["enable_expert_parallel"], False, {"enable_expert_parallel": True}, "auto"),
+        # 5.10, 5.13-5.17
+        (
+            ["tp_size", "tp_plan", "enable_expert_parallel"],
+            False,
+            {"enable_expert_parallel": True, "tp_plan": "auto"},
+            None,
+        ),
+        # 5.18 on
+        (
+            ["tp_size", "tp_plan", "enable_expert_parallel"],
+            True,
+            {"enable_expert_parallel": True, "tp_plan": "auto", "tp_size": 2},
+            None,
+        ),
+        # no release so far
+        (
+            ["tp_size", "enable_expert_parallel"],
+            True,
+            {"enable_expert_parallel": True, "tp_size": 2},
+            "auto",
+        ),
+    ],
+    ids=["no-tp-plan-field", "tp-plan-field", "own-mesh-builder", "own-mesh-builder-no-tp-plan"],
+)
+def test_each_load_argument_goes_where_this_transformers_takes_it(
+    ep_world_of_two, monkeypatch, fields, builds_its_mesh, config, tp_plan
+):
+    """``tp_plan="auto"`` goes on ``DistributedConfig`` where it has the field, else to
+    ``from_pretrained``, never both; a release that builds its own mesh gets it sized to the EP
+    degree. Each is decided by its own field, so a release with one and not the other loads too."""
+
+    class _LoadedError(Exception):
+        """Stops the load at from_pretrained."""
+
+    loaded = {}
+
+    class _Loader:
+        @staticmethod
+        def from_pretrained(path, **kwargs):
+            loaded.update(kwargs)
+            raise _LoadedError
+
+    distributed_config = dataclasses.make_dataclass(
+        "DistributedConfig", [(name, object, dataclasses.field(default=None)) for name in fields]
+    )
+    monkeypatch.setattr(hf_distributed, "DistributedConfig", distributed_config)
+    monkeypatch.setattr(
+        model_load_utils,
+        "_transformers_mesh_builder",
+        lambda: SimpleNamespace(initialize_distributed_mesh=None) if builds_its_mesh else None,
+    )
+    monkeypatch.setattr(
+        "torch.distributed.device_mesh.init_device_mesh", lambda *args, **kwargs: _Mesh()
+    )
+    monkeypatch.setattr(model_load_utils, "_auto_class", lambda *args, **kwargs: _Loader)
+
+    with pytest.raises(_LoadedError):
+        load_and_prepare_ep(
+            "org/model",
+            DistributionPlan(2, ep_size=2),
+            hf_config=SimpleNamespace(base_model_ep_plan=_EP_PLAN),
+        )
+
+    given = {k: v for k, v in vars(loaded["distributed_config"]).items() if v is not None}
+    assert given == config
+    assert loaded.get("tp_plan") == tp_plan
+
+
+# ----------------------------------------------------------------- one rank failing alone
+
+
+class _PastTheMeshesError(Exception):
+    """Raised by the meshes' stand-in: this rank got past the load's checks."""
+
+
+def _meshes_in_step(*args, **kwargs):
+    """The meshes' subgroups, as the collective they are: a rank in it waits for every peer."""
+    torch.distributed.barrier()
+    raise _PastTheMeshesError
+
+
+def _one_rank_fails_job(rank, size, ckpt, failure, failing_rank, match):
+    path, hf_config, patches = ckpt, None, []
+    if failure != "unreadable-config":
+        hf_config = SimpleNamespace(base_model_ep_plan=dict(_EP_PLAN), num_experts=4)
+    if failure != "unseen-checkpoint":
+        local = mock.patch.object(
+            model_load_utils, "ensure_local_checkpoint", side_effect=lambda p, **k: (None, p)
+        )
+        patches.append(local)
+    fails_here = rank == failing_rank
+    # No rank has a GPU, but in the no-gpu row every rank other than the failing one has its own.
+    patches += _cuda(failure == "no-gpu" and not fails_here, uuid=f"GPU-{rank}")
+
+    def fail(*args, **kwargs):
+        raise (MemoryError if failure == "load" else OSError)(match)
+
+    if failure == "unreadable-config" and fails_here:
+        path = ckpt + "-not-on-this-node"
+    elif failure == "transformers" and fails_here:
+        patches.append(mock.patch("transformers.__version__", "5.6.2"))
+    elif failure == "quantized-config" and fails_here:
+        hf_config.quantization_config = {"quant_method": "gptq"}
+    elif failure == "unseen-checkpoint" and fails_here:
+        # Rank 0 makes the checkpoint local; a rank on a node without that storage cannot see it.
+        patches.append(mock.patch.object(hf_checkpoint_utils.os.path, "isdir", return_value=False))
+    if failure in ("load", "record"):
+        loaded = model_with_decoder_at("model.")
+        load = fail if failure == "load" and fails_here else (lambda path, **kwargs: loaded)
+        loader = SimpleNamespace(__name__="Loader", from_pretrained=load)
+        record = fail if failure == "record" and fails_here else None
+        patches += [
+            mock.patch("torch.distributed.device_mesh.init_device_mesh", return_value=_Mesh()),
+            mock.patch.object(model_load_utils, "_auto_class", return_value=loader),
+            mock.patch.object(model_load_utils, "record_unplaced_source_keys", side_effect=record),
+        ]
+    else:
+        patches.append(
+            mock.patch("torch.distributed.device_mesh.init_device_mesh", _meshes_in_step)
+        )
+    refused = None
+    with bounded_collectives(), ExitStack() as stack:
+        for patch in patches:
+            stack.enter_context(patch)
+        try:  # not pytest.raises: its failure would end the pooled worker instead of the job
+            load_and_prepare_ep(path, DistributionPlan(2, ep_size=2), hf_config=hf_config)
+        except Exception as error:
+            refused = f"{type(error).__name__}: {error}"
+    # The one rank's failure on every rank, before any rank went on alone into a collective. The GPU
+    # check decides on every rank at once, from one gather, so no rank relays another's failure.
+    assert refused is not None and match in refused, (rank, refused)
+    assert fails_here or failure == "no-gpu" or f"rank {failing_rank} failed" in refused, refused
+
+
+@pytest.mark.skipif(
+    Version(transformers.__version__) < Version("5.7"), reason="the EP load needs transformers 5.7"
+)
+@pytest.mark.parametrize(
+    ("failure", "failing_rank", "match"),
+    [
+        ("unreadable-config", 1, "not-on-this-node"),
+        ("transformers", 1, "requires transformers >= 5.7, but 5.6.2"),
+        ("quantized-config", 1, "cannot dequantize a 'gptq'"),
+        ("unseen-checkpoint", 1, "which this rank cannot see"),
+        ("no-gpu", 1, "rank(s) [1] have none while the others have theirs"),
+        ("load", 1, "the GPU ran out of memory"),
+        ("record", 1, "a weight file this rank cannot read"),
+    ],
+)
+def test_a_load_failing_on_one_rank_fails_every_rank(
+    cpu_dist_workers, tmp_path, failure, failing_rank, match
+):
+    """One rank fails alone: reading the config, on its transformers, on its arguments, seeing the
+    checkpoint rank 0 made local, without a GPU of its own where the others have theirs, or in the
+    load. Every rank raises that failure, rather than leave the others waiting for it in the meshes
+    (stood in by a bounded barrier) or the warmup's collectives."""
+    Qwen3MoeConfig(num_experts=4, num_hidden_layers=1, hidden_size=8).save_pretrained(tmp_path)
+    cpu_dist_workers.run(
+        _one_rank_fails_job,
+        ckpt=str(tmp_path),
+        failure=failure,
+        failing_rank=failing_rank,
+        match=match,
+    )
+
+
+def _gpu_layout_job(rank, size, visible, bound, shared, refused):
+    hf_config = SimpleNamespace(base_model_ep_plan=dict(_EP_PLAN), num_experts=4)
+    reached = None
+    with bounded_collectives(), ExitStack() as stack:
+        # This rank is on cuda:0, bound to it or not yet; the local ranks run 0 and 1.
+        for patch in _cuda(True, bound, visible, "GPU-0" if shared else f"GPU-{rank}"):
+            stack.enter_context(patch)
+        stack.enter_context(
+            mock.patch.object(
+                model_load_utils, "ensure_local_checkpoint", side_effect=lambda p, **k: (None, p)
+            )
+        )
+        stack.enter_context(
+            mock.patch("torch.distributed.device_mesh.init_device_mesh", _meshes_in_step)
+        )
+        try:  # not pytest.raises: its failure would end the pooled worker instead of the job
+            load_and_prepare_ep("org/model", DistributionPlan(2, ep_size=2), hf_config=hf_config)
+        except Exception as error:
+            reached = error
+    if refused is None:
+        assert isinstance(reached, _PastTheMeshesError), (rank, reached)
+    else:
+        assert isinstance(reached, RuntimeError) and refused in str(reached), (rank, reached)
+
+
+@pytest.mark.skipif(
+    Version(transformers.__version__) < Version("5.7"), reason="the EP load needs transformers 5.7"
+)
+@pytest.mark.parametrize(
+    ("visible", "bound", "shared", "refused"),
+    [
+        (1, True, False, None),
+        (1, False, False, "rank(s) [1] are bound to none yet"),
+        (2, False, False, None),
+        (2, True, True, "rank(s) [0, 1] are bound to the same one"),
+    ],
+    ids=[
+        "one-gpu-each-bound",
+        "one-gpu-each-not-yet",
+        "every-gpu-not-yet",
+        "every-gpu-all-on-cuda-0",
+    ],
+)
+def test_each_rank_builds_the_meshes_on_a_gpu_of_its_own(
+    cpu_dist_workers, visible, bound, shared, refused
+):
+    """Each rank builds the meshes on the GPU it is bound to, and DeviceMesh binds one not bound yet
+    by its local rank. Shown one GPU each, every rank bound to cuda:0 goes on; not yet bound, rank 1's
+    local rank names no GPU it sees. Shown every GPU and not yet bound, DeviceMesh binds each to its
+    own. Every rank left on cuda:0, as an import that initializes CUDA leaves a launch that binds
+    none, shares one GPU. A refusal comes on every rank, from one gather."""
+    cpu_dist_workers.run(
+        _gpu_layout_job, visible=visible, bound=bound, shared=shared, refused=refused
+    )
diff --git a/tests/unit/torch/utils/test_moe_a2a.py b/tests/unit/torch/utils/test_moe_a2a.py
index 9dea6d86c67..090024ca24f 100644
--- a/tests/unit/torch/utils/test_moe_a2a.py
+++ b/tests/unit/torch/utils/test_moe_a2a.py
@@ -43,11 +43,13 @@
 
 import modelopt.torch.quantization as mtq
 import modelopt.torch.utils.moe as moe
+from modelopt.torch.utils.distributed import DistributionPlan, set_distribution_plan
 from modelopt.torch.utils.moe import (
     _choose_ep_degree,
     _num_experts_from_config,
     _routed_expert_modules,
     a2a_moe,
+    expert_parallel_mesh,
 )
 
 # Small, fully-deterministic problem. E must be divisible by every world size under test.
@@ -563,3 +565,32 @@ def test_a_transformers_below_a_features_floor_is_refused(monkeypatch, installed
         else nullcontext()
     ):
         moe.require_transformers("5.7", "EP load")
+
+
+# ------------------------------------------------------------------------ expert_parallel_mesh
+
+
+@pytest.mark.parametrize(
+    ("plan", "is_expert_parallel"),
+    [(DistributionPlan(2, ep_size=2), True), (None, False), (DistributionPlan(2), False)],
+    ids=["expert-parallel", "no-plan", "replicated"],
+)
+def test_the_expert_parallel_mesh_comes_with_an_expert_parallel_plan_only(plan, is_expert_parallel):
+    """An expert-parallel plan: the mesh transformers kept on the model. Any other model has none,
+    even with a device mesh on it, as a load through from_pretrained's own TP can leave."""
+    model = model_with_decoder_at("model.")
+    if plan is not None:
+        set_distribution_plan(model, plan)
+    model._device_mesh = mesh = object()
+
+    assert expert_parallel_mesh(model) is (mesh if is_expert_parallel else None)
+
+
+def test_an_expert_parallel_model_without_its_mesh_is_refused():
+    """A model recorded expert-parallel that load_and_prepare_ep did not load has no mesh for
+    calibration to re-install the all-to-all forwards over; it is refused by name."""
+    model = model_with_decoder_at("model.")
+    set_distribution_plan(model, DistributionPlan(2, ep_size=2))
+
+    with pytest.raises(RuntimeError, match="carries no device mesh"):
+        expert_parallel_mesh(model)