diff --git a/megatron/core/distributed/finalize_model_grads.py b/megatron/core/distributed/finalize_model_grads.py
index ca6bdd354ce..540dbbd51c5 100644
--- a/megatron/core/distributed/finalize_model_grads.py
+++ b/megatron/core/distributed/finalize_model_grads.py
@@ -281,7 +281,11 @@ def reset_model_temporary_tensors(config: TransformerConfig, model: List[torch.n
     """
     for model_chunk in model:
         for module in get_attr_wrapped_model(model_chunk, 'modules')():
-            if config.moe_router_enable_expert_bias and hasattr(module, 'expert_bias'):
+            if (
+                config.moe_router_enable_expert_bias
+                and hasattr(module, 'expert_bias')
+                and module.expert_bias is not None
+            ):
                 module.local_tokens_per_expert.zero_()
             if (
                 config.moe_router_load_balancing_type == "global_aux_loss"
@@ -303,7 +307,11 @@ def _update_router_expert_bias(model: List[torch.nn.Module], config: Transformer
             # cases where only the student is in training mode but the teacher is in eval mode
             # when using online knoweldge-distillation with Model-Optimizer. In this case, we want
             # to avoid updating teacher's expert_bias.
-            if hasattr(module, 'expert_bias') and module.training:
+            if (
+                hasattr(module, 'expert_bias')
+                and module.training
+                and module.expert_bias is not None
+            ):
                 tokens_per_expert_list.append(module.local_tokens_per_expert)
                 expert_bias_list.append(module.expert_bias)
     # For hybrid models with both MoE and Dense layers, this list can be empty.
diff --git a/megatron/core/fusions/fused_bias_swiglu.py b/megatron/core/fusions/fused_bias_swiglu.py
index 632470876c9..ec195551ffa 100644
--- a/megatron/core/fusions/fused_bias_swiglu.py
+++ b/megatron/core/fusions/fused_bias_swiglu.py
@@ -48,6 +48,23 @@ def weighted_swiglu(y, weights):
     return res.to(dtype)
 
 
+@jit_fuser
+def clamped_swiglu(y, clamp_value):
+    dtype = y.dtype
+    y_1, y_2 = torch.chunk(y.to(torch.float32), 2, -1)
+    y_1 = y_1.clamp(min=None, max=clamp_value)
+    y_2 = y_2.clamp(min=-clamp_value, max=clamp_value)
+    res = F.silu(y_1) * y_2
+    return res.to(dtype)
+
+
+@jit_fuser
+def clamped_weighted_swiglu(y, weights, clamp_value):
+    dtype = y.dtype
+    res = clamped_swiglu(y, clamp_value) * weights
+    return res.to(dtype)
+
+
 # gradient of tanh approximation of gelu
 # gradient of actual gelu is:
 # 0.5 * (1. + torch.erf(x * 0.70710678)) + 0.3989423 * x * torch.exp(-0.5 * x * x)
@@ -97,6 +114,36 @@ def weighted_swiglu_back(g, y, weights):
     return input_grad.to(input_dtype), weights_grad.to(w_dtype)
 
 
+@jit_fuser
+def clamped_swiglu_back(g, y, clamp_value):
+    dtype = y.dtype
+    y_1, y_2 = torch.chunk(y.to(torch.float32), 2, -1)
+    y_1c = y_1.clamp(min=None, max=clamp_value)
+    y_2c = y_2.clamp(min=-clamp_value, max=clamp_value)
+    res = torch.cat(
+        (
+            g
+            * torch.sigmoid(y_1c)
+            * (1 + y_1c * (1 - torch.sigmoid(y_1c)))
+            * y_2c
+            * (y_1 <= clamp_value).to(g.dtype),
+            g * F.silu(y_1c) * ((y_2 >= -clamp_value) & (y_2 <= clamp_value)).to(g.dtype),
+        ),
+        -1,
+    )
+    return res.to(dtype)
+
+
+@jit_fuser
+def clamped_weighted_swiglu_back(g, y, weights, clamp_value):
+    input_dtype = y.dtype
+    w_dtype = weights.dtype
+    input_grad = clamped_swiglu_back(g * weights, y, clamp_value)
+    weights_grad = clamped_swiglu(y, clamp_value) * g.to(w_dtype)
+    weights_grad = torch.sum(weights_grad, dim=-1, keepdim=True)
+    return input_grad.to(input_dtype), weights_grad.to(w_dtype)
+
+
 class BiasSwiGLUFunction(torch.autograd.Function):
     """Custom autograd function for SwiGLU activation with bias support."""
 
@@ -190,20 +237,27 @@ def backward(ctx, grad_output):
 
 class WeightedSwiGLUFunction(torch.autograd.Function):
     @staticmethod
-    # bias is an optional argument
-    def forward(ctx, input, weights, fp8_input_store):
+    def forward(ctx, input, weights, fp8_input_store, clamp_value):
         input_for_backward = input.to(torch.float8_e4m3fn) if fp8_input_store else input
         ctx.save_for_backward(input_for_backward, weights)
         ctx.ori_input_dtype = input.dtype
         ctx.fp8_input_store = fp8_input_store
-        return weighted_swiglu(input, weights)
+        ctx.clamp_value = clamp_value
+        if clamp_value is not None and clamp_value > 0:
+            res = clamped_weighted_swiglu(input, weights, clamp_value)
+        else:
+            res = weighted_swiglu(input, weights)
+        return res
 
     @staticmethod
     def backward(ctx, grad_output):
         input, weights = ctx.saved_tensors
         input = input.to(ctx.ori_input_dtype) if ctx.fp8_input_store else input
-        tmp, wgrad = weighted_swiglu_back(grad_output, input, weights)
-        return tmp, wgrad, None
+        if ctx.clamp_value is not None and ctx.clamp_value > 0:
+            tmp, wgrad = clamped_weighted_swiglu_back(grad_output, input, weights, ctx.clamp_value)
+        else:
+            tmp, wgrad = weighted_swiglu_back(grad_output, input, weights)
+        return tmp, wgrad, None, None
 
 
 def bias_swiglu_impl(input, bias, fp8_input_store=False, cpu_offload_input=False):
@@ -236,7 +290,7 @@ def bias_swiglu_impl(input, bias, fp8_input_store=False, cpu_offload_input=False
     return output if len(ori_shape) == 2 else output.view(ori_shape[0], ori_shape[1], -1)
 
 
-def weighted_bias_swiglu_impl(input, bias, weights, fp8_input_store=False):
+def weighted_bias_swiglu_impl(input, bias, weights, fp8_input_store=False, clamp_value=None):
     """
     Token-wise-weighted bias swiglu fusion.
     """
@@ -246,7 +300,7 @@ def weighted_bias_swiglu_impl(input, bias, weights, fp8_input_store=False):
     if bias is not None:
         raise NotImplementedError("Bias is not supported for weighted swiglu fusion")
     else:
-        output = WeightedSwiGLUFunction.apply(input, weights, fp8_input_store)
+        output = WeightedSwiGLUFunction.apply(input, weights, fp8_input_store, clamp_value)
 
     return output if len(ori_shape) == 2 else output.view(ori_shape[0], ori_shape[1], -1)
 
diff --git a/megatron/core/models/gpt/gpt_model.py b/megatron/core/models/gpt/gpt_model.py
index 19de0ed5298..7bc92069e3a 100644
--- a/megatron/core/models/gpt/gpt_model.py
+++ b/megatron/core/models/gpt/gpt_model.py
@@ -551,6 +551,11 @@ def forward(
 
         rotary_pos_cos_sin = preproc_output[6] if len(preproc_output) == 7 else None
 
+        # Pass input_ids to decoder for hash-based MoE routing
+        decoder_extra_block_kwargs = extra_block_kwargs or {}
+        if self.config.moe_n_hash_layers > 0 and input_ids is not None:
+            decoder_extra_block_kwargs['input_ids'] = input_ids
+
         # Run decoder.
         hidden_states = self.decoder(
             hidden_states=decoder_input,
@@ -563,7 +568,7 @@ def forward(
             packed_seq_params=packed_seq_params,
             sequence_len_offset=sequence_len_offset,
             padding_mask=padding_mask,
-            **(extra_block_kwargs or {}),
+            **decoder_extra_block_kwargs,
         )
 
         return self._postprocess(
diff --git a/megatron/core/transformer/moe/experts.py b/megatron/core/transformer/moe/experts.py
index 9efff3189ac..384a26c0deb 100644
--- a/megatron/core/transformer/moe/experts.py
+++ b/megatron/core/transformer/moe/experts.py
@@ -1195,6 +1195,7 @@ def remove_glu_interleaving(x: torch.Tensor) -> torch.Tensor:
                         bias_parallel,
                         permuted_probs,
                         self.config.activation_func_fp8_input_store,
+                        self.config.activation_func_clamp_value,
                     )
                 elif self.activation_func == quick_gelu and self.config.gated_linear_unit:
                     intermediate_parallel = weighted_bias_quick_geglu_impl(
diff --git a/megatron/core/transformer/moe/moe_layer.py b/megatron/core/transformer/moe/moe_layer.py
index 695e64c9681..11a4bd1a8b2 100644
--- a/megatron/core/transformer/moe/moe_layer.py
+++ b/megatron/core/transformer/moe/moe_layer.py
@@ -240,7 +240,10 @@ def __init__(
 
         # Initialize router.
         self.router = self.submodules.router(
-            config=self.config, pg_collection=pg_collection, is_mtp_layer=is_mtp_layer
+            config=self.config,
+            pg_collection=pg_collection,
+            is_mtp_layer=is_mtp_layer,
+            layer_number=layer_number,
         )
         self.tp_group = pg_collection.tp
 
@@ -421,13 +424,18 @@ def unset_inference_cuda_graphed_iteration(self):
             self.shared_expert_overlap = self._saved_shared_expert_overlap
 
     @maybe_skip_or_early_return_by_cudagraph("route")
-    def route(self, hidden_states: torch.Tensor, padding_mask: Optional[torch.Tensor] = None):
+    def route(
+        self,
+        hidden_states: torch.Tensor,
+        padding_mask: Optional[torch.Tensor] = None,
+        input_ids: Optional[torch.Tensor] = None,
+    ):
         """Compute token routing for preprocessing.
 
         This method uses the router to determine which experts to send each token to,
         producing routing probabilities and a mapping.
         """
-        probs, routing_map = apply_module(self.router)(hidden_states, padding_mask)
+        probs, routing_map = apply_module(self.router)(hidden_states, padding_mask, input_ids)
         return probs, routing_map
 
     @maybe_skip_or_early_return_by_cudagraph("preprocess")
@@ -599,6 +607,7 @@ def forward(
         hidden_states: torch.Tensor,
         intermediate_tensors=None,
         padding_mask: Optional[torch.Tensor] = None,
+        input_ids: Optional[torch.Tensor] = None,
     ):
         """Forward pass for the MoE layer.
 
@@ -613,6 +622,8 @@ def forward(
             padding_mask (torch.Tensor, optional): Boolean mask indicating non-padding tokens.
                                                    Shape [seq_length, bsz]. True for valid tokens,
                                                    False for padding tokens. Defaults to None.
+            input_ids (torch.Tensor, optional): The input IDs tensor. Shape [seq_length, bsz].
+                                                Defaults to None.
         Returns:
             A tuple containing the output tensor and the MLP bias, if any.
         """
@@ -634,7 +645,7 @@ def custom_forward(hidden_states, intermediate_tensors=None, padding_mask=None):
                         self._overload_log_num_local_tokens = (
                             self._num_token_rows_from_moe_hidden_states(hidden_states)
                         )
-                    probs, routing_map = self.route(hidden_states, padding_mask)
+                    probs, routing_map = self.route(hidden_states, padding_mask, input_ids)
                     hidden_states, probs = self.preprocess(hidden_states, probs, routing_map)
 
                     if intermediate_tensors is not None:
diff --git a/megatron/core/transformer/moe/router.py b/megatron/core/transformer/moe/router.py
index b675d33cd21..cdf968c6b12 100644
--- a/megatron/core/transformer/moe/router.py
+++ b/megatron/core/transformer/moe/router.py
@@ -34,6 +34,7 @@ def __init__(
         config: TransformerConfig,
         pg_collection: Optional[ProcessGroupCollection] = None,
         is_mtp_layer: bool = False,
+        layer_number: Optional[int] = None,
     ) -> None:
         """
         Initialize the Router module.
@@ -47,7 +48,7 @@ def __init__(
         self.config = config
         self.num_experts = self.config.num_moe_experts
         self.moe_aux_loss_func = None
-        self.layer_number = None
+        self.layer_number = layer_number
         self.is_mtp_layer = is_mtp_layer
         self.tp_group = pg_collection.tp
         self.cp_group = pg_collection.cp
@@ -155,6 +156,7 @@ def __init__(
         config: TransformerConfig,
         pg_collection: Optional[ProcessGroupCollection] = None,
         is_mtp_layer: bool = False,
+        layer_number: Optional[int] = None,
     ) -> None:
         """Initialize the zero token dropping router.
 
@@ -163,13 +165,41 @@ def __init__(
             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=config, pg_collection=pg_collection, is_mtp_layer=is_mtp_layer)
+        super().__init__(
+            config=config,
+            pg_collection=pg_collection,
+            is_mtp_layer=is_mtp_layer,
+            layer_number=layer_number,
+        )
         self.topk = self.config.moe_router_topk
         self.routing_type = self.config.moe_router_load_balancing_type
         self.score_function = self.config.moe_router_score_function
         self.input_jitter = None
 
-        self.enable_expert_bias = self.config.moe_router_enable_expert_bias
+        if self.config.moe_n_hash_layers > 0:
+            assert layer_number is not None, "layer_number is required for the hash-based router."
+        self.is_hash_layer = (
+            not self.is_mtp_layer
+            and self.config.moe_n_hash_layers > 0
+            and layer_number <= self.config.moe_n_hash_layers
+        )
+        if self.is_hash_layer:
+            # DSv4-Pro ships a pre-trained tid2eid table in its inference checkpoint;
+            # no public initialization recipe is documented. Round-robin is used here
+            # only as a placeholder so the layer is runnable from scratch.
+            vocab_size = self.config.actual_vocab_size
+            num_experts = self.config.num_moe_experts
+            ids = torch.arange(vocab_size, device=torch.cuda.current_device())
+            tid2eid = torch.stack([(ids + k) % num_experts for k in range(self.topk)], dim=1).to(
+                torch.int32
+            )
+            self.register_buffer('tid2eid', tid2eid)
+        else:
+            self.tid2eid = None
+
+        self.enable_expert_bias = (
+            self.config.moe_router_enable_expert_bias and not self.is_hash_layer
+        )
         if self.enable_expert_bias:
             self.register_buffer(
                 'local_tokens_per_expert',
@@ -583,7 +613,53 @@ def _apply_expert_bias(
                     routing_map = routing_map & (~padding_mask)
                 self.local_tokens_per_expert += routing_map.sum(dim=0)
 
-    def routing(self, logits: torch.Tensor, padding_mask: Optional[torch.Tensor] = None):
+    def _hash_routing(self, logits: torch.Tensor, input_ids: torch.Tensor):
+        """Hash-based routing: expert indices come from the tid2eid lookup table.
+
+        Scores are still computed from the gating logits for weight computation,
+        but expert selection is determined by the pre-computed hash table.
+
+        Args:
+            logits (torch.Tensor): Gating logits, shape [num_tokens, num_experts].
+            input_ids (torch.Tensor): Token IDs, shape [seq_length, bsz].
+
+        Returns:
+            Tuple[torch.Tensor, torch.Tensor]: routing_probs and routing_map.
+        """
+        num_tokens, num_experts = logits.shape
+
+        if self.score_function == "softmax":
+            scores = torch.softmax(logits, dim=-1, dtype=torch.float32).type_as(logits)
+        elif self.score_function == "sigmoid":
+            scores = torch.sigmoid(logits.float()).type_as(logits)
+        elif self.score_function == "sqrtsoftplus":
+            scores = torch.nn.functional.softplus(logits.float()).sqrt().type_as(logits)
+        else:
+            raise ValueError(f"Invalid score_function: {self.score_function}")
+
+        # input_ids is [b, s] from the model, but hidden_states are [s, b, h]
+        # and get flattened to [s*b, h]. Transpose to match.
+        flat_ids = input_ids.T.reshape(-1)
+        top_indices = self.tid2eid[flat_ids].long()  # [num_tokens, topk]
+
+        probs = scores.gather(1, top_indices)
+        if self.score_function != "softmax":
+            probs = probs / (probs.sum(dim=-1, keepdim=True) + 1e-20)
+
+        if self.config.moe_router_topk_scaling_factor:
+            probs = probs * self.config.moe_router_topk_scaling_factor
+
+        routing_probs = torch.zeros_like(logits).scatter(1, top_indices, probs)
+        routing_map = torch.zeros_like(logits).int().scatter(1, top_indices, 1).bool()
+
+        return routing_probs, routing_map
+
+    def routing(
+        self,
+        logits: torch.Tensor,
+        padding_mask: Optional[torch.Tensor] = None,
+        input_ids: Optional[torch.Tensor] = None,
+    ):
         """Top-k routing function
 
         Args:
@@ -591,6 +667,8 @@ def routing(self, logits: torch.Tensor, padding_mask: Optional[torch.Tensor] = N
             padding_mask (torch.Tensor, optional): Boolean mask indicating non-padding tokens.
                                                    Shape [seq_length, bsz]. True for valid tokens,
                                                    False for padding tokens. Defaults to None.
+            input_ids (torch.Tensor, optional): The input IDs tensor. Shape [seq_length, bsz].
+                                                Defaults to None.
 
         Returns:
             probs (torch.Tensor): The probabilities of token to experts assignment.
@@ -608,7 +686,13 @@ def routing(self, logits: torch.Tensor, padding_mask: Optional[torch.Tensor] = N
         logits = self.apply_z_loss(logits, padding_mask=padding_mask)
 
         # Calculate probs and routing_map for token dispatching
-        if self.routing_type == "sinkhorn":
+        if self.is_hash_layer:
+            assert input_ids is not None, (
+                "input_ids is required for hash-based routing but was None. "
+                "Ensure --moe-n-hash-layers is set correctly and input_ids are passed."
+            )
+            probs, routing_map = self._hash_routing(logits, input_ids)
+        elif self.routing_type == "sinkhorn":
             probs, routing_map = self.sinkhorn_load_balancing(logits)
         else:
             probs, routing_map = topk_routing_with_score_function(
@@ -677,7 +761,12 @@ def reset_global_aux_loss_tracker(self):
             self.global_tokens_per_expert.zero_()
             self.ga_steps.zero_()
 
-    def forward(self, input: torch.Tensor, padding_mask: Optional[torch.Tensor] = None):
+    def forward(
+        self,
+        input: torch.Tensor,
+        padding_mask: Optional[torch.Tensor] = None,
+        input_ids: Optional[torch.Tensor] = None,
+    ):
         """
         Forward pass of the router.
 
@@ -686,6 +775,8 @@ def forward(self, input: torch.Tensor, padding_mask: Optional[torch.Tensor] = No
             padding_mask (torch.Tensor, optional): Boolean mask indicating non-padding tokens.
                                                    Shape [seq_length, bsz]. True for valid tokens,
                                                    False for padding tokens. Defaults to None.
+            input_ids (torch.Tensor, optional): The input IDs tensor. Shape [seq_length, bsz].
+                                                Defaults to None.
         """
         self._maintain_float32_expert_bias()
 
@@ -703,7 +794,7 @@ def forward(self, input: torch.Tensor, padding_mask: Optional[torch.Tensor] = No
                 logits, self.config.moe_router_force_biased, self.layer_number
             )
 
-        probs, routing_map = self.routing(logits, padding_mask=padding_mask)
+        probs, routing_map = self.routing(logits, padding_mask=padding_mask, input_ids=input_ids)
 
         return probs, routing_map
 
diff --git a/megatron/core/transformer/transformer_block.py b/megatron/core/transformer/transformer_block.py
index 89d0b2cb75e..3ddc3415e47 100755
--- a/megatron/core/transformer/transformer_block.py
+++ b/megatron/core/transformer/transformer_block.py
@@ -455,6 +455,7 @@ def _checkpointed_forward(
         padding_mask: Optional[Tensor] = None,
         extract_layer_indices: Optional[Set[int]] = None,
         layer_offset: int = 0,
+        input_ids: Optional[Tensor] = None,
     ):
         """Forward method with activation checkpointing.
 
@@ -513,6 +514,7 @@ def custom_forward(
                             inference_context=None,
                             packed_seq_params=packed_seq_params,
                             padding_mask=padding_mask,
+                            input_ids=input_ids,
                         )
                 return hidden_states, context
 
@@ -701,6 +703,7 @@ def forward(
         sequence_len_offset: Optional[Tensor] = None,
         padding_mask: Optional[Tensor] = None,
         extract_layer_indices: Optional[Set[int]] = None,
+        input_ids: Optional[Tensor] = None,
         *,
         inference_params: Optional[BaseInferenceContext] = None,
         dynamic_inference_decode_only: Optional[bool] = None,
@@ -855,6 +858,7 @@ def forward(
                     padding_mask=padding_mask,
                     extract_layer_indices=extract_layer_indices,
                     layer_offset=layer_offset,
+                    input_ids=input_ids,
                 )
                 # Handle return value from _checkpointed_forward
                 if len(extract_layer_indices) > 0:
@@ -902,6 +906,7 @@ def forward(
                             sequence_len_offset=sequence_len_offset,
                             padding_mask=padding_mask,
                             mhc_recompute_manager=mhc_manager,
+                            input_ids=input_ids,
                         )
                     self._finalize_mhc_recompute_layer(
                         mhc_manager=mhc_manager,
diff --git a/megatron/core/transformer/transformer_config.py b/megatron/core/transformer/transformer_config.py
index b41203eb0a1..4251ee9d8a1 100644
--- a/megatron/core/transformer/transformer_config.py
+++ b/megatron/core/transformer/transformer_config.py
@@ -11,7 +11,7 @@
 
 from megatron.core.enums import Fp4Recipe, Fp8Recipe
 from megatron.core.quantization.quant_config import RecipeConfig
-from megatron.core.transformer.enums import AttnBackend, CudaGraphScope
+from megatron.core.transformer.enums import AttnBackend, CudaGraphScope, LayerType
 from megatron.core.transformer.pipeline_parallel_layer_layout import PipelineParallelLayerLayout
 from megatron.core.utils import experimental_api
 
@@ -200,7 +200,7 @@ class TransformerConfig(ModelParallelConfig):
 
     activation_func_clamp_value: Optional[float] = None
     """Clamp the output of the linear_fc1 in the activation function. Only used when activation_func
-    is quick_gelu."""
+    is quick_gelu or weighted SwiGLU (MoE only)."""
 
     num_moe_experts: Optional[int] = None
     """Number of experts to use for MoE layer. When set, it replaces MLP with MoE layer. Set to None
@@ -742,6 +742,15 @@ class TransformerConfig(ModelParallelConfig):
     If negative, generates bias once per layer and reuses it (abs value is std).
     This is an experimental feature for benchmarking purposes."""
 
+    moe_n_hash_layers: int = 0
+    """Number of leading transformer layers that use hash-based MoE routing.
+    Layers with layer_number <= moe_n_hash_layers use a pre-computed tid2eid
+    lookup table for expert selection instead of learned top-k routing."""
+
+    actual_vocab_size: Optional[int] = None
+    """Padded actual vocabulary size. Required when moe_n_hash_layers > 0 for the
+    tid2eid lookup buffer in hash-based MoE routing."""
+
     dense_grouped_gemm: bool = False
     """Use GroupedLinear(num_groups=1) for dense MLP to trigger the
     ForwardGroupedMLP_CuTeGEMMSwiGLU_MXFP8 fusion on SM100+ with MXFP8 recipe.
@@ -2000,6 +2009,24 @@ def __post_init__(self):
             if self.activation_func != F.silu or not self.gated_linear_unit:
                 raise ValueError("Storing activation input in FP8 is supported only for SwiGLU.")
 
+        if self.activation_func_clamp_value is not None:
+            # swiglu
+            if self.activation_func == F.silu and self.gated_linear_unit:
+                if self.num_moe_experts is None:
+                    raise ValueError(
+                        "activation_func_clamp_value for SwiGLU is only supported with MoE."
+                    )
+                if self.use_te_activation_func:
+                    raise ValueError(
+                        "use_te_activation_func must be False "
+                        "when activation_func_clamp_value is not None for SwiGLU"
+                    )
+                if self.use_transformer_engine_op_fuser:
+                    raise ValueError(
+                        "use_transformer_engine_op_fuser must be False "
+                        "when activation_func_clamp_value is not None for SwiGLU"
+                    )
+
         if self.apply_rope_fusion:
             if self.multi_latent_attention:
                 warnings.warn(
@@ -2129,6 +2156,38 @@ def __post_init__(self):
                 "'sqrtsoftplus', or unset --moe-router-enable-expert-bias."
             )
 
+        if self.moe_n_hash_layers > 0:
+            assert (
+                self.actual_vocab_size is not None
+            ), "actual_vocab_size must be set when moe_n_hash_layers > 0."
+            if self.pipeline_model_parallel_size > 1:
+                assert self.pipeline_model_parallel_layout is not None, (
+                    "pipeline_model_parallel_layout must be set when using hash MoE "
+                    "layers with pipeline parallelism (PP > 1)."
+                )
+                # The embedding is always in layout[0][0] (PP rank 0, VPP rank 0).
+                # All hash MoE layers must be in the same virtual pipeline stage.
+                embedding_stage = self.pipeline_model_parallel_layout.layout[0][0]
+                n_decoders_with_embedding = embedding_stage.count(LayerType.decoder)
+                assert self.moe_n_hash_layers <= n_decoders_with_embedding, (
+                    f"Currently, All hash MoE layers must be in the same virtual pipeline stage "
+                    f"as the embedding. The embedding stage has "
+                    f"{n_decoders_with_embedding} decoder layers, but "
+                    f"moe_n_hash_layers={self.moe_n_hash_layers}."
+                )
+            assert (
+                not self.overlap_moe_expert_parallel_comm
+            ), "overlap_moe_expert_parallel_comm does not support moe_n_hash_layers > 0 for now."
+            log_single_rank(
+                logger,
+                logging.WARNING,
+                f"Hash MoE layer initialized with placeholder round-robin tid2eid. "
+                f"For real training, you MUST either (a) load tid2eid from a "
+                f"pre-trained DSv4 checkpoint, or (b) provide a frequency-aware "
+                f"initialization (e.g., Sinkhorn-balanced over token frequency). "
+                f"Round-robin will cause severe expert imbalance.",
+            )
+
         if self.num_moe_experts and self.fp8:
             # TE version below 1.7.0 will raise Error when handle zeros tokens for expert
             if not is_te_min_version("1.7.0.dev0"):
diff --git a/megatron/core/transformer/transformer_layer.py b/megatron/core/transformer/transformer_layer.py
index ee20545110c..5392b87f7dd 100644
--- a/megatron/core/transformer/transformer_layer.py
+++ b/megatron/core/transformer/transformer_layer.py
@@ -390,9 +390,11 @@ def __init__(
         if isinstance(submodules.mlp, ModuleSpec):
             if submodules.mlp.module in (MoELayer, 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:
+                    # Pass is_mtp_layer flag to MoELayer to distinguish MTP MoE layers.
                     additional_mlp_kwargs["is_mtp_layer"] = self.is_mtp_layer
+                    # Pass layer number to MoELayer for router configuration.
+                    additional_mlp_kwargs["layer_number"] = self.layer_number
             elif submodules.mlp.module == MLP:
                 assert hasattr(
                     pg_collection, 'tp'
@@ -410,8 +412,6 @@ def __init__(
                     f"Unknown MLP type: {type(submodules.mlp)}. Using default kwargs.",
                 )
         self.mlp = build_module(submodules.mlp, config=self.config, **additional_mlp_kwargs)
-        if hasattr(self.mlp, 'set_layer_number'):
-            self.mlp.set_layer_number(self.layer_number)
 
         # [Module 9: BiasDropoutFusion]
         self.mlp_bda = build_module(submodules.mlp_bda)
@@ -550,6 +550,7 @@ def _forward_attention(
         packed_seq_params: Optional[PackedSeqParams] = None,
         sequence_len_offset: Optional[Tensor] = None,
         padding_mask: Optional[Tensor] = None,
+        input_ids: Optional[Tensor] = None,
         *,
         inference_params: Optional[Any] = None,
     ):
@@ -714,6 +715,7 @@ def forward(self, *args, **kwargs):
             hidden_states,
             kwargs.get("inference_context", None),
             padding_mask=kwargs.get("padding_mask", None),
+            input_ids=kwargs.get("input_ids", None),
         )
         return output, context
 
@@ -736,6 +738,7 @@ def _forward_mlp(
         hidden_states: Tensor,
         inference_context: BaseInferenceContext | None = None,
         padding_mask: Tensor | None = None,
+        input_ids: Optional[Tensor] = None,
     ) -> Tensor | list[Tensor | None]:
         """
         Perform a forward pass through the feed-forward layer.
@@ -748,6 +751,8 @@ def _forward_mlp(
                 Shape [bsz, seq_length]. True = padding (exclude), False = valid (include).
                 Only used for MoE layers to exclude padding tokens from aux loss computations.
                 The MoELayer will internally transform this to [seq_length, bsz] format.
+            input_ids (Tensor, optional): The input IDs tensor. Shape [seq_length, bsz].
+                Only used for hash-based MoE routing. Defaults to None.
         Returns:
             output (Tensor): Transformed hidden states of shape [s, b, h].
         """
@@ -784,6 +789,10 @@ def _forward_mlp(
             self.config.inference_fuse_tp_communication
         )
 
+        moe_kwargs = {}
+        if self.is_moe_layer and input_ids is not None:
+            moe_kwargs["input_ids"] = input_ids
+
         if self.recompute_mlp:
             if self.config.fp8 or self.config.fp4:
                 # import here to avoid circular import
@@ -796,10 +805,11 @@ def _forward_mlp(
                     self.pg_collection.tp,
                     pre_mlp_layernorm_output,
                     padding_mask=padding_mask,
+                    **moe_kwargs,
                 )
             else:
                 mlp_output_with_bias = tensor_parallel.checkpoint(
-                    functools.partial(self.mlp, padding_mask=padding_mask),
+                    functools.partial(self.mlp, padding_mask=padding_mask, **moe_kwargs),
                     False,
                     pre_mlp_layernorm_output,
                 )
@@ -821,7 +831,9 @@ def _forward_mlp(
                 # Set the residual for fused reduce-scatter + add + layer-norm + all-gather
                 # operation in MLP's fc2.
                 self._set_fc2_residual(residual)
-            mlp_output_with_bias = self.mlp(pre_mlp_layernorm_output, padding_mask=padding_mask)
+            mlp_output_with_bias = self.mlp(
+                pre_mlp_layernorm_output, padding_mask=padding_mask, **moe_kwargs
+            )
 
         nvtx_range_pop(suffix="mlp")
 
@@ -1005,6 +1017,19 @@ def get_layer_static_inputs(self, seq_length, micro_batch_size):
                 .reshape(1, 1, slen_per_cp, seq_length)
                 .tile(micro_batch_size, 1, 1, 1)
             )
+
+        # Add input_ids for hash-based MoE routing under CUDA graphs.
+        # Only add for layers that actually use hash routing,
+        # since other layers (e.g. on later PP stages) receive input_ids=None.
+        if (
+            self.is_moe_layer
+            and self.config.moe_n_hash_layers > 0
+            and getattr(self.mlp.router, 'is_hash_layer', False)
+        ):
+            static_inputs["input_ids"] = torch.zeros(
+                (micro_batch_size, seq_length), dtype=torch.long, device=torch.cuda.current_device()
+            )
+
         return static_inputs
 
     def _get_submodules_under_cudagraphs(self):
@@ -1075,7 +1100,9 @@ def _te_cuda_graph_capture(self, *args, **kwargs):
                 )
             )
         ):
-            hidden_states = self._forward_mlp(hidden_states)
+            hidden_states = self._forward_mlp(
+                hidden_states, input_ids=kwargs.get("input_ids", None)
+            )
         if not isinstance(hidden_states, list) and not isinstance(hidden_states, tuple):
             cuda_graph_outputs = [hidden_states]
         else:
@@ -1224,7 +1251,7 @@ def _te_cuda_graph_replay_impl(self, args, kwargs, context):
                 return residual, hidden_states, probs, shared_expert_output
 
             # CUDA Graph does not capture the MLP/MoE part at all.
-            output = self._forward_mlp(*cuda_graph_output)
+            output = self._forward_mlp(*cuda_graph_output, input_ids=kwargs.get("input_ids", None))
         return output, context
 
     def _get_te_cuda_graph_replay_args(self, *args, **kwargs):
@@ -1501,6 +1528,19 @@ def get_layer_static_inputs(self, seq_length, micro_batch_size):
             requires_grad=hs.requires_grad,
             device=hs.device,
         )
+
+        # Add input_ids for hash-based MoE routing under CUDA graphs.
+        # Only add for layers that actually use hash routing,
+        # since other layers (e.g. on later PP stages) receive input_ids=None.
+        if (
+            self.is_moe_layer
+            and self.config.moe_n_hash_layers > 0
+            and getattr(self.mlp.router, 'is_hash_layer', False)
+        ):
+            static_inputs["input_ids"] = torch.zeros(
+                (micro_batch_size, seq_length), dtype=torch.long, device=torch.cuda.current_device()
+            )
+
         return static_inputs
 
     def _get_submodules_under_cudagraphs(self):
@@ -1542,6 +1582,7 @@ def forward(self, *args, **kwargs):
             hidden_states,
             kwargs.get("inference_context", None),
             padding_mask=kwargs.get("padding_mask", None),
+            input_ids=kwargs.get("input_ids", None),
             mhc_recompute_manager=mhc_recompute_manager,
         )
         return output, context
@@ -1561,6 +1602,7 @@ def _forward_attention(
         packed_seq_params: Optional[PackedSeqParams] = None,
         sequence_len_offset: Optional[Tensor] = None,
         padding_mask: Optional[Tensor] = None,
+        input_ids: Optional[Tensor] = None,
         mhc_recompute_manager: Optional['CheckpointManager'] = None,
         *,
         inference_params: Optional[Any] = None,
@@ -1656,6 +1698,7 @@ def _forward_mlp(
         hidden_states,
         inference_context=None,
         padding_mask=None,
+        input_ids=None,
         mhc_recompute_manager: Optional['CheckpointManager'] = None,
     ):
         """Forward MLP with hyper connection pre/post processing."""
@@ -1699,6 +1742,10 @@ def _forward_mlp(
             and not self.config.transformer_impl == "inference_optimized"
         )
 
+        moe_kwargs = {}
+        if self.is_moe_layer and input_ids is not None:
+            moe_kwargs['input_ids'] = input_ids
+
         if self.recompute_mlp:
             if self.config.fp8 or self.config.fp4:
                 from megatron.core.extensions.transformer_engine import te_checkpoint
@@ -1710,10 +1757,11 @@ def _forward_mlp(
                     self.pg_collection.tp,
                     pre_mlp_layernorm_output,
                     padding_mask=padding_mask,
+                    **moe_kwargs,
                 )
             else:
                 mlp_output_with_bias = tensor_parallel.checkpoint(
-                    functools.partial(self.mlp, padding_mask=padding_mask),
+                    functools.partial(self.mlp, padding_mask=padding_mask, **moe_kwargs),
                     False,
                     pre_mlp_layernorm_output,
                 )
@@ -1726,7 +1774,9 @@ def _forward_mlp(
             bias_output = torch.stack(bias_chunks, dim=0).sum(dim=0) if bias_chunks else None
             mlp_output_with_bias = (mlp_output, bias_output)
         else:
-            mlp_output_with_bias = self.mlp(pre_mlp_layernorm_output, padding_mask=padding_mask)
+            mlp_output_with_bias = self.mlp(
+                pre_mlp_layernorm_output, padding_mask=padding_mask, **moe_kwargs
+            )
 
         nvtx_range_pop(suffix="mlp")
 
@@ -1893,7 +1943,7 @@ def _te_cuda_graph_replay_impl(self, args, kwargs, context):
             )
             self.recompute_pre_mlp_layernorm = recompute_pre_mlp_layernorm
         else:
-            output = self._forward_mlp(*cuda_graph_output)
+            output = self._forward_mlp(*cuda_graph_output, input_ids=kwargs.get("input_ids", None))
         return output, context
 
 
@@ -1991,7 +2041,7 @@ def create_mcore_cudagraph_manager(self, config):
         ):
             self.transition_cudagraph_scope('partial')
 
-    def _forward_mlp_router(self, hidden_states, padding_mask=None):
+    def _forward_mlp_router(self, hidden_states, padding_mask=None, input_ids=None):
         """
         Executes the router phase of the MoE block.
 
@@ -2016,7 +2066,10 @@ def _forward_mlp_router(self, hidden_states, padding_mask=None):
             residual = residual.float()
 
         router_outputs = self.mlp(
-            pre_mlp_layernorm_output, intermediate_tensors=(), padding_mask=padding_mask
+            pre_mlp_layernorm_output,
+            intermediate_tensors=(),
+            padding_mask=padding_mask,
+            input_ids=input_ids,
         )
 
         for attr_name in self.mlp.token_dispatcher.cudagraph_attrs:
@@ -2064,7 +2117,9 @@ def _forward_mlp_postprocess(self, residual, output, shared_expert_output, mlp_b
         output = self.mlp(None, intermediate_tensors=(output, shared_expert_output))
         return self._forward_post_mlp((output, mlp_bias), residual)
 
-    def _forward_mlp(self, hidden_states, inference_context=None, padding_mask=None):
+    def _forward_mlp(
+        self, hidden_states, inference_context=None, padding_mask=None, input_ids=None
+    ):
         """
         Orchestrates the MLP forward pass, handling partial CUDA graph execution logic.
 
@@ -2080,10 +2135,10 @@ def _forward_mlp(self, hidden_states, inference_context=None, padding_mask=None)
             )
 
         def _forward_mlp_partial_cudagraphs(
-            hidden_states, inference_context=None, padding_mask=None
+            hidden_states, inference_context=None, padding_mask=None, input_ids=None
         ):
             residual, hidden_states, probs, shared_expert_output = self._forward_mlp_router(
-                hidden_states, padding_mask=padding_mask
+                hidden_states, padding_mask=padding_mask, input_ids=input_ids
             )
 
             # After the router graph replays, the captured .copy_() operations that update
@@ -2112,16 +2167,23 @@ def _forward_mlp_partial_cudagraphs(
                         parallel_state.get_tensor_model_parallel_group(),
                         hidden_states,
                         padding_mask=padding_mask,
+                        input_ids=input_ids,
                     )
                 else:
                     return tensor_parallel.checkpoint(
                         functools.partial(
-                            _forward_mlp_partial_cudagraphs, padding_mask=padding_mask
+                            _forward_mlp_partial_cudagraphs,
+                            padding_mask=padding_mask,
+                            input_ids=input_ids,
                         ),
                         False,
                         hidden_states,
                     )
             else:
-                return _forward_mlp_partial_cudagraphs(hidden_states, padding_mask=padding_mask)
+                return _forward_mlp_partial_cudagraphs(
+                    hidden_states, padding_mask=padding_mask, input_ids=input_ids
+                )
         else:
-            return super()._forward_mlp(hidden_states, padding_mask=padding_mask)
+            return super()._forward_mlp(
+                hidden_states, padding_mask=padding_mask, input_ids=input_ids
+            )
diff --git a/megatron/training/arguments.py b/megatron/training/arguments.py
index a26cc3759ff..4b9f9790a0c 100644
--- a/megatron/training/arguments.py
+++ b/megatron/training/arguments.py
@@ -2005,6 +2005,7 @@ def core_transformer_config_from_args(args, config_class=None):
     kw_args['pipeline_dtype'] = args.params_dtype
     kw_args['batch_p2p_comm'] = not args.overlap_p2p_comm
     kw_args['num_moe_experts'] = args.num_experts
+    kw_args['actual_vocab_size'] = args.padded_vocab_size
     kw_args['rotary_interleaved'] = args.rotary_interleaved
     kw_args['num_layers_in_first_pipeline_stage'] = args.decoder_first_pipeline_num_layers
     kw_args['num_layers_in_last_pipeline_stage'] = args.decoder_last_pipeline_num_layers
@@ -2479,6 +2480,7 @@ def _add_network_size_args(parser):
         "barrier_with_L1_time",
         # args uses same var with a different name
         "num_moe_experts",
+        "actual_vocab_size",
         "fp8_param",
         "fp4_param",
         # incompatible defaults in dataclass
diff --git a/tests/unit_tests/fusions/test_swiglu_fusion.py b/tests/unit_tests/fusions/test_swiglu_fusion.py
index c72679cd047..58e7069d3f1 100644
--- a/tests/unit_tests/fusions/test_swiglu_fusion.py
+++ b/tests/unit_tests/fusions/test_swiglu_fusion.py
@@ -1,5 +1,8 @@
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
 import pytest
 import torch
+import torch.nn.functional as F
 
 from megatron.core.fusions.fused_bias_swiglu import bias_swiglu_impl, weighted_bias_swiglu_impl
 
@@ -39,3 +42,46 @@ def test_weighted_bias_swiglu(input_dtype):
     assert weights_2.grad.dtype == weights.grad.dtype
     if input_dtype == torch.float32:
         assert torch.allclose(weights.grad, weights_2.grad, **tols)
+
+
+@pytest.mark.parametrize("input_dtype", [torch.bfloat16, torch.float32])
+def test_clamped_weighted_bias_swiglu(input_dtype):
+    clamp_value = 10.0
+
+    if input_dtype == torch.float32:
+        tols = dict(rtol=1.0e-6, atol=1.0e-6)
+    elif input_dtype == torch.bfloat16:
+        tols = dict(rtol=2.0e-2, atol=1.0e-3)
+    else:
+        raise ValueError(f"Invalid input dtype: {input_dtype}")
+
+    x = torch.randn(16, 64, dtype=input_dtype, device="cuda")
+    x.requires_grad = True
+    weights = torch.randn(16, 1, dtype=torch.float32, device="cuda")
+    weights.requires_grad = True
+    bwd_input = torch.randn(16, 32, dtype=input_dtype, device="cuda")
+
+    # Reference: manual clamp + silu + weight
+    y_1, y_2 = torch.chunk(x, 2, -1)
+    y_1c = y_1.clamp(min=None, max=clamp_value)
+    y_2c = y_2.clamp(min=-clamp_value, max=clamp_value)
+    y = (F.silu(y_1c) * y_2c * weights).to(input_dtype)
+    y.backward(bwd_input)
+
+    x_2 = x.detach().clone()
+    x_2.requires_grad = True
+    weights_2 = weights.detach().clone()
+    weights_2.requires_grad = True
+    bwd_input_2 = bwd_input.detach().clone()
+
+    # Fused implementation
+    y_2_out = weighted_bias_swiglu_impl(x_2, None, weights_2, clamp_value=clamp_value)
+    y_2_out.backward(bwd_input_2)
+
+    assert y_2_out.dtype == y.dtype
+    assert torch.allclose(y, y_2_out, **tols)
+    assert x_2.grad.dtype == x.grad.dtype
+    assert torch.allclose(x.grad, x_2.grad, **tols)
+    assert weights_2.grad.dtype == weights.grad.dtype
+    if input_dtype == torch.float32:
+        assert torch.allclose(weights.grad, weights_2.grad, **tols)
diff --git a/tests/unit_tests/models/test_hybrid_moe_model.py b/tests/unit_tests/models/test_hybrid_moe_model.py
index 3fee1e52362..1df280853fc 100644
--- a/tests/unit_tests/models/test_hybrid_moe_model.py
+++ b/tests/unit_tests/models/test_hybrid_moe_model.py
@@ -33,6 +33,7 @@
     "activation_func": "megatron.core.activations.squared_relu",
     "activation_func_clamp_value": None,
     "activation_func_fp8_input_store": False,
+    "actual_vocab_size": 131072,
     "add_bias_linear": False,
     "add_qkv_bias": False,
     "apply_query_key_layer_scaling": False,
@@ -178,6 +179,7 @@
     "moe_latent_size": None,
     "moe_layer_freq": 1,
     "moe_layer_recompute": False,
+    "moe_n_hash_layers": 0,
     "moe_pad_expert_input_to_capacity": False,
     "moe_pad_experts_for_cuda_graph_inference": False,
     "moe_paged_stash": False,
diff --git a/tests/unit_tests/transformer/moe/test_aux_loss.py b/tests/unit_tests/transformer/moe/test_aux_loss.py
index ccd11bf29af..3b4697fc71a 100644
--- a/tests/unit_tests/transformer/moe/test_aux_loss.py
+++ b/tests/unit_tests/transformer/moe/test_aux_loss.py
@@ -212,8 +212,9 @@ def new_router(self, **kwargs):
         new_transformer_config = dataclasses.replace(self.default_transformer_config, **kwargs)
 
         # Create the router with the updated config
-        router = TopKRouter(config=new_transformer_config, pg_collection=pg_collection)
-        router.set_layer_number(0)
+        router = TopKRouter(
+            config=new_transformer_config, pg_collection=pg_collection, layer_number=0
+        )
         return router
 
     def teardown_method(self, method):
@@ -626,8 +627,9 @@ def new_router(self, **kwargs):
         """Create a new router with updated configuration."""
         pg_collection = get_default_pg_collection()
         new_transformer_config = dataclasses.replace(self.default_transformer_config, **kwargs)
-        router = TopKRouter(config=new_transformer_config, pg_collection=pg_collection)
-        router.set_layer_number(0)
+        router = TopKRouter(
+            config=new_transformer_config, pg_collection=pg_collection, layer_number=0
+        )
         return router
 
     @pytest.mark.internal
diff --git a/tests/unit_tests/transformer/moe/test_paged_stashing.py b/tests/unit_tests/transformer/moe/test_paged_stashing.py
index cdde5f7553f..262346d0609 100644
--- a/tests/unit_tests/transformer/moe/test_paged_stashing.py
+++ b/tests/unit_tests/transformer/moe/test_paged_stashing.py
@@ -136,11 +136,14 @@ def _create_moe_layer(self, layer_number=0):
         quantization_context = get_fp8_context(self.config, layer_number, is_init=True)
         with quantization_context:
             moe_layer = (
-                MoELayer(self.config, transformer_layer_spec.submodules.mlp.submodules)
+                MoELayer(
+                    self.config,
+                    transformer_layer_spec.submodules.mlp.submodules,
+                    layer_number=layer_number,
+                )
                 .cuda()
                 .to(dtype=self.test_dtype)
             )
-            moe_layer.set_layer_number(layer_number)
             return moe_layer
 
     def zero_grad(self):
diff --git a/tests/unit_tests/transformer/moe/test_routers.py b/tests/unit_tests/transformer/moe/test_routers.py
index 8f3dbbe96e0..40802ac65dc 100644
--- a/tests/unit_tests/transformer/moe/test_routers.py
+++ b/tests/unit_tests/transformer/moe/test_routers.py
@@ -8,8 +8,12 @@
 
 from megatron.core.models.gpt.gpt_layer_specs import get_gpt_layer_local_submodules
 from megatron.core.transformer.moe.moe_layer import MoELayer
-from megatron.core.transformer.moe.moe_utils import get_updated_expert_bias, router_gating_linear
-from megatron.core.transformer.moe.router import Router
+from megatron.core.transformer.moe.moe_utils import (
+    get_default_pg_collection,
+    get_updated_expert_bias,
+    router_gating_linear,
+)
+from megatron.core.transformer.moe.router import Router, TopKRouter
 from megatron.core.transformer.transformer_config import TransformerConfig
 from megatron.training.initialize import _set_random_seed
 from tests.unit_tests.test_utilities import Utils
@@ -563,3 +567,137 @@ def test_router_gating_linear_bias(router_dtype):
     assert torch.allclose(inp.grad, ref_inp.grad, **tols)
     assert torch.allclose(weight.grad, ref_weight.grad, **tols)
     assert torch.allclose(bias.grad, ref_bias.grad, **tols)
+
+
+# ============================================================
+# Hash-based MoE routing tests
+# ============================================================
+
+
+def _hash_routing_config(**overrides):
+    """Create a base TransformerConfig suitable for hash routing tests."""
+    defaults = dict(
+        num_layers=2,
+        hidden_size=16,
+        num_attention_heads=8,
+        num_moe_experts=4,
+        moe_router_topk=2,
+        moe_router_load_balancing_type="aux_loss",
+        moe_aux_loss_coeff=0.0,
+        moe_router_dtype="fp32",
+        add_bias_linear=False,
+        use_cpu_initialization=True,
+        moe_n_hash_layers=1,
+        actual_vocab_size=128,
+    )
+    defaults.update(overrides)
+    return TransformerConfig(**defaults)
+
+
+class TestHashRouting:
+    """Test hash-based MoE routing (_hash_routing, is_hash_layer, config validation)."""
+
+    def setup_method(self, method):
+        Utils.initialize_model_parallel(
+            tensor_model_parallel_size=1,
+            pipeline_model_parallel_size=1,
+            expert_model_parallel_size=1,
+        )
+        _set_random_seed(seed_=42, data_parallel_random_init=False)
+
+    def teardown_method(self, method):
+        Utils.destroy_model_parallel()
+
+    @pytest.mark.internal
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
+    @pytest.mark.parametrize("score_function", ["softmax", "sigmoid", "sqrtsoftplus"])
+    def test_hash_routing_correctness(self, score_function):
+        """Verify expert selection matches tid2eid and scores are computed correctly."""
+        config = _hash_routing_config(moe_router_score_function=score_function)
+        pg_collection = get_default_pg_collection()
+        router = TopKRouter(config=config, pg_collection=pg_collection, layer_number=1)
+
+        num_tokens, num_experts = 16, 4
+        logits = torch.randn(num_tokens, num_experts, device="cuda")
+        input_ids = torch.randint(0, 128, (4, 4), device="cuda")
+
+        routing_probs, routing_map = router._hash_routing(logits, input_ids)
+
+        # Compute expected
+        if score_function == "softmax":
+            scores = torch.softmax(logits, dim=-1, dtype=torch.float32).type_as(logits)
+        elif score_function == "sigmoid":
+            scores = torch.sigmoid(logits.float()).type_as(logits)
+        else:
+            scores = torch.nn.functional.softplus(logits.float()).sqrt().type_as(logits)
+
+        flat_ids = input_ids.T.reshape(-1)
+        top_indices = router.tid2eid[flat_ids].long()
+        probs = scores.gather(1, top_indices)
+        if score_function != "softmax":
+            probs = probs / (probs.sum(dim=-1, keepdim=True) + 1e-20)
+
+        # Each token routed to exactly topk experts matching tid2eid
+        assert (routing_map.sum(dim=1) == router.topk).all()
+        for i in range(num_tokens):
+            actual = routing_map[i].nonzero(as_tuple=True)[0].sort().values
+            expected = top_indices[i].sort().values
+            assert torch.equal(actual, expected)
+            for k in range(router.topk):
+                expert_idx = top_indices[i, k].item()
+                assert torch.isclose(routing_probs[i, expert_idx], probs[i, k], atol=1e-5)
+
+    @pytest.mark.internal
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
+    def test_is_hash_layer_logic(self):
+        """Test layer boundary, MTP guard, and expert bias interaction."""
+        pg_collection = get_default_pg_collection()
+
+        # Boundary: layers within/beyond moe_n_hash_layers
+        config = _hash_routing_config(moe_n_hash_layers=2)
+        r1 = TopKRouter(config=config, pg_collection=pg_collection, layer_number=1)
+        r2 = TopKRouter(config=config, pg_collection=pg_collection, layer_number=2)
+        r3 = TopKRouter(config=config, pg_collection=pg_collection, layer_number=3)
+        assert r1.is_hash_layer is True and r1.tid2eid is not None
+        assert r2.is_hash_layer is True
+        assert r3.is_hash_layer is False and r3.tid2eid is None
+
+        # MTP layers bypass hash routing
+        mtp_router = TopKRouter(
+            config=config, pg_collection=pg_collection, layer_number=1, is_mtp_layer=True
+        )
+        assert mtp_router.is_hash_layer is False and mtp_router.tid2eid is None
+
+        # Expert bias disabled on hash layers
+        bias_config = _hash_routing_config(
+            moe_n_hash_layers=1,
+            moe_router_enable_expert_bias=True,
+            moe_router_score_function="sigmoid",
+        )
+        hash_r = TopKRouter(config=bias_config, pg_collection=pg_collection, layer_number=1)
+        normal_r = TopKRouter(config=bias_config, pg_collection=pg_collection, layer_number=2)
+        assert hash_r.enable_expert_bias is False
+        assert normal_r.enable_expert_bias is True
+
+    @pytest.mark.internal
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
+    def test_moe_layer_hash_routing_integration(self):
+        """End-to-end MoELayer forward/backward with hash routing; raises without input_ids."""
+        config = _hash_routing_config(moe_n_hash_layers=1)
+        submodules = get_gpt_layer_local_submodules(
+            num_experts=config.num_moe_experts, moe_grouped_gemm=False
+        )
+        moe_layer = MoELayer(config, submodules.mlp.submodules, layer_number=1).cuda()
+
+        hidden_states = torch.randn(8, 2, 16, device="cuda", requires_grad=True)
+        input_ids = torch.randint(0, 128, (2, 8), device="cuda")
+
+        # Forward succeeds with input_ids
+        output, _ = moe_layer(hidden_states, input_ids=input_ids)
+        assert output.shape == hidden_states.shape
+        assert not torch.isnan(output).any()
+
+        # Backward succeeds
+        output.sum().backward()
+        assert hidden_states.grad is not None
+        assert not torch.isnan(hidden_states.grad).any()
diff --git a/tests/unit_tests/transformer/moe/test_token_dispatcher.py b/tests/unit_tests/transformer/moe/test_token_dispatcher.py
index ce9b8973dfd..dead2c0c12d 100644
--- a/tests/unit_tests/transformer/moe/test_token_dispatcher.py
+++ b/tests/unit_tests/transformer/moe/test_token_dispatcher.py
@@ -147,8 +147,11 @@ def new_moe_layer(self, **kargs):
             num_experts=self.config.num_moe_experts, moe_grouped_gemm=self.config.moe_grouped_gemm
         )
         new_config = dataclasses.replace(self.config, **kargs)
-        moe_layer = MoELayer(new_config, submodules.mlp.submodules).cuda().to(dtype=self.test_dtype)
-        moe_layer.set_layer_number(0)
+        moe_layer = (
+            MoELayer(new_config, submodules.mlp.submodules, layer_number=0)
+            .cuda()
+            .to(dtype=self.test_dtype)
+        )
         return moe_layer
 
     def __del__(self):