diff --git a/mamba_builders.py b/mamba_builders.py
index 6a792ba6ea5..90794f067b7 100644
--- a/mamba_builders.py
+++ b/mamba_builders.py
@@ -39,6 +39,7 @@ def mamba_builder(args, pre_process, post_process, vp_stage=None, config=None, p
         rotary_percent=args.rotary_percent,
         rotary_base=args.rotary_base,
         pg_collection=pg_collection,
+        vp_stage=vp_stage,
     )
 
     for l in range(model.decoder.num_layers_per_pipeline_rank):
diff --git a/megatron/core/models/common/language_module/language_module.py b/megatron/core/models/common/language_module/language_module.py
index 259bb716a93..296af54d6ce 100644
--- a/megatron/core/models/common/language_module/language_module.py
+++ b/megatron/core/models/common/language_module/language_module.py
@@ -23,6 +23,7 @@
 from megatron.core.process_groups_config import ProcessGroupCollection
 from megatron.core.transformer.enums import AttnBackend, CudaGraphScope
 from megatron.core.transformer.module import MegatronModule
+from megatron.core.transformer.multi_token_prediction import tie_word_embeddings_state_dict
 from megatron.core.transformer.transformer_config import TransformerConfig
 from megatron.core.transformer.utils import ensure_metadata_has_dp_cp_group
 from megatron.core.utils import (
@@ -255,12 +256,20 @@ def setup_embeddings_and_output_layer(self) -> None:
             LanguageModule.embedding_warning_printed = True
 
     def shared_embedding_or_output_weight(self) -> Tensor:
-        """Gets the emedding weight or output logit weights when share embedding and output weights set to True.
+        """Gets the embedding weight or output logit weights when share embedding and output weights set to True
+          or when use Multi-Token Prediction (MTP).
 
         Returns:
-            Tensor: During pre processing it returns the input embeddings weight while during post processing it returns the final output layers weight
+            Tensor: During pre processing or MTP process it returns the input embeddings weight while during post processing it returns the final output layers weight
         """
-        if self.pre_process:
+        if self.pre_process or getattr(self, 'mtp_process', False):
+            # Multi-Token Prediction (MTP) need both embedding layer and output layer.
+            # So there will be both embedding layer and output layer in the mtp process stage.
+            # When share_embeddings_and_output_weights is True, the embedding weight is the
+            # canonical shared weight and is passed to the output layer during forward.
+            assert hasattr(
+                self, 'embedding'
+            ), f"embedding is needed in this pipeline stage, but it is not initialized."
             return self.embedding.word_embeddings.weight
         elif self.post_process:
             return self.output_layer.weight
@@ -293,6 +302,21 @@ def sharded_state_dict(
         output_layer_weight_key = f'{prefix}output_layer.weight'
         output_layer_bias_key = f'{prefix}output_layer.bias'
 
+        # Multi-Token Prediction (MTP) needs embedding layer in mtp process stage.
+        # If MTP is not placed in the pre processing stage, we need to maintain a copy of
+        # embedding layer in the mtp process stage and tie it to the embedding in the pre
+        # processing stage.
+        # Note: MTP loss is computed at post_process stage, so the output_layer on mtp_process
+        # rank doesn't need special tying - it's not used for loss computation.
+        if getattr(self, 'mtp_process', False) and not self.pre_process:
+            emb_weight = self.embedding.word_embeddings.weight
+            tie_word_embeddings_state_dict(
+                sharded_state_dict,
+                emb_weight,
+                first_stage_word_emb_key,
+                tp_group=self.tp_group,
+                dp_cp_group=metadata['dp_cp_group'],
+            )
         if self.share_embeddings_and_output_weights:
             self.tie_embeddings_and_output_weights_state_dict(
                 sharded_state_dict, output_layer_weight_key, first_stage_word_emb_key, metadata
diff --git a/megatron/core/models/common/model_chunk_schedule_plan.py b/megatron/core/models/common/model_chunk_schedule_plan.py
index f451942ffc2..2e26e5fd1d3 100644
--- a/megatron/core/models/common/model_chunk_schedule_plan.py
+++ b/megatron/core/models/common/model_chunk_schedule_plan.py
@@ -123,7 +123,7 @@ def _build_callable_nodes(self, event, comp_stream, comm_stream, extra_args):
 
         # get flags for latter use
         is_mtp = isinstance(self.layer, MultiTokenPredictionLayer)
-        transformer_layer = self.layer.transformer_layer if is_mtp else self.layer
+        transformer_layer = self.layer.mtp_model_layer if is_mtp else self.layer
         is_moe = isinstance(transformer_layer.mlp, MoELayer)
         num_local_experts = transformer_layer.mlp.num_local_experts if is_moe else None
 
diff --git a/megatron/core/models/gpt/fine_grained_callables.py b/megatron/core/models/gpt/fine_grained_callables.py
index e77cfb71871..db5eee964ef 100644
--- a/megatron/core/models/gpt/fine_grained_callables.py
+++ b/megatron/core/models/gpt/fine_grained_callables.py
@@ -638,9 +638,9 @@ def build_mtp_layer_callables(layer):
     multi-token prediction layer nodes (attention, MLP, etc.)
     """
 
-    forward_funcs, backward_dw = build_transformer_layer_callables(layer.transformer_layer)
+    forward_funcs, backward_dw = build_transformer_layer_callables(layer.mtp_model_layer)
     attn_forward, dispatch_forward, mlp_forward, combine_forward, _ = forward_funcs
-    is_moe = isinstance(layer.transformer_layer.mlp, MoELayer)
+    is_moe = isinstance(layer.mtp_model_layer.mlp, MoELayer)
     assert is_moe, "MTP layer in a2a overlap only supports MoE layer for now."
 
     def submodule_mtp_attn_forward(node, hidden_states):
diff --git a/megatron/core/models/gpt/gpt_layer_specs.py b/megatron/core/models/gpt/gpt_layer_specs.py
index 581e3fb9207..3635b4e5fb4 100755
--- a/megatron/core/models/gpt/gpt_layer_specs.py
+++ b/megatron/core/models/gpt/gpt_layer_specs.py
@@ -753,7 +753,7 @@ def get_gpt_mtp_block_spec_for_backend(
         raise ValueError(f"Invalid spec: {spec}")
 
     mtp_layer_spec = get_mtp_layer_spec_for_backend(
-        transformer_layer_spec=transformer_layer_spec, backend=backend
+        mtp_model_layer_spec=transformer_layer_spec, backend=backend
     )
     mtp_num_layers = config.mtp_num_layers if config.mtp_num_layers else 0
     mtp_layer_specs = [mtp_layer_spec] * mtp_num_layers
diff --git a/megatron/core/models/gpt/gpt_model.py b/megatron/core/models/gpt/gpt_model.py
index a83caf998c5..8739b7b56fe 100644
--- a/megatron/core/models/gpt/gpt_model.py
+++ b/megatron/core/models/gpt/gpt_model.py
@@ -8,7 +8,7 @@
 import torch
 from torch import Tensor
 
-from megatron.core import parallel_state, tensor_parallel
+from megatron.core import tensor_parallel
 from megatron.core.config_logger import has_config_logger_enabled, log_config_to_disk
 from megatron.core.dist_checkpointing.mapping import ShardedStateDict
 from megatron.core.inference.contexts import BaseInferenceContext
@@ -29,11 +29,9 @@
 from megatron.core.transformer.enums import CudaGraphScope, ModelType
 from megatron.core.transformer.linear_cross_entropy import LinearCrossEntropyModule
 from megatron.core.transformer.multi_token_prediction import (
-    MTPLossAutoScaler,
-    MTPLossLoggingHelper,
     MultiTokenPredictionBlock,
-    roll_tensor,
-    tie_word_embeddings_state_dict,
+    mtp_on_this_rank,
+    process_mtp_loss,
 )
 from megatron.core.transformer.spec_utils import ModuleSpec
 from megatron.core.transformer.transformer_block import TransformerBlock
@@ -209,7 +207,9 @@ def __init__(
             self.rotary_base = rotary_base
         self.rotary_scaling = rope_scaling
         self.mtp_block_spec = mtp_block_spec
-        self.mtp_process = mtp_block_spec is not None
+        self.mtp_process = mtp_block_spec is not None and mtp_on_this_rank(
+            self.config, ignore_virtual=False, vp_stage=vp_stage
+        )
 
         self.fuse_linear_cross_entropy = (
             self.config.cross_entropy_loss_fusion
@@ -855,27 +855,6 @@ def _postprocess(
 
         return loss
 
-    def shared_embedding_or_output_weight(self) -> Tensor:
-        """Gets the embedding weight or output logit weights when share input embedding and
-        output weights set to True or when use Multi-Token Prediction (MTP) feature.
-
-        Returns:
-            Tensor: During pre processing or MTP process it returns the input embeddings weight.
-            Otherwise, during post processing it returns the final output layers weight.
-        """
-        if self.pre_process or self.mtp_process:
-            # Multi-Token Prediction (MTP) need both embedding layer and output layer.
-            # So there will be both embedding layer and output layer in the mtp process stage.
-            # In this case, if share_embeddings_and_output_weights is True, the shared weights
-            # will be stored in embedding layer, and output layer will not have any weight.
-            assert hasattr(
-                self, 'embedding'
-            ), f"embedding is needed in this pipeline stage, but it is not initialized."
-            return self.embedding.word_embeddings.weight
-        elif self.post_process:
-            return self.output_layer.weight
-        return None
-
     def build_schedule_plan(
         self,
         input_ids: Tensor,
@@ -966,20 +945,4 @@ def sharded_state_dict(
             output_extra_state and output_extra_state.data
         ), f'Expected output layer extra state to be empty, got: {output_extra_state}'
 
-        # Multi-Token Prediction (MTP) need embedding layer in mtp process stage.
-        # If MTP is not placed in the pre processing stage, we need to maintain a copy of
-        # embedding layer in the mtp process stage and tie it to the embedding in the pre
-        # processing stage.
-        # Now MTP loss is computed in post processing stage, so the output_layer is not needed.
-        if self.mtp_process and not self.pre_process:
-            emb_weight_key = f'{prefix}embedding.word_embeddings.weight'
-            emb_weight = self.embedding.word_embeddings.weight
-            tie_word_embeddings_state_dict(
-                sharded_state_dict,
-                emb_weight,
-                emb_weight_key,
-                tp_group=self.tp_group,
-                dp_cp_group=metadata['dp_cp_group'],
-            )
-
         return sharded_state_dict
diff --git a/megatron/core/models/mamba/mamba_layer_specs.py b/megatron/core/models/mamba/mamba_layer_specs.py
index b87124bab1d..6ca628475be 100755
--- a/megatron/core/models/mamba/mamba_layer_specs.py
+++ b/megatron/core/models/mamba/mamba_layer_specs.py
@@ -1,6 +1,7 @@
 # Copyright (c) 2023, NVIDIA CORPORATION. All rights reserved.
 
 from megatron.core.extensions.transformer_engine import (
+    TEColumnParallelLinear,
     TEDotProductAttention,
     TELayerNormColumnParallelLinear,
     TENorm,
@@ -19,6 +20,12 @@
 from megatron.core.transformer.attention import SelfAttention, SelfAttentionSubmodules
 from megatron.core.transformer.enums import AttnMaskType
 from megatron.core.transformer.mlp import MLP, MLPSubmodules
+from megatron.core.transformer.multi_token_prediction import (
+    MultiTokenPredictionBlock,
+    MultiTokenPredictionBlockSubmodules,
+    MultiTokenPredictionLayer,
+    MultiTokenPredictionLayerSubmodules,
+)
 from megatron.core.transformer.spec_utils import ModuleSpec
 from megatron.core.transformer.transformer_layer import (
     MoETransformerLayer,
@@ -26,6 +33,7 @@
     TransformerLayerSubmodules,
 )
 
+# This should be private and should not be used outside of this file.
 moe = get_moe_module_spec(
     use_te=True,
     num_experts=8,  # Can be any positive integer (must not be None).
@@ -33,6 +41,28 @@
     moe_use_legacy_grouped_gemm=False,
 )
 
+
+# MTP block spec for Mamba - provides norms and projection only.
+# Inner layers are built by MultiTokenPredictionLayer using nested MambaStack
+_mamba_mtp_block_spec = ModuleSpec(
+    module=MultiTokenPredictionBlock,
+    submodules=MultiTokenPredictionBlockSubmodules(
+        layer_specs=[
+            ModuleSpec(
+                module=MultiTokenPredictionLayer,
+                submodules=MultiTokenPredictionLayerSubmodules(
+                    enorm=TENorm,
+                    hnorm=TENorm,
+                    eh_proj=TEColumnParallelLinear,
+                    mtp_model_layer=None,  # Built via pattern + mamba_submodules
+                    layer_norm=TENorm,
+                ),
+            )
+        ]
+    ),
+)
+
+
 mamba_stack_spec = ModuleSpec(
     module=MambaStack,
     submodules=MambaStackSubmodules(
@@ -87,9 +117,11 @@
                 pre_mlp_layernorm=TENorm, mlp=moe, mlp_bda=get_bias_dropout_add
             ),
         ),
+        mtp_block_spec=_mamba_mtp_block_spec,
     ),
 )
 
+
 mamba_inference_stack_spec = ModuleSpec(
     module=MambaStack,
     submodules=MambaStackSubmodules(
@@ -147,5 +179,6 @@
                 pre_mlp_layernorm=TENorm, mlp=moe, mlp_bda=get_bias_dropout_add
             ),
         ),
+        mtp_block_spec=_mamba_mtp_block_spec,
     ),
 )
diff --git a/megatron/core/models/mamba/mamba_model.py b/megatron/core/models/mamba/mamba_model.py
index cf1002a5426..c4b744e19e0 100644
--- a/megatron/core/models/mamba/mamba_model.py
+++ b/megatron/core/models/mamba/mamba_model.py
@@ -16,6 +16,11 @@
 from megatron.core.transformer import TransformerConfig
 from megatron.core.transformer.enums import ModelType
 from megatron.core.transformer.linear_cross_entropy import LinearCrossEntropyModule
+from megatron.core.transformer.multi_token_prediction import (
+    MultiTokenPredictionBlock,
+    mtp_on_this_rank,
+    process_mtp_loss,
+)
 from megatron.core.transformer.spec_utils import ModuleSpec, build_module
 from megatron.core.utils import (
     WrappedTensor,
@@ -38,7 +43,11 @@ class MambaModel(LanguageModule):
         hybrid_attention_ratio (float, optional): The target ratio of attention
             layers to total layers
         hybrid_mlp_ratio (float, optional): The target ratio of mlp layers to total layers
-        hybrid_override_pattern (str, optional): The hybrid layer pattern to override with
+        hybrid_override_pattern (str, optional): Unified hybrid layer pattern with optional MTP.
+            Format: "<main_pattern>/<mtp_pattern>/<mtp_pattern>/..."
+            Examples:
+                - "M*M*" -> main decoder only, no MTP
+                - "M*M*/MM/MM" -> main="M*M*", mtp="MM", 2 depths
         post_process (bool, optional): Include an output layer (used with pipeline parallelism).
             Defaults to True.
         fp16_lm_cross_entropy (bool, optional): Defaults to False.
@@ -79,6 +88,7 @@ def __init__(
         scatter_embedding_sequence_parallel: bool = True,
         seq_len_interpolation_factor: Optional[float] = None,
         pg_collection: Optional[ProcessGroupCollection] = None,
+        vp_stage: Optional[int] = None,
     ) -> None:
         super().__init__(config=config, pg_collection=pg_collection)
 
@@ -97,6 +107,21 @@ def __init__(
         self.parallel_output = parallel_output
         self.share_embeddings_and_output_weights = share_embeddings_and_output_weights
         self.position_embedding_type = position_embedding_type
+        self.vp_stage = vp_stage
+
+        # Parse unified pattern to extract main and MTP components
+        from megatron.core.ssm.mamba_hybrid_layer_allocation import parse_hybrid_pattern
+
+        parsed = parse_hybrid_pattern(hybrid_override_pattern)
+        self.mtp_pattern = parsed.mtp_pattern
+        self.mtp_num_depths = parsed.mtp_num_depths
+
+        # Determine if MTP is needed (based on pattern parsing)
+        self.mtp_process = (
+            self.mtp_pattern is not None
+            and self.mtp_num_depths > 0
+            and mtp_on_this_rank(self.config, vp_stage=self.vp_stage)
+        )
 
         # megatron core pipelining currently depends on model type
         # TODO: remove this dependency ?
@@ -107,7 +132,7 @@ def __init__(
             and self.config.cross_entropy_fusion_impl == "linear"
         )
 
-        if self.pre_process:
+        if self.pre_process or self.mtp_process:
             self.embedding = LanguageModelEmbedding(
                 config=self.config,
                 vocab_size=self.vocab_size,
@@ -133,14 +158,33 @@ def __init__(
             pre_process=self.pre_process,
             hybrid_attention_ratio=self.hybrid_attention_ratio,
             hybrid_mlp_ratio=self.hybrid_mlp_ratio,
-            hybrid_override_pattern=self.hybrid_override_pattern,
+            hybrid_override_pattern=parsed.main_pattern,
             post_process=self.post_process,
             dtype=config.params_dtype,
             pg_collection=self.pg_collection,
         )
 
+        # MTP block - uses mtp_block_spec from mamba_stack_spec.submodules
+        if self.mtp_process:
+            mamba_submodules = mamba_stack_spec.submodules
+            mtp_block_spec = mamba_submodules.mtp_block_spec
+            assert mtp_block_spec is not None, (
+                "MTP pattern specified but mtp_block_spec is None in mamba_stack_spec.submodules. "
+                "Ensure mamba_stack_spec includes mtp_block_spec for MTP support."
+            )
+
+            self.mtp = MultiTokenPredictionBlock(
+                config=self.config,
+                spec=mtp_block_spec,
+                pg_collection=self.pg_collection,
+                vp_stage=self.vp_stage,
+                mtp_layer_pattern=self.mtp_pattern,
+                mtp_num_depths=self.mtp_num_depths,
+                mamba_submodules=mamba_submodules,
+            )
+
         # Output
-        if post_process:
+        if post_process or self.mtp_process:
             self.output_layer = LinearCrossEntropyModule(
                 config.hidden_size,
                 self.vocab_size,
@@ -154,7 +198,7 @@ def __init__(
                 tp_group=self.pg_collection.tp,
             )
 
-        if self.pre_process or self.post_process:
+        if self.pre_process or self.post_process or self.mtp_process:
             self.setup_embeddings_and_output_layer()
 
         for name, module in self.named_modules():
@@ -189,8 +233,10 @@ def forward(
         runtime_gather_output: Optional[bool] = None,
         *,
         inference_params: Optional[BaseInferenceContext] = None,
+        loss_mask: Optional[Tensor] = None,
         packed_seq_params: Optional[PackedSeqParams] = None,
         padding_mask: Optional[Tensor] = None,
+        mtp_labels: Optional[Tensor] = None,
     ) -> Tensor:
         """Forward function of the Mamba model. This function passes the input tensors
         through the embedding layer, and then the decoder and finally into the post
@@ -263,14 +309,45 @@ def forward(
             padding_mask=padding_mask,
         )
 
-        if not self.post_process:
-            return hidden_states
-
-        # logits and loss
         output_weight = None
         if self.share_embeddings_and_output_weights:
             output_weight = self.shared_embedding_or_output_weight()
 
+        # MTP block concatenates [main(s); mtp(s)] -> 2s; process_mtp_loss (training,
+        # labels present) splits it back to main s. In an RL logprob-only forward
+        # (labels=None) process_mtp_loss is skipped, so running MTP here would leave a
+        # 2s hidden flowing into the main lm_head and corrupt main logprobs. MTP is
+        # only needed for the training loss, so skip it when labels is None.
+        if self.mtp_process and mtp_labels is not None:
+            hidden_states = self.mtp(
+                input_ids=input_ids,
+                position_ids=position_ids,
+                hidden_states=hidden_states,
+                attention_mask=attention_mask,
+                inference_params=inference_params,
+                rotary_pos_emb=rotary_pos_emb,
+                packed_seq_params=packed_seq_params,
+                embedding=self.embedding,
+            )
+
+        if not self.post_process:
+            return hidden_states
+
+        if self.config.mtp_num_layers is not None and mtp_labels is not None:
+            hidden_states = process_mtp_loss(
+                hidden_states=hidden_states,
+                labels=mtp_labels,
+                loss_mask=loss_mask,
+                output_layer=self.output_layer,
+                output_weight=output_weight,
+                runtime_gather_output=runtime_gather_output,
+                is_training=self.training,
+                compute_language_model_loss=self.compute_language_model_loss,
+                config=self.config,
+                cp_group=self.pg_collection.cp,
+                packed_seq_params=packed_seq_params,
+            )
+
         sequence_parallel_override = False
         if in_inference_mode and inference_context.materialize_only_last_token_logits:
             if inference_context.is_static_batching():
@@ -286,7 +363,7 @@ def forward(
                     self.output_layer.sequence_parallel = False
                     sequence_parallel_override = True
 
-                # Reshape [B, 1, H] to [1, B, H] → extract each sample’s true last‐token hidden
+                # Reshape [B, 1, H] to [1, B, H] → extract each sample's true last‐token hidden
                 # state ([B, H]) → unsqueeze back to [B, 1, H]
                 # (so that the output layer, which expects S×B×H, receives only the final token)
                 hidden_states = inference_context.last_token_logits(
diff --git a/megatron/core/pipeline_parallel/schedules.py b/megatron/core/pipeline_parallel/schedules.py
index a7588e83562..687bde604bf 100644
--- a/megatron/core/pipeline_parallel/schedules.py
+++ b/megatron/core/pipeline_parallel/schedules.py
@@ -213,7 +213,7 @@ def set_current_microbatch(model, microbatch_id):
             layer.current_microbatch = microbatch_id
         if hasattr(model_with_decoder, 'mtp'):
             for layer in model_with_decoder.mtp.layers:
-                layer.transformer_layer.current_microbatch = microbatch_id
+                layer.mtp_model_layer.current_microbatch = microbatch_id
 
 
 def forward_step_calc_loss(
diff --git a/megatron/core/ssm/mamba_block.py b/megatron/core/ssm/mamba_block.py
index ef41faae143..1bbf79608af 100644
--- a/megatron/core/ssm/mamba_block.py
+++ b/megatron/core/ssm/mamba_block.py
@@ -42,6 +42,7 @@ class MambaStackSubmodules:
     attention_layer: Union[ModuleSpec, type] = IdentityOp
     mlp_layer: Union[ModuleSpec, type] = IdentityOp
     moe_layer: Union[ModuleSpec, type] = IdentityOp
+    mtp_block_spec: Optional[ModuleSpec] = None
 
 
 class MambaStack(MegatronModule):
@@ -85,12 +86,14 @@ def __init__(
         device=None,
         dtype=None,
         pg_collection: ProcessGroupCollection = None,
+        is_mtp_layer: bool = False,
     ) -> None:
         super().__init__(config=config)
         self.residual_in_fp32 = residual_in_fp32
         self.pre_process = pre_process
         self.post_layer_norm = post_layer_norm
         self.post_process = post_process
+        self.is_mtp_layer = is_mtp_layer
 
         assert pg_collection is not None, "pg_collection must be provided for MambaStack"
 
@@ -103,20 +106,31 @@ def __init__(
         self.hybrid_attention_ratio = hybrid_attention_ratio
         self.hybrid_mlp_ratio = hybrid_mlp_ratio
         self.hybrid_override_pattern = hybrid_override_pattern
+        self.pg_collection = pg_collection
+
+        # For MTP layers, always use pattern length (config.num_layers is for main decoder)
+        if self.is_mtp_layer:
+            num_layers_for_allocation = len(self.hybrid_override_pattern)
+        else:
+            num_layers_for_allocation = (
+                self.config.num_layers
+                if self.config.num_layers is not None
+                else len(self.hybrid_override_pattern)
+            )
 
         self.layer_type_list = allocate_layers(
-            self.config.num_layers,
+            num_layers_for_allocation,
             self.hybrid_attention_ratio,
             self.hybrid_mlp_ratio,
             self.hybrid_override_pattern,
         )
 
         pp_layer_offset = 0
-        if self.pp_group.size() > 1:
+        if self.pp_group.size() > 1 and not self.is_mtp_layer:
             pp_layer_offset, self.layer_type_list = self._select_layers_for_pipeline_parallel(
                 self.layer_type_list
             )
-
+        # Build main decoder layers using shared layer builder
         self.layers = nn.ModuleList()
         for i, layer_type in enumerate(self.layer_type_list):
             fp8_init_context = get_fp8_context(self.config, i + pp_layer_offset, is_init=True)
@@ -137,6 +151,7 @@ def __init__(
                         config=self.config,
                         layer_number=i + 1,
                         pg_collection=pg_collection,
+                        is_mtp_layer=is_mtp_layer,
                     )
                 elif layer_type == LayerSymbols.MLP:
                     # Transformer layers apply their own pp_layer_offset
@@ -153,6 +168,7 @@ def __init__(
                         config=self.config,
                         layer_number=i + 1,
                         pg_collection=pg_collection,
+                        is_mtp_layer=is_mtp_layer,
                     )
                 else:
                     assert False, "unexpected layer_type"
@@ -316,7 +332,7 @@ def forward(
 
         # Ensure that the tensor passed between pipeline parallel stages is
         # viewless. See related notes in TransformerBlock and TransformerLayer
-        output = make_viewless_tensor(
+        hidden_states = make_viewless_tensor(
             inp=hidden_states, requires_grad=hidden_states.requires_grad, keep_graph=True
         )
 
diff --git a/megatron/core/ssm/mamba_hybrid_layer_allocation.py b/megatron/core/ssm/mamba_hybrid_layer_allocation.py
index fe997e2249a..addecd33f2e 100644
--- a/megatron/core/ssm/mamba_hybrid_layer_allocation.py
+++ b/megatron/core/ssm/mamba_hybrid_layer_allocation.py
@@ -1,7 +1,8 @@
 # Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved.
 
 import logging
-from typing import Dict, List, Tuple
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
 
 if __name__ != "__main__":
     from megatron.core.utils import log_single_rank
@@ -29,9 +30,129 @@ class Symbols:
     ATTENTION = "*"
     MLP = "-"
     MOE = 'E'
+    MTP_SEPARATOR = "/"
     VALID = {MAMBA, ATTENTION, MLP, MOE}
 
 
+@dataclass
+class ParsedHybridPattern:
+    """Result of parsing a unified hybrid pattern string.
+
+    A unified pattern encodes both the main decoder pattern and the MTP pattern
+    in a single string using "/" as a separator.
+
+    Format: "<main_pattern>/<mtp_pattern>/<mtp_pattern>/..."
+
+    Examples:
+        - "M*M*" -> main="M*M*", mtp=None, depths=0 (no MTP)
+        - "M*M*/MM/MM" -> main="M*M*", mtp="MM", depths=2
+        - "MMMM/*M/*M/*M" -> main="MMMM", mtp="*M", depths=3
+
+    The "/" symbol introduces MTP patterns. Each repeated pattern after the main
+    decoder represents one MTP prediction depth.
+
+    Attributes:
+        main_pattern: The main decoder layer pattern (e.g., "M*M*")
+        mtp_pattern: The MTP layer pattern per depth (e.g., "MM"), or None if no MTP
+        mtp_num_depths: Number of MTP prediction depths (0 if no MTP)
+    """
+
+    main_pattern: Optional[str]
+    mtp_pattern: Optional[str]
+    mtp_num_depths: int
+
+
+def parse_hybrid_pattern(pattern: Optional[str]) -> ParsedHybridPattern:
+    """Parse a unified hybrid pattern string into main and MTP components.
+
+    The pattern uses "/" as a separator between the main decoder pattern and
+    MTP patterns. Each MTP pattern after the separator represents one prediction
+    depth.
+
+    Format: "<main_pattern>/<mtp_pattern>/<mtp_pattern>/..."
+
+    Args:
+        pattern: Unified pattern string, e.g., "M*M*/MM/MM" or just "M*M*"
+
+    Returns:
+        ParsedHybridPattern with main_pattern, mtp_pattern, and mtp_num_depths
+
+    Raises:
+        ValueError: If MTP patterns are inconsistent (all must be identical)
+        ValueError: If pattern contains invalid layer symbols
+
+    Examples:
+        >>> parse_hybrid_pattern("M*M*")
+        ParsedHybridPattern(main_pattern="M*M*", mtp_pattern=None, mtp_num_depths=0)
+
+        >>> parse_hybrid_pattern("M*M*/MM/MM")
+        ParsedHybridPattern(main_pattern="M*M*", mtp_pattern="MM", mtp_num_depths=2)
+
+        >>> parse_hybrid_pattern("MMMM/*M/*M/*M")
+        ParsedHybridPattern(main_pattern="MMMM", mtp_pattern="*M", mtp_num_depths=3)
+    """
+    if pattern is None:
+        return ParsedHybridPattern(main_pattern=None, mtp_pattern=None, mtp_num_depths=0)
+
+    parts = pattern.split(Symbols.MTP_SEPARATOR)
+
+    if len(parts) == 1:
+        # No MTP separator found - pattern is main decoder only
+        main_pattern = parts[0]
+        _validate_pattern(main_pattern, "main")
+        return ParsedHybridPattern(main_pattern=main_pattern, mtp_pattern=None, mtp_num_depths=0)
+
+    # First part is main decoder pattern
+    main_pattern = parts[0]
+    if main_pattern:
+        _validate_pattern(main_pattern, "main")
+
+    # Remaining parts are MTP patterns (one per depth)
+    mtp_parts = parts[1:]
+
+    if not mtp_parts or all(p == "" for p in mtp_parts):
+        # No MTP patterns after separator
+        return ParsedHybridPattern(
+            main_pattern=main_pattern if main_pattern else None, mtp_pattern=None, mtp_num_depths=0
+        )
+
+    # Validate all MTP patterns are identical
+    mtp_pattern = mtp_parts[0]
+    for i, part in enumerate(mtp_parts[1:], start=2):
+        if part != mtp_pattern:
+            raise ValueError(
+                f"All MTP patterns must be identical. "
+                f"Pattern 1 is '{mtp_pattern}', but pattern {i} is '{part}'. "
+                f"Full pattern: '{pattern}'"
+            )
+
+    _validate_pattern(mtp_pattern, "MTP")
+
+    return ParsedHybridPattern(
+        main_pattern=main_pattern if main_pattern else None,
+        mtp_pattern=mtp_pattern,
+        mtp_num_depths=len(mtp_parts),
+    )
+
+
+def _validate_pattern(pattern: str, pattern_name: str) -> None:
+    """Validate that a pattern contains only valid layer symbols.
+
+    Args:
+        pattern: Layer pattern string to validate
+        pattern_name: Name of pattern for error messages (e.g., "main" or "MTP")
+
+    Raises:
+        ValueError: If pattern contains invalid symbols
+    """
+    for char in pattern:
+        if char not in Symbols.VALID:
+            raise ValueError(
+                f"In {pattern_name} pattern, '{char}' is not a valid layer symbol. "
+                f"Valid symbols are: {Symbols.VALID}"
+            )
+
+
 def _allocate_auto(
     total_layers_count: int, target_attention_ratio: float, target_mlp_ratio: float
 ) -> list:
@@ -123,7 +244,9 @@ def allocate_layers(
                 logger, logging.INFO, "The override pattern matches the overridden pattern"
             )
         else:
-            log_single_rank(logger, logging.INFO, "Warning: overriding pattern A with pattern B")
+            log_single_rank(
+                logger, logging.INFO, "Warning: overriding pattern A with pattern B"
+            )
             log_single_rank(logger, logging.INFO, f"A: {''.join(layer_type_list)}")
             log_single_rank(logger, logging.INFO, f"B: {''.join(layer_type_list_override)}")
         layer_type_list = layer_type_list_override
diff --git a/megatron/core/transformer/cuda_graphs.py b/megatron/core/transformer/cuda_graphs.py
index 3643c42c3ce..dd0dad8eba5 100644
--- a/megatron/core/transformer/cuda_graphs.py
+++ b/megatron/core/transformer/cuda_graphs.py
@@ -1738,7 +1738,7 @@ def __init__(self, model, config, seq_length, micro_batch_size, optimizers=[]):
                         callables.append(layer)
                         callables_is_mtp.append(False)
                 for layer_number in range(num_mtp_layers):
-                    layer = chunk_with_decoder.mtp.layers[layer_number].transformer_layer
+                    layer = chunk_with_decoder.mtp.layers[layer_number].mtp_model_layer
                     if _layer_is_graphable(layer, config):
                         num_graphable_layers += 1
                         callables.append(layer)
@@ -1855,7 +1855,7 @@ def _get_layer_static_inputs(layer, chunk_of_the_layer):
             Get the static inputs for a layer.
             """
             assert layer in chunk_of_the_layer.decoder.layers or any(
-                layer is mtp_layer.transformer_layer for mtp_layer in chunk_of_the_layer.mtp.layers
+                layer is mtp_layer.mtp_model_layer for mtp_layer in chunk_of_the_layer.mtp.layers
             ), "Layer is not in the chunk"
 
             def get_rotary_pos_emb(transformer_module, transformer_input):
diff --git a/megatron/core/transformer/moe/moe_layer.py b/megatron/core/transformer/moe/moe_layer.py
index c5953a45b1d..0afa27cdadb 100644
--- a/megatron/core/transformer/moe/moe_layer.py
+++ b/megatron/core/transformer/moe/moe_layer.py
@@ -87,10 +87,12 @@ def __init__(
         config: TransformerConfig,
         layer_number: Optional[int] = None,
         pg_collection: Optional[ProcessGroupCollection] = None,
+        is_mtp_layer: bool = False,
     ):
         super(BaseMoELayer, self).__init__(config)
         self.config = config
         self.layer_number = layer_number
+        self.is_mtp_layer = is_mtp_layer
         self.ep_group = pg_collection.ep
         # use pg_collection.expt_tp_group as tensor parallel group in this module.
         self.attn_tp_group = pg_collection.tp
@@ -125,11 +127,6 @@ def set_layer_number(self, layer_number: int):
         self.layer_number = layer_number
         self.router.set_layer_number(layer_number)
 
-    def set_is_mtp(self):
-        """Mark this MoE layer as an MTP layer."""
-        self.router.is_mtp = True
-
-
 class MoELayer(BaseMoELayer):
     """Mixture of Experts layer.
 
@@ -144,6 +141,7 @@ def __init__(
         submodules: Optional[MoESubmodules] = None,
         layer_number: Optional[int] = None,
         pg_collection: Optional[ProcessGroupCollection] = None,
+        is_mtp_layer: bool = False,
     ):
         self.submodules = submodules
         # TODO(Hepteract): delete the usage of the global parallel_state.
@@ -151,7 +149,10 @@ def __init__(
         if pg_collection is None:
             pg_collection = get_default_pg_collection()
         super(MoELayer, self).__init__(
-            config=config, layer_number=layer_number, pg_collection=pg_collection
+            config=config,
+            layer_number=layer_number,
+            pg_collection=pg_collection,
+            is_mtp_layer=is_mtp_layer,
         )
         # If using mcore cudagraphs, recompute is handled by transformer_layer.MoETransformerLayer
         self.moe_layer_recompute = (
@@ -167,7 +168,12 @@ def __init__(
         self.tp_group = pg_collection.tp
 
         # Initialize router.
-        self.router = submodules.router(config=self.config, pg_collection=pg_collection, layer_number=layer_number)
+        self.router = submodules.router(
+            config=self.config,
+            pg_collection=pg_collection,
+            layer_number=layer_number,
+            is_mtp_layer=is_mtp_layer,
+        )
         self.tp_group = pg_collection.tp
 
         # Initialize latent projections.
diff --git a/megatron/core/transformer/moe/router.py b/megatron/core/transformer/moe/router.py
index 98057078127..0ef358a14ce 100644
--- a/megatron/core/transformer/moe/router.py
+++ b/megatron/core/transformer/moe/router.py
@@ -29,7 +29,10 @@ class Router(ABC, MegatronModule):
     """Base Router class"""
 
     def __init__(
-        self, config: TransformerConfig, pg_collection: Optional[ProcessGroupCollection] = None
+        self,
+        config: TransformerConfig,
+        pg_collection: Optional[ProcessGroupCollection] = None,
+        is_mtp_layer: bool = False,
     ) -> None:
         """
         Initialize the Router module.
@@ -37,13 +40,14 @@ def __init__(
         Args:
             config (TransformerConfig): Configuration object for the Transformer model.
             pg_collection (ProcessGroupCollection, optional): Process groups for MoE operations.
+            is_mtp_layer (bool): Flag indicating if this router is part of an MTP layer.
         """
         super().__init__(config)
         self.config = config
         self.num_experts = self.config.num_moe_experts
         self.moe_aux_loss_func = None
         self.layer_number = None
-        self.is_mtp = False
+        self.is_mtp_layer = is_mtp_layer
         self.tp_group = pg_collection.tp
         self.cp_group = pg_collection.cp
         self.tp_cp_group = pg_collection.tp_cp
@@ -156,7 +160,11 @@ class TopKRouter(Router):
     """
 
     def __init__(
-        self, config: TransformerConfig, pg_collection: Optional[ProcessGroupCollection] = None, layer_number: int = None
+        self,
+        config: TransformerConfig,
+        pg_collection: Optional[ProcessGroupCollection] = None,
+        layer_number: int = None,
+        is_mtp_layer: bool = False,
     ) -> None:
         """Initialize the zero token dropping router.
 
@@ -164,8 +172,9 @@ def __init__(
             config (TransformerConfig): The configuration for the transformer model.
             pg_collection (ProcessGroupCollection, optional): Process groups for MoE operations.
             layer_number (int, optional): The layer number for this router.
+            is_mtp_layer (bool): Flag indicating if this router is part of an MTP layer.
         """
-        super().__init__(config=config, pg_collection=pg_collection)
+        super().__init__(config=config, pg_collection=pg_collection, is_mtp_layer=is_mtp_layer)
         self.layer_number = layer_number
         self.topk = self.config.moe_router_topk
         self.routing_type = self.config.moe_router_load_balancing_type
@@ -205,7 +214,11 @@ def __init__(
             self.ga_steps = None
 
         from miles.utils.replay_base import routing_replay_manager
-        routing_replay_manager.register_to_module(self, "routing_replay")
+
+        # Rollout routing replay only records main-decoder MoE layers;
+        # an MTP slot would desync the slot/fill pairing.
+        if not self.is_mtp_layer:
+            routing_replay_manager.register_to_module(self, "routing_replay")
 
     def _init_routing_mode(self, layer_number):
         assert not self._routing_mode_initialized
@@ -489,6 +502,16 @@ def attach_and_log_load_balancing_loss(
                 padding tokens. Can be a Python int or a torch.Tensor (typically 0-d tensor).
                 If None, uses activation.shape[0]. Defaults to None.
         """
+        # When using repeated MTP layers, the loss is counted "mtp_num_layers" times.
+        # To avoid accumulating the load balancing loss multiple times, we scale it by
+        # 1/mtp_num_layers so the total loss is correct.
+        if (
+            self.is_mtp_layer
+            and self.config.mtp_use_repeated_layer
+            and self.config.mtp_num_layers is not None
+        ):
+            aux_loss = aux_loss / self.config.mtp_num_layers
+
         # TODO (zijiey): fix the per_layer_logging for MTP, currently it will incorrectly
         # add the aux loss logging value to other layer's since it is difficult to get the
         # correct layer_number for MTP. It does not affect the correctness of the calculation
@@ -496,10 +519,16 @@ def attach_and_log_load_balancing_loss(
         num_layers = self.config.num_layers
         if self.config.mtp_num_layers is not None:
             num_layers += self.config.mtp_num_layers
+
+        if self.is_mtp_layer:
+            layer_number = self.layer_number + self.config.num_layers
+        else:
+            layer_number = self.layer_number
+
         save_to_aux_losses_tracker(
             aux_loss_name,
             aux_loss / aux_loss_coeff,
-            self.layer_number,
+            layer_number,
             num_layers,
             reduce_group=reduce_group,
             reduce_group_has_dp=reduce_group_has_dp,
@@ -550,11 +579,27 @@ def apply_z_loss(self, logits, padding_mask: Optional[torch.Tensor] = None):
             else:
                 logits = MoEAuxLossAutoScaler.apply(logits, z_loss)
 
+            # When using repeated MTP layers, the same MTP layer is called mtp_num_layers times.
+            # To avoid accumulating the z_loss multiple times, we scale it by 1/mtp_num_layers
+            # so the total loss is correct.
+            if (
+                self.is_mtp_layer
+                and self.config.mtp_use_repeated_layer
+                and self.config.mtp_num_layers is not None
+            ):
+                z_loss = z_loss / self.config.mtp_num_layers
+
             num_layers = self.config.num_layers
             if self.config.mtp_num_layers is not None:
                 num_layers += self.config.mtp_num_layers
+
+            if self.is_mtp_layer:
+                layer_number = self.layer_number + self.config.num_layers
+            else:
+                layer_number = self.layer_number
+
             save_to_aux_losses_tracker(
-                "z_loss", z_loss / moe_z_loss_coeff, self.layer_number, num_layers
+                "z_loss", z_loss / moe_z_loss_coeff, layer_number, num_layers
             )
         return logits
 
@@ -635,7 +680,7 @@ def routing(self, logits: torch.Tensor, padding_mask: Optional[torch.Tensor] = N
                 score_function=self.score_function,
                 expert_bias=self.expert_bias,
                 fused=self.config.moe_router_fusion,
-                is_mtp=self.is_mtp,
+                is_mtp=self.is_mtp_layer,
                 tid2eid=self.tid2eid,
                 input_ids=input_ids.view(-1) if self.tid2eid is not None and input_ids is not None else None,
             )
diff --git a/megatron/core/transformer/multi_token_prediction.py b/megatron/core/transformer/multi_token_prediction.py
index d05971809b1..cb34aafe237 100755
--- a/megatron/core/transformer/multi_token_prediction.py
+++ b/megatron/core/transformer/multi_token_prediction.py
@@ -15,6 +15,7 @@
 from megatron.core.fp8_utils import get_fp8_context
 from megatron.core.models.backends import BackendSpecProvider, LocalSpecProvider
 from megatron.core.packed_seq_params import PackedSeqParams
+from megatron.core.pipeline_parallel.utils import is_vp_last_stage
 from megatron.core.process_groups_config import ProcessGroupCollection
 from megatron.core.tensor_parallel import (
     gather_from_tensor_model_parallel_region,
@@ -25,7 +26,6 @@
 from megatron.core.transformer.spec_utils import ModuleSpec, build_module
 from megatron.core.transformer.transformer_block import TransformerBlockSubmodules
 from megatron.core.transformer.transformer_config import TransformerConfig
-from megatron.core.transformer.transformer_layer import get_transformer_layer_offset
 from megatron.core.utils import (
     get_pg_rank,
     is_torch_min_version,
@@ -397,19 +397,19 @@ class MultiTokenPredictionLayerSubmodules:
             embedding normalization to be applied.
         eh_proj (Union[ModuleSpec, type]): Specification or instance of the
             linear projection to be applied.
-        transformer_layer (Union[ModuleSpec, type]): Specification
-            or instance of the transformer block to be applied.
+        mtp_model_layer (Union[ModuleSpec, type]): Specification
+            or instance of the transformer or mamba block to be applied.
     """
 
     enorm: Union[ModuleSpec, type] = None
     hnorm: Union[ModuleSpec, type] = None
     eh_proj: Union[ModuleSpec, type] = None
-    transformer_layer: Union[ModuleSpec, type] = None
+    mtp_model_layer: Union[ModuleSpec, type] = None
     layer_norm: Union[ModuleSpec, type] = None
 
 
 def get_mtp_layer_spec(
-    transformer_layer_spec: ModuleSpec, use_transformer_engine: bool
+    mtp_model_layer_spec: ModuleSpec, use_transformer_engine: bool
 ) -> ModuleSpec:
     """Get the MTP layer spec.
 
@@ -417,13 +417,13 @@ def get_mtp_layer_spec(
         ModuleSpec: Module specification with TE modules
     """
     return get_mtp_layer_spec_for_backend(
-        transformer_layer_spec,
+        mtp_model_layer_spec,
         backend=TESpecProvider() if use_transformer_engine else LocalSpecProvider(),
     )
 
 
 def get_mtp_layer_spec_for_backend(
-    transformer_layer_spec: ModuleSpec, backend: BackendSpecProvider
+    mtp_model_layer_spec: ModuleSpec, backend: BackendSpecProvider
 ) -> ModuleSpec:
     """Get the MTP layer spec.
 
@@ -438,7 +438,7 @@ def get_mtp_layer_spec_for_backend(
             enorm=layer_norm_impl,
             hnorm=layer_norm_impl,
             eh_proj=column_parallel_linear_impl,
-            transformer_layer=transformer_layer_spec,
+            mtp_model_layer=mtp_model_layer_spec,
             layer_norm=layer_norm_impl,
         ),
     )
@@ -587,6 +587,79 @@ def set_loss_scale(scale: torch.Tensor):
         MTPLossAutoScaler.main_loss_backward_scale = scale
 
 
+def process_mtp_loss(
+    hidden_states: Tensor,
+    labels: Tensor,
+    loss_mask: Optional[Tensor],
+    output_layer: Callable,
+    output_weight: Optional[Tensor],
+    runtime_gather_output: Optional[bool],
+    is_training: bool,
+    compute_language_model_loss: Callable,
+    config: TransformerConfig,
+    cp_group: Optional[torch.distributed.ProcessGroup] = None,
+    packed_seq_params: Optional[PackedSeqParams] = None,
+) -> Tensor:
+    """Process Multi-Token Prediction (MTP) loss computation.
+
+    This is a standalone function that handles MTP loss computation. It's used on the
+    post_process rank to split concatenated hidden states and compute MTP losses.
+
+    Args:
+        hidden_states (Tensor): Hidden states tensor (concatenated with MTP outputs).
+        labels (Tensor): Ground truth labels.
+        loss_mask (Optional[Tensor]): Mask for loss computation. If None, uses all ones.
+        output_layer (Callable): Output layer method to compute logits.
+        output_weight (Optional[Tensor]): Optional output weight for shared embeddings.
+        runtime_gather_output (Optional[bool]): Whether to gather output at runtime.
+        is_training (bool): Whether the model is in training mode.
+        compute_language_model_loss (Callable): Method to compute language model loss.
+        config (TransformerConfig): Model configuration containing mtp_num_layers etc.
+        cp_group (Optional[ProcessGroup]): Context parallelism process group.
+        packed_seq_params (Optional[PackedSeqParams]): Packed sequence parameters.
+
+    Returns:
+        Tensor: Updated hidden states after MTP loss processing (first chunk only).
+    """
+    mtp_labels = labels.clone()
+    hidden_states_list = torch.chunk(hidden_states, 1 + config.mtp_num_layers, dim=0)
+    hidden_states = hidden_states_list[0]
+
+    if loss_mask is None:
+        loss_mask = torch.ones_like(mtp_labels)
+
+    for mtp_layer_number in range(config.mtp_num_layers):
+        mtp_logits, _ = output_layer(
+            hidden_states_list[mtp_layer_number + 1],
+            weight=output_weight,
+            runtime_gather_output=runtime_gather_output,
+        )
+        mtp_labels, _ = roll_tensor(
+            mtp_labels, shifts=-1, dims=-1, cp_group=cp_group, packed_seq_params=packed_seq_params
+        )
+        loss_mask, num_tokens = roll_tensor(
+            loss_mask, shifts=-1, dims=-1, cp_group=cp_group, packed_seq_params=packed_seq_params
+        )
+        mtp_loss = compute_language_model_loss(mtp_labels, mtp_logits)
+        mtp_loss = loss_mask * mtp_loss
+        if is_training:
+            MTPLossLoggingHelper.save_loss_to_tracker(
+                torch.sum(mtp_loss) / num_tokens,
+                mtp_layer_number,
+                config.mtp_num_layers,
+                avg_group=parallel_state.get_data_parallel_group(with_context_parallel=True),
+            )
+        mtp_loss_scale = config.mtp_loss_scaling_factor / config.mtp_num_layers
+        if config.calculate_per_token_loss:
+            hidden_states = MTPLossAutoScaler.apply(hidden_states, mtp_loss_scale * mtp_loss)
+        else:
+            hidden_states = MTPLossAutoScaler.apply(
+                hidden_states, mtp_loss_scale * mtp_loss / num_tokens
+            )
+
+    return hidden_states
+
+
 class MultiTokenPredictionLayer(MegatronModule):
     """The implementation for Multi-Token Prediction (MTP) which extends
     the prediction scope to multiple future tokens at each position.
@@ -614,6 +687,9 @@ def __init__(
         layer_number: int = 1,
         vp_stage: Optional[int] = None,
         pg_collection: Optional[ProcessGroupCollection] = None,
+        # For Mamba path - pattern and submodules to build inner layers directly
+        mtp_layer_pattern: Optional[str] = None,
+        mamba_submodules: Optional["MambaStackSubmodules"] = None,
     ):
         super().__init__(config=config)
         self.sequence_parallel = config.sequence_parallel
@@ -621,14 +697,35 @@ def __init__(
         self.layer_number = layer_number + get_mtp_layer_offset(self.config, vp_stage)
         self.vp_stage = vp_stage
         self.cp_group = pg_collection.cp
+        self.mtp_layer_pattern = mtp_layer_pattern
 
-        self_attention_spec = self.submodules.transformer_layer.submodules.self_attention
-        attn_mask_type = self_attention_spec.params.get('attn_mask_type', '')
-        assert attn_mask_type in SUPPORTED_ATTN_MASK, (
-            f"Multi-Token Prediction (MTP) is not jet supported with "
-            + f"{attn_mask_type} attention mask type."
-            + f"The supported attention mask types are {SUPPORTED_ATTN_MASK}."
-        )
+        # Validate attention mask type if using transformer-based inner layers
+        if self.submodules.mtp_model_layer is not None and hasattr(
+            self.submodules.mtp_model_layer, 'submodules'
+        ):
+            from megatron.core.ssm.mamba_block import MambaStackSubmodules
+            from megatron.core.transformer.transformer_layer import TransformerLayerSubmodules
+
+            layer_submodules = None
+            if isinstance(self.submodules.mtp_model_layer.submodules, MambaStackSubmodules):
+                attention_layer_spec = self.submodules.mtp_model_layer.submodules.attention_layer
+                if hasattr(attention_layer_spec, 'submodules'):
+                    assert isinstance(attention_layer_spec.submodules, TransformerLayerSubmodules)
+                    layer_submodules = attention_layer_spec.submodules
+            elif isinstance(self.submodules.mtp_model_layer.submodules, TransformerLayerSubmodules):
+                layer_submodules = self.submodules.mtp_model_layer.submodules
+            else:
+                raise ValueError(
+                    "Unsupported mtp_model_layer submodules type for attention mask validation."
+                )
+            if layer_submodules:
+                self_attention_spec = layer_submodules.self_attention
+                attn_mask_type = self_attention_spec.params.get('attn_mask_type', '')
+                assert attn_mask_type in SUPPORTED_ATTN_MASK, (
+                    f"Multi-Token Prediction (MTP) is not yet supported with "
+                    f"{attn_mask_type} attention mask type. "
+                    f"The supported attention mask types are {SUPPORTED_ATTN_MASK}."
+                )
 
         self.enorm = build_module(
             self.submodules.enorm,
@@ -659,19 +756,37 @@ def __init__(
             bias=False,
             skip_bias_add=False,
             is_expert=False,
+            tp_comm_buffer_name="mtp_eh_proj",
         )
 
-        diff_transformer_layer_offset = self.config.num_layers - get_transformer_layer_offset(
-            self.config, vp_stage
-        )
-        self.transformer_layer = build_module(
-            self.submodules.transformer_layer,
-            config=self.config,
-            vp_stage=vp_stage,
-            layer_number=self.layer_number + diff_transformer_layer_offset,
-        )
-        if hasattr(self.transformer_layer.mlp, 'set_is_mtp'):
-            self.transformer_layer.mlp.set_is_mtp()
+        # Build inner layers: two possible paths
+        # 1. Mamba path: use MambaStack for hybrid pattern support
+        # 2. GPT path: single TransformerLayer
+        if mtp_layer_pattern is not None and mamba_submodules is not None:
+            from megatron.core.ssm.mamba_block import MambaStack
+
+            self.mtp_model_layer = MambaStack(
+                config=self.config,
+                submodules=mamba_submodules,
+                hybrid_override_pattern=mtp_layer_pattern,
+                pre_process=True,  # Always receives input from eh_proj
+                post_layer_norm=False,  # MTP has its own final_layernorm
+                post_process=True,  # MTP layer is self-contained
+                pg_collection=pg_collection,
+                is_mtp_layer=True,
+            )
+        elif self.config.mtp_num_layers is not None:
+            # GPT path: Uses the transformer block spec for MTP layer
+            # MTP inner layers use their own layer numbering (self.layer_number = 1, 2, etc.)
+            # rather than continuing from decoder layer numbers. This is consistent with the
+            # Mamba path and ensures proper aux loss tracking in router.py.
+            self.mtp_model_layer = build_module(
+                self.submodules.mtp_model_layer,
+                config=self.config,
+                vp_stage=self.vp_stage,
+                layer_number=self.layer_number,
+                is_mtp_layer=True,
+            )
 
         self.final_layernorm = build_module(
             self.submodules.layer_norm,
@@ -784,7 +899,6 @@ def _proj_and_transformer_layer(
             transformer_layer_fp8_context = nullcontext()
 
         # TODO: currently ignoring FP4 in MTP layers because we need more numerical validation
-
         with rng_context:
             with fp8_context:
                 hidden_states = self._concat_embeddings(hidden_states, decoder_input)
@@ -793,19 +907,29 @@ def _proj_and_transformer_layer(
             # transformer layer is cudagraphed, the FP8GlobalStateManager.is_first_fp8_module() is
             # True so that the fp8 weight caching can be triggered correctly.
             with transformer_layer_fp8_context:
-                hidden_states, _ = self.transformer_layer(
-                    hidden_states=hidden_states,
-                    attention_mask=attention_mask,
-                    context=context,
-                    context_mask=context_mask,
-                    rotary_pos_emb=rotary_pos_emb,
-                    rotary_pos_cos=rotary_pos_cos,
-                    rotary_pos_sin=rotary_pos_sin,
-                    attention_bias=attention_bias,
-                    inference_params=inference_params,
-                    packed_seq_params=packed_seq_params,
-                    sequence_len_offset=sequence_len_offset,
-                )
+                if self.mtp_layer_pattern is not None:
+                    hidden_states = self.mtp_model_layer(
+                        hidden_states=hidden_states,
+                        attention_mask=attention_mask,
+                        rotary_pos_emb=rotary_pos_emb,
+                        inference_context=inference_params,
+                        packed_seq_params=packed_seq_params,
+                    )
+                else:
+                    # GPT path: single TransformerLayer
+                    hidden_states, _ = self.mtp_model_layer(
+                        hidden_states=hidden_states,
+                        attention_mask=attention_mask,
+                        context=context,
+                        context_mask=context_mask,
+                        rotary_pos_emb=rotary_pos_emb,
+                        rotary_pos_cos=rotary_pos_cos,
+                        rotary_pos_sin=rotary_pos_sin,
+                        attention_bias=attention_bias,
+                        inference_params=inference_params,
+                        packed_seq_params=packed_seq_params,
+                        sequence_len_offset=sequence_len_offset,
+                    )
 
         hidden_states = self._postprocess(hidden_states)
 
@@ -946,8 +1070,7 @@ def forward(
             Union[Tensor, Tuple[Tensor, Tensor]]: The output hidden states tensor of shape
             [s, b, h], and optionally the updated context tensor if cross-attention is used.
         """
-        assert context is None, f"multi token prediction + cross attention is not yet supported."
-
+        assert context is None, "multi token prediction + cross attention is not yet supported."
         input_ids, position_ids, decoder_input, hidden_states = self._get_embeddings(
             input_ids=input_ids,
             position_ids=position_ids,
@@ -1071,6 +1194,9 @@ class MultiTokenPredictionBlock(MegatronModule):
     the linear projection. The combined serves as the input of the Transformer block at
     the k-th depth to produce the output representation.
 
+    When `mtp_use_repeated_layer=True` in config, instead of creating N separate MTP layers,
+    only 1 layer is created and applied mtp_num_layers times.
+
     for more information, please refer to DeepSeek-V3 Technical Report
     https://github.com/deepseek-ai/DeepSeek-V3/blob/main/DeepSeek_V3.pdf
     """
@@ -1081,11 +1207,26 @@ def __init__(
         spec: Union[TransformerBlockSubmodules, ModuleSpec],
         vp_stage: Optional[int] = None,
         pg_collection: Optional[ProcessGroupCollection] = None,
+        # New: For Mamba path with unified pattern syntax
+        mtp_layer_pattern: Optional[str] = None,
+        mtp_num_depths: int = 0,
+        mamba_submodules: Optional["MambaStackSubmodules"] = None,
     ):
         super().__init__(config=config)
         self.submodules = _get_mtp_block_submodules(config, spec)
         self.mtp_loss_scaling_factor = config.mtp_loss_scaling_factor
         self.vp_stage = vp_stage
+        self.mtp_layer_pattern = mtp_layer_pattern
+        self.mtp_num_depths = mtp_num_depths
+        self.mamba_submodules = mamba_submodules
+        self.mtp_use_repeated_layer = self.config.mtp_use_repeated_layer
+
+        vp_size = config.virtual_pipeline_model_parallel_size
+        assert is_vp_last_stage(vp_stage=vp_stage, vp_size=vp_size), (
+            f"MTP layers must be placed on the last virtual pipeline stage. "
+            f"Got vp_stage={vp_stage} with vp_size={vp_size}. "
+            f"Placing MTP layers on different VPP stages is not currently supported."
+        )
 
         # Initialize Context Parallelism (CP) support for MTP
         # This enables MTP to work with CP > 1 by providing the CP process group
@@ -1104,7 +1245,14 @@ def __init__(
         self.cp_group = pg_collection.cp
 
     def _build_layers(self, pg_collection):
-        def build_layer(layer_spec, layer_number):
+        # Determine number of depths to build
+        if self.mtp_num_depths > 0:
+            num_depths = self.mtp_num_depths
+        else:
+            num_depths = self.config.mtp_num_layers or len(self.submodules.layer_specs)
+
+        def build_layer_legacy(layer_spec, layer_number):
+            """Build layer using legacy spec-based approach."""
             fp8_init_context = get_fp8_context(self.config, is_init=True)
             with fp8_init_context:
                 module = build_module(
@@ -1113,15 +1261,71 @@ def build_layer(layer_spec, layer_number):
                     layer_number=layer_number,
                     vp_stage=self.vp_stage,
                     pg_collection=pg_collection,
+                    mtp_layer_pattern=self.mtp_layer_pattern,
                 )
             return module
 
-        self.layers = torch.nn.ModuleList(
-            [
-                build_layer(layer_spec, i + 1)
-                for i, layer_spec in enumerate(self.submodules.layer_specs)
-            ]
-        )
+        def build_layer_with_pattern(layer_spec, layer_number, mtp_layer_pattern, mamba_submodules):
+            """Build layer using pattern-based approach (new Mamba path)."""
+            fp8_init_context = get_fp8_context(self.config, is_init=True)
+            with fp8_init_context:
+                module = build_module(
+                    layer_spec,
+                    config=self.config,
+                    layer_number=layer_number,
+                    vp_stage=self.vp_stage,
+                    pg_collection=pg_collection,
+                    mtp_layer_pattern=mtp_layer_pattern,
+                    mamba_submodules=mamba_submodules,
+                )
+            return module
+
+        # New Mamba path: use mtp_layer_pattern and mamba_submodules
+        if self.mtp_layer_pattern is not None and self.mamba_submodules is not None:
+            if self.mtp_use_repeated_layer:
+                # Shared/repeated layer: build one layer, use it for all depths
+                layer_spec = self.submodules.layer_specs[0]
+                shared_layer = build_layer_with_pattern(
+                    layer_spec,
+                    layer_number=1,
+                    mtp_layer_pattern=self.mtp_layer_pattern,
+                    mamba_submodules=self.mamba_submodules,
+                )
+                self.layers = torch.nn.ModuleList([shared_layer])
+            else:
+                # Non-shared: each depth gets its own layers
+                self.layers = torch.nn.ModuleList(
+                    [
+                        build_layer_with_pattern(
+                            self.submodules.layer_specs[
+                                min(i, len(self.submodules.layer_specs) - 1)
+                            ],
+                            layer_number=i + 1,
+                            mtp_layer_pattern=self.mtp_layer_pattern,
+                            mamba_submodules=self.mamba_submodules,
+                        )
+                        for i in range(num_depths)
+                    ]
+                )
+        elif self.mtp_use_repeated_layer:
+            # Legacy repeated layer mode
+            if len(self.submodules.layer_specs) != 1:
+                warnings.warn(
+                    "Repeated MTP mode expects exactly 1 layer spec, got "
+                    f"{len(self.submodules.layer_specs)} instead. "
+                    f"The first layer will be applied {self.config.mtp_num_layers} times."
+                )
+            self.layers = torch.nn.ModuleList(
+                [build_layer_legacy(self.submodules.layer_specs[0], layer_number=1)]
+            )
+        else:
+            # Legacy mode: build from layer_specs
+            self.layers = torch.nn.ModuleList(
+                [
+                    build_layer_legacy(layer_spec, i + 1)
+                    for i, layer_spec in enumerate(self.submodules.layer_specs)
+                ]
+            )
 
     def forward(
         self,
@@ -1157,8 +1361,9 @@ def forward(
         offset = get_mtp_layer_offset(self.config, self.vp_stage)
         hidden_states_list = list(torch.chunk(hidden_states, 1 + offset, dim=0))
         hidden_states = hidden_states_list[offset]
-        for layer_number in range(len(self.layers)):
-            (hidden_states, input_ids, position_ids) = self.layers[layer_number](
+        for iteration in range(self.config.mtp_num_layers):
+            layer_idx = 0 if self.mtp_use_repeated_layer else iteration
+            (hidden_states, input_ids, position_ids) = self.layers[layer_idx](
                 input_ids=input_ids,
                 position_ids=position_ids,
                 hidden_states=hidden_states,
@@ -1200,7 +1405,7 @@ def sharded_state_dict(
         layer_prefix = f'{prefix}layers.'
         for layer in self.layers:
             offset = get_mtp_layer_offset(self.config, self.vp_stage)
-            sharded_prefix = f'{layer_prefix}{layer.layer_number - 1 }.'
+            sharded_prefix = f'{layer_prefix}{layer.layer_number - 1}.'
 
             state_dict_prefix = f'{layer_prefix}{layer.layer_number - 1 - offset}.'
             sharded_pp_offset = []
diff --git a/megatron/core/transformer/transformer_config.py b/megatron/core/transformer/transformer_config.py
index 5c495ea8b2c..46a3f36dc69 100644
--- a/megatron/core/transformer/transformer_config.py
+++ b/megatron/core/transformer/transformer_config.py
@@ -61,6 +61,15 @@ class TransformerConfig(ModelParallelConfig):
     which serves as an additional training objective.
     """
 
+    mtp_use_repeated_layer: bool = False
+    """Use a single MTP layer repeatedly instead of multiple separate layers."""
+
+    mtp_hybrid_override_pattern: Optional[str] = None
+    """DEPRECATED: Use unified hybrid_override_pattern instead.
+    Legacy argument for loading old checkpoints.
+    Force a specific hybrid layer pattern for MTP layers.
+    """
+
     num_layers_in_first_pipeline_stage: Optional[int] = None
     """Number of transformer layers on first pipeline stage.
     None implies equal layer division across PP ranks."""
diff --git a/megatron/core/transformer/transformer_layer.py b/megatron/core/transformer/transformer_layer.py
index b94f89583c7..bdf6d6449c1 100644
--- a/megatron/core/transformer/transformer_layer.py
+++ b/megatron/core/transformer/transformer_layer.py
@@ -275,6 +275,7 @@ def __init__(
         hidden_dropout: Optional[float] = None,
         pg_collection: Optional[ProcessGroupCollection] = None,
         vp_stage: Optional[int] = None,
+        is_mtp_layer: bool = False,
     ):
         self.submodules_config = submodules
         super().__init__(config=config, vp_stage=vp_stage)
@@ -284,10 +285,18 @@ def __init__(
         self.pg_collection = pg_collection
         self.tp_group = pg_collection.tp
 
-        self.layer_number = layer_number + get_transformer_layer_offset(
-            self.config, vp_stage, get_pg_rank(pg_collection.pp)
-        )
+        # MTP inner layers use their own layer numbering (starting from 1 within each MTP depth),
+        # so they should NOT add the decoder layer offset. The router.py handles MTP layer
+        # numbering separately by adding config.num_layers to distinguish MTP layers from decoder
+        # layers in the aux loss tracker.
+        if is_mtp_layer:
+            self.layer_number = layer_number
+        else:
+            self.layer_number = layer_number + get_transformer_layer_offset(
+                self.config, vp_stage, get_pg_rank(pg_collection.pp)
+            )
         self.hidden_dropout = config.hidden_dropout if hidden_dropout is None else hidden_dropout
+        self.is_mtp_layer = is_mtp_layer
 
         # [Module 1: Input Layernorm] Optional Layernorm on the input data
         # TODO: add pytorch only layernorm
@@ -365,6 +374,9 @@ def __init__(
         if isinstance(submodules.mlp, ModuleSpec):
             if submodules.mlp.module in (MoELayer, GroupedMLP, TEGroupedMLP, SequentialMLP):
                 additional_mlp_kwargs["pg_collection"] = pg_collection
+                # Pass is_mtp_layer flag to MoELayer to distinguish MTP MoE layers.
+                if submodules.mlp.module == MoELayer:
+                    additional_mlp_kwargs["is_mtp_layer"] = self.is_mtp_layer
             elif submodules.mlp.module == MLP:
                 assert hasattr(
                     pg_collection, 'tp'
diff --git a/megatron/training/arguments.py b/megatron/training/arguments.py
index 28fea46195d..237e9df0ea8 100644
--- a/megatron/training/arguments.py
+++ b/megatron/training/arguments.py
@@ -499,6 +499,71 @@ def validate_args(args, defaults={}):
         print_rank_0('setting global batch size to {}'.format(args.global_batch_size))
     assert args.global_batch_size > 0
 
+    # === MTP validation ===
+    # Deprecation warnings for legacy MTP arguments
+    if args.mtp_hybrid_override_pattern is not None:
+        warn_rank_0(
+            "--mtp-hybrid-override-pattern is deprecated. "
+            "For new hybrid models with MTP models, use unified --hybrid-override-pattern instead. "
+            "Example: 'M*M*/MM/MM' means main='M*M*', MTP pattern='MM' with 2 depths. "
+            "This argument is kept only for loading old checkpoints.",
+            args.rank,
+        )
+
+    # Backward compatibility: convert legacy mtp_hybrid_override_pattern to unified format
+    from megatron.core.ssm.mamba_hybrid_layer_allocation import Symbols, parse_hybrid_pattern
+    sep = Symbols.MTP_SEPARATOR
+    if (
+        getattr(args, 'mtp_hybrid_override_pattern', None) is not None
+        and args.mtp_num_layers is not None
+        and args.mtp_num_layers > 0
+        and (args.hybrid_override_pattern is None or sep not in args.hybrid_override_pattern)
+    ):
+        main_pattern = args.hybrid_override_pattern or ''
+        mtp_pattern = args.mtp_hybrid_override_pattern
+        args.hybrid_override_pattern = main_pattern + sep + sep.join([mtp_pattern] * args.mtp_num_layers)
+        args.mtp_hybrid_override_pattern = None
+        print_rank_0(f"Converted legacy MTP pattern to unified: {args.hybrid_override_pattern}")
+
+    # Infer mtp_num_layers from unified pattern
+    if args.hybrid_override_pattern and sep in args.hybrid_override_pattern:
+        parsed = parse_hybrid_pattern(args.hybrid_override_pattern)
+        if parsed.mtp_pattern and parsed.mtp_num_depths > 0:
+            inferred_mtp_num_layers = parsed.mtp_num_depths
+            if args.mtp_num_layers is None:
+                args.mtp_num_layers = inferred_mtp_num_layers
+            elif args.mtp_num_layers != inferred_mtp_num_layers:
+                warn_rank_0(
+                    f"--mtp-num-layers ({args.mtp_num_layers}) conflicts with "
+                    f"MTP depth count ({inferred_mtp_num_layers}) in pattern '{args.hybrid_override_pattern}'. "
+                    f"Using the inferred value ({inferred_mtp_num_layers}).",
+                    args.rank
+                )
+                args.mtp_num_layers = inferred_mtp_num_layers
+
+    # Validate MTP args for hybrid vs non-hybrid models
+    if args.is_hybrid_model:
+        # Mamba/hybrid model MTP validation
+        if args.mtp_num_layers and not (args.hybrid_override_pattern and sep in args.hybrid_override_pattern):
+            # Hybrid model wants MTP but no unified pattern - check for legacy args
+            if args.mtp_hybrid_override_pattern is None:
+                warn_rank_0(
+                    "Hybrid model with --mtp-num-layers but no MTP pattern. "
+                    "Use unified --hybrid-override-pattern with '/' separator (e.g., 'M*M*/MM/MM') "
+                    "or legacy --mtp-hybrid-override-pattern for old checkpoints.",
+                    args.rank
+                )
+    else:
+        # Non-hybrid (GPT) model MTP validation
+        if args.mtp_hybrid_override_pattern is not None:
+            warn_rank_0(
+                "--mtp-hybrid-override-pattern is for Mamba/hybrid models only. "
+                "For GPT models, MTP replicates the main transformer layer structure. "
+                "This argument will be ignored.",
+                args.rank
+            )
+    # === End of MTP validation ===
+
     # Uneven virtual pipeline parallelism
     assert (
         int(args.num_layers_per_virtual_pipeline_stage is not None)
diff --git a/megatron/training/checkpointing.py b/megatron/training/checkpointing.py
index 7c87eca191a..1c166a1db06 100644
--- a/megatron/training/checkpointing.py
+++ b/megatron/training/checkpointing.py
@@ -1418,6 +1418,12 @@ def _set_arg(arg_name, old_arg_name=None, force=False):
     _set_arg('hidden_dropout', force=True)
 
     _set_arg('hybrid_override_pattern', force=True)
+
+    # Legacy MTP pattern for old checkpoints
+    _set_arg('mtp_hybrid_override_pattern', force=True)
+    _set_arg('mtp_num_layers', force=True)
+    _set_arg('mtp_use_repeated_layer', force=True)
+
     _set_arg('spec', force=True)
     _set_arg('hybrid_attention_ratio', force=True)
     _set_arg('hybrid_mlp_ratio', force=True)
diff --git a/megatron/training/training.py b/megatron/training/training.py
index 847abba33f1..d8c87732662 100644
--- a/megatron/training/training.py
+++ b/megatron/training/training.py
@@ -220,10 +220,20 @@ def num_floating_point_operations(args, batch_size):
     def calculate_layer_counts():
         """Calculate the number of attention, Mamba, and MLP layers."""
         if args.hybrid_override_pattern:
-            counts = {'M': 0, '*': 0, '-': 0, 'E':0}
-            for layer_type in args.hybrid_override_pattern:
-                if layer_type in counts:
-                    counts[layer_type] += 1
+            from megatron.core.ssm.mamba_hybrid_layer_allocation import parse_hybrid_pattern
+            # Parse unified pattern to separate main and MTP components
+            parsed = parse_hybrid_pattern(args.hybrid_override_pattern)
+            counts = {'M': 0, '*': 0, '-': 0, 'E': 0}
+            # Count main decoder layers
+            if parsed.main_pattern:
+                for layer_type in parsed.main_pattern:
+                    if layer_type in counts:
+                        counts[layer_type] += 1
+            # Count MTP layers (pattern repeated mtp_num_depths times)
+            if parsed.mtp_pattern and parsed.mtp_num_depths > 0:
+                for layer_type in parsed.mtp_pattern:
+                    if layer_type in counts:
+                        counts[layer_type] += parsed.mtp_num_depths
             return counts['*'], counts['M'], counts['-'], counts['E']
         else:
             num_attn_layers = round(args.num_layers * args.hybrid_attention_ratio)
@@ -300,7 +310,7 @@ def hybrid_flops(batch_size, seq_len, hidden_size,
                      mlp_expansion=4.0, swiglu=False,
                      moe_latent_size=None,
                      moe_ffn_hidden_size=2048, shared_expert_ffn_hidden_size=2048, num_experts_routed_to=1,
-                     vocab_size=256000):
+                     vocab_size=256000, mtp_num_layers=0):
         """Calculate total FLOPs for the hybrid model."""
         flops_fwd = (
                 num_attn_layers * attn_layer_flops(batch_size, seq_len, hidden_size,
@@ -313,7 +323,7 @@ def hybrid_flops(batch_size, seq_len, hidden_size,
                 num_moe_layers * moe_layer_flops(batch_size, seq_len, hidden_size, moe_ffn_hidden_size,
                                                  shared_expert_ffn_hidden_size, num_experts_routed_to,
                                                  moe_latent_size, swiglu) +
-                (2 * batch_size * seq_len * hidden_size * vocab_size)  # logits computation
+                (2 * batch_size * seq_len * hidden_size * vocab_size * (1 + mtp_num_layers))  # logits computation
         )
         return flops_fwd * 3
 
@@ -578,6 +588,9 @@ def transformer_flops():
         # Calculate the number of each type of layer.
         num_attn_layers, num_mamba_layers, num_mlp_layers, num_moe_layers = calculate_layer_counts()
 
+        mtp_num_layers = args.mtp_num_layers
+        if mtp_num_layers is None:
+            mtp_num_layers = 0
         # Compute hybrid model FLOPs.
         return hybrid_flops(
             batch_size=batch_size,
@@ -604,6 +617,7 @@ def transformer_flops():
                                            else args.moe_shared_expert_intermediate_size),
             num_experts_routed_to=args.moe_router_topk,
             vocab_size=args.padded_vocab_size,
+            mtp_num_layers=mtp_num_layers,
         )
     else:
         # Compute standard Transformer model FLOPs.
diff --git a/pretrain_mamba.py b/pretrain_mamba.py
index e1379be63e9..c41c485c866 100644
--- a/pretrain_mamba.py
+++ b/pretrain_mamba.py
@@ -257,6 +257,7 @@ def forward_step(data_iterator, model: MambaModel):
             attention_mask,
             labels=labels,
             packed_seq_params=packed_seq_params,
+            loss_mask=loss_mask
         )
 
     # [ModelOpt]: model is needed to access ModelOpt distillation losses
diff --git a/tests/unit_tests/models/test_mamba_moe_model.py b/tests/unit_tests/models/test_mamba_moe_model.py
index 2481649bc3f..205c0e7936f 100644
--- a/tests/unit_tests/models/test_mamba_moe_model.py
+++ b/tests/unit_tests/models/test_mamba_moe_model.py
@@ -193,9 +193,11 @@
     "moe_z_loss_coeff": None,
     "moe_enable_routing_replay": False,
     "mrope_section": None,
+    "mtp_hybrid_override_pattern": None,
     "mtp_loss_scaling_factor": 0.1,
     "mtp_num_layers": None,
     "mtp_standalone": False,
+    "mtp_use_repeated_layer": False,
     "multi_latent_attention": False,
     "no_rope_freq": None,
     "no_sync_func": None,
diff --git a/tests/unit_tests/ssm/test_mamba_hybrid_layer_allocation.py b/tests/unit_tests/ssm/test_mamba_hybrid_layer_allocation.py
index 77d02c69607..77c106c3bee 100644
--- a/tests/unit_tests/ssm/test_mamba_hybrid_layer_allocation.py
+++ b/tests/unit_tests/ssm/test_mamba_hybrid_layer_allocation.py
@@ -6,7 +6,12 @@
 import pytest
 import torch
 
-from megatron.core.ssm.mamba_hybrid_layer_allocation import Symbols, allocate_layers
+from megatron.core.ssm.mamba_hybrid_layer_allocation import (
+    ParsedHybridPattern,
+    Symbols,
+    allocate_layers,
+    parse_hybrid_pattern,
+)
 
 
 @pytest.mark.internal
@@ -75,3 +80,135 @@ def test_wrong_length_override_pattern(self):
     def test_wrong_number_of_layer_types_in_override_pattern(self):
         # This override_pattern has too many mlps and not enough attention
         layer_types = allocate_layers(8, 0.5, 0.25, "M*--M**-")
+
+
+@pytest.mark.internal
+class TestParseHybridPattern:
+    """Tests for parse_hybrid_pattern with unified pattern syntax."""
+
+    def test_none_pattern(self):
+        """Test that None pattern returns all None values."""
+        result = parse_hybrid_pattern(None)
+        assert result.main_pattern is None
+        assert result.mtp_pattern is None
+        assert result.mtp_num_depths == 0
+
+    def test_main_pattern_only(self):
+        """Test patterns without MTP (no / separator)."""
+        test_cases = [
+            ("M*M*", "M*M*"),
+            ("MMMM", "MMMM"),
+            ("*M*M", "*M*M"),
+            ("MM-*", "MM-*"),
+            ("E", "E"),
+        ]
+        for pattern, expected_main in test_cases:
+            result = parse_hybrid_pattern(pattern)
+            assert result.main_pattern == expected_main, f"Failed for pattern: {pattern}"
+            assert result.mtp_pattern is None
+            assert result.mtp_num_depths == 0
+
+    def test_main_with_single_mtp_depth(self):
+        """Test patterns with 1 MTP depth."""
+        test_cases = [
+            ("M*M*/MM", "M*M*", "MM", 1),
+            ("MMMM/*M", "MMMM", "*M", 1),
+            ("M/M", "M", "M", 1),
+        ]
+        for pattern, expected_main, expected_mtp, expected_depths in test_cases:
+            result = parse_hybrid_pattern(pattern)
+            assert result.main_pattern == expected_main, f"Failed for pattern: {pattern}"
+            assert result.mtp_pattern == expected_mtp, f"Failed for pattern: {pattern}"
+            assert result.mtp_num_depths == expected_depths, f"Failed for pattern: {pattern}"
+
+    def test_main_with_multiple_mtp_depths(self):
+        """Test patterns with multiple MTP depths."""
+        test_cases = [
+            ("M*M*/MM/MM", "M*M*", "MM", 2),
+            ("M*M*/MM/MM/MM", "M*M*", "MM", 3),
+            ("MMMM/*M/*M/*M", "MMMM", "*M", 3),
+            ("M*/*/*/*", "M*", "*", 3),
+            ("M/M/M/M/M", "M", "M", 4),
+        ]
+        for pattern, expected_main, expected_mtp, expected_depths in test_cases:
+            result = parse_hybrid_pattern(pattern)
+            assert result.main_pattern == expected_main, f"Failed for pattern: {pattern}"
+            assert result.mtp_pattern == expected_mtp, f"Failed for pattern: {pattern}"
+            assert result.mtp_num_depths == expected_depths, f"Failed for pattern: {pattern}"
+
+    def test_mtp_patterns_must_be_identical(self):
+        """Test that mismatched MTP patterns raise ValueError."""
+        invalid_patterns = [
+            "M*M*/MM/M*",  # MM != M*
+            "M*M*/MM/MM/M",  # MM != M
+            "MMMM/*M/M*",  # *M != M*
+        ]
+        for pattern in invalid_patterns:
+            with pytest.raises(ValueError, match="All MTP patterns must be identical"):
+                parse_hybrid_pattern(pattern)
+
+    def test_invalid_symbols_in_main_pattern(self):
+        """Test that invalid symbols in main pattern raise ValueError."""
+        invalid_patterns = [
+            "M*X*",  # X is not valid
+            "MaMM",  # a is not valid
+            "M*M*1",  # 1 is not valid
+        ]
+        for pattern in invalid_patterns:
+            with pytest.raises(ValueError, match="not a valid layer symbol"):
+                parse_hybrid_pattern(pattern)
+
+    def test_invalid_symbols_in_mtp_pattern(self):
+        """Test that invalid symbols in MTP pattern raise ValueError."""
+        # Single MTP depth with invalid symbol - should raise "not a valid layer symbol"
+        with pytest.raises(ValueError, match="not a valid layer symbol"):
+            parse_hybrid_pattern("M*M*/MX")  # X is not valid
+
+        # Multiple MTP depths with invalid symbol and matching patterns
+        with pytest.raises(ValueError, match="not a valid layer symbol"):
+            parse_hybrid_pattern("M*M*/Ma/Ma")  # a is not valid
+
+        # Multiple MTP depths with invalid symbol but mismatched patterns
+        # This raises "All MTP patterns must be identical" before checking symbols
+        with pytest.raises(ValueError, match="All MTP patterns must be identical"):
+            parse_hybrid_pattern("M*M*/MM/Ma")
+
+    def test_empty_main_pattern_with_mtp(self):
+        """Test pattern that starts with / (empty main pattern)."""
+        result = parse_hybrid_pattern("/MM/MM")
+        assert result.main_pattern is None
+        assert result.mtp_pattern == "MM"
+        assert result.mtp_num_depths == 2
+
+    def test_trailing_separator(self):
+        """Test patterns with trailing separator."""
+        # "M*M*/" means main="M*M*", one empty MTP pattern
+        result = parse_hybrid_pattern("M*M*/")
+        assert result.main_pattern == "M*M*"
+        # Empty string after separator means no valid MTP pattern
+        assert result.mtp_pattern is None
+        assert result.mtp_num_depths == 0
+
+    def test_complex_patterns(self):
+        """Test more complex realistic patterns."""
+        test_cases = [
+            # Main decoder with attention, MTP with mamba only
+            ("M*M*M*M*/MMM/MMM", "M*M*M*M*", "MMM", 2),
+            # Main decoder with MLP, MTP with attention+mamba
+            ("MM-MM-/*M/*M", "MM-MM-", "*M", 2),
+            # All attention main, mamba MTP
+            ("*****/M/M/M/M", "*****", "M", 4),
+            # MoE in main pattern
+            ("MEME/MM/MM", "MEME", "MM", 2),
+        ]
+        for pattern, expected_main, expected_mtp, expected_depths in test_cases:
+            result = parse_hybrid_pattern(pattern)
+            assert result.main_pattern == expected_main, f"Failed for pattern: {pattern}"
+            assert result.mtp_pattern == expected_mtp, f"Failed for pattern: {pattern}"
+            assert result.mtp_num_depths == expected_depths, f"Failed for pattern: {pattern}"
+
+    def test_dataclass_equality(self):
+        """Test that ParsedHybridPattern supports equality comparison."""
+        p1 = parse_hybrid_pattern("M*M*/MM/MM")
+        p2 = ParsedHybridPattern(main_pattern="M*M*", mtp_pattern="MM", mtp_num_depths=2)
+        assert p1 == p2
diff --git a/tests/unit_tests/transformer/test_multi_token_prediction.py b/tests/unit_tests/transformer/test_multi_token_prediction.py
index ddfa9bfba16..18df40301c6 100644
--- a/tests/unit_tests/transformer/test_multi_token_prediction.py
+++ b/tests/unit_tests/transformer/test_multi_token_prediction.py
@@ -13,6 +13,8 @@
     get_gpt_mtp_block_spec,
 )
 from megatron.core.models.gpt.gpt_model import GPTModel
+from megatron.core.models.mamba.mamba_layer_specs import mamba_stack_spec
+from megatron.core.models.mamba.mamba_model import MambaModel
 from megatron.core.num_microbatches_calculator import destroy_num_microbatches_calculator
 from megatron.core.packed_seq_params import PackedSeqParams
 from megatron.core.parallel_state import get_context_parallel_group
@@ -94,7 +96,7 @@ def test_constructor_local(self, tp):
             assert mtp.layers[i].hnorm.weight.shape[0] == config.hidden_size
             assert mtp.layers[i].eh_proj.weight.shape[0] == config.hidden_size / tp
             assert mtp.layers[i].eh_proj.weight.shape[1] == config.hidden_size * 2
-            assert mtp.layers[i].transformer_layer is not None
+            assert mtp.layers[i].mtp_model_layer is not None
         num_weights = sum([p.numel() for p in mtp.parameters()])
         if tp == 1:
             assert num_weights == 58560 * config.mtp_num_layers
@@ -120,7 +122,7 @@ def test_constructor_ues_te(self, tp, cp):
             assert mtp.layers[i].hnorm.weight.shape[0] == config.hidden_size
             assert mtp.layers[i].eh_proj.weight.shape[0] == config.hidden_size / tp
             assert mtp.layers[i].eh_proj.weight.shape[1] == config.hidden_size * 2
-            assert mtp.layers[i].transformer_layer is not None
+            assert mtp.layers[i].mtp_model_layer is not None
         num_weights = sum([p.numel() for p in mtp.parameters()])
         if tp == 1:
             assert num_weights == 58560 * config.mtp_num_layers
@@ -162,7 +164,7 @@ def model_provider(
             config=config,
             transformer_layer_spec=transformer_layer_spec,
             mtp_block_spec=mtp_block_spec,
-            vocab_size=args.vocal_size,
+            vocab_size=args.vocab_size,
             max_sequence_length=args.max_position_embeddings,
             pre_process=pre_process,
             post_process=post_process,
@@ -186,7 +188,7 @@ def create_test_args(
         args.num_layers = 2
         args.mtp_num_layers = 2
         args.mtp_loss_scaling_factor = 0.1
-        args.vocal_size = 128800
+        args.vocab_size = 128800
         args.hidden_size = 128
         args.num_attention_heads = 8
         args.max_position_embeddings = 256
@@ -683,3 +685,256 @@ def log(self, metrics, iteration):
         assert torch.all(MTPLossLoggingHelper.tracker["values"] == 0)
         assert MTPLossLoggingHelper.tracker["reduce_group"] is None
         assert MTPLossLoggingHelper.tracker["avg_group"] is None
+
+
+class TestMultiTokenPredictionMamba:
+    """Test Multi-Token Prediction with Mamba hybrid models."""
+
+    def setup_method(self, method):
+        self.seq_length = 32
+        self.micro_batch_size = 2
+        os.environ['CUDA_DEVICE_MAX_CONNECTIONS'] = '1'
+
+    def teardown_method(self, method):
+        Utils.destroy_model_parallel()
+        destroy_global_vars()
+        destroy_num_microbatches_calculator()
+        MTPLossLoggingHelper.tracker = {}
+
+    def model_provider(self, pre_process=True, post_process=True, **config_kwargs):
+        """Model provider for Mamba hybrid models with MTP.
+
+        Uses the unified pattern syntax where MTP is configured via hybrid_override_pattern:
+        Format: "<main_pattern>/<mtp_pattern>/<mtp_pattern>/..."
+        Example: "M*M*/M*/M*" = main decoder "M*M*", MTP pattern "M*" with 2 depths
+        """
+        model_parallel_cuda_manual_seed(_SEED)
+        args = get_args()
+        config = core_transformer_config_from_args(args)
+
+        # MTP is configured via unified pattern in hybrid_override_pattern
+        # MambaModel creates the MTP block internally based on the parsed pattern
+        model = MambaModel(
+            config=config,
+            mamba_stack_spec=mamba_stack_spec,
+            vocab_size=args.vocab_size,
+            max_sequence_length=args.max_position_embeddings,
+            pre_process=pre_process,
+            post_process=post_process,
+            hybrid_attention_ratio=args.hybrid_attention_ratio,
+            hybrid_mlp_ratio=args.hybrid_mlp_ratio,
+            hybrid_override_pattern=args.hybrid_override_pattern,
+            fp16_lm_cross_entropy=args.fp16_lm_cross_entropy,
+            parallel_output=True,
+            share_embeddings_and_output_weights=not args.untie_embeddings_and_output_weights,
+            position_embedding_type=args.position_embedding_type,
+            rotary_percent=args.rotary_percent,
+        )
+        return model
+
+    def create_test_args(
+        self, tp, cp, sequence_length, micro_batch_size, fp8=None, full_recompute=False
+    ):
+        destroy_global_vars()
+        destroy_num_microbatches_calculator()
+
+        sys.argv = ['test_multi_token_prediction_mamba.py']
+        args = parse_args()
+        args.num_layers = 4
+        args.mtp_num_layers = 2
+        args.mtp_loss_scaling_factor = 0.1
+        args.vocab_size = 128800
+        args.hidden_size = 128
+        args.num_attention_heads = 8
+        args.num_query_groups = 8
+        args.mamba_num_groups = 4
+        args.max_position_embeddings = 256
+        args.micro_batch_size = micro_batch_size
+        args.create_attention_mask_in_dataloader = True
+        args.seq_length = sequence_length
+        args.tensor_model_parallel_size = tp
+        args.sequence_parallel = True if tp > 1 else False
+        args.context_parallel_size = cp
+        args.position_embedding_type = 'rope'
+        args.train_iters = 1
+        args.ckpt_format = 'torch_dist'
+        args.lr = 3e-5
+        args.attention_dropout = 0.0
+        args.hidden_dropout = 0.0
+        args.async_tensor_model_parallel_allreduce = False
+        args.no_save_optim = True
+        args.no_load_optim = True
+        args.no_load_rng = True
+        args.bf16 = True
+        args.hybrid_attention_ratio = 0.5
+        args.hybrid_mlp_ratio = 0.0
+        # Unified pattern: "main/mtp/mtp" - main decoder "M*M*", MTP pattern "M*" with 2 depths
+        args.hybrid_override_pattern = "M*M*/M*/M*"
+        args.spec = "megatron.core.models.mamba.mamba_layer_specs.mamba_stack_spec"
+
+        if fp8 is not None:
+            args.fp8 = 'e4m3'
+        if full_recompute:
+            args.recompute_granularity = 'full'
+            args.recompute_method = 'uniform'
+            args.recompute_num_layers = 1
+        else:
+            args.recompute_granularity = None
+        args.add_bias_linear = False
+        args.swiglu = True
+
+        validate_args(args)
+        set_global_variables(args, False)
+        return args
+
+    def get_batch(self, seq_length, micro_batch_size):
+        data = list(range(seq_length))
+        input_ids = torch.tensor(data, dtype=torch.int64).repeat((micro_batch_size, 1)).cuda()
+        labels = 1 + torch.tensor(data, dtype=torch.int64).repeat((micro_batch_size, 1)).cuda()
+        position_ids = torch.tensor(data, dtype=torch.int64).repeat((micro_batch_size, 1)).cuda()
+        attention_mask = torch.ones(
+            (micro_batch_size, 1, seq_length, seq_length), dtype=bool
+        ).cuda()
+        loss_mask = torch.ones(seq_length).repeat((micro_batch_size, 1)).cuda()
+        batch = {
+            'tokens': input_ids,
+            'labels': labels,
+            'loss_mask': loss_mask,
+            'attention_mask': attention_mask,
+            'position_ids': position_ids,
+        }
+        return batch
+
+    @pytest.mark.skipif(not HAVE_TE, reason="transformer_engine not available")
+    @pytest.mark.parametrize(("tp", "cp"), [(1, 1), (2, 1)])
+    def test_sharded_state_dict_mamba(self, tp, cp):
+        """Test MTP with Mamba hybrid model - sharded state dict."""
+        args = self.create_test_args(tp, cp, self.seq_length, self.micro_batch_size)
+        set_args(args)
+        torch.manual_seed(_SEED)
+        Utils.initialize_model_parallel(tensor_model_parallel_size=tp, context_parallel_size=cp)
+        mamba_model = get_model(self.model_provider, ModelType.encoder_or_decoder)
+        mamba_model = unwrap_model(mamba_model)
+        sharded_state_dict = mamba_model[0].sharded_state_dict()
+
+        # Verify MTP layers are in the state dict
+        for i in range(args.mtp_num_layers):
+            assert f"mtp.layers.{i}.enorm.weight" in sharded_state_dict.keys()
+            assert f"mtp.layers.{i}.hnorm.weight" in sharded_state_dict.keys()
+            assert f"mtp.layers.{i}.eh_proj.weight" in sharded_state_dict.keys()
+
+    @pytest.mark.skipif(not HAVE_TE, reason="transformer_engine not available")
+    @pytest.mark.parametrize(("tp", "cp"), [(1, 1), (2, 1)])
+    def test_forward_backward_mamba(self, tmp_path_dist_ckpt, tp, cp):
+        """Test MTP forward and backward with Mamba hybrid model."""
+        tp_ref = 1
+        cp_ref = 1
+        args = self.create_test_args(tp_ref, cp_ref, self.seq_length, self.micro_batch_size)
+        set_args(args)
+        torch.manual_seed(_SEED)
+        Utils.initialize_model_parallel(
+            tensor_model_parallel_size=tp_ref, context_parallel_size=cp_ref
+        )
+        batch = self.get_batch(self.seq_length, self.micro_batch_size)
+        tokens, labels, loss_mask, attention_mask, position_ids = batch.values()
+
+        mamba_model_ref, optimizer, opt_param_scheduler = setup_model_and_optimizer(
+            self.model_provider, ModelType.encoder_or_decoder
+        )
+
+        output_ref = mamba_model_ref[0].forward(
+            input_ids=tokens,
+            position_ids=position_ids,
+            attention_mask=attention_mask,
+            labels=labels,
+            # MTP loss/block are gated on mtp_labels (RL passes labels=None for the
+            # PG path), so MTP only runs when mtp_labels is provided.
+            mtp_labels=labels,
+            loss_mask=loss_mask,
+        )
+        tracker = MTPLossLoggingHelper.tracker
+        mtp_loss_ref = None
+        assert "values" in tracker
+        mtp_loss_ref = tracker['values'].clone()
+        MTPLossLoggingHelper.clean_loss_in_tracker()
+
+        iteration = 123
+        num_floating_point_operations_so_far = 456
+
+        def set_ckpt_path(ckpt_path):
+            args.save = ckpt_path
+            args.load = ckpt_path
+
+        with TempNamedDir(tmp_path_dist_ckpt / 'test_mtp_mamba_model_reconfiguration') as ckpt_dir:
+            set_ckpt_path(ckpt_dir)
+            save_checkpoint(
+                iteration,
+                mamba_model_ref,
+                optimizer,
+                opt_param_scheduler,
+                num_floating_point_operations_so_far,
+            )
+
+            expected_ckpt_path = args.save / "iter_0000123" / ".metadata"
+            assert os.path.exists(expected_ckpt_path)
+
+            Utils.destroy_model_parallel()
+            args = self.create_test_args(tp, cp, self.seq_length, self.micro_batch_size)
+            set_args(args)
+            set_ckpt_path(ckpt_dir)
+            torch.manual_seed(_SEED)
+            Utils.initialize_model_parallel(tensor_model_parallel_size=tp, context_parallel_size=cp)
+            mamba_model, optimizer, opt_param_scheduler = setup_model_and_optimizer(
+                self.model_provider, ModelType.encoder_or_decoder
+            )
+            load_checkpoint(mamba_model, optimizer, opt_param_scheduler, strict=False)
+
+            batch["output_ref"] = output_ref
+            batch = get_batch_on_this_cp_rank(batch)
+            tokens, labels, loss_mask, attention_mask, position_ids, output_ref = batch.values()
+            output = mamba_model[0].forward(
+                input_ids=tokens,
+                position_ids=position_ids,
+                attention_mask=attention_mask,
+                labels=labels,
+                mtp_labels=labels,
+                loss_mask=loss_mask,
+            )
+            tracker = MTPLossLoggingHelper.tracker
+            assert "values" in tracker
+            mtp_loss = tracker['values'].clone()
+            pg_collection = ProcessGroupCollection.use_mpu_process_groups(required_pgs=['cp'])
+            torch.distributed.all_reduce(
+                mtp_loss, group=pg_collection.cp, op=torch.distributed.ReduceOp.AVG
+            )
+            MTPLossLoggingHelper.clean_loss_in_tracker()
+            assert torch.allclose(output_ref, output, rtol=1e-03, atol=1e-03)
+            assert torch.allclose(mtp_loss, mtp_loss_ref, rtol=1e-02, atol=1e-02)
+
+            assert output.shape[0] == self.micro_batch_size
+            assert output.shape[1] == self.seq_length / cp
+
+            loss = output.mean()
+            loss.backward()
+            for name, param in mamba_model[0].named_parameters():
+                assert param.main_grad is not None
+
+    @pytest.mark.skipif(not HAVE_TE, reason="transformer_engine not available")
+    def test_attention_mask_validation_mamba(self):
+        """Test that attention mask type validation works for Mamba hybrid models."""
+        tp = 1
+        cp = 1
+        args = self.create_test_args(tp, cp, self.seq_length, self.micro_batch_size)
+        set_args(args)
+        torch.manual_seed(_SEED)
+        Utils.initialize_model_parallel(tensor_model_parallel_size=tp, context_parallel_size=cp)
+        try:
+            mamba_model = get_model(self.model_provider, ModelType.encoder_or_decoder)
+            mamba_model = unwrap_model(mamba_model)
+            assert isinstance(mamba_model[0], MambaModel)
+            assert mamba_model[0].mtp is not None
+        except AssertionError as e:
+            if "Multi-Token Prediction (MTP) is not yet supported" in str(e):
+                pytest.fail(f"Attention mask validation failed for Mamba hybrid model: {e}")
+            else:
+                raise