diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index e32358de351..4a6116df670 100755
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -40,6 +40,7 @@ Changelog
 - Add an end-to-end W4A4 NVFP4 PTQ and QAD tutorial for Qwen3.6-35B-A3B also covering evaluation and vLLM throughput benchmarking. See `examples/megatron_bridge/tutorials/Qwen3.6-35B-A3B/README.md <https://github.com/NVIDIA/Model-Optimizer/tree/main/examples/megatron_bridge/tutorials/Qwen3.6-35B-A3B/>`_ for details.
 - Add ``--mlflow <tracking-uri>`` to the ``examples/megatron_bridge`` scripts that write a checkpoint -- ``prune_minitron.py``, ``quantize.py``, ``distill.py``, ``export_quantized_megatron_to_hf.py`` and ``export_distilled_megatron_to_hf.py`` (MLflow's own ``MLFLOW_TRACKING_URI`` is honoured too). Each run records the invocation, its arguments as searchable params and its log -- a distillation records its training metrics instead of the rank-0 log -- and writes ``.experiment.json`` into the checkpoint it produced, so a pruning, a quantization, the distillation that refines its checkpoint and the export that deploys it can be traced to one another; uploading the checkpoints themselves stays off unless ``--mlflow_log_checkpoints`` is passed.
 - Add ``--ep_size`` to ``examples/megatron_bridge/export_quantized_megatron_to_hf.py`` so large MoE models with grouped-GEMM experts can be exported with their experts sharded across GPUs. Checkpoints built with ``--no_moe_grouped_gemm`` must still be exported at ``--ep_size 1``.
+- Add unified HF export of GLM-5 / GLM-5.2 (``glm_moe_dsa``) checkpoints quantized with ``examples/megatron_bridge``. The MTP layer, which Megatron-Bridge does not build for these models, is copied through unquantized from the source checkpoint. DSA sparse-attention models also gain FP8 KV-cache quantization on Megatron.
 
 *Misc*
 
diff --git a/modelopt/torch/distill/plugins/megatron.py b/modelopt/torch/distill/plugins/megatron.py
index 1bb9a3db791..5f6a1de4fb7 100644
--- a/modelopt/torch/distill/plugins/megatron.py
+++ b/modelopt/torch/distill/plugins/megatron.py
@@ -40,6 +40,7 @@
 import modelopt.torch.distill as mtd
 from modelopt.torch.distill.config import Criterion
 from modelopt.torch.quantization.nn import TensorQuantizer
+from modelopt.torch.utils import warn_rank_0
 
 if TYPE_CHECKING:
     from megatron.core.dist_checkpointing.mapping import ShardedStateDict
@@ -592,9 +593,11 @@ def _sharded_state_dict(self, *args, **kwargs) -> "ShardedStateDict":
     skip_mtp_loss = _mtp_excluded_from_quantization(model)
     if distill_cfg.skip_lm_loss and skip_mtp_loss:
         # Freeze the untrained MTP head: DDP's overlapped grad reduce asserts on params with no grad.
-        for name, param in model.named_parameters():
-            if "mtp" in name.split("."):
-                param.requires_grad_(False)
+        warn_rank_0("MTP head is outside quantization and its loss is skipped: freezing it.")
+        with model.hide_teacher_model():
+            for name, param in model.named_parameters():
+                if "mtp" in name.split("."):
+                    param.requires_grad_(False)
 
     def _compute_student_lm_loss(self, labels, logits) -> Tensor:
         self._lm_loss_call_count += 1
diff --git a/modelopt/torch/export/plugins/mcore_common.py b/modelopt/torch/export/plugins/mcore_common.py
index d275aee4c0a..b74a4fc6aaa 100644
--- a/modelopt/torch/export/plugins/mcore_common.py
+++ b/modelopt/torch/export/plugins/mcore_common.py
@@ -18,6 +18,7 @@
 from typing import Any
 
 from .mcore_deepseek import deepseek_causal_lm_export, deepseek_causal_lm_import
+from .mcore_glm import glm_moe_dsa_causal_lm_export
 from .mcore_gptoss import gptoss_causal_lm_export, gptoss_causal_lm_import
 from .mcore_llama import (
     eagle3_deep_llama_causal_lm_export,
@@ -63,6 +64,7 @@
     "Qwen3VLForConditionalGeneration": qwen3vl_causal_lm_export,
     "Qwen3_5ForConditionalGeneration": qwen3_5_vl_causal_lm_export,
     "Qwen3_5MoeForConditionalGeneration": qwen3_5_vl_causal_lm_export,
+    "GlmMoeDsaForCausalLM": glm_moe_dsa_causal_lm_export,
 }
 
 # VLM architectures whose Megatron export covers the language model only: the vision tower is copied
diff --git a/modelopt/torch/export/plugins/mcore_glm.py b/modelopt/torch/export/plugins/mcore_glm.py
new file mode 100644
index 00000000000..25a967860ac
--- /dev/null
+++ b/modelopt/torch/export/plugins/mcore_glm.py
@@ -0,0 +1,41 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Custom mappings from Megatron Core models to GLM-5.x Hugging Face models.
+
+GLM-5 / GLM-5.2 (``glm_moe_dsa``) is DeepSeek-V3-style MLA with a DSA indexer. It keeps the MTP
+layer at HF index ``num_hidden_layers`` under the decoder's names; Megatron-Bridge does not build
+it, so the exporter copies it from the source checkpoint.
+"""
+
+from .mcore_custom import (
+    GroupedGatedMLPSlicing,
+    GroupedMLPSlicing,
+    NameRemapping,
+    SelfAttentionScaling,
+)
+from .mcore_deepseek import deepseek_causal_lm_export
+
+glm_moe_dsa_causal_lm_export: dict = {
+    **deepseek_causal_lm_export,
+    "mtp_in_decoder_layers": True,
+    "core_attention": SelfAttentionScaling("model.layers.{}.self_attn."),
+    "indexer.linear_wq_b": NameRemapping("model.layers.{}.self_attn.indexer.wq_b."),
+    "indexer.linear_wk": NameRemapping("model.layers.{}.self_attn.indexer.wk."),
+    "indexer.k_norm": NameRemapping("model.layers.{}.self_attn.indexer.k_norm."),
+    "indexer.linear_weights_proj": NameRemapping("model.layers.{}.self_attn.indexer.weights_proj."),
+    "experts.linear_fc1": GroupedGatedMLPSlicing("model.layers.{}.mlp.experts.{{}}"),
+    "experts.linear_fc2": GroupedMLPSlicing("model.layers.{}.mlp.experts.{{}}.down_proj"),
+}
diff --git a/modelopt/torch/export/unified_export_megatron.py b/modelopt/torch/export/unified_export_megatron.py
index 8cf2e7b0001..f4b52c56bb2 100644
--- a/modelopt/torch/export/unified_export_megatron.py
+++ b/modelopt/torch/export/unified_export_megatron.py
@@ -29,7 +29,7 @@
 
 import torch
 import torch.distributed
-from huggingface_hub import get_safetensors_metadata, hf_hub_download
+from huggingface_hub import get_safetensors_metadata, hf_hub_download, snapshot_download
 from huggingface_hub.errors import EntryNotFoundError
 from safetensors import safe_open
 from safetensors.torch import save_file
@@ -175,6 +175,7 @@ def __init__(
         self._hf_text_config = getattr(self._hf_config, "text_config", self._hf_config)
 
         # Update hf_config
+        self._src_num_hidden_layers = self._hf_text_config.num_hidden_layers
         self._hf_text_config.num_hidden_layers = language_model.config.num_layers
         self._hf_text_config.hidden_size = language_model.config.hidden_size
         # MLA's kv_channels is the V head dim, not HF's head_dim (e.g. glm5_next derives it from RoPE).
@@ -203,6 +204,12 @@ def __init__(
                 del self._hf_config.quantization_config
         self.all_rules = self._populate_rule_book()
         self.rules = self.all_rules[self.arch]
+        self.all_mcore_mappings = all_mcore_hf_export_mapping[self.arch]
+        if self.rules.get("mtp_in_decoder_layers", False) and language_model.config.mtp_num_layers:
+            # Only the copy from the source checkpoint is supported (the bridge builds no MTP).
+            raise NotImplementedError(
+                f"Exporting a Megatron-built MTP for {self.arch} is not supported yet."
+            )
         # The vision tower is copied through unquantized, so deployments must not treat it as such.
         self.exclude_modules = [
             prefix.removesuffix(".") + "*" for prefix in self.vision_passthrough_prefixes or ()
@@ -351,9 +358,11 @@ def save_pretrained(
                 "Export at expert parallel size > 1 needs grouped-GEMM experts; "
                 "export SequentialMLP (--no_moe_grouped_gemm) checkpoints at EP=1."
             )
-        # One writer per pipeline stage: other TP / DP / EP ranks hold the same layers (EP>1 ranks
-        # hold no gathered experts at all), and writing them too would race on the same files.
-        writes_layers = (
+        # One layer-shard writer per pipeline stage: other TP / DP / EP ranks hold the same layers
+        # (EP>1 ranks hold no gathered experts at all), and writing them too would race on the same
+        # files. Unlike is_writer_rank (one global writer of save_directory metadata), this has no
+        # PP term on purpose: each stage owns a disjoint slice of the layer shards.
+        is_stage_layer_writer = (
             tp_rank == 0 and get_data_parallel_rank() == 0 and get_expert_model_parallel_rank() == 0
         )
 
@@ -421,7 +430,7 @@ def save_pretrained(
 
             # The live MTP export runs EP collectives, so every last-stage main rank joins it; the
             # collective-free copy from the source checkpoint only runs on the writer.
-            mtp_state_dict = self._get_mtp_state_dict(copy_from_pretrained=writes_layers)
+            mtp_state_dict = self._get_mtp_state_dict(copy_from_pretrained=is_stage_layer_writer)
             if len(mtp_state_dict) > 0:
                 layer_state_dicts[self.model.config.num_layers].update(mtp_state_dict)
                 print(f"Successfully loaded {len(mtp_state_dict)} MTP tensors")
@@ -459,7 +468,7 @@ def save_pretrained(
         # Add multimodal components to state_dict. Since only support decoder model quantization,
         # no changes will be made to the multimodal components. We copy the multimodal components
         # from the pretrained model directly to the state_dict to avoid implementing the export logic.
-        if is_first_stage_main_rank and writes_layers:
+        if is_first_stage_main_rank and is_stage_layer_writer:
             # layer_state_dicts is keyed by layer_number (1-indexed), so the first
             # decoder layer on this (first) PP stage is the smallest key, not 0.
             # Merge the multimodal components into that shard so they land in a file
@@ -488,7 +497,7 @@ def save_pretrained(
         torch.distributed.barrier()
 
         save_safetensors_by_layer_index(
-            layer_state_dicts=layer_state_dicts if writes_layers else {},
+            layer_state_dicts=layer_state_dicts if is_stage_layer_writer else {},
             total_layers=self.model.config.num_layers,
             save_directory=save_directory,
             name_template="model-{:05d}-of-{:05d}",
@@ -550,7 +559,11 @@ def _verify_exported_keys(self, save_directory, pretrained_model_name_or_path) -
         # Narrow on purpose: compare module prefixes, not tensor names, since a quantized source
         # carries extras with no export counterpart, and only inside decoder layers, whose naming
         # is stable. A dropped decoder module is the case that loads fine and produces garbage.
-        num_layers = self.model.config.num_layers
+        # HF decoder depth of this export: Megatron may build several physical layers per HF layer.
+        hf_depth = self._hf_text_config.num_hidden_layers
+        num_mtp = 0
+        if self.rules.get("mtp_in_decoder_layers", False):
+            num_mtp = getattr(self._hf_text_config, "num_nextn_predict_layers", 0) or 0
         # Ancestors too: an export may expand one source module into several (Qwen3.5 packs
         # routed experts; the quantized export writes them per expert). Expansion is not a drop.
         exported_modules = set()
@@ -562,12 +575,18 @@ def _verify_exported_keys(self, save_directory, pretrained_model_name_or_path) -
                     break
                 exported_modules.add(prefix)
         missing = set()
-        for key in source - exported:
+        for key in source:
             layer = re.search(r"\.layers\.(\d+)\.", key)
             if layer is None:
                 continue  # see the note above: decoder layers only
-            if int(layer.group(1)) >= num_layers:
-                continue  # depth-pruned model: the source has layers this export does not
+            if int(layer.group(1)) >= hf_depth:
+                mtp_id = int(layer.group(1)) - self._src_num_hidden_layers
+                if not 0 <= mtp_id < num_mtp:
+                    continue  # depth-pruned model: the source has layers this export does not
+                # An MTP stored as extra decoder layers follows the exported decoder layers.
+                key = f"{key[: layer.start(1)]}{hf_depth + mtp_id}{key[layer.end(1) :]}"
+            if key in exported:
+                continue
             if key.rsplit(".", 1)[0] in exported_modules:
                 continue  # module is exported; this name is a source-side quantization artifact
             if "rotary_emb" in key:
@@ -693,6 +712,12 @@ def _get_transformer_layer_state_dict(self, layer, layer_id, is_mtp=False):
                     layer.self_attention.linear_kv_up_proj, layer_id, is_mtp=is_mtp
                 )
                 self.rules["linear_proj"](layer.self_attention.linear_proj, layer_id, is_mtp=is_mtp)
+                core_attention = getattr(layer.self_attention, "core_attention", None)
+                if core_attention is not None and "core_attention" in self.rules:
+                    self.rules["core_attention"](core_attention, layer_id, is_mtp=is_mtp)
+                indexer = getattr(core_attention, "indexer", None)
+                if indexer is not None:
+                    self._get_dsa_indexer_state_dict(indexer, layer_id, is_mtp)
             elif "linear_attn" in self.rules and hasattr(layer.self_attention, "in_proj"):
                 # GatedDeltaNet (Qwen3.5 linear attention): no q/k layernorm, no core_attention.
                 self._get_gated_delta_net_state_dict(layer, layer_id, is_mtp=is_mtp)
@@ -857,6 +882,8 @@ def _copy_mtp_state_dict_from_pretrained(self) -> dict[str, torch.Tensor]:
         mtp_state_dict = {}
         if not self._hf_pretrained_model_name:
             return mtp_state_dict
+        if self.rules.get("mtp_in_decoder_layers", False):
+            return self._copy_decoder_mtp_layers_from_pretrained()
 
         mtp_exists = False
 
@@ -910,6 +937,55 @@ def _copy_mtp_state_dict_from_pretrained(self) -> dict[str, torch.Tensor]:
             self.exclude_modules.append("mtp*")
         return mtp_state_dict
 
+    def _copy_decoder_mtp_layers_from_pretrained(self) -> dict[str, torch.Tensor]:
+        """Copy MTP layers stored as extra decoder layers (GLM-5.x) from the source, dequantized.
+
+        Used when Megatron did not build the MTP (e.g. Megatron-Bridge's GLM-5 bridge); the copies
+        stay BF16 and are excluded from quantization, like the released NVFP4 checkpoints.
+        """
+        num_mtp = getattr(self._hf_text_config, "num_nextn_predict_layers", 0) or 0
+        source = self._hf_pretrained_model_name
+        if num_mtp == 0 or source is None:
+            return {}
+        layers_prefix = self.all_mcore_mappings["input_layernorm"].target_name_or_prefix
+        layers_prefix = layers_prefix.split("{}")[0]  # e.g. "model.layers."
+        src_prefixes = [
+            f"{layers_prefix}{self._src_num_hidden_layers + i}." for i in range(num_mtp)
+        ]
+        if not os.path.isdir(source):
+            source = self._download_hub_shards(str(source), tuple(src_prefixes))
+        keys = _read_checkpoint_keys(source)
+        mtp_state_dict = {}
+        for i in range(num_mtp):
+            src = src_prefixes[i]
+            dst = f"{layers_prefix}{self._hf_text_config.num_hidden_layers + i}."
+            for key in sorted(
+                k for k in keys if k.startswith(src) and not k.endswith("_scale_inv")
+            ):
+                mtp_state_dict[dst + key[len(src) :]] = get_safetensor(
+                    str(source), key, dequantize=True
+                )
+            self.exclude_modules.append(dst + "*")
+        if mtp_state_dict:
+            print(f"Copied {len(mtp_state_dict)} MTP tensors from {source}")
+        else:
+            warn_rank_0(f"No MTP tensors under {src_prefixes} in {source}; exporting without MTP.")
+        return mtp_state_dict
+
+    @staticmethod
+    def _download_hub_shards(repo_id: str, key_prefixes: tuple[str, ...]) -> str:
+        """Download only the Hub shards holding tensors under ``key_prefixes``; return the local dir."""
+        try:
+            index_file = hf_hub_download(repo_id, "model.safetensors.index.json")
+        except EntryNotFoundError:  # unsharded checkpoint
+            return snapshot_download(repo_id, allow_patterns=["model.safetensors"])
+        with open(index_file) as f:
+            weight_map = json.load(f)["weight_map"]
+        shards = sorted(
+            {shard for key, shard in weight_map.items() if key.startswith(key_prefixes)}
+        )
+        return snapshot_download(repo_id, allow_patterns=["model.safetensors.index.json", *shards])
+
     def _get_gated_delta_net_state_dict(self, layer, layer_id, is_mtp=False):
         """Export a GatedDeltaNet (Qwen3.5 linear-attention) layer's ``self_attention``."""
         gdn = layer.self_attention
@@ -920,6 +996,13 @@ def _get_gated_delta_net_state_dict(self, layer, layer_id, is_mtp=False):
         self.rules["linear_attn.out_norm"](gdn.out_norm, layer_id, is_mtp=is_mtp)
         self.rules["linear_attn.out_proj"](gdn.out_proj, layer_id, is_mtp=is_mtp)
 
+    def _get_dsa_indexer_state_dict(self, indexer, layer_id, is_mtp=False):
+        """Export the DSA indexer of a sparse MLA layer."""
+        if "indexer.linear_wq_b" not in self.rules:
+            raise NotImplementedError(f"No export rule for the DSA indexer of {self.arch}.")
+        for name in ("linear_wq_b", "linear_wk", "k_norm", "linear_weights_proj"):
+            self.rules[f"indexer.{name}"](getattr(indexer, name), layer_id, is_mtp=is_mtp)
+
     def _get_mamba_layer_state_dict(self, layer, layer_id, is_mtp=False):
         if not isinstance(layer.norm, IdentityOp):
             self.rules["norm"](layer.norm, layer_id, is_mtp=is_mtp)
diff --git a/modelopt/torch/quantization/plugins/megatron.py b/modelopt/torch/quantization/plugins/megatron.py
index aaacf198a12..5e422235e14 100644
--- a/modelopt/torch/quantization/plugins/megatron.py
+++ b/modelopt/torch/quantization/plugins/megatron.py
@@ -30,6 +30,15 @@
 import megatron.core.transformer.moe.experts as megatron_moe
 import torch
 from megatron.core.dist_checkpointing.utils import replace_prefix_for_sharding
+from megatron.core.extensions.transformer_engine import (
+    TEColumnParallelGroupedLinear,
+    TEColumnParallelLinear,
+    TEDotProductAttention,
+    TELayerNormColumnParallelLinear,
+    TELinear,
+    TERowParallelGroupedLinear,
+    TERowParallelLinear,
+)
 from megatron.core.models.gpt import GPTModel
 from megatron.core.parallel_state import get_data_parallel_group
 from megatron.core.tensor_parallel.mappings import gather_from_sequence_parallel_region
@@ -62,23 +71,14 @@
 from ..utils.layerwise_calib import LayerActivationCollector
 from .custom import CUSTOM_MODEL_PLUGINS, _ParallelLinear
 from .gated_delta_net import GatedDeltaNetStateQuantMixin
+from .transformer_engine import _QuantTEGroupedLinear, _QuantTELayerNormLinear, _QuantTELinear
 
 try:
-    from megatron.core.extensions.transformer_engine import (
-        TEColumnParallelGroupedLinear,
-        TEColumnParallelLinear,
-        TEDotProductAttention,
-        TELayerNormColumnParallelLinear,
-        TELinear,
-        TERowParallelGroupedLinear,
-        TERowParallelLinear,
-    )
-
-    from .transformer_engine import _QuantTEGroupedLinear, _QuantTELayerNormLinear, _QuantTELinear
-
-    HAS_TE = True
+    from megatron.core.transformer.experimental_attention_variant.dsa import DSAttention
+
+    HAS_DSA = True
 except ImportError:
-    HAS_TE = False
+    HAS_DSA = False
 
 try:
     from megatron.core.ssm.gated_delta_net import GatedDeltaNet
@@ -835,251 +835,275 @@ def sharded_state_dict(self, prefix="", sharded_offsets=(), metadata=None):
         return sharded_state_dict
 
 
-if HAS_TE:
+@QuantModuleRegistry.register({TERowParallelLinear: "te_mcore_RowParallelLinear"})
+class _QuantTEMCoreRowParallelLinear(_QuantTELinear, _MegatronRowParallelLinear):
+    pass
 
-    @QuantModuleRegistry.register({TERowParallelLinear: "te_mcore_RowParallelLinear"})
-    class _QuantTEMCoreRowParallelLinear(_QuantTELinear, _MegatronRowParallelLinear):
-        pass
 
-    @QuantModuleRegistry.register({TEColumnParallelLinear: "te_mcore_ColumnParallelLinear"})
-    class _QuantTEMCoreColumnParallelLinear(_QuantTELinear, _MegatronColumnParallelLinear):
-        pass
+@QuantModuleRegistry.register({TEColumnParallelLinear: "te_mcore_ColumnParallelLinear"})
+class _QuantTEMCoreColumnParallelLinear(_QuantTELinear, _MegatronColumnParallelLinear):
+    pass
 
-    @QuantModuleRegistry.register({TELinear: "te_mcore_Linear"})
-    class _QuantTEMCoreLinear(_QuantTELinear):
-        pass
 
-    @QuantModuleRegistry.register(
-        {TELayerNormColumnParallelLinear: "te_mcore_LayerNormColumnParallelLinear"}
-    )
-    class _QuantTELayerNormColumnParallelLinear(
-        _QuantTELayerNormLinear, _MegatronColumnParallelLinear
-    ):
-        pass
-
-    # Quantized subclasses to support TEGroupedLinear quantization
-    class _QuantMegatronTEGroupedLinear(_QuantTEGroupedLinear, _MegatronParallelLinear):
-        def modelopt_post_load_extra_state(self):
-            _initialize_grouped_weight_quantizer_state(self)
-
-        def _load_from_state_dict(self, state_dict, prefix, *args, **kwargs):
-            # _sharded_state_dict_grouped adds _extra_state{gemm_idx} for gemm_idx:[1, num_gemms] in
-            # sharded_state_dict which is same as _extra_state. The _extra_state{gemm_idx} is used for
-            # TE Fp8 checkpoint, we need to remove the _extra_state{gemm_idx} for gemm_idx:[1, num_gemms]
-            # for modelopt checkpoint restore
-            filtered_state_dict = {
-                k: v
-                for k, v in state_dict.items()
-                if not any(k.endswith(f"_extra_state{num}") for num in range(1, self.num_gemms))
-            }
-            return super()._load_from_state_dict(filtered_state_dict, prefix, *args, **kwargs)
+@QuantModuleRegistry.register({TELinear: "te_mcore_Linear"})
+class _QuantTEMCoreLinear(_QuantTELinear):
+    pass
+
+
+@QuantModuleRegistry.register(
+    {TELayerNormColumnParallelLinear: "te_mcore_LayerNormColumnParallelLinear"}
+)
+class _QuantTELayerNormColumnParallelLinear(_QuantTELayerNormLinear, _MegatronColumnParallelLinear):
+    pass
+
+
+# Quantized subclasses to support TEGroupedLinear quantization
+class _QuantMegatronTEGroupedLinear(_QuantTEGroupedLinear, _MegatronParallelLinear):
+    def modelopt_post_load_extra_state(self):
+        _initialize_grouped_weight_quantizer_state(self)
+
+    def _load_from_state_dict(self, state_dict, prefix, *args, **kwargs):
+        # _sharded_state_dict_grouped adds _extra_state{gemm_idx} for gemm_idx:[1, num_gemms] in
+        # sharded_state_dict which is same as _extra_state. The _extra_state{gemm_idx} is used for
+        # TE Fp8 checkpoint, we need to remove the _extra_state{gemm_idx} for gemm_idx:[1, num_gemms]
+        # for modelopt checkpoint restore
+        filtered_state_dict = {
+            k: v
+            for k, v in state_dict.items()
+            if not any(k.endswith(f"_extra_state{num}") for num in range(1, self.num_gemms))
+        }
+        return super()._load_from_state_dict(filtered_state_dict, prefix, *args, **kwargs)
+
+    def _process_quantizer_amax(self, k, v, quantizer_state_dict):
+        # Per-expert quantizers have independent checkpoint keys. Preserve their native
+        # scalar, channel, or block shape instead of flattening them through the legacy
+        # single-quantizer path.
+        if re.match(r"weight_quantizer\.\d+\..+_amax$", k):
+            quantizer_state_dict[k] = v
+        else:
+            quantizer_state_dict[k] = v.view(-1) if v.numel() == 1 else v
+
+    def _expert_parallel_groups(self):
+        """Return ``(ep_group, expt_tp_group, expt_dp_group)`` for fused experts."""
+        pg_collection = getattr(self, "_pg_collection", None)
+        if pg_collection is not None:
+            return pg_collection.ep, pg_collection.expt_tp, pg_collection.expt_dp
+        return (
+            mcore_parallel.get_expert_model_parallel_group(),
+            mcore_parallel.get_expert_tensor_parallel_group(),
+            mcore_parallel.get_expert_data_parallel_group(),
+        )
+
+    def sharded_state_dict(self, prefix="", sharded_offsets=(), metadata=None):
+        """Emit per-expert quantizer amax with the same global expert identity as the weights.
+
+        The base linear emits ``weight_quantizer.{local_i}._amax`` with the local index and no
+        expert offset, so every EP rank writes identical keys and ``torch_dist`` dedup keeps only
+        one rank's experts. Here we mirror Megatron ``TEGroupedLinear._sharded_state_dict_grouped``:
+        each fused expert comes with its ``global_expert_idx`` (baked into the key prefix under
+        ``singleton_local_shards``, otherwise an EP sharded-offset) so all ``num_global_experts``
+        persist and reshard to any EP. Shared, whole-linear quantizer buffers (e.g.
+        ``input_quantizer``) keep the plain replicated path.
+        """
+        metadata = ensure_metadata_has_dp_cp_group(metadata)
+        singleton_local_shards = bool((metadata or {}).get("singleton_local_shards", False))
+
+        # Weights/bias/_extra_state come from the wrapped TE grouped linear, which already
+        # assigns each expert its global identity. Skip _MegatronParallelLinear's local-index
+        # amax emission by starting from the base MCore module's sharded_state_dict.
+        sharded_state_dict = super(_MegatronParallelLinear, self).sharded_state_dict(
+            prefix, sharded_offsets, metadata
+        )
 
-        def _process_quantizer_amax(self, k, v, quantizer_state_dict):
-            # Per-expert quantizers have independent checkpoint keys. Preserve their native
-            # scalar, channel, or block shape instead of flattening them through the legacy
-            # single-quantizer path.
-            if re.match(r"weight_quantizer\.\d+\..+_amax$", k):
+        # Collect the quantizer buffers exactly like _MegatronParallelLinear.sharded_state_dict.
+        quantizer_state_dict = {}
+        for k, v in self.state_dict(prefix="", keep_vars=True).items():
+            if "_quantizer" in k and "_amax" in k:
+                self._process_quantizer_amax(k, v, quantizer_state_dict)
+            elif k == "input_quantizer._pre_quant_scale":
+                self._process_activation_quantizer_pre_quant_scale(k, v, quantizer_state_dict)
+            elif self._parameter_to_keep_in_quantizer_state_dict(k):
                 quantizer_state_dict[k] = v
+            elif "quantizer" in k:
+                warn_rank_0(
+                    f"Quantizer state {k} is not supported for sharded_state_dict. "
+                    "Please use regular state_dict."
+                )
+
+        # Channel shard axes (per real key); _global_amax stays un-sharded along channels but
+        # still rides with the expert identity below.
+        shard_axis_dict = self._get_shard_axis_dict(quantizer_state_dict)
+        ep_group, expt_tp_group, expt_dp_group = self._expert_parallel_groups()
+
+        # Split per-expert weight_quantizer.{i}.* from shared (input/output) quantizer buffers.
+        expert_re = re.compile(r"^weight_quantizer\.(\d+)\.(.+)$")
+        per_expert_subs = [[] for _ in range(self.num_gemms)]
+        shared_state = {}
+        for k, v in quantizer_state_dict.items():
+            m = expert_re.match(k)
+            if m:
+                per_expert_subs[int(m.group(1))].append((m.group(2), v, shard_axis_dict.get(k)))
             else:
-                quantizer_state_dict[k] = v.view(-1) if v.numel() == 1 else v
-
-        def _expert_parallel_groups(self):
-            """Return ``(ep_group, expt_tp_group, expt_dp_group)`` for fused experts."""
-            pg_collection = getattr(self, "_pg_collection", None)
-            if pg_collection is not None:
-                return pg_collection.ep, pg_collection.expt_tp, pg_collection.expt_dp
-            return (
-                mcore_parallel.get_expert_model_parallel_group(),
-                mcore_parallel.get_expert_tensor_parallel_group(),
-                mcore_parallel.get_expert_data_parallel_group(),
-            )
+                shared_state[k] = v
 
-        def sharded_state_dict(self, prefix="", sharded_offsets=(), metadata=None):
-            """Emit per-expert quantizer amax with the same global expert identity as the weights.
-
-            The base linear emits ``weight_quantizer.{local_i}._amax`` with the local index and no
-            expert offset, so every EP rank writes identical keys and ``torch_dist`` dedup keeps only
-            one rank's experts. Here we mirror Megatron ``TEGroupedLinear._sharded_state_dict_grouped``:
-            each fused expert comes with its ``global_expert_idx`` (baked into the key prefix under
-            ``singleton_local_shards``, otherwise an EP sharded-offset) so all ``num_global_experts``
-            persist and reshard to any EP. Shared, whole-linear quantizer buffers (e.g.
-            ``input_quantizer``) keep the plain replicated path.
-            """
-            metadata = ensure_metadata_has_dp_cp_group(metadata)
-            singleton_local_shards = bool((metadata or {}).get("singleton_local_shards", False))
-
-            # Weights/bias/_extra_state come from the wrapped TE grouped linear, which already
-            # assigns each expert its global identity. Skip _MegatronParallelLinear's local-index
-            # amax emission by starting from the base MCore module's sharded_state_dict.
-            sharded_state_dict = super(_MegatronParallelLinear, self).sharded_state_dict(
-                prefix, sharded_offsets, metadata
+        # Shared quantizer buffers have no expert identity in their keys or offsets. Keep the
+        # dense TP/DP defaults so replica IDs distinguish EP ranks; using expt_tp/expt_dp here
+        # would collide across EP ranks. Expert-axis sharding would require an EP-aware
+        # replica ID in addition to the expert process groups.
+        shared_axis_dict = {k: shard_axis_dict[k] for k in shared_state if k in shard_axis_dict}
+        sharded_state_dict.update(
+            make_sharded_tensors_for_checkpoint(
+                shared_state, prefix, shared_axis_dict, sharded_offsets
             )
+        )
 
-            # Collect the quantizer buffers exactly like _MegatronParallelLinear.sharded_state_dict.
-            quantizer_state_dict = {}
-            for k, v in self.state_dict(prefix="", keep_vars=True).items():
-                if "_quantizer" in k and "_amax" in k:
-                    self._process_quantizer_amax(k, v, quantizer_state_dict)
-                elif k == "input_quantizer._pre_quant_scale":
-                    self._process_activation_quantizer_pre_quant_scale(k, v, quantizer_state_dict)
-                elif self._parameter_to_keep_in_quantizer_state_dict(k):
-                    quantizer_state_dict[k] = v
-                elif "quantizer" in k:
-                    warn_rank_0(
-                        f"Quantizer state {k} is not supported for sharded_state_dict. "
-                        "Please use regular state_dict."
-                    )
-
-            # Channel shard axes (per real key); _global_amax stays un-sharded along channels but
-            # still rides with the expert identity below.
-            shard_axis_dict = self._get_shard_axis_dict(quantizer_state_dict)
-            ep_group, expt_tp_group, expt_dp_group = self._expert_parallel_groups()
-
-            # Split per-expert weight_quantizer.{i}.* from shared (input/output) quantizer buffers.
-            expert_re = re.compile(r"^weight_quantizer\.(\d+)\.(.+)$")
-            per_expert_subs = [[] for _ in range(self.num_gemms)]
-            shared_state = {}
-            for k, v in quantizer_state_dict.items():
-                m = expert_re.match(k)
-                if m:
-                    per_expert_subs[int(m.group(1))].append((m.group(2), v, shard_axis_dict.get(k)))
-                else:
-                    shared_state[k] = v
-
-            # Shared quantizer buffers have no expert identity in their keys or offsets. Keep the
-            # dense TP/DP defaults so replica IDs distinguish EP ranks; using expt_tp/expt_dp here
-            # would collide across EP ranks. Expert-axis sharding would require an EP-aware
-            # replica ID in addition to the expert process groups.
-            shared_axis_dict = {k: shard_axis_dict[k] for k in shared_state if k in shard_axis_dict}
-            sharded_state_dict.update(
-                make_sharded_tensors_for_checkpoint(
-                    shared_state, prefix, shared_axis_dict, sharded_offsets
+        # Per-expert amax: assign the same global expert identity the weights use.
+        num_global_experts = get_pg_size(ep_group) * self.num_gemms
+        local_expert_indices_offset = get_pg_rank(ep_group) * self.num_gemms
+        ep_axis = len(sharded_offsets)
+        for gemm_idx, subs in enumerate(per_expert_subs):
+            if not subs:
+                continue
+            global_expert_idx = local_expert_indices_offset + gemm_idx
+            if singleton_local_shards:
+                expert_prefix = f"{global_expert_idx}.{prefix}"
+                new_sharded_offsets = sharded_offsets
+            else:
+                expert_prefix = prefix
+                new_sharded_offsets = (
+                    *sharded_offsets,
+                    (ep_axis, global_expert_idx, num_global_experts),
                 )
+            expert_state = {f"{gemm_idx}.weight_quantizer.{sub}": v for sub, v, _ in subs}
+            expert_axis = {
+                f"{gemm_idx}.weight_quantizer.{sub}": axis
+                for sub, _, axis in subs
+                if axis is not None
+            }
+            sub_sd = make_sharded_tensors_for_checkpoint(
+                expert_state,
+                "",
+                expert_axis,
+                new_sharded_offsets,
+                tp_group=expt_tp_group,
+                dp_cp_group=expt_dp_group,
             )
+            # Rewrite each ShardedTensor.key to carry the global expert identity (dict keys,
+            # which map to the local buffers on restore, are left untouched).
+            replace_prefix_for_sharding(sub_sd, f"{gemm_idx}.", expert_prefix)
+            for sub, _, _ in subs:
+                sh_ten = sub_sd[f"{gemm_idx}.weight_quantizer.{sub}"]
+                sharded_state_dict[f"{prefix}weight_quantizer.{gemm_idx}.{sub}"] = sh_ten
+        return sharded_state_dict
 
-            # Per-expert amax: assign the same global expert identity the weights use.
-            num_global_experts = get_pg_size(ep_group) * self.num_gemms
-            local_expert_indices_offset = get_pg_rank(ep_group) * self.num_gemms
-            ep_axis = len(sharded_offsets)
-            for gemm_idx, subs in enumerate(per_expert_subs):
-                if not subs:
-                    continue
-                global_expert_idx = local_expert_indices_offset + gemm_idx
-                if singleton_local_shards:
-                    expert_prefix = f"{global_expert_idx}.{prefix}"
-                    new_sharded_offsets = sharded_offsets
-                else:
-                    expert_prefix = prefix
-                    new_sharded_offsets = (
-                        *sharded_offsets,
-                        (ep_axis, global_expert_idx, num_global_experts),
-                    )
-                expert_state = {f"{gemm_idx}.weight_quantizer.{sub}": v for sub, v, _ in subs}
-                expert_axis = {
-                    f"{gemm_idx}.weight_quantizer.{sub}": axis
-                    for sub, _, axis in subs
-                    if axis is not None
-                }
-                sub_sd = make_sharded_tensors_for_checkpoint(
-                    expert_state,
-                    "",
-                    expert_axis,
-                    new_sharded_offsets,
-                    tp_group=expt_tp_group,
-                    dp_cp_group=expt_dp_group,
-                )
-                # Rewrite each ShardedTensor.key to carry the global expert identity (dict keys,
-                # which map to the local buffers on restore, are left untouched).
-                replace_prefix_for_sharding(sub_sd, f"{gemm_idx}.", expert_prefix)
-                for sub, _, _ in subs:
-                    sh_ten = sub_sd[f"{gemm_idx}.weight_quantizer.{sub}"]
-                    sharded_state_dict[f"{prefix}weight_quantizer.{gemm_idx}.{sub}"] = sh_ten
-            return sharded_state_dict
-
-    @QuantModuleRegistry.register(
-        {TEColumnParallelGroupedLinear: "megatron_TEColumnParallelGroupedLinear"}
-    )
-    class _MegatronTEGroupedColumnParallelLinear(
-        _QuantMegatronTEGroupedLinear, _MegatronColumnParallelLinear
-    ):
-        pass
 
-    @QuantModuleRegistry.register(
-        {TERowParallelGroupedLinear: "megatron_TERowParallelGroupedLinear"}
-    )
-    class _MegatronTEGroupedRowParallelLinear(
-        _QuantMegatronTEGroupedLinear, _MegatronRowParallelLinear
-    ):
-        pass
+@QuantModuleRegistry.register(
+    {TEColumnParallelGroupedLinear: "megatron_TEColumnParallelGroupedLinear"}
+)
+class _MegatronTEGroupedColumnParallelLinear(
+    _QuantMegatronTEGroupedLinear, _MegatronColumnParallelLinear
+):
+    pass
 
-    @QuantModuleRegistry.register({megatron_moe.TEGroupedMLP: "megatron_moe_TEGroupedMLP"})
-    class _MegatronTEGroupedMLP(_MegatronMLP):
-        def _setup(self):
-            if not hasattr(self, "parallel_state") or self.parallel_state is None:
-                self.parallel_state = ParallelState(
-                    mcore_parallel.get_expert_data_parallel_group(),
-                    tensor_parallel_group=mcore_parallel.get_expert_tensor_parallel_group(),
-                    expert_model_parallel_group=mcore_parallel.get_expert_model_parallel_group(),
-                )
-            # These child linears are still native MCore modules here. Seed `_parallel_state`
-            # directly so the later QuantModule conversion sees the intended parallel state.
-            self.linear_fc1._parallel_state = self.parallel_state
-            self.linear_fc2._parallel_state = self.parallel_state
-
-    @QuantModuleRegistry.register({TEDotProductAttention: "TEDotProductAttention"})
-    class _QuantTEDotProductAttention(QuantModule):
-        """Quantized version of TEDotProductAttention for Megatron models with KV cache quantization.
-
-        This class adds KV cache quantization support to Transformer Engine's TEDotProductAttention
-        module used in Megatron-Core models. It introduces three quantizers (q_bmm_quantizer,
-        k_bmm_quantizer, v_bmm_quantizer) that quantize the query, key, and value tensors after
-        RoPE has been applied.
-        """
 
-        def _setup(self):
-            """Initialize quantizers for Q, K, V tensors."""
-            self.q_bmm_quantizer = TensorQuantizer()
-            self.k_bmm_quantizer = TensorQuantizer()
-            self.v_bmm_quantizer = TensorQuantizer()
+@QuantModuleRegistry.register({TERowParallelGroupedLinear: "megatron_TERowParallelGroupedLinear"})
+class _MegatronTEGroupedRowParallelLinear(
+    _QuantMegatronTEGroupedLinear, _MegatronRowParallelLinear
+):
+    pass
 
-            # Set parallel_state for distributed sync of BMM quantizers
-            try:
-                data_parallel_group = get_data_parallel_group(with_context_parallel=True)
-            except AssertionError:
-                data_parallel_group = get_data_parallel_group()
+
+@QuantModuleRegistry.register({megatron_moe.TEGroupedMLP: "megatron_moe_TEGroupedMLP"})
+class _MegatronTEGroupedMLP(_MegatronMLP):
+    def _setup(self):
+        if not hasattr(self, "parallel_state") or self.parallel_state is None:
             self.parallel_state = ParallelState(
-                data_parallel_group,
-                mcore_parallel.get_tensor_model_parallel_group(),
+                mcore_parallel.get_expert_data_parallel_group(),
+                tensor_parallel_group=mcore_parallel.get_expert_tensor_parallel_group(),
+                expert_model_parallel_group=mcore_parallel.get_expert_model_parallel_group(),
             )
+        # These child linears are still native MCore modules here. Seed `_parallel_state`
+        # directly so the later QuantModule conversion sees the intended parallel state.
+        self.linear_fc1._parallel_state = self.parallel_state
+        self.linear_fc2._parallel_state = self.parallel_state
+
+
+@QuantModuleRegistry.register({TEDotProductAttention: "TEDotProductAttention"})
+class _QuantCoreAttention(QuantModule):
+    """Core attention (TEDotProductAttention, DSAttention) with KV cache quantization.
+
+    Adds q/k/v_bmm_quantizers that quantize the post-RoPE query, key and value tensors.
+    """
 
-        def forward(self, query, key, value, *args, **kwargs):
-            """Apply post-RoPE quantization to KV cache."""
-            # Quantize Q, K, V
-            query = self.q_bmm_quantizer(query)
-            key = self.k_bmm_quantizer(key)
+    def _setup(self):
+        """Initialize quantizers for Q, K, V tensors."""
+        self.q_bmm_quantizer = TensorQuantizer()
+        self.k_bmm_quantizer = TensorQuantizer()
+        self.v_bmm_quantizer = TensorQuantizer()
+
+        # Set parallel_state for distributed sync of BMM quantizers
+        try:
+            data_parallel_group = get_data_parallel_group(with_context_parallel=True)
+        except AssertionError:
+            data_parallel_group = get_data_parallel_group()
+        self.parallel_state = ParallelState(
+            data_parallel_group,
+            mcore_parallel.get_tensor_model_parallel_group(),
+        )
+
+    def forward(self, query, key, value, *args, **kwargs):
+        """Apply post-RoPE quantization to KV cache."""
+        # Quantize Q, K, V
+        query = self.q_bmm_quantizer(query)
+        if value is None:
+            # Absorbed MLA (DSAttention) passes value=None: the key is the KV latent that both K
+            # and V are read from, so calibrate V on it too (output unused) to export a V scale.
+            if self.v_bmm_quantizer._if_calib:
+                self.v_bmm_quantizer(key)
+        else:
             value = self.v_bmm_quantizer(value)
-            return super().forward(query, key, value, *args, **kwargs)
-
-        def modelopt_post_restore(self, name=""):
-            """Restore quantizer states after model loading."""
-            for tq in [self.q_bmm_quantizer, self.k_bmm_quantizer, self.v_bmm_quantizer]:
-                # TODO: Add support for non-scalar states such as
-                # Affine KVCache  bias vector which is per head per channel
-                if not all(v.numel() == 1 for v in tq.state_dict().values()):
-                    raise NotImplementedError(
-                        "Only scalar states are supported for KV Cache/BMM Quantizers"
-                    )
-            # dtype and device should have been set in `megatron_replace_quant_module_hook`
-            # via `_configure_attention_for_kv_cache_quant`
-            assert hasattr(self, "device") and hasattr(self, "dtype")
-            self.to(device=self.device, dtype=self.dtype)
-
-        def sharded_state_dict(self, prefix="", sharded_offsets=(), metadata=None):
-            # Currently we do not need sharded_state_dict for TEDotProductAttention since the amax are scalar values.
-            # However we would need this in future to support non-scalar states such as
-            # Affine KVCache Quant bias vector.
-            state_dict = self.state_dict(prefix="", keep_vars=True)
-            return make_sharded_tensors_for_checkpoint(state_dict, prefix, {}, sharded_offsets)
+        key = self.k_bmm_quantizer(key)
+        return super().forward(query, key, value, *args, **kwargs)
+
+    def modelopt_post_restore(self, name=""):
+        """Restore quantizer states after model loading."""
+        for tq in [self.q_bmm_quantizer, self.k_bmm_quantizer, self.v_bmm_quantizer]:
+            # TODO: Add support for non-scalar states such as
+            # Affine KVCache  bias vector which is per head per channel
+            if not all(v.numel() == 1 for v in tq.state_dict().values()):
+                raise NotImplementedError(
+                    "Only scalar states are supported for KV Cache/BMM Quantizers"
+                )
+        # dtype and device should have been set in `megatron_replace_quant_module_hook`
+        # via `_configure_attention_for_kv_cache_quant`
+        assert hasattr(self, "device") and hasattr(self, "dtype")
+        self.to(device=self.device, dtype=self.dtype)
+
+    def sharded_state_dict(self, prefix="", sharded_offsets=(), metadata=None):
+        # Currently we do not need sharded_state_dict for core attention since the amax are scalar values.
+        # However we would need this in future to support non-scalar states such as
+        # Affine KVCache Quant bias vector.
+        state_dict = self.state_dict(prefix="", keep_vars=True)
+        return make_sharded_tensors_for_checkpoint(state_dict, prefix, {}, sharded_offsets)
+
+
+if HAS_DSA:
+
+    @QuantModuleRegistry.register({DSAttention: "megatron_DSAttention"})
+    class _QuantDSAttention(_QuantCoreAttention):
+        """DSAttention with KV-cache quantization.
+
+        torch's state_dict() / load_state_dict() route ``_extra_state`` only through classes that
+        override these; TEDotProductAttention does, DSAttention does not, so without them the
+        quantizer state (including amax) was dropped from checkpoints. They exist for that class
+        check: at runtime the instance-level modelopt callbacks shadow them, so don't remove them.
+        """
+
+        def get_extra_state(self):
+            return quant_module_get_extra_state(self)
+
+        def set_extra_state(self, state):
+            quant_module_set_extra_state(self, state)
 
 
 if HAS_GDN:
@@ -1093,6 +1117,14 @@ class _QuantGatedDeltaNet(GatedDeltaNetStateQuantMixin):
         paths (``ssm_prefill`` / ``ssm_decode``) are left untouched.
         """
 
+        # Class-level overrides so torch routes the quantizer state through ``_extra_state``
+        # (GatedDeltaNet has none); see _QuantDSAttention.
+        def get_extra_state(self):
+            return quant_module_get_extra_state(self)
+
+        def set_extra_state(self, state):
+            quant_module_set_extra_state(self, state)
+
         def _setup(self):
             super()._setup()
             try:
diff --git a/tests/gpu_megatron/torch/export/plugins/test_mcore_export_mappings.py b/tests/gpu_megatron/torch/export/plugins/test_mcore_export_mappings.py
index 9debbc2231d..8b048e87bcd 100644
--- a/tests/gpu_megatron/torch/export/plugins/test_mcore_export_mappings.py
+++ b/tests/gpu_megatron/torch/export/plugins/test_mcore_export_mappings.py
@@ -34,6 +34,7 @@
     "Qwen3VLForConditionalGeneration": "model.language_model.layers.{}.self_attn.",
     "Qwen3_5ForConditionalGeneration": "model.language_model.layers.{}.self_attn.",
     "Qwen3_5MoeForConditionalGeneration": "model.language_model.layers.{}.self_attn.",
+    "GlmMoeDsaForCausalLM": "model.layers.{}.self_attn.",
 }
 
 
@@ -54,6 +55,7 @@ def test_export_mapping_emits_kv_cache_scales(arch, prefix):
     "Qwen3_5MoeForConditionalGeneration": "model.language_model.layers.{}.mlp.experts.{}",
     "Qwen3MoeForCausalLM": "model.layers.{}.mlp.experts.{}",
     "NemotronHForCausalLM": "backbone.layers.{}.mixer.experts.{}",
+    "GlmMoeDsaForCausalLM": "model.layers.{}.mlp.experts.{}",
 }
 
 
diff --git a/tests/gpu_megatron/torch/export/test_unified_export_megatron.py b/tests/gpu_megatron/torch/export/test_unified_export_megatron.py
index 480be83018a..1060e1d6ea1 100644
--- a/tests/gpu_megatron/torch/export/test_unified_export_megatron.py
+++ b/tests/gpu_megatron/torch/export/test_unified_export_megatron.py
@@ -47,6 +47,7 @@
 import modelopt.torch.quantization.ggml as ggml
 import modelopt.torch.speculative as mtsp
 from modelopt.torch.export import KV_CACHE_FP8, export_mcore_gpt_to_hf, import_mcore_gpt_from_hf
+from modelopt.torch.export.plugins.mcore_common import all_mcore_hf_export_mapping
 from modelopt.torch.export.quant_format import IQ_FORMATS
 from modelopt.torch.export.unified_export_megatron import GPTModelExporter
 from modelopt.torch.quantization.config import QuantizerAttributeConfig
@@ -976,6 +977,7 @@ def _make_exporter_for_mtp(model_dir: Path) -> GPTModelExporter:
     exporter._hf_pretrained_model_name = str(model_dir)
     exporter._state_dict = {}  # MTP keys are absent — they should be picked up
     exporter.exclude_modules = []
+    exporter.rules = {}  # a Qwen-style ``mtp.*`` checkpoint, not MTP stored as decoder layers
     return exporter
 
 
@@ -1052,6 +1054,104 @@ def test_mtp_state_dict_index_file(tmp_path):
     assert "mtp*" in exporter.exclude_modules
 
 
+def _test_live_decoder_mtp_export_rejected(model_dir, rank, size):
+    model = get_mcore_gpt_model(
+        tensor_model_parallel_size=size,
+        pipeline_model_parallel_size=1,
+        initialize_megatron=True,
+        num_layers=2,
+        hidden_size=64,
+        num_attention_heads=4,
+        vocab_size=128,
+        max_sequence_length=32,
+        mtp_num_layers=1,
+    ).cuda()
+    with pytest.raises(NotImplementedError, match="Megatron-built MTP"):
+        GPTModelExporter(model, str(model_dir))
+
+
+def test_live_decoder_mtp_export_rejected(dist_workers_size_1, tmp_path):
+    """GLM-5 MTP is only copied from the source; a Megatron-built one must not export misnamed."""
+    transformers.GlmMoeDsaConfig(
+        num_hidden_layers=2, architectures=["GlmMoeDsaForCausalLM"]
+    ).save_pretrained(tmp_path)
+    dist_workers_size_1.run(partial(_test_live_decoder_mtp_export_rejected, tmp_path))
+
+
+def test_mtp_state_dict_copies_decoder_mtp_layers(tmp_path):
+    """GLM-5 keeps MTP as an extra decoder layer; copy it dequantized when Megatron did not build it."""
+    model_dir = tmp_path / "fake_glm5"
+    model_dir.mkdir()
+    fp8 = torch.full((128, 128), 2.0).to(torch.float8_e4m3fn)
+    save_file(
+        {
+            "model.layers.1.input_layernorm.weight": torch.ones(8),  # pruned decoder layer
+            "model.layers.2.enorm.weight": torch.full((8,), 3.0),  # MTP layer of the source
+            "model.layers.2.eh_proj.weight": fp8,
+            "model.layers.2.eh_proj.weight_scale_inv": torch.full((1, 1), 0.5),
+        },
+        str(model_dir / "model.safetensors"),
+    )
+    exporter = _make_exporter_for_mtp(model_dir)
+    exporter.rules = {"mtp_in_decoder_layers": True}
+    exporter.all_mcore_mappings = all_mcore_hf_export_mapping["GlmMoeDsaForCausalLM"]
+    # Depth-pruned from 2 to 1 decoder layers: the source MTP (layer 2) lands at layer 1.
+    exporter._src_num_hidden_layers = 2
+    exporter._hf_text_config = SimpleNamespace(num_hidden_layers=1, num_nextn_predict_layers=1)
+
+    mtp_state_dict = exporter._get_mtp_state_dict()
+
+    assert sorted(mtp_state_dict) == [
+        "model.layers.1.eh_proj.weight",
+        "model.layers.1.enorm.weight",
+    ]
+    assert mtp_state_dict["model.layers.1.eh_proj.weight"].dtype == torch.bfloat16
+    torch.testing.assert_close(
+        mtp_state_dict["model.layers.1.eh_proj.weight"].float(), torch.ones(128, 128)
+    )
+    assert exporter.exclude_modules == ["model.layers.1.*"]
+
+
+def test_mtp_state_dict_copies_decoder_mtp_layers_from_hub(tmp_path, monkeypatch):
+    """A Hub-ID source downloads only the shards holding the MTP layer, then copies it."""
+    hub = tmp_path / "hub"
+    hub.mkdir()
+    shards = {
+        "model.layers.0.input_layernorm.weight": "model-00001-of-00002.safetensors",
+        "model.layers.1.enorm.weight": "model-00002-of-00002.safetensors",
+    }
+    save_file(
+        {"model.layers.0.input_layernorm.weight": torch.ones(8)},
+        str(hub / shards["model.layers.0.input_layernorm.weight"]),
+    )
+    save_file(
+        {"model.layers.1.enorm.weight": torch.full((8,), 3.0)},
+        str(hub / shards["model.layers.1.enorm.weight"]),
+    )
+    (hub / "model.safetensors.index.json").write_text(json.dumps({"weight_map": shards}))
+    requested = {}
+
+    def fake_snapshot_download(repo_id, allow_patterns):
+        requested[repo_id] = allow_patterns
+        return str(hub)
+
+    monkeypatch.setattr(uem, "hf_hub_download", lambda repo_id, filename: str(hub / filename))
+    monkeypatch.setattr(uem, "snapshot_download", fake_snapshot_download)
+    exporter = _make_exporter_for_mtp(Path("zai-org/GLM-5.2"))
+    exporter.rules = {"mtp_in_decoder_layers": True}
+    exporter.all_mcore_mappings = all_mcore_hf_export_mapping["GlmMoeDsaForCausalLM"]
+    exporter._src_num_hidden_layers = 1
+    exporter._hf_text_config = SimpleNamespace(num_hidden_layers=1, num_nextn_predict_layers=1)
+
+    mtp_state_dict = exporter._get_mtp_state_dict()
+
+    assert requested == {
+        "zai-org/GLM-5.2": ["model.safetensors.index.json", "model-00002-of-00002.safetensors"]
+    }
+    assert list(mtp_state_dict) == ["model.layers.1.enorm.weight"]
+    assert exporter.exclude_modules == ["model.layers.1.*"]
+
+
 class _FakeTEGroupedMLP:
     """Minimal TEGroupedMLP stand-in exposing num_gemms, weight{i}, and state_dict()."""
 
@@ -1260,12 +1360,39 @@ def test_is_sidecar_writer_rank_pins_to_dp0_ep0(monkeypatch):
     assert GPTModelExporter._is_sidecar_writer_rank(True) is False
 
 
-def _make_exporter_for_key_check(num_layers: int) -> GPTModelExporter:
+def _make_exporter_for_key_check(
+    num_layers: int, src_num_layers: int | None = None, num_mtp: int = 0
+) -> GPTModelExporter:
+    """``num_layers`` is the exported HF decoder depth; ``num_mtp`` MTP layers follow it."""
     exporter = object.__new__(GPTModelExporter)
-    exporter.model = SimpleNamespace(config=SimpleNamespace(num_layers=num_layers))
+    exporter._hf_text_config = SimpleNamespace(
+        num_hidden_layers=num_layers, num_nextn_predict_layers=num_mtp
+    )
+    exporter._src_num_hidden_layers = num_layers if src_num_layers is None else src_num_layers
+    exporter.rules = {"mtp_in_decoder_layers": num_mtp > 0}
     return exporter
 
 
+@pytest.mark.parametrize(("export_mtp", "raises"), [(True, False), (False, True)])
+def test_verify_exported_keys_depth_pruned_with_decoder_mtp(tmp_path, export_mtp, raises):
+    """Pruned 3 -> 2 layers: source layer 2 is not required, and its MTP (layer 3) must land at 2."""
+    source, export = tmp_path / "src", tmp_path / "exp"
+    _write_index(
+        source,
+        [f"model.layers.{i}.input_layernorm.weight" for i in range(3)]
+        + ["model.layers.3.eh_proj.weight"],
+    )
+    exported = [f"model.layers.{i}.input_layernorm.weight" for i in range(2)]
+    _write_index(export, exported + (["model.layers.2.eh_proj.weight"] if export_mtp else []))
+    exporter = _make_exporter_for_key_check(num_layers=2, src_num_layers=3, num_mtp=1)
+
+    if raises:
+        with pytest.raises(RuntimeError, match=r"model\.layers\.2\.eh_proj\.weight"):
+            exporter._verify_exported_keys(str(export), str(source))
+    else:
+        exporter._verify_exported_keys(str(export), str(source))
+
+
 def _write_index(dir_path: Path, keys) -> None:
     dir_path.mkdir(parents=True, exist_ok=True)
     (dir_path / "model.safetensors.index.json").write_text(
diff --git a/tests/gpu_megatron/torch/quantization/plugins/test_megatron.py b/tests/gpu_megatron/torch/quantization/plugins/test_megatron.py
index 9a9d0e61977..e2702f7cf30 100644
--- a/tests/gpu_megatron/torch/quantization/plugins/test_megatron.py
+++ b/tests/gpu_megatron/torch/quantization/plugins/test_megatron.py
@@ -60,12 +60,14 @@
 )
 from megatron.core.tensor_parallel.layers import ColumnParallelLinear, RowParallelLinear
 from megatron.core.transformer import MegatronModule, TransformerConfig
+from megatron.core.transformer.mlp import MLP
 from megatron.core.transformer.moe.experts import SequentialMLP, TEGroupedMLP
 from megatron.core.transformer.moe.router import TopKRouter
 
 import modelopt
 import modelopt.torch.opt as mto
 import modelopt.torch.quantization as mtq
+from modelopt.torch.opt.dynamic import DynamicModule
 from modelopt.torch.opt.plugins.mcore_dist_checkpointing import (
     restore_sharded_modelopt_state,
     save_sharded_modelopt_state,
@@ -2076,6 +2078,117 @@ def test_csa_indexer_quant(dist_workers_size_1, tmp_path):
     dist_workers_size_1.run(partial(_test_csa_indexer_quant_helper, tmp_path))
 
 
+def test_registered_megatron_quant_modules_checkpoint_quantizer_state():
+    """Quantizer state rides in ``_extra_state``, which torch saves only for classes overriding it.
+
+    ``register_modelopt_extra_state_callbacks`` binds the hooks per instance, so a registered class
+    without a class-level ``get/set_extra_state`` silently drops its amax from checkpoints (as
+    DSAttention did). Containers whose child linears hold all quantizers are exempt.
+    """
+    containers = {MLP, SequentialMLP, TEGroupedMLP}
+    base = torch.nn.Module
+    missing = []
+    for cls, quant_cls in QuantModuleRegistry._registry.items():
+        if not issubclass(cls, MegatronModule) or cls in containers:
+            continue
+        if issubclass(cls, DynamicModule):
+            continue  # e.g. LoRA wrappers: the runtime class also includes the wrapped Megatron layer
+        routes = all(
+            any(getattr(c, attr) is not getattr(base, attr) for c in (quant_cls, cls))
+            for attr in ("get_extra_state", "set_extra_state")
+        )
+        if not routes:
+            missing.append(cls.__name__)
+    assert not missing, f"quantizer state would be dropped from checkpoints for: {missing}"
+
+
+def _get_tiny_dsa_gpt_model():
+    """Tiny GPT with DSA sparse attention (AbsorbedMLASelfAttention + DSAttention + indexer)."""
+    # Local: DSA is optional; test_dsa_kv_cache_quant importorskips it before calling this.
+    from megatron.core.models.gpt.experimental_attention_variant_module_specs import (
+        get_transformer_block_with_experimental_attention_variant_spec,
+    )
+    from megatron.core.transformer.transformer_config import MLATransformerConfig
+
+    config = MLATransformerConfig(
+        num_layers=2,
+        hidden_size=64,
+        num_attention_heads=4,
+        ffn_hidden_size=128,
+        q_lora_rank=32,
+        kv_lora_rank=16,
+        qk_head_dim=16,
+        qk_pos_emb_head_dim=16,
+        v_head_dim=16,
+        experimental_attention_variant="dsa",
+        dsa_indexer_n_heads=2,
+        dsa_indexer_head_dim=32,
+        dsa_indexer_topk=8,
+        normalization="RMSNorm",
+        add_bias_linear=False,
+        hidden_dropout=0.0,  # deterministic forward for the checkpoint round-trip comparison
+        attention_dropout=0.0,
+        bf16=True,
+        params_dtype=torch.bfloat16,
+    )
+    spec = get_transformer_block_with_experimental_attention_variant_spec(config)
+    return GPTModel(
+        config=config,
+        transformer_layer_spec=spec,
+        vocab_size=64,
+        max_sequence_length=64,
+        position_embedding_type="rope",
+        # Tied so a checkpoint round-trip restores the output layer too.
+        share_embeddings_and_output_weights=True,
+    ).cuda()
+
+
+def _test_dsa_kv_cache_quant_helper(tmp_path, rank, size):
+    # Local: DSA is optional; test_dsa_kv_cache_quant importorskips it before calling this.
+    from megatron.core.transformer.experimental_attention_variant.dsa import DSAttention
+
+    initialize_for_megatron(tensor_model_parallel_size=1, pipeline_model_parallel_size=1, seed=SEED)
+    model, model_test = _get_tiny_dsa_gpt_model(), _get_tiny_dsa_gpt_model()
+    forward = get_forward(model)
+    # Calibrated (not constant-amax) FP8 KV cache: the case that needs a real V amax to export.
+    kv_fp8_calibrated = {
+        "quant_cfg": [
+            {"quantizer_name": "*", "enable": False},
+            {"quantizer_name": "*[kv]_bmm_quantizer", "cfg": {"num_bits": (4, 3), "axis": None}},
+        ],
+        "algorithm": "max",
+    }
+    model = mtq.quantize(model, kv_fp8_calibrated, forward)
+
+    dsa_modules = [m for m in model.modules() if isinstance(m, DSAttention)]
+    assert dsa_modules, "DSAttention was not converted for KV-cache quantization"
+    for module in dsa_modules:
+        assert module.k_bmm_quantizer.is_enabled and module.v_bmm_quantizer.is_enabled
+        # Absorbed MLA passes value=None; V is calibrated on the shared KV latent like K, so a
+        # calibrated FP8 KV cache still exports a v_scale.
+        assert module.k_bmm_quantizer.amax is not None
+        assert torch.equal(module.v_bmm_quantizer.amax, module.k_bmm_quantizer.amax)
+
+    # The KV quantizer state (incl. amax) survives a torch-dist round-trip alongside the indexer.
+    # DSA trains its indexer through a separate indexer loss, so the LM-loss backward in the helper
+    # gives it no gradient; exclude it from that check.
+    for name, param in model_test.named_parameters():
+        if ".indexer." in name:
+            param.requires_grad_(False)
+    sharded_state_dict_test_helper(tmp_path, model, model_test, forward)
+    # The helper compares every restored tensor (incl. K/V amax), but V is unused in the forward
+    # pass, so also check both quantizers came back enabled.
+    restored = [m for m in model_test.modules() if isinstance(m, DSAttention)]
+    assert len(restored) == len(dsa_modules)
+    assert all(m.k_bmm_quantizer.is_enabled and m.v_bmm_quantizer.is_enabled for m in restored)
+
+
+def test_dsa_kv_cache_quant(dist_workers_size_1, tmp_path):
+    """DSAttention gets calibrated K/V KV-cache quantizers that survive a checkpoint round-trip."""
+    pytest.importorskip("megatron.core.transformer.experimental_attention_variant.dsa")
+    dist_workers_size_1.run(partial(_test_dsa_kv_cache_quant_helper, tmp_path))
+
+
 def _test_kv_cache_amax_sync_helper(config, rank, size, tensor_model_parallel_size=1):
     """Helper function for testing KV cache quantizer amax sync across distributed world."""
     # Use rank in seed to produce different amax values across ranks