diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index e30bb9e245e..88685f95dd9 100755
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -41,6 +41,7 @@ Changelog
 *Misc*
 
 - The ``evaluation`` agent skill can stop looping ("doom loop") reasoning early: ask it to avoid doom loops and it adds vLLM's ``repetition_detection`` request field (vLLM 0.17.0 or later) to every task, ending a request once its output repeats. Use the same setting for baseline and candidate runs.
+- ``modelopt.torch.utils.plugins.model_load_utils.load_pretrained`` now loads a model distributed under ``torchrun``, FSDP2-sharded or whole on every rank, by the plan ``resolve_distribution_plan`` decides or one passed as its new ``distribution`` argument. Until the expert-parallel load lands, an MoE planned expert-parallel is refused; set ``MODELOPT_EP_DEGREE=1`` to load it.
 
 **Backward Breaking Changes**
 
@@ -49,6 +50,7 @@ Changelog
 - Bump minimum required PyTorch version to 2.10.
 - Bump minimum required ``diffusers`` version to 0.37, ``peft`` to 0.18.1, ``accelerate`` to 1.6 and ``nltk`` to 3.9 for the ``[hf]`` extra.
 - Drop support for Transformer Engine 1.x. Quantizing Transformer Engine modules now requires Transformer Engine 2.0 or later.
+- ``modelopt.torch.utils.distributed.fsdp2_wrap`` moved to ``modelopt.torch.utils.plugins.model_load_utils``, with the other loaders. The old import path still works, but calling it now needs the ``hf`` extra.
 - Diffusion HF exports now omit attention quantizer buffers by default, except for FLUX.1, which retains enabled Q/K/V ``_amax`` keys for runtime compatibility; other consumers of those keys can opt in with ``export_attention_quantizers=True`` (CLI: ``--extra-param export_attention_quantizers=true``). Disabled attention quantizers remain excluded, and diffusion components, including LTX-2, no longer emit KV-cache quantization metadata.
 - ``hf_quant_config.json`` of an INT4 or W4A8 AWQ checkpoint now sets ``pre_quant_scale`` per layer, only for layers that have the tensor. A model whose fused-MoE experts have none (AWQ never applies to them) is now exported as ``MIXED_PRECISION`` with per-layer entries instead of a uniform algorithm.
 - The Windows FVD example rejects I3D checkpoints requiring unrestricted pickle loading. Custom TorchScript archives now require ``--weights-sha256`` with a trusted checksum; tensor state dictionaries continue to load directly.
diff --git a/modelopt/torch/utils/distributed.py b/modelopt/torch/utils/distributed.py
index 26892ef112d..9ac89c2eb52 100644
--- a/modelopt/torch/utils/distributed.py
+++ b/modelopt/torch/utils/distributed.py
@@ -30,7 +30,7 @@
 
 import torch
 import torch.distributed
