diff --git a/modelopt/torch/export/distributed_writer.py b/modelopt/torch/export/distributed_writer.py
index 21ad6da9f34..d380b3c7735 100644
--- a/modelopt/torch/export/distributed_writer.py
+++ b/modelopt/torch/export/distributed_writer.py
@@ -17,7 +17,8 @@
 
 Every tensor is sent to exactly one owner rank over batched point-to-point sends and
 written from there, so no rank ever materializes the whole model and no unit is
-replicated across ranks. Selected by ``MODELOPT_FSDP2_EXPORT=nogather``.
+replicated across ranks. Used for every expert-parallel model; a model sharded by FSDP2 alone opts in
+with ``MODELOPT_FSDP2_EXPORT=nogather``.
 """
 
 import json
@@ -35,7 +36,8 @@
 from torch.distributed.checkpoint.state_dict import StateDictOptions, get_model_state_dict
 from torch.distributed.tensor import DTensor
 
-from modelopt.torch.utils.distributed import all_reduce
+from modelopt.torch.utils.distributed import all_reduce, distribution_plan, expert_parallel_state
+from modelopt.torch.utils.moe import _num_experts_from_config
 
 from .model_utils import TiedWeightMap
 from .moe_utils import _FUSED_PROJ, _fused_experts_prefixes, _split_local_fused_module
@@ -690,7 +692,7 @@ def _distributed_save_hf_checkpoint_impl(
 
       1. Fused expert weights (when present): each rank splits its LOCAL shard into per-expert keys
          with global indices and keeps them in place -- no communication. Experts are therefore
-         already disjoint across ranks.
+         already disjoint across ranks. dp-replica ranks under DP x EP skip this.
       2. Dense / non-expert weights: each fqn is assigned to exactly ONE owner rank and moved there
          (:func:`_materialize_owned`), so every tensor becomes whole on its owner while the set stays
          spread across ranks. Replicated and plain values need no transport; dim-0 shards travel
@@ -716,6 +718,54 @@ def _distributed_save_hf_checkpoint_impl(
 
     export_dir = Path(export_dir)
     rank, world = dist.get_rank(), dist.get_world_size()
+
+    # Where this rank sits in expert parallelism, if any.
+    #
+    # Under FSDP2 alone the fused 3-D expert weight is sharded on dim 0, so each per-expert key's
+    # global index is just the DTensor's own dim-0 offset and the answer below stays (0, 0).
+    #
+    # Under expert parallelism it is not: transformers pre-slices each expert param to this rank's
+    # EP group, so the EP split is INVISIBLE to the per-param DTensor -- its dim-0 offset is relative
+    # to the group, not global. Without ep_rank every group would write experts 0..n-1 over each
+    # other and the checkpoint would be quietly wrong rather than broken. The coordinate comes from
+    # the routed experts' ParallelState -- the same source calibration reads, not transformers'
+    # internal ``model._device_mesh``. An expert's data-parallel group is its expert-data-parallel one.
+    ep_rank, dp_idx = 0, 0
+    plan = distribution_plan(model)
+    if plan.is_expert_parallel:
+        parallel_state = expert_parallel_state(model)
+        if parallel_state is None:
+            raise RuntimeError(
+                "The model's plan is expert-parallel, but no routed expert carries its expert-parallel "
+                "group: load it with an expert-parallel loader, which seeds each routed expert's "
+                "ParallelState."
+            )
+        ep_group = parallel_state.expert_model_parallel_group
+        edp_group = parallel_state.data_parallel_group
+        # The coordinates index the plan's own groups. A ParallelState left with its default
+        # data-parallel group holds the whole world there: taken as the replica index, it would leave
+        # every rank but global rank 0 out of the expert write, with no error. The groups have the
+        # same sizes on every rank, so every rank refuses alike.
+        edp_size = edp_group.world_size() if edp_group.is_initialized() else 1
+        if ep_group.world_size() != plan.ep_size or edp_size != plan.edp:
+            raise RuntimeError(
+                f"The routed experts' expert-parallel group holds {ep_group.world_size()} ranks and "
+                f"their data-parallel group {edp_size}, but the model's plan has "
+                f"ep_size={plan.ep_size} and edp={plan.edp}, so the writer cannot tell which rank "
+                "writes which experts."
+            )
+        ep_rank = ep_group.rank()
+        if edp_group.is_initialized():
+            # Experts are replicated across EDP, so only the first replica writes them.
+            dp_idx = edp_group.rank()
+    # Total config expert count, to tell a per-EP-group expert tensor (size < total) from one that
+    # spans all experts (classic DP x EP) when computing the per-expert global index offset.
+    total_experts = _num_experts_from_config(getattr(model, "config", None))
+    # DP x EP: experts are dp-REPLICATED across the EDP group (each holds a full EP-sharded copy). Only
+    # the first replica (dp_idx == 0) writes them; replica ranks (dp_idx > 0) skip the per-expert split.
+    # dp_idx is the EDP-group rank resolved above -- 0 when experts are not EDP-replicated (dense /
+    # FSDP2 / EP-without-DP), so every such rank writes.
+    write_experts = dp_idx == 0
     sharded_sd = get_model_state_dict(model, options=StateDictOptions(full_state_dict=False))
 
     # Re-materialize every inference tensor as a NORMAL one UP FRONT. A caller in inference mode hands
@@ -789,17 +839,21 @@ def _is_expert_key(key: str) -> bool:
         for j, k in enumerate(in_keys):
             sharded_sd[k] = stacked[j].clone()
     # (1) Experts: split this rank's local shard into per-expert keys (global idx); stays local.
-    # FSDP2 shards a fused 3-D expert weight on dim 0, the expert axis, so each per-expert key's global
-    # index is the DTensor's own dim-0 offset, which _split_local_fused_module reads.
+    # Skipped on dp-replica ranks under DP x EP (their experts are written by dp group 0).
     local_sd: dict = {}
-    for prefix in prefixes:
-        local_sd.update(_split_local_fused_module(prefix, sharded_sd))
-    local_sd = postprocess_state_dict(
-        local_sd, maxbound, kv_cache_format, is_modelopt_qlora, tied_map=tied_map
-    )
-    # In place: the split's tensors are its own copies, so one made contiguous replaces the old.
-    for _k in list(local_sd.keys()):
-        local_sd[_k] = local_sd[_k].detach().contiguous()
+    if write_experts:
+        for prefix in prefixes:
+            local_sd.update(
+                _split_local_fused_module(
+                    prefix, sharded_sd, ep_rank=ep_rank, total_experts=total_experts
+                )
+            )
+        local_sd = postprocess_state_dict(
+            local_sd, maxbound, kv_cache_format, is_modelopt_qlora, tied_map=tied_map
+        )
+        # In place: the split's tensors are its own copies, so one made contiguous replaces the old.
+        for _k in list(local_sd.keys()):
+            local_sd[_k] = local_sd[_k].detach().contiguous()
     # At this point local_sd holds ONLY this rank's (already split, per-expert) experts -- capture the
     # set so _materialize_owned can tell disjoint experts from replicated dense.
     expert_key_set = set(local_sd)
diff --git a/modelopt/torch/export/moe_utils.py b/modelopt/torch/export/moe_utils.py
index e12ed147abf..50b8d909f3b 100644
--- a/modelopt/torch/export/moe_utils.py
+++ b/modelopt/torch/export/moe_utils.py
@@ -804,17 +804,29 @@ def _dtensor_dim0_offset(dt) -> int:
         return 0
 
 
-def _split_local_fused_module(prefix: str, sharded_sd: dict) -> dict:
+def _split_local_fused_module(
+    prefix: str, sharded_sd: dict, ep_rank: int = 0, total_experts: int = 0
+) -> dict:
     """Split this rank's local shard of a fused-experts module into per-expert keys.
 
     Split THIS rank's *local* shard of one fused-experts module into per-expert keys with GLOBAL
     expert indices (plain, local tensors -- no all-gather).
 
-    FSDP2 shards the fused 3-D expert weight on dim 0 (the expert axis), so every expert this rank
+    FSDP2/EP shard the fused 3-D expert weight on dim 0 (the expert axis), so every expert this rank
     owns is whole and local. We reuse :func:`split_fused_experts_state_dict` on the ``to_local()``
     shard (which numbers experts ``0..E_local-1``) and then shift each expert index by the rank's
     global dim-0 offset, so the resulting per-expert keys match the single-process export and stay
     on this rank for a direct distributed write (no consolidation).
+
+    The fused expert tensor can arrive in two EP layouts, distinguished by its dim-0 size vs
+    ``total_experts`` (the config expert count):
+      * **per-EP-group** (size < total): transformers pre-sliced the param to this rank's EP group
+        (FSDP x EP -> a dp-sharded DTensor of the group; or a plain ep-shard). ``_dtensor_dim0_offset``
+        only gives the WITHIN-group offset, so we add the EP block base ``ep_rank * group_size``.
+      * **all-experts** (size == total): an EP DTensor sharded across the ep mesh spanning all experts
+        (classic DP x EP). ``_dtensor_dim0_offset`` already returns the TRUE global offset -> no base.
+    ``total_experts == 0``, a count the caller could not read off the model's config, takes the
+    tensor for a per-EP-group slice.
     """
     gated = f"{prefix}.gate_up_proj" in sharded_sd
     first = "gate_up_proj" if gated else "up_proj"
@@ -822,6 +834,13 @@ def _split_local_fused_module(prefix: str, sharded_sd: dict) -> dict:
     # dp_offset: this rank's dim-0 start WITHIN its DTensor; n_local: experts it owns.
     dp_offset = _dtensor_dim0_offset(fw) if isinstance(fw, DTensor) else 0
     n_local = fw.to_local().shape[0] if isinstance(fw, DTensor) else fw.shape[0]
+    group_size = int(fw.shape[0])
+    # Add the EP block base only when the tensor is a per-EP-group slice (size < total_experts) or
+    # total is unknown (0). When it spans all experts, dp_offset is already global.
+    add_ep_base = (
+        bool(ep_rank) and group_size > 0 and (total_experts == 0 or group_size < total_experts)
+    )
+    offset = dp_offset + (ep_rank * group_size if add_ep_base else 0)
 
     def _loc(key):
         v = sharded_sd.get(key)
@@ -864,12 +883,12 @@ def _loc_scale(key):
     split_local = split_fused_experts_state_dict(
         local_fused
     )  # keys {prefix}.{i}.* (i in 0..E_local-1)
-    if dp_offset == 0:
+    if offset == 0:
         return split_local
 
     plen = len(prefix) + 1
     out: dict = {}
     for key, val in split_local.items():
         i_str, tail = key[plen:].split(".", 1)
-        out[f"{prefix}.{int(i_str) + dp_offset}.{tail}"] = val
+        out[f"{prefix}.{int(i_str) + offset}.{tail}"] = val
     return out
diff --git a/modelopt/torch/export/quant_aware_conversion.py b/modelopt/torch/export/quant_aware_conversion.py
index 1005a3a904f..36bcb7522f4 100644
--- a/modelopt/torch/export/quant_aware_conversion.py
+++ b/modelopt/torch/export/quant_aware_conversion.py
@@ -62,6 +62,8 @@
 (in-memory-name) behavior rather than emit a silently-wrong checkpoint.
 """
 
+import contextlib
+import functools
 import logging
 import re
 from collections.abc import Iterable
@@ -98,6 +100,18 @@
 )
 _LEAF_SUFFIXES = (".weight", *_QUANT_STATE_SUFFIXES, ".bias")
 
+# Checkpoint tensors a load-time rewrite folds into a weight -- compressed-tensors' stored
+# parameters, FP8's activation scale -- and so lists as sources next to it. They name the same
+# module as that weight; here they only matter for reading the module's leaf off a pattern.
+_LOAD_SIBLING_SUFFIXES = (
+    ".weight_packed",
+    ".weight_global_scale",
+    ".weight_shape",
+    ".weight_zero_point",
+    ".weight_g_idx",
+    ".activation_scale",
+)
+
 # Leaves that are per-tensor scalars (0-d) and must be *duplicated*, not split, when
 # a fused module is un-fused.
 _SCALAR_LEAF_SUFFIXES = (".weight_scale_2", ".input_scale")