-from torch.distributed.fsdp import CPUOffloadPolicy, FSDPModule, fully_shard
+from torch.distributed.fsdp import FSDPModule
 from torch.distributed.tensor import DTensor
 
 __all__ = [
@@ -355,10 +355,10 @@ def set_distribution_plan(model: torch.nn.Module, plan: DistributionPlan) -> Non
     """Record that ``model`` is distributed as ``plan``, on it and on every submodule it has now.
 
     Code handed a submodule, such as a VLM's language model, then reads the same plan as code handed
-    the whole model; a module added afterwards records none. Code that distributes a model records
-    its plan here before calibrating or exporting the model. The plan says how the model is placed
-    over the run; the :class:`ParallelState` a quantized module holds names the process groups its
-    collectives run on.
+    the whole model; a module added afterwards records none. ``load_pretrained`` records the plan it
+    loads by once the model is built; code that distributes a model itself records it here before
+    calibrating or exporting the model. The plan says how the model is placed over the run; the
+    :class:`ParallelState` a quantized module holds names the process groups its collectives run on.
 
     A model is distributed once, and its plan recorded once: recording a plan on a model that, or
     any of whose submodules, already records one raises, the same plan included, and leaves the
@@ -403,9 +403,10 @@ def distribution_plan(model: torch.nn.Module) -> DistributionPlan:
     if sharded or size() > 1:
         how = "is FSDP2-sharded" if sharded else f"is held in a run of {size()} ranks"
         raise RuntimeError(
-            f"{type(model).__name__} {how}, but has no distribution plan recorded. Record how it "
-            "is distributed with set_distribution_plan(); a module added to a model after its plan "
-            "was recorded has none, so read the plan off that model."
+            f"{type(model).__name__} {how}, but has no distribution plan recorded. Load it with "
+            "load_pretrained, or record how it is distributed with set_distribution_plan(); a "
+            "module added to a model after its plan was recorded has none, so read the plan off "
+            "that model."
         )
     return DistributionPlan()
 
@@ -776,49 +777,15 @@ def _move_to_fsdp_device(model, params: set[torch.nn.Parameter]) -> None:
 
 
 def fsdp2_wrap(model, shard_root=True, mp_policy=None, cpu_offload: bool = False):
-    """Auto-detect a HF causal-LM's decoder layers and FSDP2 ``fully_shard`` each one.
+    """FSDP2-shard a HF causal LM's decoder layers, and the root unless ``shard_root=False``.
 
-    By default (``shard_root=True``) the root module is wrapped too, so embed/lm_head/norm are
-    sharded instead of replicated per rank; pass ``shard_root=False`` to leave the root replicated
-    (only decoder layers sharded). Returns the detected decoder layers so callers can reuse the
-    detection result.
-
-    Parameters whose dtype differs from the model's dominant one are excluded from the wrap (see
-    :func:`_off_dtype_params`), since FSDP2 rejects a shard group that mixes dtypes. They stay
-    replicated and are moved onto the shards' device, which ``fully_shard`` does not do for the
-    params it ignores.
+    The FSDP2 load wraps its model with it, so it lives with the loaders now:
+    :func:`modelopt.torch.utils.plugins.model_load_utils.fsdp2_wrap`. This keeps the old path, which
+    now needs the ``hf`` extra, as the loaders do.
     """
-    # Lazy import: layerwise_calib imports this module at top level (circular).
-    from modelopt.torch.quantization.utils.layerwise_calib import LayerActivationCollector
-
-    decoder_layers = LayerActivationCollector.get_decoder_layers(model)
-    if decoder_layers is None:
-        raise RuntimeError(
-            "Could not auto-detect decoder layers; FSDP2 wrap requires a standard HF causal-LM layout."
-        )
+    from modelopt.torch.utils.plugins.model_load_utils import fsdp2_wrap as wrap
 
-    fsdp_kwargs: dict[str, Any] = {"reshard_after_forward": True}
-    if mp_policy is not None:
-        fsdp_kwargs["mp_policy"] = mp_policy
-    if cpu_offload:
-        fsdp_kwargs["offload_policy"] = CPUOffloadPolicy()
-    ignored_params = _off_dtype_params(model)
-    if ignored_params:
-        fsdp_kwargs["ignored_params"] = ignored_params
-
-    # Snapshot/restore config.architectures: some HF builders mutate it during fully_shard.
-    config = getattr(model, "config", None)
-    architectures = list(getattr(config, "architectures", []) or [])
-    for layer in decoder_layers:
-        fully_shard(layer, **fsdp_kwargs)
-    if shard_root:
-        fully_shard(model, **fsdp_kwargs)
-    if ignored_params:
-        _move_to_fsdp_device(model, ignored_params)
-    if config is not None and architectures:
-        config.architectures = architectures
-
-    return decoder_layers
+    return wrap(model, shard_root=shard_root, mp_policy=mp_policy, cpu_offload=cpu_offload)
 
 
 def broadcast_state_dict(
diff --git a/modelopt/torch/utils/plugins/model_load_utils.py b/modelopt/torch/utils/plugins/model_load_utils.py
index babf7f11054..3e2ca0a4d08 100644
--- a/modelopt/torch/utils/plugins/model_load_utils.py
+++ b/modelopt/torch/utils/plugins/model_load_utils.py
@@ -13,12 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""HuggingFace model loading: :func:`load_pretrained`, and the plan to distribute a model by."""
+"""HuggingFace model loading: :func:`load_pretrained`, by a plan to distribute the model."""
 
+import copy
 import logging
 import os
 import re
 from collections import defaultdict
+from contextlib import contextmanager
 from itertools import chain
 from typing import TYPE_CHECKING, Any
 
@@ -31,7 +33,7 @@
     from collections.abc import Iterable
 from torch.distributed.checkpoint import HuggingFaceStorageReader
 from torch.distributed.checkpoint.state_dict import StateDictOptions, set_model_state_dict
-from torch.distributed.fsdp import FSDPModule
+from torch.distributed.fsdp import CPUOffloadPolicy, FSDPModule, fully_shard
 from torch.distributed.tensor import DTensor
 from transformers import (
     AutoConfig,
@@ -49,10 +51,12 @@
 from modelopt.torch.utils import env
 from modelopt.torch.utils.distributed import (
     DistributionPlan,
+    _move_to_fsdp_device,
+    _off_dtype_params,
     barrier,
     broadcast_state_dict,
-    fsdp2_wrap,
     is_initialized,
+    set_distribution_plan,
 )
 from modelopt.torch.utils.logging import print_rank_0, warn_rank_0
 from modelopt.torch.utils.moe import (
@@ -428,74 +432,416 @@ def record_unplaced_source_keys(
     return keys
 
 
+_FSDP2 = "FSDP2"
+# What each sharded load takes besides the checkpoint. These loads build the model class and place
+# it themselves, so from_pretrained's own arguments (device_map, max_memory, ...) do not reach them.
+_SHARDED_LOAD_OPTIONS = {
+    _FSDP2: frozenset(
+        {
+            "trust_remote_code",
+            "attn_implementation",
+            "hf_config",
+            "cpu_offload",
+            "mp_policy",
+            "broadcast_chunk_size",
+        }
+    ),
+}
+
+
+def _whole_load_options(kwargs) -> list[str]:
+    """The load ``kwargs`` that place a whole model, which the FSDP2 load refuses."""
+    names = set(kwargs) - _SHARDED_LOAD_OPTIONS[_FSDP2] - {"config", "dtype", "torch_dtype"}
+    dtypes = {name for name in ("dtype", "torch_dtype") if kwargs.get(name) not in (None, "auto")}
+    return sorted(names | dtypes)
+
+
+def _route(plan: DistributionPlan) -> str | None:
+    """The sharded load ``plan`` takes, or ``None`` when every rank loads the whole model."""
+    if plan.world_size != dist.size():
+        raise ValueError(
+            f"{plan!r} is for {plan.world_size} ranks, but this run has {dist.size()}."
+        )
+    if plan.is_expert_parallel:
+        raise NotImplementedError(
+            f"{plan!r} shards the routed experts, which load_pretrained has no load for. Plan "
+            "ep_size=1 (MODELOPT_EP_DEGREE=1) to give every rank the whole model, or shard it "
+            "with FSDP2."
+        )
+    return _FSDP2 if plan.is_fsdp_dense else None
+
+
+@contextmanager
+def _refused_together(label: str, alone: bool):
+    """Raise on every rank if the block raises on any, through :func:`dist.all_reduce`.
+
+    A rank refusing a load alone would leave the others waiting in the load's next collective (the
+    checkpoint's fetch, the sharded load). A process of a launch that set no group up (``alone``) has
+    no peers to tell.
+    """
+    error = None
+    try:
+        yield
+    # A SystemExit too: the others would wait for this rank all the same.
+    except BaseException as e:
+        error = e
+    if alone:
+        if error is not None:
+            raise error
+        return
+
+    def refuse() -> None:
+        if error is not None:
+            raise error
+
+    dist.all_reduce(label, refuse, combine=lambda _: None)
+
+
+def _read_config(path, auto_class, trust_remote_code: bool = False, **hub_options):
+    """``path``'s config as ``auto_class.from_pretrained`` reads it: config_class, or AutoConfig."""
+    if (reader := getattr(auto_class, "config_class", None)) is not None:
+        return reader.from_pretrained(path, **hub_options)
+    return AutoConfig.from_pretrained(path, trust_remote_code=trust_remote_code, **hub_options)
+
+
+def _refuse_sharded_load_options(route, auto_class, kwargs) -> None:
+    """Refuse the arguments the sharded load ``route`` cannot honor: the arguments alone decide."""
+    if auto_class not in (None, AutoModelForCausalLM):
+        raise NotImplementedError(
+            f"The {route} load builds its model class itself; auto_class={auto_class.__name__} "
+            "does not apply to it."
+        )
+    # torch_dtype is from_pretrained's older name for dtype, and None means "auto" to it.
+    for name in ("dtype", "torch_dtype"):
+        value = kwargs.get(name)
+        if value not in (None, "auto"):  # A sharded load keeps the checkpoint's dtype.
+            raise TypeError(
+                f"The {route} load reads the checkpoint's tensors as stored and keeps their dtype, "
+                f"so {name}={value!r} does not apply to it; FSDP2's mp_policy sets the dtype it "
+                "computes in."
+            )
+    unknown = sorted(set(kwargs) - _SHARDED_LOAD_OPTIONS[route] - {"dtype", "torch_dtype"})
+    if unknown:
+        whole = (
+            " To place the model yourself, have every rank load it whole: MODELOPT_FSDP_SIZE=1, or "
+            f"distribution=DistributionPlan({dist.size()}, dense='replicate')."
+            if route == _FSDP2
+            else ""
+        )
+        raise TypeError(
+            f"{unknown} do not apply to the {route} load, which places the model itself; it "
+            f"takes {sorted(_SHARDED_LOAD_OPTIONS[route])}.{whole}"
+        )
+
+
+def _sharded_load_kwargs(route, ckpt_path, auto_class, model_type, kwargs, hub_options) -> dict:
+    """What the sharded load ``route`` names takes, refusing what it cannot honor."""
+    from modelopt.torch.models.hf import checkpoint_has_mtp
+
+    _refuse_sharded_load_options(route, auto_class, kwargs)
+    kwargs = {k: v for k, v in kwargs.items() if k not in ("dtype", "torch_dtype")}
+    if checkpoint_has_mtp(model_type, ckpt_path):
+        # prepare_model_for_loading builds the head on a whole model only; carried over, the
+        # head would be exported unquantized.
+        raise NotImplementedError(
+            f"The {route} load cannot build the MTP head this {model_type} checkpoint carries; "
+            "load it unsharded."
+        )
+    trust_remote_code = kwargs.pop("trust_remote_code", False)
+    hf_config = kwargs.pop("hf_config", None)
+    if hf_config is None:
+        hf_config = AutoConfig.from_pretrained(
+            ckpt_path, trust_remote_code=trust_remote_code, **hub_options
+        )
+    if getattr(hf_config, "quantization_config", None) is not None:
+        # It reads the tensors as stored: quantized weights would load as if they were not.
+        raise NotImplementedError(
+            f"The {route} load cannot dequantize this {model_type} checkpoint; load it unsharded."
+        )
+    if _auto_class(hf_config, trust_remote_code=trust_remote_code) is not AutoModelForCausalLM:
+        # It builds through AutoModelForCausalLM, which refuses most VLMs' configs and, for some,
+        # builds the language model alone, whose other towers the export would carry unquantized.
+        raise NotImplementedError(
+            f"The {route} load builds a causal LM, which this {model_type} checkpoint is not. To load "
+            "it whole on every rank, pass its model class as auto_class, with MODELOPT_FSDP_SIZE=1 "
+            f"or distribution=DistributionPlan({dist.size()}, dense='replicate')."
+        )
+    if not dist.is_initialized() and "RANK" not in os.environ:
+        # FSDP2 shards through a process group even on one rank; a launch without its group was
+        # refused before the load.
+        raise RuntimeError(
+            f"The {route} load shards through a process group, even on one rank, which a process "
+            "started without a launcher cannot set up; launch it with `torchrun --nproc_per_node=1`."
+        )
+    return {"trust_remote_code": trust_remote_code, "hf_config": hf_config, **kwargs}
+
+
+def _load_sharded(ckpt_path, load_kwargs) -> nn.Module:
+    """The FSDP2 load, by the arguments :func:`_sharded_load_kwargs` checked."""
+    if not dist.is_initialized():
+        # A single process's group of one; a run (distributed_run) destroys it on exit.
+        dist.setup()
+    return parallel_load_and_prepare_fsdp2(
+        ckpt_path,
+        torch.device("cuda", dist.local_rank()),
+        dist.rank(),
+        dist.size(),
+        **load_kwargs,
+    )
+
+
 def load_pretrained(
     ckpt_path: str,
     auto_class: Any = None,
     *,
     model_type: str | None = None,
+    distribution: DistributionPlan | None = None,
     **kwargs,
 ) -> nn.Module:
-    """Load a Hugging Face model for ModelOpt to optimize and export.
-
-    ``auto_class.from_pretrained(ckpt_path, **kwargs)`` (``auto_class`` defaults to
-    ``AutoModelForCausalLM``), plus what ModelOpt needs around the load:
+    """Load a Hugging Face model for ModelOpt to optimize and export, whole or distributed.
+
+    ``distribution`` is the :class:`~modelopt.torch.utils.distributed.DistributionPlan` to spread
+    the model over the run's ranks by. Every rank takes rank 0's, or, when rank 0 passes none,
+    :func:`resolve_distribution_plan`'s. By the plan:
+
+    * ``dense="fsdp"``: a causal LM FSDP2-sharded over the whole process group, each rank reading its
+      own layers (:func:`parallel_load_and_prepare_fsdp2`, whose ``cpu_offload``, ``mp_policy`` and
+      ``broadcast_chunk_size`` ``kwargs`` may set). A single process launched with
+      ``torchrun --nproc_per_node=1`` has no group to shard through, so the load sets up one of one
+      rank, which :func:`~modelopt.torch.utils.distributed.distributed_run` destroys with the run; a
+      process started without a launcher cannot set one up and is refused. It reads the
+      checkpoint's tensors as stored, so it refuses a quantized checkpoint.
+    * otherwise ``auto_class.from_pretrained(ckpt_path, **kwargs)`` (``auto_class`` defaults to
+      ``AutoModelForCausalLM``), placed as ``kwargs`` say. Every rank of a distributed run gets the
+      whole model, ``device_map`` defaulting to the rank's GPU.
+
+    A plan with ``ep_size > 1`` raises: no load here shards the routed experts. On a distributed run
+    every rank loads the plan rank 0 was handed, and a refusal on any rank, of its own arguments, of a
+    checkpoint it cannot see or of the config it reads, raises on every rank. A launch whose processes
+    set no process group up (a launcher that only exports ``WORLD_SIZE``) has each process load as a
+    single process does, the whole model with no plan to agree on; a plan for several ranks and a
+    sharded load need the group, so there they are refused: set it up first, with
+    :func:`~modelopt.torch.utils.distributed.distributed_run`. The plan is recorded on the model
+    (:func:`~modelopt.torch.utils.distributed.set_distribution_plan`), where calibration and export
+    read it back.
+
+    A sharded load builds the model class and places it itself. Besides its own options it takes
+    ``trust_remote_code``, ``attn_implementation`` and ``hf_config``, and refuses any other keyword
+    rather than ignore it. With neither a plan nor ``MODELOPT_FSDP_SIZE``, rank 0's GPU memory
+    decides whether the dense part is sharded, so a call that places the model (``device_map``,
+    ``quantization_config``, a ``dtype``) loads only where it fits one GPU: pass ``distribution=``
+    or set ``MODELOPT_FSDP_SIZE=1`` to keep it whole. The model is weighed in the checkpoint's own
+    dtype (its config's, else bfloat16), not one the load asks for: hand a load that widens it,
+    ``dtype=torch.float32`` say, its plan. ``hf_config`` is the config
+    :func:`resolve_distribution_plan` returned with the plan handed in as ``distribution``: the load
+    builds from it as given, so without that plan it is refused. A ``config`` passed for the load, or
+    a path to one as ``from_pretrained`` takes, is the one the plan is decided from, and a sharded
+    load takes it as its ``hf_config``.
+
+    Whatever the placement:
 
     * A Hub model ID is made local first (:func:`ensure_local_checkpoint`, which takes the
       ``revision``, ``token``, ``cache_dir`` and ``local_files_only`` ``kwargs``): the export reads
       the source checkpoint from disk. Under ``torch.distributed`` every rank must call this. The
-      config and model loads still take all but ``revision``: an ``auto_map`` entry can name
-      custom code in another Hub repo, which Transformers fetches with them. A ``subfolder`` is
-      part of where the checkpoint is: the config, the loading hook and the record below read it,
-      as the load does.
+      config read and the whole-model load still take all but ``revision``: an ``auto_map`` entry
+      can name custom code in another Hub repo, which Transformers fetches with them. A sharded
+      load builds its model from the config with ``from_config``, which takes none, so such code
+      comes with the environment's Hub settings. A ``subfolder`` is part of where the checkpoint
+      is: the config, the loading hook and the record below read it, as the load does.
     * The model type's loading hook runs
       (:func:`~modelopt.torch.models.hf.prepare_model_for_loading`), e.g. to keep the MTP head a
       Nemotron-H checkpoint carries. ``model_type`` selects it; when omitted, the type of the
-      ``config`` passed for the load, else of the checkpoint's config.
+      config passed for the load, else of the checkpoint's config. A sharded load cannot run it, so
+      it refuses a checkpoint the hook would build modules for.
     * The checkpoint tensors the model has no parameter for are recorded
       (:func:`record_unplaced_source_keys`), so the export carries them over unchanged. The model
       is returned alone: ``output_loading_info=True`` is refused. So is a ``variant``: the record
       and the export read the checkpoint's default weight files.
     """
-    if kwargs.pop("output_loading_info", False):
-        raise ValueError(
-            "load_pretrained returns the model alone and does not take output_loading_info=True; "
-            "the checkpoint tensors the model has no parameter for are recorded on it instead "
-            "(see record_unplaced_source_keys)."
-        )
-    if kwargs.get("variant") is not None:
-        raise ValueError(
-            f"load_pretrained does not take variant={kwargs['variant']!r}: the checkpoint tensors "
-            "the model has no parameter for are recorded from, and carried into the export from, "
-            "the checkpoint's default weight files."
-        )
     # Importing modelopt.torch.models registers every model spec; only callers of this need them.
     from modelopt.torch.models.hf import prepare_model_for_loading
 
-    auto_class = auto_class or AutoModelForCausalLM
+    # A launcher can start several processes that set no process group up (hf_ptq without
+    # --use_fsdp2, say): each then loads as a single process does, the whole model with no plan to
+    # agree on, as before plans existed. A plan for several ranks needs the group.
+    alone = dist.is_distributed_launch() and not dist.is_initialized()
+    if alone and isinstance(distribution, DistributionPlan) and distribution.world_size > 1:
+        raise RuntimeError(
+            f"{distribution!r} is for {distribution.world_size} ranks, whose process group this "
+            f"{os.environ['WORLD_SIZE']}-rank launch has not set up; load inside distributed_run(), "
+            "which sets it up."
+        )
+    if alone and isinstance(distribution, DistributionPlan) and _route(distribution) is not None:
+        # One rank of a launch of several cannot set the group up for itself, and a one-rank plan
+        # for a sharded load is no plan for the launch.
+        launch = os.environ["WORLD_SIZE"]
+        raise RuntimeError(
+            f"The {_route(distribution)} load shards through this {launch}-rank launch's process "
+            "group, which is not set up; load inside distributed_run(), which sets it up, with a "
+            f"plan for the launch: DistributionPlan({launch}, dense='fsdp')."
+        )
+    if alone and distribution is None:
+        # The knobs ask to shard over the launch, rather than leave each process the whole model. A
+        # value that is not an EP degree is refused, as the plan decision refuses it.
+        ep, world = env.ep_degree().strip(), int(os.environ["WORLD_SIZE"])
+        if ep.lower() not in ("auto", "0") and not ep.isdigit():
+            raise ValueError(f"MODELOPT_EP_DEGREE={ep!r}: expected auto, 0 (auto) or an EP degree.")
+        asked = [f"MODELOPT_EP_DEGREE={ep}"] if ep.isdigit() and int(ep) > 1 else []
+        if _dense_from_env(world) == "fsdp":
+            asked.append(f"MODELOPT_FSDP_SIZE={env.fsdp_size()}")
+        if asked:
+            raise RuntimeError(
+                f"{' and '.join(asked)} ask to shard the model over this "
+                f"{os.environ['WORLD_SIZE']}-rank launch, whose process group is not set up; load "
+                "inside distributed_run(), which sets it up."
+            )
+    # The config to load with, or, as from_pretrained also takes it, a path to read it from.
+    config = kwargs.get("config")
+    config = config if isinstance(config, (PretrainedConfig, str, os.PathLike)) else None
+
+    def handed() -> DistributionPlan | None:
+        if distribution is None:
+            return None
+        if not isinstance(distribution, DistributionPlan):
+            raise TypeError(f"distribution={distribution!r}: expected a DistributionPlan or None.")
+        _route(distribution)
+        return distribution
+
+    # Every rank takes rank 0's distribution, a plan or none, so every rank runs the same load and
+    # its collectives: a rank going by its own could wait in collectives the others never join, and
+    # ranks loading different plans would build mismatched meshes. Rank 0 checks it before it sends,
+    # so one it refuses raises on every rank instead of leaving them in the broadcast, and before
+    # the checkpoint is fetched.
+    plan = handed() if alone else dist.broadcast_from("distribution", handed)
     hub_options = {
         k: kwargs.pop(k) for k in ("token", "cache_dir", "local_files_only") if k in kwargs
     }
     revision = {"revision": kwargs.pop("revision")} if "revision" in kwargs else {}
-    _, ckpt_path = ensure_local_checkpoint(ckpt_path, **revision, **hub_options)
+    # Each rank's arguments are its own, so each rank checks its own, and a refusal on one fails
+    # every rank before the checkpoint is fetched.
+    with _refused_together("load arguments", alone):
+        if kwargs.pop("output_loading_info", False):
+            raise ValueError(
+                "load_pretrained returns the model alone and does not take "
+                "output_loading_info=True; the checkpoint tensors the model has no parameter for "
+                "are recorded on it instead (see record_unplaced_source_keys)."
+            )
+        if kwargs.get("variant") is not None:
+            raise ValueError(
+                f"load_pretrained does not take variant={kwargs['variant']!r}: the checkpoint "
+                "tensors the model has no parameter for are recorded from, and carried into the "
+                "export from, the checkpoint's default weight files."
+            )
+        if config is not None and "hf_config" in kwargs:
+            raise TypeError("config and hf_config both name the config to load with; pass one.")
+        if "hf_config" in kwargs and (plan is None or _route(plan) is None):
+            # Only a sharded load builds from hf_config, as given, so the plan must be the one it
+            # came with, not one decided here from another config nor one that loads it whole.
+            raise TypeError(
+                "hf_config is the config resolve_distribution_plan returned with a plan that "
+                "shards the model: pass that plan as distribution=, or pass config= to have the "
+                "plan decided from your config."
+            )
+        if plan is not None and _route(plan) is not None:
+            # A handed plan names its load before anything is fetched, and the arguments alone
+            # decide what that load refuses. A config passed for it becomes its hf_config;
+            # subfolder is where the checkpoint is.
+            _refuse_sharded_load_options(
+                _route(plan),
+                auto_class,
+                {
+                    k: v
+                    for k, v in kwargs.items()
+                    if k != "subfolder" and not (k == "config" and config is not None)
+                },
+            )
+    # Rank 0 fetches the checkpoint, and each rank then checks it can see it: one that cannot fails
+    # every rank, rather than leave the others in the load's next collective.
+    with _refused_together("checkpoint fetch", alone):
+        _, ckpt_path = ensure_local_checkpoint(ckpt_path, **revision, **hub_options)
+        if config is not None and not isinstance(config, PretrainedConfig):
+            # Read as the load reads it, so the plan and the loading hook follow the config the
+            # model is built from, which need not be in the checkpoint's directory.
+            trust = kwargs.get("trust_remote_code", False)
+            config = _read_config(config, auto_class, trust, **hub_options)
     subfolder = kwargs.pop("subfolder", None)
     if subfolder:
         ckpt_path = os.path.join(ckpt_path, subfolder)
     trust_remote_code = kwargs.get("trust_remote_code", False)
-    if model_type is None:
-        # The load uses the config passed for it, if any, so the loading hook follows that one.
-        model_type = getattr(kwargs.get("config"), "model_type", None) or (
-            AutoConfig.from_pretrained(
-                ckpt_path, trust_remote_code=trust_remote_code, **hub_options
-            ).model_type
+    hf_config = kwargs.get("hf_config")
+    decided = plan is None and dist.size() > 1
+    if decided:
+        plan, hf_config = resolve_distribution_plan(
+            ckpt_path,
+            config=config,
+            auto_class=auto_class,
+            trust_remote_code=trust_remote_code,
+            **hub_options,
         )
-    with prepare_model_for_loading(
-        model_type,
-        ckpt_path,
-        trust_remote_code,
-        model_class=None if auto_class in (AutoModel, AutoModelForCausalLM) else auto_class,
-    ):
-        model = auto_class.from_pretrained(ckpt_path, **hub_options, **kwargs)
-    unplaced = record_unplaced_source_keys(model, ckpt_path)
+        if _route(plan) is not None:
+            kwargs.setdefault("hf_config", hf_config)
+    elif plan is None:
+        # One rank holds the whole model, whatever the model: there is nothing to decide or read.
+        plan = DistributionPlan()
+    route = _route(plan)
+    # Each rank reads the checkpoint's config itself, so a refusal of what it reads, or of its own
+    # arguments, fails every rank before the load's collectives.
+    with _refused_together("load checks", alone):
+        placed = _whole_load_options(kwargs)
+        if decided and route == _FSDP2 and env.fsdp_size().lower() == "auto" and placed:
+            # Rank 0's GPU memory chose to shard: say so, rather than refuse the options below as
+            # if the caller had asked for the sharded load.
+            raise TypeError(
+                f"{placed} place a whole model, but no plan or MODELOPT_FSDP_SIZE asked for a "
+                f"layout, and by rank 0's GPU memory the model does not fit one GPU, so {plan!r} "
+                "shards it. To load it whole on every rank and place it yourself, set "
+                "MODELOPT_FSDP_SIZE=1 or pass "
+                f"distribution=DistributionPlan({plan.world_size}, dense='replicate')."
+            )
+        if route is not None and config is not None:
+            # The sharded load builds the model from hf_config: the plan's copy of this one, or a
+            # copy of it, since building writes to the config.
+            del kwargs["config"]
+            if "hf_config" not in kwargs:
+                kwargs["hf_config"] = copy.deepcopy(config)
+        if model_type is None:
+            # The load uses the config passed for it, if any, so the loading hook follows that one.
+            if route == _FSDP2 and hf_config is None and config is None:
+                # The FSDP2 load builds from the checkpoint's config too: read it once, for both.
+                hf_config = AutoConfig.from_pretrained(
+                    ckpt_path, trust_remote_code=trust_remote_code, **hub_options
+                )
+                kwargs["hf_config"] = hf_config
+            model_type = getattr(hf_config or config, "model_type", None) or (
+                AutoConfig.from_pretrained(
+                    ckpt_path, trust_remote_code=trust_remote_code, **hub_options
+                ).model_type
+            )
+        load_kwargs = (
+            {}
+            if route is None
+            else _sharded_load_kwargs(route, ckpt_path, auto_class, model_type, kwargs, hub_options)
+        )
+    if route is not None:
+        model = _load_sharded(ckpt_path, load_kwargs)
+        # A sharded load finds the unplaced keys while it places the weights.
+        keys = model._modelopt_unplaced_source_keys
+    else:
+        if plan.world_size > 1:
+            # Each rank takes the whole model onto its own GPU. Left to "auto", every rank's
+            # accelerate could pick the same idle GPU.
+            kwargs.setdefault("device_map", {"": f"cuda:{dist.local_rank()}"})
+        auto_class = auto_class or AutoModelForCausalLM
+        with prepare_model_for_loading(
+            model_type,
+            ckpt_path,
+            trust_remote_code,
+            model_class=None if auto_class in (AutoModel, AutoModelForCausalLM) else auto_class,
+        ):
+            model = auto_class.from_pretrained(ckpt_path, **hub_options, **kwargs)
+        keys = None
+    set_distribution_plan(model, plan)
+    unplaced = record_unplaced_source_keys(model, ckpt_path, keys)
     if unplaced:
         print_rank_0(
             f"{len(unplaced)} checkpoint tensor(s) the model has no parameter for "
@@ -504,6 +850,56 @@ def load_pretrained(
     return model
 
 
+def fsdp2_wrap(model, shard_root=True, mp_policy=None, cpu_offload: bool = False):
+    """Auto-detect a HF causal-LM's decoder layers and FSDP2 ``fully_shard`` each one.
+
+    By default (``shard_root=True``) the root module is wrapped too, so embed/lm_head/norm are
+    sharded instead of replicated per rank; pass ``shard_root=False`` to leave the root replicated
+    (only decoder layers sharded). Returns the detected decoder layers so callers can reuse the
+    detection result.
+
+    Parameters whose dtype differs from the model's dominant one are excluded from the wrap (see
+    :func:`~modelopt.torch.utils.distributed._off_dtype_params`), since FSDP2 rejects a shard group
+    that mixes dtypes. They stay replicated and are moved onto the shards' device, which
+    ``fully_shard`` does not do for the params it ignores.
+
+    It records no distribution plan: :func:`load_pretrained` records the plan it loads by, and a
+    caller that wraps a model itself records one
+    (:func:`~modelopt.torch.utils.distributed.set_distribution_plan`) before calibrating or
+    exporting it.
+    """
+    # Importing the quantization package registers every mode and plugin; only a wrap needs it.
+    from modelopt.torch.quantization.utils.layerwise_calib import LayerActivationCollector
+
+    decoder_layers = LayerActivationCollector.get_decoder_layers(model)
+    if decoder_layers is None:
+        raise RuntimeError(
+            "Could not auto-detect decoder layers; FSDP2 wrap requires a standard HF causal-LM layout."
+        )
+
+    fsdp_kwargs: dict[str, Any] = {"reshard_after_forward": True}
+    if mp_policy is not None:
+        fsdp_kwargs["mp_policy"] = mp_policy
+    if cpu_offload:
+        fsdp_kwargs["offload_policy"] = CPUOffloadPolicy()
+    ignored_params = _off_dtype_params(model)
+    if ignored_params:
+        fsdp_kwargs["ignored_params"] = ignored_params
+
+    # Snapshot/restore config.architectures: some HF builders mutate it during fully_shard.
+    config = getattr(model, "config", None)
+    architectures = list(getattr(config, "architectures", []) or [])
+    for layer in decoder_layers:
+        fully_shard(layer, **fsdp_kwargs)
+    if shard_root:
+        fully_shard(model, **fsdp_kwargs)
+    if ignored_params:
+        _move_to_fsdp_device(model, ignored_params)
+    if config is not None and architectures:
+        config.architectures = architectures
+    return decoder_layers
+
+
 def _map_ckpt_keys_to_model(model: "torch.nn.Module", ckpt_keys, model_keys) -> dict:
     """Map each checkpoint key to the model ``state_dict`` key it feeds.
 
@@ -830,6 +1226,8 @@ def parallel_load_and_prepare_fsdp2(
     ``broadcast_chunk_size`` sets how many of a source's owned layers are broadcast per collective:
     a smaller value lowers the peak transient GPU memory at the cost of more collectives (default 8;
     pass ``None`` to broadcast all of a source's layers at once).
+
+    It records no distribution plan, as :func:`fsdp2_wrap` does not; :func:`load_pretrained` does.
     """
     resolved_path = _resolve_checkpoint_dir(ckpt_path, rank)
     weight_map = indexed_weight_map(resolved_path)
@@ -1005,7 +1403,7 @@ def _ep_degree(ep_size) -> int | str:
 
 
 def _decide_distribution(
-    ckpt_path: str, hf_config, world_size: int, ep_size, dense, trust_remote_code: bool
+    ckpt_path: str, hf_config, world_size: int, ep_size, dense, trust_remote_code: bool, auto_class
 ) -> DistributionPlan:
     """The plan :func:`resolve_distribution_plan` returns, decided by this rank alone."""
     # What the caller set the EP degree with, for a refusal to name it.
@@ -1021,6 +1419,27 @@ def _decide_distribution(
         )
     ep_size = _ep_degree(ep_size)
     dense = _dense_from_env(world_size) if dense is None else dense
+    if (
+        auto_class not in (None, AutoModelForCausalLM)
+        and ep_size in ("auto", 1)
+        and dense in ("auto", "replicate")
+    ):
+        # The sharded loads build their own model class and refuse another, so with nothing asking
+        # to shard, a load with this one gives every rank the whole model: there is nothing to
+        # measure, nor a model to build for it.
+        return DistributionPlan(world_size)
+    if auto_class not in (None, AutoModelForCausalLM):
+        # Something asks to shard, and every sharded load refuses this class: refuse it here, rather
+        # than build the model of the decision's own pick of class, which may not build at all.
+        raise NotImplementedError(
+            f"A sharded load builds its model class itself, so auto_class={auto_class.__name__} "
+            f"does not apply to it, and ep_size={ep_size!r}, dense={dense!r} ask to shard. To give "
+            "every rank the whole model, plan ep_size=1 and dense='replicate' "
+            "(MODELOPT_EP_DEGREE=1, MODELOPT_FSDP_SIZE=1)."
+        )
+    # Building writes to the config it builds from, and only rank 0 decides: build from a copy, so
+    # the config returned with the plan, which a sharded load builds from, is alike on every rank.
+    hf_config = copy.deepcopy(hf_config)
     try:
         model = build_meta_causal_lm(
             ckpt_path,
@@ -1088,6 +1507,8 @@ def _decide_distribution(
 def resolve_distribution_plan(
     ckpt_path: str,
     *,
+    config: PretrainedConfig | None = None,
+    auto_class: Any = None,
     trust_remote_code: bool = False,
     ep_size: int | str | None = None,
     dense: str | None = None,
@@ -1104,7 +1525,9 @@ def resolve_distribution_plan(
     every later step reads it back from there
     (:func:`~modelopt.torch.utils.distributed.distribution_plan`). Returns ``(plan, hf_config)``,
     ``hf_config`` being this rank's own copy of the config (rank 0 decides the plan from its copy),
-    for a sharded load to build from.
+    for a sharded load to build from: hand it to :func:`load_pretrained` only with a plan that
+    shards the model, as a whole-model load reads the checkpoint's config itself, and refuses
+    ``hf_config``.
 
     Rank 0 decides, from the model built on the meta device: from its parameters rather than
     estimates of them, and from its routed experts as an expert-parallel load finds them. Every rank
@@ -1131,6 +1554,14 @@ def resolve_distribution_plan(
     returns no path, so a caller that loads the checkpoint resolves it first, with
     :func:`ensure_local_checkpoint`, and passes the local directory: the plan is then decided for
     the snapshot the load reads.
+
+    ``auto_class`` is the class the load builds the model with, :func:`load_pretrained`'s. The
+    sharded loads build their own and refuse any other than ``AutoModelForCausalLM``, so with
+    another, and no knob or argument asking to shard, every rank loads the whole model, planned
+    without building it.
+
+    ``config``, the config the load will use when the caller has one, is decided from instead of
+    the checkpoint's, which is then not read: each rank takes a copy, as building writes to it.
     """
     # Each rank reads the checkpoint for itself. One that cannot must fail every rank here, rather
     # than leave the others waiting in a broadcast below for a rank that already raised.
@@ -1146,14 +1577,18 @@ def checkpoint_read() -> bool:
         )
         if subfolder:
             local_path = os.path.join(local_path, subfolder)
-        config = AutoConfig.from_pretrained(
+        if config is not None:
+            read.append((local_path, copy.deepcopy(config)))
+            return True
+        hf_config = _read_config(
             local_path,
+            auto_class,
             trust_remote_code=trust_remote_code,
             token=token,
             cache_dir=cache_dir,
             local_files_only=local_files_only,
         )
-        read.append((local_path, config))
+        read.append((local_path, hf_config))
         return True
 
     dist.all_reduce("checkpoint read", checkpoint_read, all)
@@ -1162,7 +1597,7 @@ def checkpoint_read() -> bool:
     plan = dist.broadcast_from(
         "plan",
         lambda: _decide_distribution(
-            ckpt_path, hf_config, world_size, ep_size, dense, trust_remote_code
+            ckpt_path, hf_config, world_size, ep_size, dense, trust_remote_code, auto_class
         ),
     )
     if plan != DistributionPlan():
diff --git a/tests/examples/hf_ptq/test_example_utils.py b/tests/examples/hf_ptq/test_example_utils.py
index caaa625805b..990126b3ef0 100644
--- a/tests/examples/hf_ptq/test_example_utils.py
+++ b/tests/examples/hf_ptq/test_example_utils.py
@@ -253,7 +253,7 @@ def test_get_model_uses_expected_dtype_kwarg(
         torch_dtype=torch.bfloat16,
     )
 
-    class FakeModel:
+    class FakeModel(torch.nn.Module):  # load_pretrained records its plan on every module
         def eval(self):
             calls["eval"] = True
 
@@ -338,7 +338,7 @@ def test_get_model_device_map_for_diffusion_gemma(
         torch_dtype=torch.bfloat16,
     )
 
-    class FakeModel:
+    class FakeModel(torch.nn.Module):  # load_pretrained records its plan on every module
         def eval(self):
             calls["eval"] = True
 
@@ -467,7 +467,7 @@ def _load_config(*args, **kwargs):
             raise builtin_config_error
         return hf_config
 
-    class FakeModel:
+    class FakeModel(torch.nn.Module):  # load_pretrained records its plan on every module
         def eval(self):
             return None
 
diff --git a/tests/gpu/torch/export/test_export_carry_over.py b/tests/gpu/torch/export/test_export_carry_over.py
index fa528f4cdf9..34a54d23f5e 100644
--- a/tests/gpu/torch/export/test_export_carry_over.py
+++ b/tests/gpu/torch/export/test_export_carry_over.py
@@ -40,7 +40,12 @@
 import modelopt.torch.quantization as mtq
 from modelopt.torch.export.unified_export_hf import export_hf_checkpoint
 from modelopt.torch.quantization.utils import patch_fsdp_mp_dtypes
-from modelopt.torch.utils.distributed import fsdp2_wrap, is_fsdp2_model
+from modelopt.torch.utils.distributed import (
+    DistributionPlan,
+    fsdp2_wrap,
+    is_fsdp2_model,
+    set_distribution_plan,
+)
 
 # Small enough to force the tiny model across several shards, so the multi-file layout -- and the
 # weight index that makes it loadable -- are exercised rather than collapsing to one model.safetensors.
@@ -108,6 +113,8 @@ def _ptq_and_export(rank, size, *, src_dir, export_dir, quant_cfg, **export_kwar
         model.eval()
 
         fsdp2_wrap(model)
+        # Wrapped by hand, the model records how it is distributed, as load_pretrained would.
+        set_distribution_plan(model, DistributionPlan(size, dense="fsdp"))
         # The FSDP2 export path is selected by exactly this predicate, so assert it here: without it
         # the export would silently fall back to the rank-0 gather and the test would pass while
         # never touching the code under test.
@@ -201,6 +208,7 @@ def _ptq_language_model_and_export(rank, size, *, src_dir, export_dir):
         )
         model.eval()
         fsdp2_wrap(model)
+        set_distribution_plan(model, DistributionPlan(size, dense="fsdp"))
         assert is_fsdp2_model(model), "fsdp2_wrap did not shard the VLM"
         torch.distributed.barrier()
 
diff --git a/tests/gpu/torch/utils/test_model_load_utils.py b/tests/gpu/torch/utils/test_model_load_utils.py
index 3296ccbfc86..54b96461485 100644
--- a/tests/gpu/torch/utils/test_model_load_utils.py
+++ b/tests/gpu/torch/utils/test_model_load_utils.py
@@ -13,7 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""GPU/distributed tests for the FSDP2 load path and its helpers."""
+"""GPU/distributed tests for the loads: ``load_pretrained`` by its plan, the FSDP2 load, and their
+helpers."""
 
 import json
 import os
@@ -24,10 +25,19 @@
 from _test_utils.torch.transformers_models import create_tiny_llama_dir
 from safetensors.torch import load_file
 from torch.distributed.tensor import DTensor
+from transformers import AutoModelForCausalLM, LlamaConfig
 
 from modelopt.torch.export.unified_export_hf import export_hf_checkpoint
-from modelopt.torch.utils.distributed import broadcast_state_dict
-from modelopt.torch.utils.plugins.model_load_utils import parallel_load_and_prepare_fsdp2
+from modelopt.torch.utils.distributed import (
+    DistributionPlan,
+    broadcast_state_dict,
+    distribution_plan,
+    set_distribution_plan,
+)
+from modelopt.torch.utils.plugins.model_load_utils import (
+    load_pretrained,
+    parallel_load_and_prepare_fsdp2,
+)
 
 VOCAB_SIZE = 64
 
@@ -61,6 +71,7 @@ def _test_parallel_load_and_export(rank, size, ckpt_dir, export_dir, cpu_offload
         size,
         cpu_offload=cpu_offload,
     )
+    set_distribution_plan(model, DistributionPlan(size, dense="fsdp"))
 
     # Decoder layers AND root params (embed/lm_head) are sharded DTensors under shard_root=True.
     decoder_params = list(model.model.layers[0].parameters())
@@ -106,6 +117,7 @@ def _test_carry_over_unplaced_weights(rank, size, ckpt_dir, export_dir, orphan_p
     """Load, export, and require the weights the model could not place to survive."""
     device = torch.device(f"cuda:{rank}")
     model = parallel_load_and_prepare_fsdp2(ckpt_dir, device, rank, size)
+    set_distribution_plan(model, DistributionPlan(size, dense="fsdp"))
 
     unplaced = getattr(model, "_modelopt_unplaced_source_keys", None)
     assert unplaced, (
@@ -171,3 +183,68 @@ def test_carry_over_unplaced_weights(dist_workers, tmp_path):
             orphan_prefix=f"model.layers.{orphan_idx}.",
         )
     )
+
+
+# ------------------------------------------------------------------ load_pretrained, left to decide
+
+_TINY = {
+    "hidden_size": 64,
+    "intermediate_size": 128,
+    "num_hidden_layers": 2,
+    "num_attention_heads": 4,
+    "num_key_value_heads": 2,
+    "vocab_size": 128,
+    "max_position_embeddings": 64,
+}
+
+
+def _save(config, ckpt_dir):
+    torch.manual_seed(0)
+    AutoModelForCausalLM.from_config(config).to(torch.bfloat16).save_pretrained(ckpt_dir)
+
+
+def _full(tensor):
+    return tensor.full_tensor() if isinstance(tensor, DTensor) else tensor
+
+
+def _load_and_check(rank, size, ckpt_dir, knobs, expected):
+    """Load on every rank of the run with the plan left to decide, and check what each rank holds.
+
+    Left to decide, ``load_pretrained`` FSDP2-shards a dense model when asked to, and otherwise gives
+    every rank the whole model on its own GPU. Whichever load runs, every rank ends up with the
+    checkpoint's weights and the load's record of what it could not place.
+    """
+    device = torch.device(f"cuda:{rank}")
+    os.environ.update(knobs)
+    try:
+        model = load_pretrained(ckpt_dir)
+        whole = AutoModelForCausalLM.from_pretrained(ckpt_dir, dtype=torch.bfloat16)
+
+        params = dict(model.named_parameters())
+        whole_params = dict(whole.named_parameters())
+        assert params.keys() == whole_params.keys()
+        sharded = {name for name, param in params.items() if isinstance(param, DTensor)}
+        if expected == "replicated":
+            assert not sharded
+            assert all(param.device == device for param in params.values())
+        else:
+            assert {name for name in params if ".layers." in name} <= sharded
+        for name, param in params.items():  # the same order on every rank: full_tensor gathers
+            torch.testing.assert_close(_full(param).cpu(), whole_params[name])
+        assert model._modelopt_unplaced_source_keys == []
+        assert model._modelopt_source_checkpoint == ckpt_dir
+        dense = "replicate" if expected == "replicated" else "fsdp"
+        assert distribution_plan(model) == DistributionPlan(size, dense=dense)
+    finally:
+        for knob in knobs:
+            del os.environ[knob]
+
+
+def test_a_dense_model_loads_whole_on_every_rank(need_2_gpus, dist_workers, tmp_path):
+    _save(LlamaConfig(**_TINY), tmp_path)
+    dist_workers.run(_load_and_check, str(tmp_path), {}, "replicated")
+
+
+def test_a_dense_model_asked_to_shard_loads_fsdp2_sharded(need_2_gpus, dist_workers, tmp_path):
+    _save(LlamaConfig(**_TINY), tmp_path)
+    dist_workers.run(_load_and_check, str(tmp_path), {"MODELOPT_FSDP_SIZE": "world"}, "fsdp2")
diff --git a/tests/unit/torch/utils/test_distributed.py b/tests/unit/torch/utils/test_distributed.py
index 25b3bc12268..df1d0066f20 100644
--- a/tests/unit/torch/utils/test_distributed.py
+++ b/tests/unit/torch/utils/test_distributed.py
@@ -192,6 +192,21 @@ def test_a_model_with_no_plan_raises_in_a_run_of_several_ranks(cpu_dist_workers)
     cpu_dist_workers.run(_no_plan_on_several_ranks_job)
 
 
+def test_fsdp2_wrap_keeps_its_old_path(monkeypatch):
+    """fsdp2_wrap lives with the loaders that wrap with it; callers of the old path still reach it."""
+    model_load_utils = pytest.importorskip("modelopt.torch.utils.plugins.model_load_utils")
+    calls = []
+    monkeypatch.setattr(
+        model_load_utils,
+        "fsdp2_wrap",
+        lambda model, **kwargs: calls.append((model, kwargs)) or "layers",
+    )
+    model = nn.Linear(2, 2)
+
+    assert dist.fsdp2_wrap(model, shard_root=False, cpu_offload=True) == "layers"
+    assert calls == [(model, {"shard_root": False, "mp_policy": None, "cpu_offload": True})]
+
+
 # ------------------------------------------------------------------------------------ collectives
 
 
diff --git a/tests/unit/torch/utils/test_model_load_utils.py b/tests/unit/torch/utils/test_model_load_utils.py
index 2c225dafc14..ba56e4635d0 100644
--- a/tests/unit/torch/utils/test_model_load_utils.py
+++ b/tests/unit/torch/utils/test_model_load_utils.py
@@ -15,12 +15,14 @@
 
 """Tests for ``modelopt.torch.utils.plugins.model_load_utils``.
 
-Nothing here needs a GPU. The tests of ``resolve_distribution_plan`` answer the process group's rank
-and size as a torchrun launch would, and serve tiny real transformers configs from memory, so the
-decision builds their real models on the meta device; only the last ones use a process group, two
-gloo ranks.
+Nothing here needs a GPU. The tests of ``resolve_distribution_plan`` and of the loads
+``load_pretrained`` routes a plan to answer the process group's rank and size as a torchrun launch
+would, and serve tiny real transformers configs from memory: the decision builds their real models on
+the meta device, and stand-ins for the sharded loaders record how they were called. Only the last
+tests of each part use a process group, two gloo ranks.
 """
 
+import inspect
 import math
 import os
 from contextlib import contextmanager
@@ -49,6 +51,7 @@
     AutoModelForSeq2SeqLM,
     AutoModelForSpeechSeq2Seq,
     LlamaConfig,
+    LlamaForCausalLM,
     Phi4MultimodalConfig,
     Qwen2_5_VLConfig,
     Qwen3MoeConfig,
@@ -56,9 +59,10 @@
     WhisperConfig,
 )
 
+import modelopt.torch.models.hf as models_hf
 import modelopt.torch.utils.distributed as dist
 from modelopt.torch.quantization.utils.layerwise_calib import LayerActivationCollector
-from modelopt.torch.utils.distributed import DistributionPlan
+from modelopt.torch.utils.distributed import DistributionPlan, distribution_plan
 from modelopt.torch.utils.plugins import model_load_utils
 from modelopt.torch.utils.plugins.model_load_utils import (
     _auto_class,
@@ -66,6 +70,7 @@
     _convert_keys,
     _resolve_target,
     build_meta_causal_lm,
+    load_pretrained,
     record_unplaced_source_keys,
     resolve_distribution_plan,
     unplaced_source_keys,
@@ -236,17 +241,27 @@ def _gathered_alike(gathered, mine, group=None):
 def run_as(monkeypatch):
     """Run as rank ``rank`` of a ``world_size``-rank launch, deciding for itself.
 
-    Every rank reads the checkpoint as this one does. How the ranks agree on one plan is tested at
-    the end of this file.
+    Every rank acts as its group's first, and reads the checkpoint as this one does. How the ranks
+    agree on one plan, and on one load, is tested on two gloo ranks at the end of each part.
     """
     monkeypatch.setattr(torch.distributed, "get_rank", lambda group=None: 0)
     monkeypatch.setattr(torch.distributed, "broadcast_object_list", lambda *a, **k: None)
     monkeypatch.setattr(torch.distributed, "all_gather_object", _gathered_alike)
+    broadcast_from = dist.broadcast_from
+
+    def broadcast_as_the_first(label, fn, src=0):
+        # broadcast_from sends from dist.rank(), which process_group sets for the FSDP2 load.
+        with monkeypatch.context() as sending:
+            sending.setattr(dist, "rank", lambda group=None: src)
+            return broadcast_from(label, fn, src)
+
+    monkeypatch.setattr(dist, "broadcast_from", broadcast_as_the_first)
 
     def enter(world_size, rank=0):
         monkeypatch.setattr(dist, "is_distributed_run", lambda: world_size > 1)
         monkeypatch.setattr(dist, "size", lambda group=None: world_size)
         monkeypatch.setattr(dist, "is_master", lambda group=None: rank == 0)
+        monkeypatch.setattr(dist, "local_rank", lambda: rank)
         monkeypatch.setattr(torch.distributed, "get_world_size", lambda group=None: world_size)
 
     return enter
@@ -714,6 +729,17 @@ def remote_from_config(config, **kwargs):
     assert ProbeConfig not in mapping
 
 
+def test_the_config_returned_with_the_plan_is_as_every_rank_read_it(run_as, serve, tmp_path):
+    """Rank 0 alone decides, building the model, which writes to the config it builds from; a
+    sharded load builds from the config returned, so rank 0's must be the one the others read."""
+    run_as(2)
+    serve(LlamaConfig(**_TINY))
+
+    _, hf_config = resolve_distribution_plan(str(tmp_path))
+
+    assert vars(hf_config) == vars(LlamaConfig(**_TINY))
+
+
 def test_the_hub_options_locate_the_checkpoint_the_load_reads(tmp_path):
     """``revision``, ``cache_dir``, ``local_files_only`` and ``subfolder`` find the checkpoint as
     ``load_pretrained`` takes them, so the plan and its config are those of what the load reads."""
@@ -770,23 +796,647 @@ def test_every_rank_takes_the_plan_rank0_decided(cpu_dist_workers, tmp_path):
     )
 
 
-def _one_rank_cannot_see_the_checkpoint_job(rank, size, ckpt):
+def _one_rank_cannot_see_the_checkpoint_job(rank, size, ckpt, load):
     isdir = os.path.isdir
     with (
         bounded_collectives(),
         mock.patch("os.path.isdir", lambda path: (rank == 0 or path != ckpt) and isdir(path)),
     ):
         try:
-            resolve_distribution_plan(ckpt)
+            if load:
+                load_pretrained(ckpt, distribution=DistributionPlan(size), device_map="cpu")
+            else:
+                resolve_distribution_plan(ckpt)
         except RuntimeError as error:
             raised = str(error)
         else:
             raised = "no error"
-    # Rank 1's own error, carried to rank 0, rather than rank 0 timing out in the plan's broadcast.
+    # Rank 1's own error, carried to rank 0, rather than rank 0 timing out in the next collective.
     assert "which this rank cannot see" in raised, raised
 
 
-def test_a_rank_that_cannot_see_the_checkpoint_fails_every_rank(cpu_dist_workers, tmp_path):
-    """Before the plan's broadcast, which rank 0 would otherwise wait in for a rank that raised."""
+@pytest.mark.parametrize("load", [False, True], ids=["plan", "load"])
+def test_a_rank_that_cannot_see_the_checkpoint_fails_every_rank(cpu_dist_workers, tmp_path, load):
+    """Deciding the plan, or loading by a handed one: rank 0 would otherwise wait for the rank that
+    raised in the next collective, the plan's broadcast or the load's checks. The group is the
+    caller's and the error is caught, so nothing else ends rank 0's wait."""
+    LlamaConfig(**_TINY).save_pretrained(tmp_path)
+    cpu_dist_workers.run(_one_rank_cannot_see_the_checkpoint_job, ckpt=str(tmp_path), load=load)
+
+
+# ------------------------------------------------------------------------------ loading by the plan
+
+
+class _Loaded(torch.nn.Module):
+    """What a sharded loader returns: a model whose load recorded what it could not place."""
+
+    def __init__(self):
+        super().__init__()
+        self._modelopt_unplaced_source_keys = []
+
+
+@pytest.fixture
+def loaders(monkeypatch):
+    """Stand-ins for the sharded loaders, each recording its call by parameter name."""
+    calls = {}
+
+    def recording(name, loader):
+        def stub(*args, **kwargs):
+            calls[name] = inspect.signature(loader).bind(*args, **kwargs).arguments
+            return _Loaded()
+
+        return stub
+
+    monkeypatch.setattr(
+        model_load_utils,
+        "parallel_load_and_prepare_fsdp2",
+        recording("fsdp2", model_load_utils.parallel_load_and_prepare_fsdp2),
+    )
+    return calls
+
+
+@pytest.fixture
+def auto_class():
+    """An ``auto_class`` whose from_pretrained records its arguments."""
+
+    class Auto:
+        calls: list = []
+
+        @classmethod
+        def from_pretrained(cls, ckpt_path, **kwargs):
+            cls.calls.append(kwargs)
+            return torch.nn.Linear(1, 1)
+
+    return Auto
+
+
+@pytest.fixture
+def process_group(monkeypatch):
+    """Rank 1 of a 2-rank process group, as the FSDP2 load reads it."""
+    monkeypatch.setattr(dist, "is_initialized", lambda: True)
+    monkeypatch.setattr(dist, "rank", lambda group=None: 1)
+    monkeypatch.setattr(dist, "size", lambda group=None: 2)
+    monkeypatch.setattr(dist, "local_rank", lambda: 1)
+
+
+# ----------------------------------------------------------------------------------- the loads
+
+
+@pytest.mark.parametrize(
+    ("device_map", "placed"),
+    [(None, {"": "cuda:1"}), ("cpu", "cpu")],
+    ids=["defaults_to_the_rank_gpu", "caller_placement_wins"],
+)
+def test_a_replicated_plan_loads_the_whole_model_on_every_rank(
+    run_as, serve, loaders, auto_class, tmp_path, device_map, placed
+):
+    """Each rank takes the whole model onto its own GPU; "auto" would let every rank's accelerate
+    pick the same idle GPU."""
+    run_as(4, rank=1)
+    serve(LlamaConfig(**_TINY))
+    kwargs = {} if device_map is None else {"device_map": device_map}
+
+    model = load_pretrained(str(tmp_path), auto_class, **kwargs)
+
+    assert auto_class.calls == [{"device_map": placed}]
+    assert distribution_plan(model) == DistributionPlan(4)
+    assert loaders == {}
+
+
+@pytest.mark.parametrize(
+    "dtype",
+    [{"dtype": "auto"}, {"dtype": None}, {"torch_dtype": "auto"}],
+    ids=["auto", "none", "torch_dtype"],
+)
+def test_an_fsdp_plan_loads_through_the_parallel_fsdp2_loader(
+    run_as, loaders, process_group, monkeypatch, tmp_path, dtype
+):
+    """FSDP2 shards over the process group it reads its rank and GPU from, by a config read once,
+    with the Hub options the load was given (custom code an ``auto_map`` names in another repo).
+    It keeps the checkpoint's dtype, as from_pretrained does for "auto" and for None, under either
+    name of the argument."""
+    run_as(2, rank=1)
+    served, reads = LlamaConfig(**_TINY), []
+    monkeypatch.setattr(
+        transformers.AutoConfig, "from_pretrained", lambda *a, **k: reads.append(k) or served
+    )
+    plan = DistributionPlan(2, dense="fsdp")
+
+    model = load_pretrained(
+        str(tmp_path), distribution=plan, cpu_offload=True, local_files_only=True, **dtype
+    )
+
+    call = loaders.pop("fsdp2")
+    assert (call["device"], call["rank"], call["world_size"]) == (torch.device("cuda:1"), 1, 2)
+    assert call["cpu_offload"] is True
+    assert call["hf_config"] is served
+    assert [read.get("local_files_only") for read in reads] == [True]
+    assert distribution_plan(model) == plan
+    assert loaders == {}
+
+
+def test_a_load_it_decides_reads_the_config_once(
+    run_as, loaders, process_group, monkeypatch, tmp_path
+):
+    """The config the plan was decided from is the one the FSDP2 load gets, not a second read."""
+    monkeypatch.setenv("MODELOPT_FSDP_SIZE", "world")
+    run_as(2, rank=1)
+    config, reads = LlamaConfig(**_TINY), []
+    monkeypatch.setattr(
+        transformers.AutoConfig, "from_pretrained", lambda *a, **k: reads.append(a) or config
+    )
+
+    model = load_pretrained(str(tmp_path))
+
+    assert len(reads) == 1
+    assert loaders["fsdp2"]["hf_config"] is config
+    assert distribution_plan(model) == DistributionPlan(2, dense="fsdp")
+
+
+@pytest.mark.parametrize("handed", [False, True], ids=["decided", "handed"])
+def test_a_config_passed_for_the_load_is_the_one_it_decides_from(
+    run_as, loaders, process_group, monkeypatch, tmp_path, handed
+):
+    """No config is read: the plan is decided from the one passed for the load, which the FSDP2
+    load then builds from, as it does with a plan handed in. Building writes to that config, so the
+    load works on a copy of it."""
+    monkeypatch.setenv("MODELOPT_FSDP_SIZE", "world")
+    run_as(2, rank=1)
+    reads: list = []
+    monkeypatch.setattr(transformers.AutoConfig, "from_pretrained", lambda *a, **k: reads.append(a))
+    config = LlamaConfig(**{**_TINY, "num_hidden_layers": 2})
+    as_passed = config.to_dict()
+    plan = {"distribution": DistributionPlan(2, dense="fsdp")} if handed else {}
+
+    model = load_pretrained(str(tmp_path), config=config, **plan)
+
+    assert reads == []
+    assert loaders["fsdp2"]["hf_config"].num_hidden_layers == 2
+    assert loaders["fsdp2"]["hf_config"] is not config  # building it writes to it
+    assert config.to_dict() == as_passed
+    assert distribution_plan(model) == DistributionPlan(2, dense="fsdp")
+
+
+def test_hf_config_is_refused_without_the_plan_it_came_with(
+    run_as, serve, loaders, process_group, monkeypatch, tmp_path
+):
+    """A sharded load builds from hf_config as given, so the plan must be the one it came with: a
+    plan decided from the checkpoint's own config would load a model it was not decided for."""
+    monkeypatch.setenv("MODELOPT_FSDP_SIZE", "world")
+    run_as(2, rank=1)
+    serve(LlamaConfig(**_TINY))
+
+    with pytest.raises(TypeError, match="returned with a plan that shards the model: pass that"):
+        load_pretrained(str(tmp_path), hf_config=LlamaConfig(**{**_TINY, "num_hidden_layers": 2}))
+    assert loaders == {}
+
+
+class _OwnConfig(LlamaConfig):
+    """A config class of the caller's own, which AutoConfig has no mapping for."""
+
+    model_type = "modelopt_test_own_llama"
+
+
+@pytest.mark.parametrize(
+    ("fsdp_size", "own_config"),
+    [(None, False), ("world", False), (None, True)],
+    ids=["no-knob", "knob-shards", "own-config-class"],
+)
+def test_a_model_class_of_the_callers_own_loads_whole(
+    run_as, serve, loaders, auto_class, monkeypatch, tmp_path, fsdp_size, own_config
+):
+    """The sharded loads build their own model class and refuse another, so with nothing asking to
+    shard, a class of the caller's own loads whole on every rank, and with a knob asking to shard
+    it is refused. Neither builds the model on the meta device, which the decision's own pick of
+    class may not build (an encoder-decoder loaded through AutoModel). The decision reads the
+    checkpoint's config as the class does, through its own config class, which AutoConfig need not
+    know."""
+    run_as(2)
+    if own_config:
+        _OwnConfig(**_TINY).save_pretrained(tmp_path)
+        auto_class.config_class = _OwnConfig
+    else:
+        serve(LlamaConfig(**_TINY))
+    monkeypatch.setattr(
+        model_load_utils, "build_meta_causal_lm", lambda *a, **k: pytest.fail("built a model")
+    )
+    if fsdp_size is None:
+        model = load_pretrained(str(tmp_path), auto_class)
+        assert distribution_plan(model) == DistributionPlan(2)
+        assert len(auto_class.calls) == 1
+    else:
+        monkeypatch.setenv("MODELOPT_FSDP_SIZE", fsdp_size)
+        with pytest.raises(NotImplementedError, match="builds its model class itself"):
+            load_pretrained(str(tmp_path), auto_class)
+    assert loaders == {}
+
+
+def test_an_expert_parallel_plan_is_refused(run_as, serve, loaders, auto_class, tmp_path):
+    """No load here shards the routed experts: an EP plan fails before anything loads."""
+    run_as(2)
+    serve(Qwen3MoeConfig(num_experts=8, moe_intermediate_size=8, **_TINY))
+
+    with pytest.raises(NotImplementedError, match="shards the routed experts"):
+        load_pretrained(str(tmp_path), auto_class, distribution=DistributionPlan(2, ep_size=2))
+    assert loaders == {}
+    assert auto_class.calls == []
+
+
+def _every_rank_loads_rank0s_plan_job(rank, size, ckpt, handed):
+    model, refused = None, None
+    with bounded_collectives():
+        try:  # not pytest.raises: its failure would end the pooled worker instead of the job
+            model = load_pretrained(ckpt, distribution=handed[rank], device_map="cpu")
+        except Exception as error:
+            refused = str(error)
+    if isinstance(handed[0], DistributionPlan):
+        assert refused is None and distribution_plan(model) == handed[0], (rank, refused)
+    else:
+        # Rank 0's refusal, on every rank, rather than a rank left waiting in the broadcast.
+        assert "expected a DistributionPlan or None" in str(refused), (rank, refused)
+
+
+@pytest.mark.parametrize(
+    "handed",
+    [
+        (DistributionPlan(2), DistributionPlan(2, dense="fsdp")),
+        (DistributionPlan(2), None),
+        ("fsdp2", DistributionPlan(2)),
+    ],
+    ids=["another-plan", "none", "not-a-plan"],
+)
+def test_every_rank_loads_the_plan_rank0_was_handed(cpu_dist_workers, tmp_path, handed):
+    """Over a real broadcast, whatever rank 1 was handed. Ranks loading different plans would build
+    mismatched meshes, and one handed none would decide a plan through collectives the others never
+    join: both would hang. A distribution rank 0 refuses fails every rank."""
+    LlamaForCausalLM(LlamaConfig(**_TINY)).save_pretrained(tmp_path)
+    cpu_dist_workers.run(_every_rank_loads_rank0s_plan_job, ckpt=str(tmp_path), handed=handed)
+
+
+def _fetched_in_step(ckpt_path, **hub_options):
+    """The checkpoint's fetch, as the collective it is: a rank not in it leaves the others waiting."""
+    torch.distributed.barrier()
+    return None, ckpt_path
+
+
+def _loaded_in_step(*args, **kwargs):
+    """The FSDP2 load, as the collectives it runs."""
+    torch.distributed.barrier()
+    return _Loaded()
+
+
+def _one_rank_refused_job(rank, size, ckpt, refused_on_1, match):
+    refused = None
+    with (
+        bounded_collectives(),
+        mock.patch.object(model_load_utils, "ensure_local_checkpoint", _fetched_in_step),
+        mock.patch.object(model_load_utils, "parallel_load_and_prepare_fsdp2", _loaded_in_step),
+    ):
+        try:  # not pytest.raises: its failure would end the pooled worker instead of the job
+            load_pretrained(
+                ckpt,
+                distribution=DistributionPlan(2, dense="fsdp"),
+                **(refused_on_1 if rank == 1 else {}),
+            )
+        except Exception as error:
+            refused = str(error)
+    # Rank 1's refusal on both ranks, rather than rank 0 failing at the group's timeout.
+    assert refused is not None and match in refused, (rank, refused)
+    assert rank == 1 or "rank 1 failed" in refused, refused
+
+
+@pytest.mark.parametrize(
+    ("refused_on_1", "match"),
+    [
+        ({"config": LlamaConfig(**_TINY), "hf_config": LlamaConfig(**_TINY)}, "both name"),
+        ({"dtype": torch.float32}, "keeps their dtype"),
+        (
+            {"hf_config": LlamaConfig(**_TINY, quantization_config={"quant_method": "fp8"})},
+            "cannot dequantize",
+        ),
+    ],
+    ids=["two-configs", "dtype", "quantized-config"],
+)
+def test_a_load_refused_on_one_rank_fails_every_rank(
+    cpu_dist_workers, tmp_path, refused_on_1, match
+):
+    """Rank 1 alone is refused, by its arguments before the checkpoint is fetched or by the config
+    it was handed before the FSDP2 load: every rank raises the refusal, rather than leave rank 0
+    waiting in the fetch's or the load's collectives, which stand-ins on a bounded group run here."""
     LlamaConfig(**_TINY).save_pretrained(tmp_path)
-    cpu_dist_workers.run(_one_rank_cannot_see_the_checkpoint_job, ckpt=str(tmp_path))
+    cpu_dist_workers.run(
+        _one_rank_refused_job, ckpt=str(tmp_path), refused_on_1=refused_on_1, match=match
+    )
+
+
+@pytest.mark.parametrize(
+    ("change", "error", "match"),
+    [
+        ({"kwargs": {"device_map": "auto"}}, TypeError, r"\['device_map'\] do not apply"),
+        ({"kwargs": {"dtype": torch.bfloat16}}, TypeError, "keeps their dtype"),
+        (
+            {"kwargs": {"torch_dtype": torch.bfloat16}},
+            TypeError,
+            r"keeps their dtype, so torch_dtype=torch\.bfloat16",
+        ),
+        ({"auto_class": AutoModel}, NotImplementedError, "builds its model class itself"),
+        ({"mtp": True}, NotImplementedError, "cannot build the MTP head"),
+        (
+            {"kwargs": {"config": LlamaConfig(**_TINY), "hf_config": LlamaConfig(**_TINY)}},
+            TypeError,
+            "config and hf_config both name",
+        ),
+    ],
+    ids=["placement", "dtype", "torch_dtype", "model_class", "mtp_head", "two_configs"],
+)
+def test_a_sharded_load_refuses_what_it_cannot_honor(
+    run_as, serve, loaders, process_group, monkeypatch, tmp_path, change, error, match
+):
+    """Placement, a model class, a head only a whole model builds, two configs to load with: none
+    reaches a sharded load, which fails before it starts rather than load something other than
+    what was asked."""
+    run_as(2)
+    serve(LlamaConfig(**_TINY))
+    if change.get("mtp"):
+        monkeypatch.setattr(models_hf, "checkpoint_has_mtp", lambda *a: True)
+
+    with pytest.raises(error, match=match):
+        load_pretrained(
+            str(tmp_path),
+            change.get("auto_class"),
+            distribution=DistributionPlan(2, dense="fsdp"),
+            **change.get("kwargs", {}),
+        )
+    assert loaders == {}
+
+
+@pytest.mark.parametrize(
+    ("distribution", "checkpoint", "whole_model_kwargs", "error", "match"),
+    [
+        (
+            DistributionPlan(4, dense="fsdp"),
+            "causal-lm",
+            False,
+            ValueError,
+            "is for 4 ranks, but this run has 2",
+        ),
+        (
+            DistributionPlan(2, dense="fsdp"),
+            "quantized",
+            False,
+            NotImplementedError,
+            "cannot dequantize",
+        ),
+        (
+            DistributionPlan(2, dense="fsdp"),
+            "vlm",
+            False,
+            NotImplementedError,
+            "builds a causal LM, which this llava checkpoint is not",
+        ),
+        ("fsdp2", "causal-lm", False, TypeError, "expected a DistributionPlan or None"),
+        (
+            DistributionPlan(2),
+            "causal-lm",
+            True,
+            TypeError,
+            "returned with a plan that shards the model",
+        ),
+    ],
+    ids=[
+        "another-world",
+        "quantized-checkpoint",
+        "vlm-checkpoint",
+        "not-a-plan",
+        "hf-config-on-a-whole-load",
+    ],
+)
+def test_a_load_refuses_what_it_cannot_load(
+    run_as,
+    serve,
+    loaders,
+    auto_class,
+    process_group,
+    tmp_path,
+    distribution,
+    checkpoint,
+    whole_model_kwargs,
+    error,
+    match,
+):
+    """Before anything loads: a plan for another world; a quantized checkpoint, which FSDP2 would
+    read as stored; a VLM, which FSDP2 would build through AutoModelForCausalLM, refused or as its
+    language model alone; anything but a plan; and the sharded loads' config, which from_pretrained
+    would take for a model argument and fail far from here."""
+    run_as(2)
+    config = serve(transformers.LlavaConfig() if checkpoint == "vlm" else LlamaConfig(**_TINY))
+    if checkpoint == "quantized":
+        config.quantization_config = {"quant_method": "fp8"}
+    whole = {"auto_class": auto_class, "hf_config": config} if whole_model_kwargs else {}
+
+    with pytest.raises(error, match=match):
+        load_pretrained(str(tmp_path), distribution=distribution, **whole)
+    assert loaders == {}
+    assert auto_class.calls == []
+
+
+def test_an_automatic_layout_says_so_when_it_refuses_a_placement(
+    run_as, serve, loaders, monkeypatch, tmp_path
+):
+    """With neither a plan nor MODELOPT_FSDP_SIZE, rank 0's GPU memory decides to shard a model too
+    big for it, and the sharded load refuses options that place a whole model. The refusal says the
+    layout was decided rather than asked for, and how to keep the load whole."""
+    run_as(2)
+    serve(LlamaConfig(**_TINY))
+    monkeypatch.setattr(model_load_utils, "_gpu_memory", lambda: 1)
+
+    with pytest.raises(TypeError, match=r"\['device_map'\] place a whole model, but no plan"):
+        load_pretrained(str(tmp_path), device_map="auto")
+    assert loaders == {}
+
+
+@pytest.mark.parametrize(
+    ("distribution", "options", "error", "match"),
+    [
+        (DistributionPlan(4), {}, ValueError, "is for 4 ranks"),
+        (DistributionPlan(2), {"hf_config": LlamaConfig(**_TINY)}, TypeError, "plan that shards"),
+        (
+            DistributionPlan(2, dense="fsdp"),
+            {"device_map": "auto"},
+            TypeError,
+            r"\['device_map'\] do not apply to the FSDP2 load",
+        ),
+        (
+            DistributionPlan(2, dense="fsdp"),
+            {"auto_class": AutoModel},
+            NotImplementedError,
+            "auto_class=AutoModel does not apply",
+        ),
+        (
+            DistributionPlan(2, dense="fsdp"),
+            {"dtype": torch.float32},
+            TypeError,
+            "keeps their dtype",
+        ),
+    ],
+    ids=[
+        "another-world",
+        "hf-config-on-a-whole-load",
+        "whole-load-option-on-a-sharded-plan",
+        "auto-class-on-a-sharded-plan",
+        "dtype-on-a-sharded-plan",
+    ],
+)
+def test_a_handed_plan_is_checked_before_the_checkpoint_is_fetched(
+    run_as, monkeypatch, tmp_path, distribution, options, error, match
+):
+    """Making the checkpoint local can mean downloading it: a plan the load refuses fails first, as
+    do hf_config with a plan that loads the model whole and an option a sharded plan's load
+    refuses."""
+    run_as(2)
+    monkeypatch.setattr(
+        model_load_utils, "ensure_local_checkpoint", lambda *a, **k: pytest.fail("fetched it")
+    )
+
+    with pytest.raises(error, match=match):
+        load_pretrained(str(tmp_path), distribution=distribution, **options)
+
+
+@pytest.mark.parametrize("quantized", [False, True])
+def test_a_single_process_fsdp2_load_sets_up_its_group_of_one(
+    monkeypatch, serve, loaders, tmp_path, quantized
+):
+    """FSDP2 on one rank still shards through a process group, of one, which the load sets up once
+    nothing can refuse it: a refused load leaves no group behind."""
+    config = serve(LlamaConfig(**_TINY))
+    if quantized:
+        config.quantization_config = {"quant_method": "fp8"}
+    monkeypatch.setenv("RANK", "0")  # torchrun --nproc_per_node=1
+    live = []
+    monkeypatch.setattr(dist, "is_initialized", lambda: bool(live))
+    monkeypatch.setattr(dist, "setup", lambda timeout=None: live.append(0))
+    # What a group of one answers once it is up.
+    monkeypatch.setattr(dist, "rank", lambda group=None: 0)
+    monkeypatch.setattr(dist, "size", lambda group=None: 1)
+    monkeypatch.setattr(dist, "local_rank", lambda: 0)
+    plan = DistributionPlan(dense="fsdp")
+
+    if quantized:
+        with pytest.raises(NotImplementedError, match="cannot dequantize"):
+            load_pretrained(str(tmp_path), distribution=plan)
+        assert live == []
+        return
+    load_pretrained(str(tmp_path), distribution=plan)
+
+    assert live == [0]
+    call = loaders.pop("fsdp2")
+    assert (call["rank"], call["world_size"]) == (0, 1)
+
+
+def test_a_single_process_fsdp2_load_needs_a_launcher(monkeypatch, serve, loaders, tmp_path):
+    """Started without one, the process cannot set its group of one up; it is told to launch with
+    torchrun rather than fail inside init_process_group."""
+    serve(LlamaConfig(**_TINY))
+    monkeypatch.delenv("RANK", raising=False)
+    monkeypatch.setattr(dist, "is_initialized", lambda: False)
+    monkeypatch.setattr(dist, "setup", lambda timeout=None: pytest.fail("set a group up"))
+
+    with pytest.raises(RuntimeError, match="torchrun --nproc_per_node=1"):
+        load_pretrained(str(tmp_path), distribution=DistributionPlan(dense="fsdp"))
+    assert loaders == {}
+
+
+@pytest.fixture
+def launch_without_its_group(monkeypatch):
+    """Rank 0 of an 8-rank launch whose processes set no process group up."""
+    monkeypatch.setenv("RANK", "0")
+    monkeypatch.setenv("WORLD_SIZE", "8")
+    monkeypatch.setattr(dist, "is_initialized", lambda: False)
+    monkeypatch.setattr(dist, "setup", lambda timeout=None: pytest.fail("set a group up"))
+
+
+def test_a_launch_that_sets_no_group_up_loads_whole_on_each_process(
+    launch_without_its_group, serve, loaders, auto_class, tmp_path
+):
+    """A launcher can start several processes that never set a group up (hf_ptq without
+    --use_fsdp2): each loads the whole model as a single process does, with no collective to wait
+    in and no plan to agree on."""
+    serve(LlamaConfig(**_TINY))
+
+    model = load_pretrained(str(tmp_path), auto_class)
+
+    assert auto_class.calls == [{}]
+    assert distribution_plan(model) == DistributionPlan()
+    assert loaders == {}
+
+
+@pytest.mark.parametrize(
+    ("distribution", "knobs", "error", "match"),
+    [
+        (
+            DistributionPlan(8),
+            {},
+            RuntimeError,
+            "is for 8 ranks, whose process group this 8-rank launch has not",
+        ),
+        (
+            DistributionPlan(dense="fsdp"),
+            {},
+            RuntimeError,
+            r"shards through this 8-rank launch's .* DistributionPlan\(8, dense='fsdp'\)",
+        ),
+        (
+            None,
+            {"MODELOPT_FSDP_SIZE": "world"},
+            RuntimeError,
+            "MODELOPT_FSDP_SIZE=world ask to shard",
+        ),
+        (None, {"MODELOPT_EP_DEGREE": "4"}, RuntimeError, "MODELOPT_EP_DEGREE=4 ask to shard"),
+        (None, {"MODELOPT_EP_DEGREE": "eigth"}, ValueError, "MODELOPT_EP_DEGREE='eigth'"),
+    ],
+    ids=["several-ranks", "sharded", "fsdp-knob", "ep-knob", "ep-knob-not-a-degree"],
+)
+def test_what_needs_the_group_is_refused_in_a_launch_that_sets_none_up(
+    launch_without_its_group,
+    serve,
+    loaders,
+    monkeypatch,
+    tmp_path,
+    distribution,
+    knobs,
+    error,
+    match,
+):
+    """A plan for several ranks, a sharded load, and knobs asking to shard over the launch need its
+    group, which one rank cannot set up for itself: refused before anything is fetched, rather than
+    leave each process the whole model the knobs did not ask for. A knob that is not one is refused
+    as the plan decision refuses it, not dropped."""
+    serve(LlamaConfig(**_TINY))
+    for name, value in knobs.items():
+        monkeypatch.setenv(name, value)
+    monkeypatch.setattr(
+        model_load_utils, "ensure_local_checkpoint", lambda *a, **k: pytest.fail("fetched it")
+    )
+
+    with pytest.raises(error, match=match):
+        load_pretrained(str(tmp_path), distribution=distribution)
+    assert loaders == {}
+
+
+# --------------------------------------------------------------------------------- fsdp2_wrap
+
+
+def test_fsdp2_wrap_leaves_the_plan_to_the_load(monkeypatch):
+    """load_pretrained records the plan once, at the end of the load; the wrap it calls records none,
+    so a caller that wraps a model itself must record one before anything reads it."""
+    model = torch.nn.Sequential(torch.nn.Linear(2, 2), torch.nn.Linear(2, 2))
+    monkeypatch.setattr(
+        LayerActivationCollector, "get_decoder_layers", staticmethod(lambda model: list(model))
+    )
+    monkeypatch.setattr(model_load_utils, "fully_shard", lambda module, **kwargs: None)
+    monkeypatch.setattr(dist, "size", lambda group=None: 4)
+
+    model_load_utils.fsdp2_wrap(model)
+
+    with pytest.raises(RuntimeError, match="no distribution plan recorded"):
+        distribution_plan(model)
diff --git a/tests/unit/torch/utils/test_unplaced_source_keys.py b/tests/unit/torch/utils/test_unplaced_source_keys.py
index 31920ea6799..4ddbffeedec 100644
--- a/tests/unit/torch/utils/test_unplaced_source_keys.py
+++ b/tests/unit/torch/utils/test_unplaced_source_keys.py
@@ -41,6 +41,7 @@
     Mxfp4Config,
 )
 
+from modelopt.torch.utils.distributed import DistributionPlan, distribution_plan
 from modelopt.torch.utils.plugins import model_load_utils
 from modelopt.torch.utils.plugins.model_load_utils import (
     load_pretrained,
@@ -105,6 +106,7 @@ def test_records_the_stray_tensors_and_the_checkpoint(tmp_path, extra, subfolder
     assert model._modelopt_unplaced_source_keys == sorted(extra), "sorted, for a stable export"
     assert model._modelopt_source_checkpoint == str(checkpoint)
     assert model.dtype == torch.bfloat16, "from_pretrained kwargs must pass through"
+    assert distribution_plan(model) == DistributionPlan(), "a single process holds it whole"
 
 
 class _Recorder:
@@ -192,17 +194,64 @@ def test_an_option_it_cannot_honor_is_refused_before_anything_loads(option, refu
         load_pretrained("no/such/checkpoint", **option)
 
 
-def test_a_config_passed_for_the_load_needs_none_on_disk(tmp_path):
-    """The load uses the config it is passed, as ``from_pretrained`` does, and so does the choice of
-    its loading hook: a directory without a config.json loads."""
+class _OwnConfig(LlamaConfig):
+    """A config class of the caller's own, which AutoConfig has no mapping for."""
+
+    model_type = "modelopt_test_own_llama"
+
+
+class _OwnModel(LlamaForCausalLM):
+    config_class = _OwnConfig
+
+
+@pytest.mark.parametrize(
+    ("as_path", "model_type", "model_class"),
+    [
+        (False, None, None),
+        (True, None, LlamaForCausalLM),
+        (True, "llama", LlamaForCausalLM),
+        (True, None, _OwnModel),
+    ],
+    ids=["config", "config-path", "config-path-and-model-type", "own-config-class-path"],
+)
+def test_a_config_passed_for_the_load_needs_none_on_disk(
+    tmp_path, as_path, model_type, model_class
+):
+    """The load uses the config it is passed, or reads it from the directory it is passed, as
+    ``from_pretrained`` does, through the model class's own config class, and so do its plan and the
+    choice of its loading hook: a checkpoint directory without a config.json loads, through a model
+    class of the caller's own too."""
     config = _tiny_llama_config()
+    if model_class is _OwnModel:
+        config = _OwnConfig(**config.to_diff_dict())
+    weights = tmp_path / "weights"
+    weights.mkdir()
+    _save(weights, config, {**LlamaForCausalLM(config).state_dict(), STRAY: torch.ones(32)})
+    if as_path:
+        (tmp_path / "config").mkdir()
+        (weights / "config.json").rename(tmp_path / "config" / "config.json")
+        passed = {"auto_class": model_class, "config": str(tmp_path / "config")}
+    else:
+        (weights / "config.json").unlink()
+        passed = {"config": config}
+    (weights / "generation_config.json").write_text("{}")  # from_pretrained reads it instead
+
+    model = _load(weights, model_type=model_type, **passed)
+
+    assert model._modelopt_unplaced_source_keys == [STRAY]
+
+
+def test_a_model_class_of_the_callers_own_reads_its_config_itself(tmp_path):
+    """Its model type given, a single process reads no config for a model class of the caller's
+    own: the class reads the checkpoint's through its own config class, which AutoConfig need not
+    know, and the plan of a single process is the same whatever the model."""
+    config = _OwnConfig(**_tiny_llama_config().to_diff_dict())
     _save(tmp_path, config, {**LlamaForCausalLM(config).state_dict(), STRAY: torch.ones(32)})
-    (tmp_path / "config.json").unlink()
-    (tmp_path / "generation_config.json").write_text("{}")  # from_pretrained reads it instead
 
-    model = _load(tmp_path, config=config)
+    model = _load(tmp_path, _OwnModel, model_type=_OwnConfig.model_type)
 
     assert model._modelopt_unplaced_source_keys == [STRAY]
+    assert distribution_plan(model) == DistributionPlan()
 
 
 def test_an_ignore_rule_does_not_hide_an_unplaced_tensor(tmp_path):