@@ -745,6 +759,46 @@ def _drop_shadowed_prefix_renames(model, rules: list[RenameRule]) -> list[Rename
     return kept
 
 
+@functools.cache
+def _quantization_codec_ops() -> tuple[type, ...]:
+    """Transformers' ops that decode or encode a weight's on-disk quantization in place.
+
+    Includes the identity op, which is what older releases (5.5, 5.6) reverse ``Fp8Dequantize``
+    into; later ones reverse it into ``Fp8Quantize``.
+    """
+    ops: list[Any] = []
+    # Imported apart: 5.5 imports triton atop its FP8 integration, so without triton that import
+    # fails, while the identity op those releases reverse into lives in core_model_loading.
+    with contextlib.suppress(Exception):  # transformers without the FP8 integration
+        from transformers.integrations import finegrained_fp8
+
+        names = ("Fp8Dequantize", "Fp8Quantize", "Fp8DecodeScale", "Fp8EncodeScale")
+        ops += [getattr(finegrained_fp8, n, None) for n in names]
+    with contextlib.suppress(Exception):
+        from transformers import core_model_loading
+
+        ops.append(getattr(core_model_loading, "_IdentityOp", None))
+    return tuple(op for op in ops if op is not None)
+
+
+def _decodes_in_place(rev, ops) -> bool:
+    """Whether a reversed converter only re-encodes a weight's quantization, under the same name.
+
+    A dequantizing load (FineGrainedFP8 ``dequantize=True``) decodes each weight in place, and
+    its reverse would re-encode the weight in the source's format. The export never wants that:
+    it writes ModelOpt's own quantized weights and scales. Such a converter renames nothing: the
+    weight maps onto itself, and every pattern names a tensor of the same module, the weight or a
+    scale the load folded into it. One naming another module would rename it, so it is not skipped.
+    """
+    codec_ops = _quantization_codec_ops()
+    if not ops or not codec_ops or not all(isinstance(op, codec_ops) for op in ops):
+        return False
+    sources = [p.strip("^$") for p in _as_list(rev.source_patterns)]
+    targets = [p.strip("^$") for p in _as_list(rev.target_patterns)]
+    modules = {p.rpartition(".")[0] for p in (*sources, *targets)}
+    return bool(sources) and sources[:1] == targets[:1] and len(modules) == 1
+
+
 def _build_reverse_rules(
     model,
 ) -> tuple[list[SplitRule], list[_MergeRule], list[RenameRule], list[str], set[str] | None]:
@@ -767,6 +821,8 @@ def _build_reverse_rules(
       survives in the state dict, so it is un-fused via a :class:`SplitRule`.
     * Dense splitting ``WeightConverter`` (reverse is ``Concatenate`` only): the split
       tensors are re-fused under the transform's original sub-model scope.
+    * A dequantizing load's ``WeightConverter`` (quantization codec ops only, same name): no
+      rule -- the export writes ModelOpt's own quantization, not the source's.
 
     Raises :class:`QuantConversionUnsupportedError` for any op shape not covered, so
     the caller falls back to the legacy (in-memory-name) behavior.
@@ -842,6 +898,8 @@ def _build_reverse_rules(
                         dim=ops[0].dim,
                     )
                 )
+            elif _decodes_in_place(rev, ops):
+                continue
             else:
                 raise QuantConversionUnsupportedError(
                     f"unsupported reverse ops: {[type(o).__name__ for o in ops]}"
@@ -884,7 +942,7 @@ def _expert_leaf_renames(rev) -> list[RenameRule]:
     fused leaf as per-expert parts, mapped back to the hub leaves positionally.
     """
     src_leaf = _leaf(_as_list(rev.source_patterns)[0])
-    hub_leaves = [_leaf(t) for t in _as_list(rev.target_patterns)]
+    hub_leaves = _distinct_leaves(_as_list(rev.target_patterns))
     part_leaves = _FUSED_EXPERT_PART_NAMES.get(src_leaf, [src_leaf])
     if len(part_leaves) != len(hub_leaves):
         raise QuantConversionUnsupportedError(
@@ -903,7 +961,7 @@ def _expert_leaf_renames(rev) -> list[RenameRule]:
 def _dense_split_rule(rev, ops) -> SplitRule:
     """Un-fuse a dense (non-expert) fused linear that survives in the state dict."""
     fused = _leaf_suffix(_as_list(rev.source_patterns)[0])
-    parts = tuple(_leaf_suffix(t) for t in _as_list(rev.target_patterns))
+    parts = tuple("." + leaf for leaf in _distinct_leaves(_as_list(rev.target_patterns)))
     dim = next((op.dim for op in ops if hasattr(op, "dim")), 0)
     return SplitRule(fused_suffix=fused, part_suffixes=parts, dim=dim)
 
@@ -912,10 +970,21 @@ def _as_list(x) -> list:
     return list(x) if isinstance(x, (list, tuple)) else [x]
 
 
+def _distinct_leaves(patterns: list[str]) -> list[str]:
+    """The module leaves ``patterns`` name, each once, in order.
+
+    A converter that loading rewrote lists every checkpoint tensor it consumes per module: a
+    dequantizing quantizer adds each weight's ``weight_scale_inv``, an on-the-fly decompressor the
+    stored ``weight_packed`` / ``weight_scale`` / ``weight_global_scale``. Those all name the same
+    module, so repeated leaves collapse to it.
+    """
+    return list(dict.fromkeys(_leaf(p) for p in patterns))
+
+
 def _leaf(pattern: str) -> str:
     """Bare leaf name from a conversion pattern, e.g. ``.experts.*.w1.weight`` -> ``w1``."""
-    p = pattern
-    for suffix in _LEAF_SUFFIXES:
+    p = pattern.removesuffix("$")
+    for suffix in (*_LEAF_SUFFIXES, *_LOAD_SIBLING_SUFFIXES):
         if p.endswith(suffix):
             p = p[: -len(suffix)]
             break
diff --git a/modelopt/torch/export/unified_export_hf.py b/modelopt/torch/export/unified_export_hf.py
index 9c63130bfd1..357cb994ac8 100644
--- a/modelopt/torch/export/unified_export_hf.py
+++ b/modelopt/torch/export/unified_export_hf.py
@@ -22,6 +22,7 @@
 import os
 import re
 import tempfile
+import traceback
 import warnings
 from builtins import ValueError
 from collections import defaultdict
@@ -66,11 +67,13 @@
     HAS_DIFFUSERS = False
 
 try:
-    from transformers import PreTrainedModel
+    from transformers import PretrainedConfig, PreTrainedModel
 
     _TRANSFORMERS_MODEL_TYPES: tuple[type, ...] = (PreTrainedModel,)
+    _TRANSFORMERS_CONFIG_TYPES: tuple[type, ...] = (PretrainedConfig,)
 except ImportError:
     _TRANSFORMERS_MODEL_TYPES = ()
+    _TRANSFORMERS_CONFIG_TYPES = ()
 
 
 from modelopt.torch.opt.conversion import ModeloptStateManager, modelopt_state
@@ -97,7 +100,7 @@
 from modelopt.torch.utils import print_rank_0
 from modelopt.torch.utils.dataset_utils import _disable_use_cache
 from modelopt.torch.utils.distributed import (
-    DistributionPlan,
+    all_reduce,
     broadcast_from,
     distribution_plan,
     is_fsdp2_model,
@@ -1087,6 +1090,38 @@ def pack_quantized_weights(
     _reconstruct_fused_moe_linear(model)
 
 
+def _hf_shard_flags(model: nn.Module) -> tuple[bool, bool]:
+    """Return ``(fsdp2_sharded, ep_sharded)`` for an HF model about to be exported, from its plan.
+
+    Expert parallelism shards the experts too, just not by FSDP2: transformers pre-slices the expert
+    params across the EP group. The two are reported separately because a model can be both at once,
+    FSDP2-dense with expert-parallel experts. A diffusers pipeline is not a module, and nothing here
+    distributes one.
+    """
+    if not isinstance(model, nn.Module):
+        return False, False
+    plan = distribution_plan(model)
+    return plan.is_fsdp_dense, plan.is_expert_parallel
+
+
+def _process_hf_model_for_export(
+    model: nn.Module, dtype: torch.dtype | None = None, is_modelopt_qlora: bool = False
+) -> dict[str, Any]:
+    """Fold a quantized HF model's quantizers into its weights in place; return its quant config.
+
+    The processing half of an export, for a caller that serializes the model itself: the no-gather
+    writer, which writes each rank's shard. Afterwards the model's own state dict holds the
+    deployment weights. Nothing is gathered, so it runs on a live FSDP2 or expert-parallel model;
+    there the fused MoE experts stay fused, since each rank folds only its local shard, and the
+    writer then splits each rank's experts into per-expert tensors.
+    """
+    dtype, _, quant_config = _prepare_model_for_export(model, dtype, is_modelopt_qlora)
+    # Only the no-gather writer calls this, and only for a sharded model, so the experts stay fused
+    # without reading the plan a second time.
+    pack_quantized_weights(model, dtype, is_modelopt_qlora, keep_fused_experts=True)
+    return quant_config
+
+
 def _export_transformers_checkpoint(
     model: nn.Module,
     dtype: torch.dtype | None = None,
@@ -1933,6 +1968,69 @@ def read_unplaced_weights(model: nn.Module, *, keys_only: bool = False) -> dict[
     return out
 
 
+# Config attributes a distributed load sets to describe THIS RUN rather than the checkpoint: the
+# DistributedConfig transformers stamps on the config when it shards the model (the EP degree and tp
+# plan the export happened to use; not JSON-serializable either), and the base_model_ep_plan the EP
+# loader swapped to the torch_a2a styles, which exist only while ModelOpt is imported.
+_DISTRIBUTED_RUN_CONFIG_ATTRS = ("distributed_config", "base_model_ep_plan")
+
+
+@contextlib.contextmanager
+def _source_distributed_config_attrs(model: nn.Module):
+    """Within the block, every config holds the run attributes its source ``config.json`` declared.
+
+    transformers drops ``base_model_tp_plan`` and ``base_model_pp_plan`` when it serializes a config,
+    but not ``base_model_ep_plan``, so without this the export ships a plan whose styles nothing but
+    ModelOpt can load. Each config -- the root and every nested one, e.g. a VLM's ``text_config``,
+    where the plan lives -- gets the value its source ``config.json`` holds at the same place, or
+    loses the attribute when the source has none: the config a single-GPU export writes. A source
+    that cannot be read counts as having none. The run's values come back on exit, since the
+    sharded model still reads them (transformers' ``tp_plan`` checks ``distributed_config``).
+    """
+    source: Any = {}
+    ckpt = _source_checkpoint(model)
+    if ckpt and Path(ckpt, "config.json").is_file():
+        with contextlib.suppress(OSError, ValueError):
+            source = json.loads(Path(ckpt, "config.json").read_text())
+
+    _missing = object()
+    saved: list[tuple[Any, str, Any]] = []
+
+    def use_source(config: Any, source: Any) -> None:
+        source = source if isinstance(source, dict) else {}
+        for attr in _DISTRIBUTED_RUN_CONFIG_ATTRS:
+            if attr in vars(config) or attr in source:
+                saved.append((config, attr, vars(config).get(attr, _missing)))
+                if attr in source:
+                    setattr(config, attr, source[attr])
+                elif attr in vars(config):
+                    delattr(config, attr)
+        for key, sub_config in list(vars(config).items()):
+            if isinstance(sub_config, _TRANSFORMERS_CONFIG_TYPES):
+                use_source(sub_config, source.get(key))
+
+    try:
+        use_source(model.config, source)
+        yield
+    finally:
+        for config, attr, value in reversed(saved):
+            if value is _missing:
+                if attr in vars(config):
+                    delattr(config, attr)
+            else:
+                setattr(config, attr, value)
+
+
+def _write_base_config(model: nn.Module, export_dir: Path) -> None:
+    """Write the no-gather export's non-weight files, once every rank has written its shards.
+
+    ``save_non_weight_artifacts`` writes them, as it does for the FSDP2 streaming writer; this
+    keeps the config attributes an expert-parallel run sets out of config.json.
+    """
+    with _source_distributed_config_attrs(model):
+        save_non_weight_artifacts(model, export_dir)
+
+
 def _no_gather_export_asked() -> bool:
     """Whether ``MODELOPT_FSDP2_EXPORT`` picks the no-gather writer for an FSDP2-sharded export.
 
@@ -1995,19 +2093,63 @@ def export_hf_checkpoint(
     # emits it from rank 0 alone, so on the other ranks a full read is host memory spent and thrown
     # away -- and the carried set is the large stuff. The KEY list is still resolved everywhere,
     # because get_quant_config runs per rank and the configs have to agree.
-    # How the model is distributed decides how it is exported. A diffusers pipeline is not a module,
-    # and nothing here distributes one.
-    plan = distribution_plan(model) if isinstance(model, nn.Module) else DistributionPlan()
-    _writes_extra = not plan.is_fsdp_dense or torch.distributed.get_rank() == 0
-    _carried = read_unplaced_weights(model, keys_only=not _writes_extra)
+    # Expert parallelism shards the experts too, just not by FSDP2 -- transformers pre-slices the
+    # expert params across the EP group. Computed here rather than with the other predicates below
+    # because the carry-over gate needs it and runs ahead of them (the layerwise and diffusers paths
+    # return before those are reached).
+    _fsdp2_sharded, _ep_sharded = _hf_shard_flags(model)
+    # Under expert parallelism each rank holds only its own experts' ModelOpt state, so the
+    # rank-0 save would write an incomplete one. Refused before any collective, alike on every rank.
+    if _ep_sharded and save_modelopt_state:
+        raise NotImplementedError(
+            "save_modelopt_state=True is not supported for an expert-parallel export: each rank "
+            "holds only its own experts' ModelOpt state."
+        )
+    if _fsdp2_sharded and _ep_sharded:
+        plan = distribution_plan(model)
+        # With fewer expert-parallel ranks than the world, FSDP2 shards the dense part within each
+        # expert-parallel group and replicates it across them, and the no-gather writer every
+        # expert-parallel export takes assembles a dense tensor only from a dim-0 shard over the
+        # whole world. Refused here, before anything is folded, not after the export has begun.
+        if plan.edp > 1:
+            raise NotImplementedError(
+                f"The export cannot write this plan's dense part: with ep_size={plan.ep_size} of "
+                f"world_size={plan.world_size} and dense='fsdp', it is sharded within each of the "
+                f"{plan.edp} expert-parallel groups and replicated across them. Set "
+                f"MODELOPT_EP_DEGREE={plan.world_size} if that divides the expert count, or "
+                "MODELOPT_FSDP_SIZE=1 to replicate the dense part."
+            )
+    if _ep_sharded:
+        # Each rank holds only its own experts' quantizers, so they are agreed on before any folds.
+        _agree_on_fused_expert_quantization(model)
+    # Either form of sharding, including the hybrid where both hold: a model can be FSDP2-dense AND
+    # expert-parallel, so these are tracked separately rather than as an either/or.
+    _sharded_export = _ep_sharded or _fsdp2_sharded
+    # A sharded export reads and hands over the carried weights on rank 0 alone, since both sharded
+    # writers take extra_state_dict from that one sender: on any other rank, reading them (889
+    # tensors on GLM-5.3-Flash, its MTP layer) would be host memory spent for nothing. This gates
+    # the tensors only. Every rank still resolves their names, which _modelopt_carried_over_names
+    # below records alike on every rank for the quant config.
+    _writes_extra = not _sharded_export or torch.distributed.get_rank() == 0
+    _carried: dict = {}
+
+    def carried_over_names() -> list[str]:
+        # Everything the export writes in original precision straight from the source, by either
+        # mechanism: tensors carried here, and the off-index weight files copied verbatim alongside.
+        # get_quant_config reads this to seed exclude_modules; recorded here because it runs before
+        # that, and because only this point knows what was actually written rather than what was
+        # merely unplaced.
+        _carried.update(read_unplaced_weights(model, keys_only=not _writes_extra))
+        return sorted({*_carried, *off_index_tensor_names(model)})
+
+    if _sharded_export:
+        # Rank 0 reads the source here and sends every rank its names, so a failure of that read
+        # fails every rank rather than leave the others waiting in the export's first collective.
+        model._modelopt_carried_over_names = broadcast_from("carried weights", carried_over_names)
+    else:
+        model._modelopt_carried_over_names = carried_over_names()
     if _writes_extra and _carried:
         extra_state_dict = {**_carried, **(extra_state_dict or {})}
-    # Everything the export writes in original precision straight from the source, by either
-    # mechanism: tensors carried above, and the off-index weight files copied verbatim alongside.
-    # get_quant_config reads this to seed exclude_modules; recorded here because it runs before
-    # that, and because only this point knows what was actually written rather than what was
-    # merely unplaced.
-    model._modelopt_carried_over_names = sorted({*_carried, *off_index_tensor_names(model)})
 
     from .layerwise_export import LAYERWISE_EXPORTER_ATTR
 
@@ -2044,19 +2186,33 @@ def export_hf_checkpoint(
         )
         return
 
-    is_fsdp2_sharded = plan.is_fsdp_dense
-    # Rank 0's choice: the two writers issue different collectives, so ranks that read their
-    # environments differently would wait in collectives the others never join.
-    use_no_gather_writer = is_fsdp2_sharded and broadcast_from(
-        "MODELOPT_FSDP2_EXPORT", _no_gather_export_asked
+    is_fsdp2_sharded = _fsdp2_sharded
+
+    # An expert-parallel model is sharded too, just not by FSDP2: transformers pre-slices the expert
+    # params across the EP group, so save_pretrained sees a tensor-parallel model and refuses it
+    # outright. There is no gather fallback for that shape, so EP always takes the no-gather writer,
+    # FSDP2-sharded dense included (the hybrid is both at once). A model sharded by FSDP2 alone keeps
+    # the existing path unless MODELOPT_FSDP2_EXPORT=nogather opts it in, as rank 0 reads it: the
+    # two writers issue different collectives, so ranks that read their environments differently
+    # would wait in collectives the others never join. Derived from _ep_sharded above so the
+    # carry-over gate and this cannot disagree about what EP is.
+    is_ep_sharded = _ep_sharded
+    use_no_gather_writer = is_ep_sharded or (
+        is_fsdp2_sharded and broadcast_from("MODELOPT_FSDP2_EXPORT", _no_gather_export_asked)
     )
-    # Not the global rank: a non-FSDP2 export writes the whole checkpoint from every process.
-    rank = torch.distributed.get_rank() if is_fsdp2_sharded else 0
+
+    # The rank used to gate the single-writer steps below. It must be the real rank whenever the
+    # checkpoint is written cooperatively -- under expert parallelism every process would otherwise
+    # read 0 here, all of them would rewrite config.json at once, and one would truncate the file
+    # another was reading. A non-sharded export does write the whole checkpoint from every process,
+    # which is what the 0 is for.
+    rank = torch.distributed.get_rank() if (is_fsdp2_sharded or is_ep_sharded) else 0
     # Offloaded models take the streaming path: it materializes one layer at a time and
     # writes each straight to a shard file, so peak memory is one layer plus one shard
     # buffer instead of the whole quantized state dict.
     _offloaded = has_accelerate_offload(model)
 
+    error: BaseException | None = None
     try:
         if _offloaded:
             # Imported here rather than at module scope: the streaming exporter imports the
@@ -2089,12 +2245,10 @@ def export_hf_checkpoint(
             from .distributed_writer import distributed_save_hf_checkpoint
 
             _qlora = kwargs.get("is_modelopt_qlora", False)
-            # The writer builds its own name-based tied-weight map, so this one is not handed down.
-            _dtype, _, hf_quant_config = _prepare_model_for_export(model, dtype, _qlora)
-            # Preparing the model does not pack its weights, so pack them here. keep_fused_experts
-            # leaves the MoE experts fused and sharded through the fold, so each rank folds only
-            # its local experts; the writer then splits each rank's experts into per-expert tensors.
-            pack_quantized_weights(model, _dtype, _qlora, keep_fused_experts=True)
+            # Folds in place; the model is sharded here, so the MoE experts stay fused and each rank
+            # folds only its local experts. The writer then splits each rank's experts into
+            # per-expert tensors.
+            hf_quant_config = _process_together(model, dtype, _qlora)
             if getattr(model, "hf_quantizer", None) is not None:
                 model.hf_quantizer = None
             reverted = distributed_save_hf_checkpoint(
@@ -2109,7 +2263,7 @@ def export_hf_checkpoint(
                 extra_state_dict=extra_state_dict,
             )
             if rank == 0:
-                save_non_weight_artifacts(model, export_dir)
+                _write_base_config(model, export_dir)
                 # As the streaming path below: every FSDP2 rank holds the same ModelOpt state.
                 if save_modelopt_state and ModeloptStateManager.is_converted(model):
                     torch.save(modelopt_state(model), export_dir / _MODELOPT_STATE_SAVE_NAME)
@@ -2206,16 +2360,119 @@ def export_hf_checkpoint(
         if rank == 0:
             _write_hf_export_config(model, hf_quant_config, export_dir)
             copy_off_index_safetensors(_source_checkpoint(model), export_dir)
-            if not (_offloaded or is_fsdp2_sharded):
-                # The streaming paths already did, from save_non_weight_artifacts.
+            if not (_offloaded or is_fsdp2_sharded or use_no_gather_writer):
+                # The streaming paths and the no-gather writer already did, from
+                # save_non_weight_artifacts.
                 _copy_non_model_files_from_source(model, export_dir)
 
     except Exception as e:
+        error = e
+        # Print (not just warnings.warn, which the default filter can dedupe/suppress) BEFORE the
+        # barrier below: under FSDP2/EP every other rank is still waiting there, so a failing rank
+        # that raises straight into `finally` blocks on that barrier forever without this rank's
+        # traceback ever reaching the log -- the job just hangs with no visible error. A single
+        # process has no barrier, and the interpreter prints the traceback.
+        if is_fsdp2_sharded or is_ep_sharded:
+            print(
+                f"[export_hf_checkpoint] rank {torch.distributed.get_rank()} failed:\n"
+                f"{traceback.format_exc()}",
+                flush=True,
+            )
         warnings.warn(
             "Cannot export model to the model_config. The modelopt-optimized model state_dict"
             " can be saved with torch.save for further inspection."
         )
         raise e
+    except BaseException as e:
+        # An exit or an interrupt on one rank must fail the others too, or they return as if the
+        # checkpoint were whole.
+        error = e
+        raise
     finally:
-        if is_fsdp2_sharded:
-            torch.distributed.barrier()
+        # Every rank waits for the writer to finish. Without this the ranks that wrote nothing exit
+        # first and the job hangs on the ones still writing.
+        if is_fsdp2_sharded or is_ep_sharded:
+            _finish_export_together(error)
+
+
+def _agree_on_fused_expert_quantization(model: nn.Module) -> None:
+    """Refuse, on every rank, a fused-experts projection quantized for some experts and not others.
+
+    The no-gather export writes each projection's experts as one fused tensor, so they must be all
+    quantized or all not. Under expert parallelism a rank holds only its own experts' quantizers
+    and sees only its share: ranks that disagree would fold differently, and only the quantized
+    ones would join the writer's input-scale all-reduce. So every rank's view is gathered, and every
+    rank raises, before any of them folds. Collective.
+    """
+
+    def states() -> dict[str, str]:
+        found: dict[str, str] = {}
+        for name, module in model.named_modules():
+            if _hf_export_handlers._has_fused_experts_quantizers(module):
+                for proj in (getattr(module, "_first_proj_attr", "gate_up_proj"), "down_proj"):
+                    # As iter_weights_for_calibration does: a layout may register only the first.
+                    quantizers = getattr(module, f"{proj}_weight_quantizers", None)
+                    if quantizers is None:
+                        continue
+                    enabled = [q.is_enabled for q in quantizers]
+                    found[f"{name}.{proj}"] = (
+                        "quantized" if all(enabled) else "mixed" if any(enabled) else "unquantized"
+                    )
+        return found
+
+    def agree(per_rank: list[dict[str, str]]) -> dict[str, str]:
+        # A projection's state where every rank reports the same one, else what each rank reported.
+        merged: dict[str, str] = {}
+        for key in sorted(set().union(*per_rank)):
+            seen = [found.get(key, "absent") for found in per_rank]
+            merged[key] = (
+                seen[0]
+                if len(set(seen)) == 1
+                else ", ".join(f"rank {r}: {state}" for r, state in enumerate(seen))
+            )
+        return merged
+
+    agreed = all_reduce("fused expert quantization", states, agree)
+    unsupported = {k: v for k, v in agreed.items() if v not in ("quantized", "unquantized")}
+    if unsupported:
+        raise NotImplementedError(
+            "The expert-parallel export writes each fused-experts projection as one tensor, so its "
+            "experts must be all quantized or all unquantized, on every rank: "
+            + "; ".join(f"{key} ({state})" for key, state in unsupported.items())
+        )
+
+
+def _process_together(
+    model: nn.Module, dtype: torch.dtype | None, is_modelopt_qlora: bool
+) -> dict[str, Any]:
+    """:func:`_process_hf_model_for_export` on every rank, raising on all of them if any failed.
+
+    Each rank folds only its own shard, so processing can fail on one rank alone -- one rank's
+    experts mixing quantized and unquantized ones, say. That rank would leave for the closing
+    consensus while its peers start the writer's collectives, and the two would not match.
+    """
+    quant_config: list[dict[str, Any]] = []
+
+    def processed() -> bool:
+        quant_config.append(_process_hf_model_for_export(model, dtype, is_modelopt_qlora))
+        return True
+
+    all_reduce("export processed", processed, all)
+    return quant_config[0]
+
+
+def _finish_export_together(error: BaseException | None) -> None:
+    """Wait for every rank to finish writing, and raise on all of them if any failed.
+
+    A barrier alone lets the ranks whose writes succeeded return, and the run carries on as if the
+    checkpoint were whole while another rank's export raised. Every rank's outcome is gathered
+    instead (:func:`~modelopt.torch.utils.distributed.all_reduce`), so every rank raises with the
+    failing rank's error.
+    """
+
+    def exported() -> bool:
+        if error is not None:
+            raise error
+        return True
+
+    all_reduce("export_ok", exported, all)
diff --git a/modelopt/torch/utils/distributed.py b/modelopt/torch/utils/distributed.py
index 2a82598dbb4..bb3dca0b260 100644
--- a/modelopt/torch/utils/distributed.py
+++ b/modelopt/torch/utils/distributed.py
@@ -57,6 +57,7 @@
     "distributed_forward_loop",
     "distributed_run",
     "distribution_plan",
+    "expert_parallel_state",
     "fsdp2_wrap",
     "gather_bytes",
     "gather_state_dicts",
@@ -690,6 +691,19 @@ def distributed_run(
             _destroy_group(exc)
 
 
+def expert_parallel_state(model: torch.nn.Module) -> "ParallelState | None":
+    """Return the ``ParallelState`` of the model's routed experts, or ``None`` if none has one.
+
+    An expert-parallel load seeds every routed-expert module with a ``ParallelState`` that carries the
+    expert-model-parallel group; no other module of a HuggingFace model has one.
+    """
+    for module in model.modules():
+        state = getattr(module, "_parallel_state", None)
+        if state is not None and state.expert_model_parallel_group.is_initialized():
+            return state
+    return None
+
+
 class _StridedIterableShard(IterableDataset):
     """Rank-strided shard of a streaming ``IterableDataset`` (no ``len()`` required).
 
diff --git a/tests/unit/torch/export/test_distributed_writer.py b/tests/unit/torch/export/test_distributed_writer.py
index a35e47086c9..810eff08870 100644
--- a/tests/unit/torch/export/test_distributed_writer.py
+++ b/tests/unit/torch/export/test_distributed_writer.py
@@ -64,7 +64,7 @@
     _export_transformers_checkpoint,
     export_hf_checkpoint,
 )
-from modelopt.torch.utils.distributed import DistributionPlan, set_distribution_plan
+from modelopt.torch.utils.distributed import DistributionPlan, ParallelState, set_distribution_plan
 
 _WRITER = "modelopt.torch.export.distributed_writer"
 
@@ -177,6 +177,51 @@ def _fsdp2_shard(model, mesh, wrap_root=True):
     return model
 
 
+def test_an_expert_parallel_plan_needs_its_experts_expert_parallel_group(
+    gloo_world_of_one, tmp_path
+):
+    """Without the group the writer could not offset each EP rank's experts, and every rank would
+    write experts 0..n-1 over the others'."""
+    model = _load(_save(_llama_config(), tmp_path / "src"))
+    set_distribution_plan(model, DistributionPlan(2, ep_size=2))
+
+    with pytest.raises(RuntimeError, match="no routed expert carries its expert-parallel group"):
+        _write(model, tmp_path / "export")
+
+
+def _expert_parallel_groups_job(rank, world_size, tmp_dir):
+    for name, data_parallel_group, refused in (
+        # Left at its default, the data-parallel group is the whole world.
+        ("default-data-parallel-group", None, True),
+        # The plan's single replica leaves the experts' data-parallel group unused.
+        ("unused-data-parallel-group", -1, False),
+    ):
+        model = _load(tmp_dir / "src")
+        set_distribution_plan(model, DistributionPlan(world_size, ep_size=world_size))
+        model.model.layers[0].mlp._parallel_state = ParallelState(
+            data_parallel_group=data_parallel_group, expert_model_parallel_group=dist.group.WORLD
+        )
+        try:
+            with bounded_collectives():
+                _write(model, tmp_dir / name)
+        except RuntimeError as error:
+            raised = str(error)
+        else:
+            raised = "no error"
+        assert ("ep_size=2 and edp=1" in raised) if refused else raised == "no error", raised
+
+
+def test_an_expert_parallel_export_checks_its_experts_groups_against_the_plan(
+    cpu_dist_workers, tmp_path
+):
+    """Each rank's expert coordinates come from the routed experts' groups. Taken as the replica
+    index, a data-parallel group left at its default, the whole world, would leave every rank but
+    rank 0 out of the expert write with no error, so every rank refuses it; the plan's own groups
+    pass."""
+    _save(_llama_config(), tmp_path / "src")
+    cpu_dist_workers.run(_expert_parallel_groups_job, tmp_dir=tmp_path)
+
+
 # FP8 weights, with an FP8 KV cache in layer 0 and an NVFP4 one in layer 1: a mixed-precision KV
 # cache, whose per-layer formats the export reads from the quantization config.
 _FP8_MIXED_KV_CFG = copy.deepcopy(mtq.FP8_DEFAULT_CFG)
diff --git a/tests/unit/torch/export/test_fused_experts_keep_fused.py b/tests/unit/torch/export/test_fused_experts_keep_fused.py
index a646c3743a3..6ed2b81ad57 100644
--- a/tests/unit/torch/export/test_fused_experts_keep_fused.py
+++ b/tests/unit/torch/export/test_fused_experts_keep_fused.py
@@ -565,6 +565,32 @@ def test_fused_experts_prefixes_lists_each_fused_module_once_in_order():
     ]
 
 
+@pytest.mark.parametrize(
+    ("ep_rank", "total_experts", "first_index"),
+    [
+        (0, 0, 0),
+        # This EP group's slice of a larger expert set: shift by the group base.
+        (1, 2 * LOCAL_EXPERTS, LOCAL_EXPERTS),
+        (3, 4 * LOCAL_EXPERTS, 3 * LOCAL_EXPERTS),
+        # The tensor already spans every expert, so its indices are global.
+        (1, LOCAL_EXPERTS, 0),
+        # Total unknown: the legacy per-group shift.
+        (1, 0, LOCAL_EXPERTS),
+    ],
+)
+def test_split_local_numbers_experts_globally(ep_rank, total_experts, first_index):
+    sd = _fused_state_dict(gated=True)
+    unshifted = split_fused_experts_state_dict(dict(sd))
+
+    out = _split_local_fused_module(PREFIX, sd, ep_rank=ep_rank, total_experts=total_experts)
+
+    want = {}
+    for key, value in unshifted.items():
+        expert, tail = key[len(PREFIX) + 1 :].split(".", 1)
+        want[f"{PREFIX}.{int(expert) + first_index}.{tail}"] = value
+    _assert_byte_identical(out, want)
+
+
 # ------------------------------------------------------------------ sharded (DTensor) experts
 
 
diff --git a/tests/unit/torch/export/test_quant_aware_conversion.py b/tests/unit/torch/export/test_quant_aware_conversion.py
index c584c085b8d..d6feac8799b 100644
--- a/tests/unit/torch/export/test_quant_aware_conversion.py
+++ b/tests/unit/torch/export/test_quant_aware_conversion.py
@@ -23,6 +23,7 @@
 """
 
 import json
+import sys
 import types
 import warnings
 from fnmatch import fnmatchcase
@@ -40,6 +41,7 @@
     SplitRule,
     _assert_experts_pre_expanded,
     _build_reverse_rules,
+    _dense_split_rule,
     _source_checkpoint_keys,
     _SourceAwareRenamer,
     apply_reverse_rules,
@@ -1281,3 +1283,145 @@ def test_revert_quant_config_names_mapper():
     q2 = {"exclude_modules": ["x*"]}
     revert_quant_config_names(q2, None)
     assert q2["exclude_modules"] == ["x*"]
+
+
+def _as_loading_rewrote(conversions, siblings):
+    """``conversions`` with every expert converter's ``.weight`` sources expanded the way a
+    load-time rewrite expands them: one anchored source per checkpoint tensor it consumes."""
+    # Local import: optional dependency; every caller importorskips transformers first.
+    from transformers.core_model_loading import WeightConverter
+
+    rewritten = []
+    for conv in conversions:
+        sources = conv.source_patterns if isinstance(conv.source_patterns, list) else []
+        if not isinstance(conv, WeightConverter) or not any(".experts." in s for s in sources):
+            rewritten.append(conv)
+            continue
+        stems = [s.removesuffix("$").removesuffix(".weight") for s in sources]
+        rewritten.append(
+            WeightConverter(
+                source_patterns=[f"{stem}.{leaf}$" for leaf in siblings for stem in stems],
+                target_patterns=getattr(conv, "_original_target_patterns", conv.target_patterns),
+                operations=list(conv.operations),
+            )
+        )
+    return rewritten
+
+
+@pytest.mark.parametrize(
+    "siblings",
+    [("weight", "weight_scale_inv"), ("weight_packed", "weight_scale", "weight_global_scale")],
+    ids=["fp8-dequantize", "compressed-tensors-decompress"],
+)
+def test_experts_revert_through_converters_loading_rewrote(siblings):
+    """The export replays the conversions ``from_pretrained`` recorded. After a dequantizing or
+    decompressing load, each expert converter lists every checkpoint tensor it consumed, so the
+    same leaf repeats; the expert reverse used to bail on that and leave every name in-memory.
+
+    A FineGrainedFP8 ``dequantize=True`` load also adds a converter that only decodes each weight
+    in place. It renames nothing, so it must neither abort the reverse of the model's real
+    conversions (MiniMax-M2.7 kept its in-memory expert names) nor re-encode a weight.
+    """
+    pytest.importorskip("transformers")
+    # Local import: optional dependency, guarded by the importorskip above.
+    from _test_utils.torch.transformers_models import get_tiny_mixtral
+
+    try:
+        from transformers.conversion_mapping import get_model_conversion_mapping
+    except ImportError:
+        pytest.skip("transformers build has no conversion_mapping API")
+    model = get_tiny_mixtral(**_MIXTRAL_KWARGS)
+    stock = get_model_conversion_mapping(model, add_legacy=False)
+    if not any(".experts." in str(c.source_patterns) for c in stock):
+        pytest.skip("transformers build has no mixtral expert conversion")
+    model._weight_conversions = _as_loading_rewrote(stock, siblings)
+    if "weight_scale_inv" in siblings:
+        fp8 = pytest.importorskip("transformers.integrations.finegrained_fp8")
+        from transformers.core_model_loading import WeightConverter
+
+        model._weight_conversions.append(
+            WeightConverter(
+                source_patterns=["weight$", "weight_scale_inv", "activation_scale"],
+                target_patterns="weight",
+                operations=[fp8.Fp8Dequantize(None)],
+            )
+        )
+
+    p = "model.layers.0"
+    sd = {
+        f"{p}.mlp.gate.weight": torch.randn(2, 32),
+        f"{p}.self_attn.q_proj.weight": torch.randn(8, 16, dtype=torch.bfloat16),
+    }
+    for e in range(2):
+        sd.update(_nvfp4_linear(f"{p}.mlp.experts.{e}.gate_proj", 64, 32))
+        sd.update(_nvfp4_linear(f"{p}.mlp.experts.{e}.up_proj", 64, 32))
+        sd.update(_nvfp4_linear(f"{p}.mlp.experts.{e}.down_proj", 32, 64))
+
+    out = revert_weight_conversion_quant_aware(model, sd)
+
+    for e in range(2):
+        base = f"{p}.block_sparse_moe.experts.{e}"
+        assert f"{base}.w1.weight" in out
+        assert f"{base}.w3.weight" in out
+        assert f"{base}.w2.weight" in out
+        assert f"{base}.w1.weight_scale_2" in out
+    assert out[f"{p}.self_attn.q_proj.weight"].dtype == torch.bfloat16
+    assert not any(".mlp.experts." in k or k.endswith("weight_scale_inv") for k in out)
+
+
+def test_the_identity_decode_is_known_without_the_fp8_integration(monkeypatch):
+    """transformers 5.5 imports triton atop its FP8 integration, so without triton that import
+    fails. The identity op 5.5 and 5.6 reverse ``Fp8Dequantize`` into is ``core_model_loading``'s,
+    and must still read as a decode in place, or the whole reverse of a dequantizing load aborts."""
+    core = pytest.importorskip("transformers.core_model_loading")
+    if not hasattr(core, "_IdentityOp"):
+        pytest.skip("transformers build has no _IdentityOp")
+    integrations = pytest.importorskip("transformers.integrations")
+    monkeypatch.delattr(integrations, "finegrained_fp8", raising=False)
+    monkeypatch.setitem(sys.modules, "transformers.integrations.finegrained_fp8", None)
+    quant_aware_conversion._quantization_codec_ops.cache_clear()
+    try:
+        rev = types.SimpleNamespace(source_patterns=["weight$"], target_patterns="weight")
+        assert quant_aware_conversion._decodes_in_place(rev, [core._IdentityOp()])
+    finally:
+        quant_aware_conversion._quantization_codec_ops.cache_clear()
+
+
+@pytest.mark.parametrize(
+    ("sources", "targets", "in_place"),
+    [
+        # A dequantizing FP8 load, reversed: the weight and the scales folded into it.
+        (["weight$"], ["weight$", "weight_scale_inv", "activation_scale"], True),
+        (["mlp.up_proj.weight$"], ["mlp.up_proj.weight$", "mlp.up_proj.weight_scale_inv"], True),
+        # A later target names another module, so the converter renames it.
+        (["mlp.up_proj.weight$"], ["mlp.up_proj.weight$", "mlp.gate_proj.weight"], False),
+    ],
+    ids=["bare-siblings", "module-siblings", "another-module"],
+)
+def test_a_codec_converter_decodes_in_place_only_if_it_names_one_module(sources, targets, in_place):
+    """A converter skipped as a decode in place gives up its reverse rules, so it must rename
+    nothing: a target naming another module is kept, to be reversed or refused."""
+    core = pytest.importorskip("transformers.core_model_loading")
+    if not hasattr(core, "_IdentityOp"):
+        pytest.skip("transformers build has no _IdentityOp")
+    rev = types.SimpleNamespace(source_patterns=sources, target_patterns=targets)
+    assert quant_aware_conversion._decodes_in_place(rev, [core._IdentityOp()]) is in_place
+
+
+def test_dense_split_names_each_part_once_through_load_siblings():
+    """A dense fused converter that a dequantizing load rewrote still un-fuses into one part per
+    module, not one per checkpoint tensor the load consumed."""
+    pytest.importorskip("transformers")
+    # Local import: optional dependency, guarded by the importorskip above.
+    from transformers.core_model_loading import Concatenate, WeightConverter
+
+    modules = ("self_attn.q_proj", "self_attn.k_proj", "self_attn.v_proj")
+    leaves = ("weight", "weight_scale_inv")
+    conv = WeightConverter(
+        source_patterns=[f"{m}.{leaf}$" for leaf in leaves for m in modules],
+        target_patterns="self_attn.qkv_proj",
+        operations=[Concatenate(dim=0)],
+    )
+    rev = conv.reverse_transform()
+    rule = _dense_split_rule(rev, list(rev.operations))
+    assert rule.part_suffixes == (".q_proj", ".k_proj", ".v_proj")
diff --git a/tests/unit/torch/export/test_unified_export_hf.py b/tests/unit/torch/export/test_unified_export_hf.py
index ef113dbcc4d..0273ffd31ce 100644
--- a/tests/unit/torch/export/test_unified_export_hf.py
+++ b/tests/unit/torch/export/test_unified_export_hf.py
@@ -19,9 +19,11 @@
 import json
 from copy import deepcopy
 from types import SimpleNamespace
+from unittest import mock
 
 import pytest
 import torch
+from _test_utils.torch.distributed.utils import bounded_collectives
 from _test_utils.torch.quantization.tied_modules import (
     make_tied_linear_pair,
     wrap_in_parent_with_tied_keys,
@@ -29,6 +31,7 @@
 from safetensors.torch import save_file
 
 import modelopt.torch.quantization as mtq
+from modelopt.torch.export import distributed_writer, unified_export_hf
 from modelopt.torch.export.model_utils import (
     TiedWeightMap,
     get_language_model_from_vl,
@@ -44,10 +47,13 @@
     _fuse_shared_input_modules,
     _resolve_export_dtype,
     _source_checkpoint,
+    _write_base_config,
     collect_shared_input_modules,
+    export_hf_checkpoint,
     read_unplaced_weights,
 )
 from modelopt.torch.quantization.nn import TensorQuantizer
+from modelopt.torch.utils.distributed import DistributionPlan, set_distribution_plan
 
 
 def test_multimodal_detection_accepts_null_architectures():
@@ -1107,3 +1113,249 @@ def test_source_checkpoint_is_none_for_a_model_built_from_config(tmp_path, monke
 
     assert _source_checkpoint(model) is None
     assert not any(export_dir.iterdir())
+
+
+@pytest.mark.parametrize("source_plan", [None, {"layers.*.feed_forward.experts": "grouped_gemm"}])
+def test_distributed_export_writes_the_config_a_single_gpu_export_writes(tmp_path, source_plan):
+    """The EP load's DistributedConfig and torch_a2a plan stay out of config.json, the source's own
+    plan survives, and the live configs keep the run's values afterwards."""
+    pytest.importorskip("transformers")
+    # Local import: optional dependency, guarded by the importorskip above.
+    from transformers import AutoConfig, Llama4Config
+    from transformers.distributed import DistributedConfig
+
+    source = tmp_path / "source"
+    Llama4Config().save_pretrained(source)
+    if source_plan is not None:
+        config_json = json.loads((source / "config.json").read_text())
+        config_json["text_config"]["base_model_ep_plan"] = source_plan
+        (source / "config.json").write_text(json.dumps(config_json))
+    AutoConfig.from_pretrained(source).save_pretrained(tmp_path / "single_gpu")
+
+    # What the EP load leaves behind: the stamped DistributedConfig and the swapped plan.
+    config = AutoConfig.from_pretrained(source)
+    distributed_config = DistributedConfig(enable_expert_parallel=True)
+    swapped_plan = {"layers.*.feed_forward.experts": "torch_a2a_experts"}
+    config.distributed_config = distributed_config
+    config.text_config.base_model_ep_plan = swapped_plan
+    model = SimpleNamespace(config=config, _modelopt_source_checkpoint=str(source))
+
+    _write_base_config(model, tmp_path / "export")
+
+    written = json.loads((tmp_path / "export" / "config.json").read_text())
+    assert written == json.loads((tmp_path / "single_gpu" / "config.json").read_text())
+    assert written["text_config"].get("base_model_ep_plan") == source_plan
+    assert config.distributed_config is distributed_config
+    assert config.text_config.base_model_ep_plan is swapped_plan
+
+
+def _one_rank_fails_after_the_write_job(rank, size, tmp_dir, failure):
+    model = torch.nn.Linear(2, 2)
+    set_distribution_plan(model, DistributionPlan(size, ep_size=size))
+    raised = None
+    with (
+        bounded_collectives(),
+        mock.patch.object(unified_export_hf, "_process_hf_model_for_export", return_value={}),
+        mock.patch.object(distributed_writer, "distributed_save_hf_checkpoint") as write,
+        # Rank 0 alone writes config.json, once every rank has written its shards.
+        mock.patch.object(unified_export_hf, "_write_base_config", side_effect=failure),
+    ):
+        # Not pytest.raises: when nothing is raised it fails with a BaseException, which would take
+        # the shared worker down instead of failing this test alone.
+        try:
+            unified_export_hf.export_hf_checkpoint(model, export_dir=tmp_dir)
+        except (OSError, RuntimeError, SystemExit) as e:
+            raised = e
+    assert write.called
+    if isinstance(failure, SystemExit):  # an exit exits every rank, with its code
+        assert isinstance(raised, SystemExit) and raised.code == failure.code, raised
+    else:
+        assert raised is not None and "disk full" in str(raised), raised
+
+
+@pytest.mark.parametrize("failure", [OSError("disk full"), SystemExit(3)], ids=["error", "exit"])
+def test_every_rank_fails_an_export_that_one_rank_failed_after_the_write(
+    cpu_dist_workers, tmp_path, failure
+):
+    """The ranks that wrote their shards must not return as if the checkpoint were whole while rank 0
+    failed to finish it: a barrier alone would let them, and so would an exit, which is no Exception."""
+    cpu_dist_workers.run(_one_rank_fails_after_the_write_job, tmp_dir=tmp_path, failure=failure)
+
+
+def _rank_0_cannot_read_the_carried_weights_job(rank, size, tmp_dir):
+    model = torch.nn.Linear(2, 2)
+    set_distribution_plan(model, DistributionPlan(size, ep_size=size))
+    read = unified_export_hf.read_unplaced_weights
+
+    def unreadable_on_rank_0(*args, **kwargs):
+        if rank == 0:
+            raise OSError("a carried shard of the source is unreadable")
+        return read(*args, **kwargs)
+
+    raised = None
+    with (
+        bounded_collectives(),
+        mock.patch.object(unified_export_hf, "read_unplaced_weights", unreadable_on_rank_0),
+        mock.patch.object(unified_export_hf, "_process_hf_model_for_export", return_value={}),
+        mock.patch.object(distributed_writer, "distributed_save_hf_checkpoint") as write,
+    ):
+        # Not pytest.raises: when nothing is raised it fails with a BaseException, which would take
+        # the shared worker down instead of failing this test alone.
+        try:
+            unified_export_hf.export_hf_checkpoint(model, export_dir=tmp_dir)
+        except (OSError, RuntimeError) as e:
+            raised = e
+    # Rank 0 raises its own error, and rank 1 one that names it, before either writes.
+    assert raised is not None and "a carried shard of the source is unreadable" in str(raised), (
+        raised
+    )
+    write.assert_not_called()
+
+
+def test_every_rank_fails_an_export_whose_carried_weights_rank_0_cannot_read(
+    cpu_dist_workers, tmp_path
+):
+    """Rank 0 alone reads the carried tensors off the source, so its failure there must fail every
+    rank, not leave the others waiting in the export's first collective."""
+    cpu_dist_workers.run(_rank_0_cannot_read_the_carried_weights_job, tmp_dir=tmp_path)
+
+
+def _one_rank_fails_processing_job(rank, size, tmp_dir):
+    model = torch.nn.Linear(2, 2)
+    set_distribution_plan(model, DistributionPlan(size, ep_size=size))
+    failure = NotImplementedError("rank 1's experts mix quantized and unquantized ones")
+
+    def process(*args):
+        if rank == 1:
+            raise failure
+        return {}
+
+    raised = None
+    with (
+        bounded_collectives(),
+        mock.patch.object(unified_export_hf, "_process_hf_model_for_export", process),
+        mock.patch.object(distributed_writer, "distributed_save_hf_checkpoint") as write,
+    ):
+        # Not pytest.raises: when nothing is raised it fails with a BaseException, which would take
+        # the shared worker down instead of failing this test alone.
+        try:
+            unified_export_hf.export_hf_checkpoint(model, export_dir=tmp_dir)
+        except (NotImplementedError, RuntimeError) as e:
+            raised = e
+    # Rank 1 raises its own error, and rank 0 one that names rank 1, before either writes.
+    if rank == 1:
+        assert raised is failure, raised
+    else:
+        assert isinstance(raised, RuntimeError) and "rank 1" in str(raised), raised
+    write.assert_not_called()
+
+
+def test_every_rank_stops_before_the_write_when_one_ranks_processing_failed(
+    cpu_dist_workers, tmp_path
+):
+    """Each rank folds only its own shard, so its processing can fail alone; the others must not
+    go on into the writer's collectives without it."""
+    cpu_dist_workers.run(_one_rank_fails_processing_job, tmp_dir=tmp_path)
+
+
+class _FusedExpertsQuantizers(torch.nn.Module):
+    """What the export reads of a fused-experts container: each projection's weight quantizers."""
+
+    def __init__(self, enabled, down_enabled=None):
+        super().__init__()
+        down_enabled = enabled if down_enabled is None else down_enabled
+        self.gate_up_proj_weight_quantizers = [SimpleNamespace(is_enabled=e) for e in enabled]
+        if down_enabled != "absent":  # a layout may register only the first projection's
+            self.down_proj_weight_quantizers = [SimpleNamespace(is_enabled=e) for e in down_enabled]
+
+
+def _expert_quantization_job(rank, size, tmp_dir, enabled_per_rank, down_enabled_per_rank, refused):
+    model = torch.nn.Module()
+    down_enabled = None if down_enabled_per_rank is None else down_enabled_per_rank[rank]
+    model.experts = _FusedExpertsQuantizers(enabled_per_rank[rank], down_enabled)
+    set_distribution_plan(model, DistributionPlan(size, ep_size=size))
+    raised = None
+    with (
+        mock.patch.object(
+            unified_export_hf, "_process_hf_model_for_export", return_value={}
+        ) as fold,
+        mock.patch.object(distributed_writer, "distributed_save_hf_checkpoint") as write,
+        mock.patch.object(unified_export_hf, "_write_base_config"),
+        mock.patch.object(unified_export_hf, "_write_hf_export_config"),
+        bounded_collectives(),
+    ):
+        # Not pytest.raises: when nothing is raised it fails with a BaseException, which would take
+        # the shared worker down instead of failing this test alone.
+        try:
+            unified_export_hf.export_hf_checkpoint(model, export_dir=tmp_dir)
+        except NotImplementedError as e:
+            raised = str(e)
+    if refused is None:
+        assert raised is None, raised
+    else:
+        assert raised is not None and refused in raised, raised
+    # Refused before any rank folds or writes; agreed states export as usual.
+    assert fold.called == write.called == (refused is None)
+
+
+@pytest.mark.parametrize(
+    ("enabled_per_rank", "down_enabled_per_rank", "refused"),
+    [
+        (((False, False), (True, True)), None, "experts.gate_up_proj"),
+        (((True, False), (False, True)), None, "experts.gate_up_proj"),
+        (((True, True), (True, True)), ((False, False), (True, True)), "experts.down_proj"),
+        (((True, True), (True, True)), ("absent", "absent"), None),
+        (((True, True), (True, True)), None, None),
+        (((False, False), (False, False)), None, None),
+    ],
+    ids=[
+        "ranks-disagree",
+        "every-rank-mixed",
+        "down-proj-disagrees",
+        "no-down-proj-quantizers",
+        "all-quantized",
+        "all-unquantized",
+    ],
+)
+def test_expert_parallel_ranks_agree_on_their_experts_quantization_before_folding(
+    cpu_dist_workers, tmp_path, enabled_per_rank, down_enabled_per_rank, refused
+):
+    """Each rank sees only its own experts' quantizers. Ranks that disagree would fold differently,
+    and only the quantized ones would join the writer's input-scale all-reduce."""
+    cpu_dist_workers.run(
+        _expert_quantization_job,
+        tmp_dir=tmp_path,
+        enabled_per_rank=enabled_per_rank,
+        down_enabled_per_rank=down_enabled_per_rank,
+        refused=refused,
+    )
+
+
+def test_an_expert_parallel_export_refuses_to_save_the_modelopt_state(tmp_path):
+    """Each expert-parallel rank holds only its own experts' ModelOpt state, so saving rank 0's
+    would write an incomplete one: refused before anything is written."""
+    model = torch.nn.Sequential(torch.nn.Linear(16, 16))
+    mtq.quantize(model, mtq.FP8_DEFAULT_CFG, lambda m: m(torch.randn(2, 16)))
+    set_distribution_plan(model, DistributionPlan(2, ep_size=2))
+
+    with pytest.raises(NotImplementedError, match="expert-parallel export"):
+        export_hf_checkpoint(model, export_dir=tmp_path / "export", save_modelopt_state=True)
+
+    assert not (tmp_path / "export").exists()
+
+
+def test_an_expert_parallel_plan_with_replicated_dense_shards_is_refused_before_folding(tmp_path):
+    """Below the world size, EP leaves FSDP2's dense shards replicated across the expert-parallel
+    groups, which the no-gather writer cannot write: refused before the export folds anything."""
+    model = torch.nn.Sequential(torch.nn.Linear(16, 16))
+    mtq.quantize(model, mtq.FP8_DEFAULT_CFG, lambda m: m(torch.randn(2, 16)))
+    set_distribution_plan(model, DistributionPlan(4, ep_size=2, dense="fsdp"))
+    before = {k: v.clone() for k, v in model.state_dict().items()}
+
+    with pytest.raises(NotImplementedError, match=r"MODELOPT_EP_DEGREE=4 .* MODELOPT_FSDP_SIZE=1"):
+        export_hf_checkpoint(model, export_dir=tmp_path / "export")
+
+    after = model.state_dict()
+    assert after.keys() == before.keys()
+    assert all(torch.equal(after[k], v) for k, v in before.items())
+    assert not (tmp_path / "export").exists()
diff --git a/tests/unit/torch/utils/test_distributed.py b/tests/unit/torch/utils/test_distributed.py
index f292843c89a..7dddb8e2379 100644
--- a/tests/unit/torch/utils/test_distributed.py
+++ b/tests/unit/torch/utils/test_distributed.py
@@ -22,6 +22,7 @@
 from contextlib import nullcontext
 from datetime import timedelta
 from functools import partial
+from types import SimpleNamespace
 from unittest import mock
 
 import pytest
@@ -43,6 +44,7 @@
     broadcast_from,
     distributed_run,
     distribution_plan,
+    expert_parallel_state,
     set_distribution_plan,
 )
 
@@ -548,3 +550,24 @@ def test_the_run_leaves_a_group_it_found_alone(monkeypatch, recwarn, world_size,
         if raises:
             raise ValueError("calibration failed")
     assert "timeout=1:00:00) does not apply" in str(recwarn.pop(UserWarning).message)
+
+
+# ---------------------------------------------------------------------------- expert parallel state
+
+
+def _state(expert_parallel: bool):
+    return SimpleNamespace(
+        expert_model_parallel_group=SimpleNamespace(is_initialized=lambda: expert_parallel)
+    )
+
+
+def test_the_routed_experts_state_is_found_by_its_expert_parallel_group():
+    model = torch.nn.Module()
+    model.attn = torch.nn.Linear(2, 2)
+    model.experts = torch.nn.Module()
+    assert expert_parallel_state(model) is None
+    for module in model.modules():
+        module._parallel_state = _state(expert_parallel=False)
+    assert expert_parallel_state(model) is None
+    model.experts._parallel_state = expert = _state(expert_parallel=True)
+    assert expert_parallel_state(model) is expert