diff --git a/megatron/core/distributed/finalize_model_grads.py b/megatron/core/distributed/finalize_model_grads.py
index e494bbde2b6..d1633a2e992 100644
--- a/megatron/core/distributed/finalize_model_grads.py
+++ b/megatron/core/distributed/finalize_model_grads.py
@@ -21,7 +21,10 @@
 from megatron.core.process_groups_config import ProcessGroupCollection
 
 from .. import parallel_state
-from ..transformer.moe.moe_utils import get_updated_expert_bias
+from ..transformer.moe.moe_utils import (
+    get_updated_expert_bias,
+    get_updated_expert_bias_with_quantile,
+)
 from ..transformer.transformer_config import TransformerConfig
 from ..utils import (
     get_attr_wrapped_model,
@@ -325,6 +328,8 @@ def reset_model_temporary_tensors(config: TransformerConfig, model: List[torch.n
                 and module.expert_bias is not None
             ):
                 module.local_tokens_per_expert.zero_()
+            if getattr(module, 'qb_histogram', None) is not None:
+                module.qb_histogram.zero_()
             if (
                 config.moe_router_load_balancing_type == "global_aux_loss"
                 or "global_aux_loss" in config.moe_router_load_balancing_type
@@ -373,6 +378,44 @@ def _update_router_expert_bias(
         expert_bias.copy_(updated_expert_bias)
 
 
+def _update_router_expert_bias_with_quantile(
+    model: List[torch.nn.Module],
+    config: TransformerConfig,
+    tp_dp_cp_group: Optional[torch.distributed.ProcessGroup] = None,
+):
+    """Pool K3 histograms and update every local router once per global batch."""
+    expert_bias_list = []
+    qb_histogram_list = []
+    qb_bin_bounds_list = []
+    for model_chunk in model:
+        for module in get_attr_wrapped_model(model_chunk, 'modules')():
+            if (
+                getattr(module, 'qb_histogram', None) is not None
+                and module.training
+                and not getattr(module, 'frozen_expert_bias', False)
+            ):
+                expert_bias_list.append(module.expert_bias)
+                qb_histogram_list.append(module.qb_histogram)
+                qb_bin_bounds_list.append(module.qb_bin_bounds)
+
+    if not expert_bias_list:
+        return
+
+    stacked_bias = torch.stack(expert_bias_list, dim=0)
+    stacked_histogram = torch.stack(qb_histogram_list, dim=0)
+    stacked_bin_bounds = torch.stack(qb_bin_bounds_list, dim=0)
+    if get_pg_size(tp_dp_cp_group) > 1:
+        torch.distributed.all_reduce(stacked_histogram, group=tp_dp_cp_group)
+    updated_bias, updated_bin_bounds = get_updated_expert_bias_with_quantile(
+        stacked_histogram, stacked_bin_bounds, stacked_bias, config.moe_router_topk
+    )
+    for bias, bounds, next_bias, next_bounds in zip(
+        expert_bias_list, qb_bin_bounds_list, updated_bias, updated_bin_bounds
+    ):
+        bias.copy_(next_bias)
+        bounds.copy_(next_bounds)
+
+
 def _allreduce_non_tensor_model_parallel_grads(
     model: List[torch.nn.Module],
     config: TransformerConfig,
@@ -483,10 +526,13 @@ def finalize_model_grads(
             "If you don't need pos_embd_group, you need to explicitly set it to None."
         )
         assert hasattr(pg_collection, 'dp_cp')
-        if config.moe_router_enable_expert_bias:
-            assert hasattr(pg_collection, 'tp_dp_cp') and pg_collection.tp_dp_cp is not None, (
-                "pg_collection must have tp_dp_cp when " "moe_router_enable_expert_bias is enabled."
-            )
+        if (
+            config.moe_router_enable_expert_bias
+            or config.moe_router_load_balancing_type == "quantile_balancing"
+        ):
+            assert (
+                hasattr(pg_collection, 'tp_dp_cp') and pg_collection.tp_dp_cp is not None
+            ), "pg_collection must have tp_dp_cp when router bias updates are enabled."
             tp_dp_cp_group = pg_collection.tp_dp_cp
         tp_group = pg_collection.tp
         pp_group = pg_collection.pp
@@ -547,6 +593,14 @@ def finalize_model_grads(
             )
         _update_router_expert_bias(model, config, tp_dp_cp_group=tp_dp_cp_group)
 
+    if config.moe_router_load_balancing_type == "quantile_balancing":
+        if pg_collection is None:
+            # Legacy compatibility; modern callers provide pg_collection.tp_dp_cp above.
+            tp_dp_cp_group = parallel_state.get_tensor_and_data_parallel_group(
+                with_context_parallel=True
+            )
+        _update_router_expert_bias_with_quantile(model, config, tp_dp_cp_group=tp_dp_cp_group)
+
     reset_model_temporary_tensors(config, model)
 
     # normalize gradients for per-token loss normalization.
diff --git a/megatron/core/extensions/transformer_engine.py b/megatron/core/extensions/transformer_engine.py
index 348847e7399..82d7fca30ea 100644
--- a/megatron/core/extensions/transformer_engine.py
+++ b/megatron/core/extensions/transformer_engine.py
@@ -3420,10 +3420,22 @@ def te_general_gemm(
         fused_topk_with_score_function,
     )
 
+    try:
+        _fused_topk_sig = inspect.signature(fused_topk_with_score_function)
+        fused_topk_with_score_function_supports_qb = {
+            "qb_histogram",
+            "qb_bin_bounds",
+            "qb_histogram_mode",
+        }.issubset(_fused_topk_sig.parameters)
+        del _fused_topk_sig
+    except (TypeError, ValueError):
+        fused_topk_with_score_function_supports_qb = False
+
 else:
     fused_topk_with_score_function = None
     fused_compute_score_for_moe_aux_loss = None
     fused_moe_aux_loss = None
+    fused_topk_with_score_function_supports_qb = False
 
 
 def set_save_original_input(module):
diff --git a/megatron/core/transformer/moe/README.md b/megatron/core/transformer/moe/README.md
index b96bf623631..61c4af82b25 100644
--- a/megatron/core/transformer/moe/README.md
+++ b/megatron/core/transformer/moe/README.md
@@ -297,6 +297,7 @@ Routers determine which expert(s) handle each token. A lightweight MLP scores ev
 | **seq_aux_loss** | Sequence-level auxiliary loss for balancing expert usage on each sequence| `--moe-router-load-balancing-type seq_aux_loss` |
 | **global_aux_loss** | Global auxiliary loss for balancing expert usage on a global batch across all ranks | `--moe-router-load-balancing-type global_aux_loss` |
 | **sinkhorn** | Optimal transport formulation for balancing expert usage | `--moe-router-load-balancing-type sinkhorn` |
+| **quantile_balancing** | Kimi K3 aux-loss-free global-batch histogram quantile bias updates | `--moe-router-load-balancing-type quantile_balancing --moe-router-score-function sigmoid --moe-aux-loss-coeff 0` |
 | **aux loss free** | Dynamic bias-based load balancing strategy without auxiliary loss | `--moe-router-enable-expert-bias --moe-router-bias-update-rate 1e-3`|
 | **none** | No load balancing | `--moe-router-load-balancing-type none` |
 
@@ -528,7 +529,9 @@ For MoE models, certain configurations may prevent CUDA Graph capture of MoE lay
 ### Router Arguments
 | Argument | Description | Default |
 |----------|-------------|---------|
-| --moe-router-load-balancing-type | Load balancing: aux_loss, sinkhorn, seq_aux_loss, none | aux_loss |
+| --moe-router-load-balancing-type | Load balancing: aux_loss, seq_aux_loss, global_aux_loss, sinkhorn, quantile_balancing, none | aux_loss |
+| --moe-router-quantile-balancing-estimation-scope | Quantile population; dev supports global_batch | global_batch |
+| --moe-router-qb-num-bins | Uniform histogram bins per expert for global-batch quantile balancing | 1000 |
 | --moe-router-topk | Number of experts per token | 2 |
 | --moe-router-score-function | Score function: softmax, sigmoid | softmax |
 | --moe-router-pre-softmax | Softmax before top-k | False |
diff --git a/megatron/core/transformer/moe/moe_utils.py b/megatron/core/transformer/moe/moe_utils.py
index c8e197d2a3b..dcb8c9f3b2a 100644
--- a/megatron/core/transformer/moe/moe_utils.py
+++ b/megatron/core/transformer/moe/moe_utils.py
@@ -37,6 +37,7 @@
         fused_sort_chunks_by_index,
         fused_sort_chunks_by_index_with_probs,
         fused_topk_with_score_function,
+        fused_topk_with_score_function_supports_qb,
         fused_unpermute,
         te_general_gemm,
     )
@@ -53,6 +54,7 @@
         fused_unpermute,
         te_general_gemm,
     ) = (None, None, None, None, None, None, None, None, None, None)
+    fused_topk_with_score_function_supports_qb = False
 
 
 def switch_load_balancing_loss_func(
@@ -692,6 +694,8 @@ def topk_routing_with_score_function(
     fused: bool = False,
     router_replay: Optional['RouterReplay'] = None,
     dense_output: bool = False,
+    qb_histogram: Optional[torch.Tensor] = None,
+    qb_bin_bounds: Optional[torch.Tensor] = None,
 ) -> Tuple[torch.Tensor, torch.Tensor]:
     """Compute the routing probabilities and map for top-k selection with score function.
 
@@ -716,6 +720,10 @@ def topk_routing_with_score_function(
                                               Defaults to None.
         dense_output (bool, optional): If True, return dense tensors [num_tokens, topk] instead of
                                        sparse tensors [num_tokens, num_experts]. Defaults to False.
+        qb_histogram (torch.Tensor, optional): Caller-owned int32 K3 Quantile Balancing histogram
+                                               with shape [num_experts, num_bins].
+        qb_bin_bounds (torch.Tensor, optional): FP32 CUDA tensor containing the lower and upper
+                                                K3 Quantile Balancing histogram bounds.
 
     Returns:
         Tuple[torch.Tensor, torch.Tensor]:
@@ -734,6 +742,31 @@ def topk_routing_with_score_function(
     """
     assert logits.dim() == 2, f"Expected 2D logits [num_tokens, num_experts], got {logits.dim()}."
     num_tokens, num_experts = logits.shape
+    use_quantile_balancing = qb_histogram is not None or qb_bin_bounds is not None
+    if use_quantile_balancing and (qb_histogram is None or qb_bin_bounds is None):
+        raise ValueError("qb_histogram and qb_bin_bounds must be provided together.")
+    if use_quantile_balancing:
+        if expert_bias is None:
+            raise ValueError("Quantile Balancing requires an expert bias.")
+        if score_function != "sigmoid":
+            raise ValueError("Quantile Balancing currently requires score_function='sigmoid'.")
+        if use_pre_softmax:
+            raise ValueError("Quantile Balancing does not use pre-softmax routing.")
+        if topk >= num_experts:
+            raise ValueError("Quantile Balancing requires topk < num_experts.")
+        if num_groups is not None or group_topk is not None:
+            raise ValueError("Quantile Balancing does not support group-limited routing.")
+        if router_replay is not None:
+            raise ValueError("Quantile Balancing does not support router replay.")
+        if qb_histogram.dim() != 2 or qb_histogram.shape[0] != num_experts:
+            raise ValueError(
+                "qb_histogram must have shape [num_experts, num_bins], got "
+                f"{tuple(qb_histogram.shape)}."
+            )
+        if qb_histogram.dtype != torch.int32:
+            raise ValueError("qb_histogram must have dtype torch.int32.")
+        if qb_bin_bounds.shape != (2,) or qb_bin_bounds.dtype != torch.float32:
+            raise ValueError("qb_bin_bounds must be an FP32 tensor with shape [2].")
     if fused:
         if not HAVE_TE or fused_topk_with_score_function is None:
             raise ValueError(
@@ -744,16 +777,29 @@ def topk_routing_with_score_function(
                 "Fused sqrtsoftplus score function requires TE >= 2.13.0. "
                 "Please upgrade Transformer Engine or disable moe_router_fusion."
             )
-        return fused_topk_with_score_function(
-            logits=logits,
-            topk=topk,
-            use_pre_softmax=use_pre_softmax,
-            num_groups=num_groups,
-            group_topk=group_topk,
-            scaling_factor=scaling_factor,
-            score_function=score_function,
-            expert_bias=expert_bias,
-        )
+        kwargs = {
+            "logits": logits,
+            "topk": topk,
+            "use_pre_softmax": use_pre_softmax,
+            "num_groups": num_groups,
+            "group_topk": group_topk,
+            "scaling_factor": scaling_factor,
+            "score_function": score_function,
+            "expert_bias": expert_bias,
+        }
+        if use_quantile_balancing:
+            if not fused_topk_with_score_function_supports_qb:
+                raise ValueError(
+                    "The installed Transformer Engine fused router does not expose Quantile "
+                    "Balancing histogram outputs. Upgrade Transformer Engine or disable "
+                    "moe_router_fusion."
+                )
+            kwargs.update(
+                qb_histogram=qb_histogram,
+                qb_bin_bounds=qb_bin_bounds,
+                qb_histogram_mode="fused_atomic",
+            )
+        return fused_topk_with_score_function(**kwargs)
 
     def _compute_topk(
         scores: torch.Tensor,
@@ -815,7 +861,27 @@ def compute_topk(scores, topk, num_groups=None, group_topk=None):
             scores = torch.nn.functional.softplus(logits.float()).sqrt()
         if expert_bias is not None:
             scores_for_routing = scores + expert_bias.float()
-            _, top_indices = compute_topk(scores_for_routing, topk, num_groups, group_topk)
+            if use_quantile_balancing:
+                topk_result = torch.topk(scores_for_routing, topk + 1, dim=1, sorted=True)
+                cutoff = topk_result.values[:, -1:]
+                top_indices = topk_result.indices[:, :-1]
+                with torch.no_grad():
+                    num_bins = qb_histogram.shape[1]
+                    lower, upper = qb_bin_bounds.unbind()
+                    bin_indices = torch.floor(
+                        (cutoff - scores.detach() - lower) * (num_bins / (upper - lower))
+                    ).to(torch.int64)
+                    bin_indices.clamp_(0, num_bins - 1)
+                    expert_offsets = (
+                        torch.arange(num_experts, device=logits.device, dtype=torch.int64)
+                        * num_bins
+                    )
+                    flat_indices = (bin_indices + expert_offsets).reshape(-1)
+                    qb_histogram.view(-1).scatter_add_(
+                        0, flat_indices, torch.ones_like(flat_indices, dtype=torch.int32)
+                    )
+            else:
+                _, top_indices = compute_topk(scores_for_routing, topk, num_groups, group_topk)
             scores = torch.gather(scores, dim=1, index=top_indices)
         else:
             scores, top_indices = compute_topk(scores, topk, num_groups, group_topk)
@@ -1196,6 +1262,61 @@ def get_updated_expert_bias(
         return updated_expert_bias
 
 
+def get_updated_expert_bias_with_quantile(
+    histogram: torch.Tensor, bin_bounds: torch.Tensor, expert_bias: torch.Tensor, topk: int
+) -> Tuple[torch.Tensor, torch.Tensor]:
+    """Recover Kimi K3 Quantile Balancing biases from a pooled global-batch histogram."""
+    if histogram.dim() < 2:
+        raise ValueError("QB histogram must have shape [..., num_experts, num_bins].")
+    num_experts, num_bins = histogram.shape[-2:]
+    if num_experts <= 0 or num_bins <= 0:
+        raise ValueError("QB histogram dimensions must be positive.")
+    if expert_bias.shape != histogram.shape[:-1]:
+        raise ValueError(
+            f"QB expert_bias shape {expert_bias.shape} does not match expected "
+            f"{histogram.shape[:-1]} for histogram shape {histogram.shape}."
+        )
+    if bin_bounds.shape != histogram.shape[:-2] + (2,):
+        raise ValueError(
+            f"QB bin_bounds shape {bin_bounds.shape} does not match expected "
+            f"{histogram.shape[:-2] + (2,)} for histogram shape {histogram.shape}."
+        )
+    if not 0 < topk < num_experts:
+        raise ValueError(f"QB topk must be in [1, {num_experts}), got {topk}.")
+
+    with torch.no_grad():
+        cumulative_counts = torch.cumsum(histogram, dim=-1, dtype=torch.int64)
+        tokens_per_expert = cumulative_counts[..., -1]
+        # Every token contributes one margin sample to every expert, so all expert totals match.
+        total_tokens = tokens_per_expert[..., 0]
+        target_quantile = total_tokens.to(torch.float32) * (topk / num_experts)
+        target_rank = torch.ceil(target_quantile).to(torch.int64).clamp_min_(1)
+        selected_bins = (
+            (cumulative_counts >= target_rank[..., None, None]).to(torch.int64).argmax(dim=-1)
+        )
+        selected_counts = torch.gather(histogram, -1, selected_bins.unsqueeze(-1)).squeeze(-1)
+        previous_bins = (selected_bins - 1).clamp_min(0)
+        counts_before = torch.gather(cumulative_counts, -1, previous_bins.unsqueeze(-1)).squeeze(-1)
+        counts_before = torch.where(
+            selected_bins == 0, torch.zeros_like(counts_before), counts_before
+        )
+        interpolation = (
+            (target_quantile[..., None] - counts_before.to(torch.float32))
+            / selected_counts.clamp_min(1).to(torch.float32)
+        ).clamp_(0.0, 1.0)
+        lower = bin_bounds[..., 0, None]
+        upper = bin_bounds[..., 1, None]
+        bin_width = (upper - lower) / num_bins
+        updated_expert_bias = lower + (selected_bins.to(torch.float32) + interpolation) * bin_width
+        updated_expert_bias -= updated_expert_bias.mean(dim=-1, keepdim=True)
+        has_tokens = total_tokens > 0
+        updated_expert_bias = torch.where(has_tokens[..., None], updated_expert_bias, expert_bias)
+        bias_min, bias_max = torch.aminmax(updated_expert_bias, dim=-1)
+        updated_bin_bounds = torch.stack((bias_min - 1.0, bias_max + 1.0), dim=-1)
+        updated_bin_bounds = torch.where(has_tokens[..., None], updated_bin_bounds, bin_bounds)
+        return updated_expert_bias, updated_bin_bounds
+
+
 def maybe_move_tensor_to_cpu(
     tensor: torch.Tensor, as_numpy: bool = False, record_stream: bool = False
 ) -> torch.Tensor:
diff --git a/megatron/core/transformer/moe/paged_stash.py b/megatron/core/transformer/moe/paged_stash.py
index d68216690cd..23c6de96de9 100644
--- a/megatron/core/transformer/moe/paged_stash.py
+++ b/megatron/core/transformer/moe/paged_stash.py
@@ -1033,6 +1033,14 @@ def _set_moe_paged_stash_all(self, value: bool) -> None:
         for c in self._configs_to_sync_moe_paged_stash:
             c.moe_paged_stash = value
 
+    def _reset_qb_histograms(self) -> None:
+        """Discard histogram observations from a failed full-iteration attempt."""
+        for mlp in self.moe_layers:
+            router = getattr(mlp, 'router', None)
+            qb_histogram = getattr(router, 'qb_histogram', None)
+            if qb_histogram is not None:
+                qb_histogram.zero_()
+
     def data_read(self, data_iterator, model, training, num_microbatches):
         """Read all microbatch inputs from Dataloader and copy to static buffers."""
         data_iterator_saved = []
@@ -1105,6 +1113,11 @@ def prepare_for_rerun(self, is_training=True):
             self.stash_manager.host_spill.zero_()
         self._set_moe_paged_stash_all(False)
 
+        # The dropless retry replays every microbatch. Preserve CUDA addresses while
+        # discarding the failed attempt so each token contributes exactly once.
+        if is_training:
+            self._reset_qb_histograms()
+
         # Set grad to zero.
         for model_chunk in self.model:
             model_chunk.zero_grad_buffer()
diff --git a/megatron/core/transformer/moe/router.py b/megatron/core/transformer/moe/router.py
index b26bf1a0d3e..6ac0891b9a5 100644
--- a/megatron/core/transformer/moe/router.py
+++ b/megatron/core/transformer/moe/router.py
@@ -218,6 +218,9 @@ def __init__(
         else:
             self.tid2eid = None
 
+        self.use_quantile_balancing = (
+            self.routing_type == "quantile_balancing" and not self.is_hash_layer
+        )
         self.enable_expert_bias = (
             self.config.moe_router_enable_expert_bias and not self.is_hash_layer
         )
@@ -231,6 +234,10 @@ def __init__(
                 ),
                 persistent=False,
             )
+        else:
+            self.local_tokens_per_expert = None
+
+        if self.enable_expert_bias or self.use_quantile_balancing:
             self.register_buffer(
                 'expert_bias',
                 torch.zeros(
@@ -240,9 +247,27 @@ def __init__(
                 ),
             )
         else:
-            self.local_tokens_per_expert = None
             self.expert_bias = None
 
+        if self.use_quantile_balancing:
+            self.register_buffer(
+                'qb_histogram',
+                torch.zeros(
+                    self.config.num_moe_experts,
+                    self.config.moe_router_qb_num_bins,
+                    dtype=torch.int32,
+                    device=torch.cuda.current_device(),
+                ),
+                persistent=False,
+            )
+            self.register_buffer(
+                'qb_bin_bounds',
+                torch.tensor([-1.0, 1.0], dtype=torch.float32, device=torch.cuda.current_device()),
+            )
+        else:
+            self.qb_histogram = None
+            self.qb_bin_bounds = None
+
         # Initialize global tokens per expert for global aux loss
         if self.get_aux_loss_coeff("global_aux_loss") > 0:
             self.register_buffer(
@@ -277,6 +302,9 @@ def _maintain_float32_expert_bias(self):
         if hasattr(self, 'expert_bias') and self.expert_bias is not None:
             if self.expert_bias.dtype != torch.float32:
                 self.expert_bias.data = self.expert_bias.data.to(torch.float32)
+        if hasattr(self, 'qb_bin_bounds') and self.qb_bin_bounds is not None:
+            if self.qb_bin_bounds.dtype != torch.float32:
+                self.qb_bin_bounds.data = self.qb_bin_bounds.data.to(torch.float32)
 
     def sinkhorn_load_balancing(self, logits: torch.Tensor):
         """Apply sinkhorn routing to the logits tensor.
@@ -810,6 +838,19 @@ def routing(
         elif self.routing_type == "sinkhorn":
             probs, routing_map = self.sinkhorn_load_balancing(logits)
         else:
+            # Activation checkpointing runs the original forward under no-grad and its
+            # recompute under enable-grad, so this gate records each token exactly once.
+            accumulate_qb_histogram = (
+                self.use_quantile_balancing
+                and self.training
+                and torch.is_grad_enabled()
+                and not self.frozen_expert_bias
+            )
+            if accumulate_qb_histogram and padding_mask is not None:
+                raise RuntimeError(
+                    "Quantile Balancing does not yet support padding masks because the "
+                    "histogram APIs do not accept a valid-token mask."
+                )
             probs, routing_map = topk_routing_with_score_function(
                 logits,
                 self.topk,
@@ -821,6 +862,8 @@ def routing(
                 expert_bias=self.expert_bias,
                 fused=self.config.moe_router_fusion,
                 router_replay=self.router_replay,
+                qb_histogram=self.qb_histogram if accumulate_qb_histogram else None,
+                qb_bin_bounds=self.qb_bin_bounds if accumulate_qb_histogram else None,
             )
 
         # Dropless HybridEP consumes the sparse routing map directly, so exclude padding
diff --git a/megatron/core/transformer/transformer_config.py b/megatron/core/transformer/transformer_config.py
index 275bb1720c3..541d3a98298 100644
--- a/megatron/core/transformer/transformer_config.py
+++ b/megatron/core/transformer/transformer_config.py
@@ -807,6 +807,8 @@ class TransformerConfig(ModelParallelConfig):
     for each individual sample.
     - "global_aux_loss": Load balancing loss calculated at global batch level.
     - "sinkhorn": Balancing algorithm used in S-BASE.
+    - "quantile_balancing": Kimi K3 histogram Quantile Balancing. The histogram is accumulated
+    across the global batch and converted into the next-step per-expert routing bias.
     - "none": No load balancing.
     A list of strings can be provided to combine multiple aux-loss load balancing types.
     The default is "aux_loss".
@@ -879,6 +881,16 @@ class TransformerConfig(ModelParallelConfig):
     and decreased for the experts with more assigned tokens.
     The default value 1e-3 is same as that used in DeepSeekV3."""
 
+    moe_router_quantile_balancing_estimation_scope: Literal['global_batch'] = "global_batch"
+    """Population used to estimate the Quantile Balancing bias.
+
+    The ``dev`` implementation provides Kimi K3's ``global_batch`` histogram estimator. The
+    identically named option on ``main`` also accepts ``micro_batch`` for its older exact estimator.
+    """
+
+    moe_router_qb_num_bins: int = 1000
+    """Number of persistent uniform histogram bins per expert for global-batch QB."""
+
     moe_router_force_load_balancing: bool = False
     """[Experimental] Force load balancing with random logits for MoE router, supports naive topk 
     and group-limited topk. This is an experimental feature and only for benchmark."""
@@ -2161,6 +2173,16 @@ def __post_init__(self):
                     f"moe_shared_expert_overlap only works with alltoall or flex token dispatcher."
                 )
 
+        if self.moe_router_load_balancing_type == ["quantile_balancing"]:
+            assert (
+                isinstance(self.moe_aux_loss_coeff, list) and len(self.moe_aux_loss_coeff) == 1
+            ), (
+                "moe_aux_loss_coeff must be a list of the same length as "
+                "moe_router_load_balancing_type"
+            )
+            self.moe_router_load_balancing_type = "quantile_balancing"
+            self.moe_aux_loss_coeff = self.moe_aux_loss_coeff[0]
+
         if isinstance(self.moe_router_load_balancing_type, list):
             assert isinstance(self.moe_aux_loss_coeff, list) and len(
                 self.moe_aux_loss_coeff
@@ -2169,6 +2191,54 @@ def __post_init__(self):
                 "moe_router_load_balancing_type"
             )
 
+        if (
+            isinstance(self.moe_router_load_balancing_type, list)
+            and "quantile_balancing" in self.moe_router_load_balancing_type
+        ):
+            raise ValueError("quantile_balancing must be the sole moe_router_load_balancing_type.")
+        if self.moe_router_load_balancing_type == "quantile_balancing":
+            aux_coeffs = (
+                self.moe_aux_loss_coeff
+                if isinstance(self.moe_aux_loss_coeff, list)
+                else [self.moe_aux_loss_coeff]
+            )
+            if any(float(coeff) != 0.0 for coeff in aux_coeffs):
+                raise ValueError(
+                    "quantile_balancing requires moe_aux_loss_coeff=0 because it replaces "
+                    "the auxiliary load-balancing loss."
+                )
+            if self.moe_router_quantile_balancing_estimation_scope != "global_batch":
+                raise ValueError(
+                    "Megatron-LM dev supports only "
+                    "moe_router_quantile_balancing_estimation_scope='global_batch'."
+                )
+            if self.moe_router_score_function != "sigmoid":
+                raise ValueError("quantile_balancing requires moe_router_score_function='sigmoid'.")
+            if self.moe_router_pre_softmax:
+                raise ValueError("quantile_balancing does not use pre-softmax routing.")
+            if self.moe_router_enable_expert_bias:
+                raise ValueError(
+                    "quantile_balancing selects the expert-bias update rule; do not also enable "
+                    "the DeepSeek-style moe_router_enable_expert_bias."
+                )
+            if self.moe_router_num_groups is not None or self.moe_router_group_topk is not None:
+                raise ValueError("quantile_balancing does not support group-limited routing.")
+            if self.moe_enable_routing_replay:
+                raise ValueError("quantile_balancing does not support routing replay.")
+            if self.moe_expert_capacity_factor is not None and self.moe_expert_capacity_factor >= 0:
+                raise ValueError("quantile_balancing does not support per-expert token dropping.")
+            if self.moe_expert_rank_capacity_factor is not None and not self.moe_paged_stash:
+                raise ValueError(
+                    "quantile_balancing with expert-rank capacity requires moe_paged_stash "
+                    "so an over-budget attempt can be retried without token dropping."
+                )
+            if self.num_moe_experts is None or not 0 < self.moe_router_topk < self.num_moe_experts:
+                raise ValueError(
+                    "quantile_balancing requires 0 < moe_router_topk < num_moe_experts."
+                )
+            if self.moe_router_qb_num_bins <= 1:
+                raise ValueError("moe_router_qb_num_bins must be greater than one.")
+
         if self.moe_expert_capacity_factor is not None:
             if self.moe_expert_capacity_factor < 0:
                 self.moe_expert_capacity_factor = None
@@ -2189,7 +2259,10 @@ def __post_init__(self):
                 "seq_aux_loss",
                 "global_aux_loss",
                 "none",
-            ]:
+            ] and not (
+                self.moe_router_load_balancing_type == "quantile_balancing"
+                and self.moe_expert_capacity_factor is None
+            ):
                 raise ValueError(
                     "moe_expert_capacity_factor only works with aux_loss, "
                     "seq_aux_loss, global_aux_loss or none load balancing"
diff --git a/megatron/training/arguments.py b/megatron/training/arguments.py
index 9c379511b20..4bdb757cf11 100644
--- a/megatron/training/arguments.py
+++ b/megatron/training/arguments.py
@@ -2761,6 +2761,8 @@ def _add_network_size_args(parser):
         "csa_compress_ratios",
         "moe_router_load_balancing_type",
         "moe_aux_loss_coeff",
+        "moe_router_quantile_balancing_estimation_scope",
+        "moe_router_qb_num_bins",
         "cp_comm_type",
         "cuda_graph_modules",
         "cuda_graph_scope",  # deprecated alias; handled manually by --cuda-graph-scope flag
@@ -4858,9 +4860,16 @@ def _add_moe_args(parser):
         '--moe-router-load-balancing-type',
         nargs='+',
         type=str,
-        choices=['aux_loss', 'seq_aux_loss', 'global_aux_loss', 'sinkhorn', 'none'],
+        choices=[
+            'aux_loss',
+            'seq_aux_loss',
+            'global_aux_loss',
+            'sinkhorn',
+            'quantile_balancing',
+            'none',
+        ],
         default='aux_loss',
-        help='Determines the load balancing strategy for the router. "aux_loss" corresponds to the load balancing loss used in GShard and SwitchTransformer; "seq_aux_loss" corresponds to the load balancing loss used in DeepSeekV2, which computes the loss for each individual sample; "sinkhorn" corresponds to the balancing algorithm used in S-BASE, and "none" implies no load balancing. The default is "aux_loss".',
+        help='Determines the load balancing strategy for the router. "aux_loss" corresponds to the load balancing loss used in GShard and SwitchTransformer; "seq_aux_loss" corresponds to the load balancing loss used in DeepSeekV2, which computes the loss for each individual sample; "sinkhorn" corresponds to the balancing algorithm used in S-BASE; "quantile_balancing" uses Kimi K3 global-batch histogram bias updates; and "none" implies no load balancing. The default is "aux_loss".',
     )
     group.add_argument(
         '--moe-aux-loss-coeff',
@@ -4869,6 +4878,22 @@ def _add_moe_args(parser):
         default=0.0,
         help='Scaling coefficient for the aux loss: a starting value of 1e-2 is recommended.',
     )
+    group.add_argument(
+        '--moe-router-quantile-balancing-estimation-scope',
+        type=str,
+        choices=['global_batch'],
+        default='global_batch',
+        help=(
+            'Population used to estimate quantile-balancing biases. The dev branch supports '
+            'Kimi K3 global-batch histogram estimation.'
+        ),
+    )
+    group.add_argument(
+        '--moe-router-qb-num-bins',
+        type=int,
+        default=1000,
+        help='Number of uniform histogram bins per expert for global-batch quantile balancing.',
+    )
     # Token dispatcher arguments
     # MoE communication overlap arguments
 
diff --git a/megatron/training/checkpointing.py b/megatron/training/checkpointing.py
index 395d5a57eca..b3850906279 100644
--- a/megatron/training/checkpointing.py
+++ b/megatron/training/checkpointing.py
@@ -1955,6 +1955,10 @@ def _set_arg(arg_name, old_arg_name=None, force=False):
     _set_arg('moe_single_grouped_weight', force=True)
     _set_arg('moe_single_grouped_bias', force=True)
     _set_arg('moe_shared_expert_intermediate_size', force=True)
+    _set_arg('moe_router_load_balancing_type', force=True)
+    _set_arg('moe_aux_loss_coeff', force=True)
+    _set_arg('moe_router_quantile_balancing_estimation_scope', force=True)
+    _set_arg('moe_router_qb_num_bins', force=True)
     _set_arg('moe_router_score_function', force=True)
     _set_arg('moe_router_enable_expert_bias', force=True)
     _set_arg('moe_router_topk_scaling_factor', force=True)
diff --git a/tests/unit_tests/models/test_hybrid_moe_model.py b/tests/unit_tests/models/test_hybrid_moe_model.py
index a9c015b4011..337568b13e4 100644
--- a/tests/unit_tests/models/test_hybrid_moe_model.py
+++ b/tests/unit_tests/models/test_hybrid_moe_model.py
@@ -233,6 +233,8 @@
     "moe_router_padding_for_fp8": False,
     "moe_router_padding_for_quantization": False,
     "moe_router_pre_softmax": False,
+    "moe_router_qb_num_bins": 1000,
+    "moe_router_quantile_balancing_estimation_scope": "global_batch",
     "moe_router_score_function": "sigmoid",
     "moe_router_topk": 6,
     "moe_router_topk_limited_devices": None,
diff --git a/tests/unit_tests/test_checkpointing.py b/tests/unit_tests/test_checkpointing.py
index 5e35d3bda9f..eb59219d9f2 100644
--- a/tests/unit_tests/test_checkpointing.py
+++ b/tests/unit_tests/test_checkpointing.py
@@ -24,6 +24,7 @@
     _build_sharded_state_dict_metadata,
     _load_base_checkpoint,
     get_checkpoint_tracker_filename,
+    load_args_from_checkpoint,
     load_checkpoint,
     read_metadata,
     save_checkpoint,
@@ -74,6 +75,38 @@ def sharded_state_dict(self, *args, metadata: Optional[dict] = None, **kwargs):
         return self.state_dict()
 
 
+def test_load_args_restores_quantile_balancing_from_checkpoint():
+    """QB mode and histogram shape are reconstructed before the router is built."""
+    checkpoint_args = SimpleNamespace(
+        moe_router_load_balancing_type="quantile_balancing",
+        moe_aux_loss_coeff=0.0,
+        moe_router_quantile_balancing_estimation_scope="global_batch",
+        moe_router_qb_num_bins=257,
+    )
+    args = SimpleNamespace(
+        load="checkpoint",
+        iteration=0,
+        moe_router_load_balancing_type="aux_loss",
+        moe_aux_loss_coeff=0.01,
+        moe_router_quantile_balancing_estimation_scope="micro_batch",
+        moe_router_qb_num_bins=64,
+        use_tokenizer_model_from_checkpoint_args=False,
+        use_mp_args_from_checkpoint_args=False,
+    )
+    state_dict = {"args": checkpoint_args, "iteration": 12}
+
+    with mock.patch(
+        "megatron.training.checkpointing._load_base_checkpoint",
+        return_value=(state_dict, "checkpoint", False, CheckpointType.LEGACY),
+    ):
+        restored_args, _ = load_args_from_checkpoint(args)
+
+    assert restored_args.moe_router_load_balancing_type == "quantile_balancing"
+    assert restored_args.moe_aux_loss_coeff == 0.0
+    assert restored_args.moe_router_quantile_balancing_estimation_scope == "global_batch"
+    assert restored_args.moe_router_qb_num_bins == 257
+
+
 def create_checkpoint(load_path, ckpt_format):
     """Setup a dummy checkpoint directory."""
     iteration = 123
diff --git a/tests/unit_tests/transformer/moe/test_quantile_balancing.py b/tests/unit_tests/transformer/moe/test_quantile_balancing.py
new file mode 100644
index 00000000000..6a2a5553a18
--- /dev/null
+++ b/tests/unit_tests/transformer/moe/test_quantile_balancing.py
@@ -0,0 +1,566 @@
+# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
+import argparse
+import os
+from types import SimpleNamespace
+
+import pytest
+import torch
+import torch.distributed as dist
+
+from megatron.core import parallel_state
+from megatron.core.distributed import DistributedDataParallelConfig
+from megatron.core.distributed.finalize_model_grads import (
+    _update_router_expert_bias_with_quantile,
+    finalize_model_grads,
+    reset_model_temporary_tensors,
+)
+from megatron.core.extensions.transformer_engine import fused_topk_with_score_function_supports_qb
+from megatron.core.transformer.moe.moe_utils import (
+    get_updated_expert_bias_with_quantile,
+    topk_routing_with_score_function,
+)
+from megatron.core.transformer.transformer_config import TransformerConfig
+
+
+class _QBFinalizeModel(torch.nn.Module):
+    """Minimal model wrapper that runs the production gradient finalizer."""
+
+    def __init__(self, router: torch.nn.Module, config: TransformerConfig):
+        super().__init__()
+        self.router = router
+        self.config = config
+        self.ddp_config = DistributedDataParallelConfig()
+        self.finish_grad_sync_calls = 0
+
+    def finish_grad_sync(self, force_all_reduce: bool = False):
+        del force_all_reduce
+        self.finish_grad_sync_calls += 1
+
+
+def _config(**overrides):
+    kwargs = dict(
+        num_layers=1,
+        hidden_size=16,
+        num_attention_heads=4,
+        num_moe_experts=4,
+        moe_router_topk=1,
+        moe_router_score_function="sigmoid",
+        moe_router_load_balancing_type="quantile_balancing",
+        moe_router_quantile_balancing_estimation_scope="global_batch",
+        moe_aux_loss_coeff=0.0,
+    )
+    kwargs.update(overrides)
+    return TransformerConfig(**kwargs)
+
+
+def test_qb_global_batch_config():
+    config = _config()
+    assert config.moe_router_quantile_balancing_estimation_scope == "global_batch"
+
+
+def test_qb_accepts_singleton_load_balancing_list():
+    config = _config(
+        moe_router_load_balancing_type=["quantile_balancing"], moe_aux_loss_coeff=[0.0]
+    )
+    assert config.moe_router_load_balancing_type == "quantile_balancing"
+    assert config.moe_aux_loss_coeff == 0.0
+
+
+def test_qb_cli_exposes_global_batch_scope_and_histogram_bins():
+    from megatron.training.arguments import add_megatron_arguments
+
+    parser = argparse.ArgumentParser()
+    add_megatron_arguments(parser)
+    args = parser.parse_args(
+        [
+            "--moe-router-load-balancing-type",
+            "quantile_balancing",
+            "--moe-router-quantile-balancing-estimation-scope",
+            "global_batch",
+            "--moe-router-qb-num-bins",
+            "257",
+        ]
+    )
+
+    assert args.moe_router_load_balancing_type == ["quantile_balancing"]
+    assert args.moe_router_quantile_balancing_estimation_scope == "global_batch"
+    assert args.moe_router_qb_num_bins == 257
+
+
+def test_qb_dev_rejects_micro_batch_scope():
+    with pytest.raises(ValueError, match="dev supports only"):
+        _config(moe_router_quantile_balancing_estimation_scope="micro_batch")
+
+
+def test_qb_rejects_nonzero_aux_loss():
+    with pytest.raises(ValueError, match="moe_aux_loss_coeff=0"):
+        _config(moe_aux_loss_coeff=0.01)
+
+
+def test_qb_rejects_pre_softmax_routing():
+    with pytest.raises(ValueError, match="does not use pre-softmax"):
+        _config(moe_router_pre_softmax=True)
+
+
+def test_qb_treats_negative_capacity_as_disabled():
+    config = _config(moe_expert_capacity_factor=-1.0)
+    assert config.moe_expert_capacity_factor is None
+
+
+def test_qb_negative_capacity_preserves_sinkhorn_validation():
+    with pytest.raises(ValueError, match="moe_expert_capacity_factor only works"):
+        _config(moe_router_load_balancing_type="sinkhorn", moe_expert_capacity_factor=-1.0)
+
+
+def test_qb_rejects_active_expert_capacity():
+    with pytest.raises(ValueError, match="does not support per-expert token dropping"):
+        _config(moe_expert_capacity_factor=1.0)
+
+
+def test_qb_histogram_recovery():
+    histogram = torch.tensor([[0, 2, 2, 0], [2, 2, 0, 0]], dtype=torch.int32)
+    bin_bounds = torch.tensor([-1.0, 1.0], dtype=torch.float32)
+    expert_bias = torch.zeros(2, dtype=torch.float32)
+
+    updated_bias, updated_bounds = get_updated_expert_bias_with_quantile(
+        histogram, bin_bounds, expert_bias, topk=1
+    )
+
+    torch.testing.assert_close(updated_bias, torch.tensor([0.25, -0.25]))
+    torch.testing.assert_close(updated_bounds, torch.tensor([-1.25, 1.25]))
+
+
+def test_qb_uses_pooled_global_batch_quantile_not_mean_microbatch_quantile():
+    bounds = torch.tensor([0.0, 10.0], dtype=torch.float32)
+    bias = torch.zeros(2, dtype=torch.float32)
+    mb1 = torch.zeros(2, 10, dtype=torch.int32)
+    mb2 = torch.zeros_like(mb1)
+    mb1[0, 0], mb1[0, 9], mb1[1, 0] = 3, 1, 4
+    mb2[0, 9], mb2[1, 0] = 4, 4
+
+    pooled_bias, _ = get_updated_expert_bias_with_quantile(mb1 + mb2, bounds, bias, topk=1)
+    mb1_bias, _ = get_updated_expert_bias_with_quantile(mb1, bounds, bias, topk=1)
+    mb2_bias, _ = get_updated_expert_bias_with_quantile(mb2, bounds, bias, topk=1)
+
+    assert not torch.allclose(pooled_bias, (mb1_bias + mb2_bias) / 2)
+
+
+def test_qb_unfused_histogram_accumulates_microbatches():
+    logits = torch.tensor([[-1.0, 0.5, 1.5], [2.0, -0.5, 0.25]], dtype=torch.float32)
+    bias = torch.tensor([-0.2, 0.1, 0.0], dtype=torch.float32)
+    bounds = torch.tensor([-1.0, 1.0], dtype=torch.float32)
+    histogram = torch.zeros(3, 8, dtype=torch.int32)
+
+    scores = torch.sigmoid(logits)
+    topk_result = torch.topk(scores + bias, 2, dim=1, sorted=True)
+    cutoff = topk_result.values[:, -1:]
+    expected_bins = torch.floor((cutoff - scores - bounds[0]) * (8 / (bounds[1] - bounds[0])))
+    expected_bins = expected_bins.to(torch.int64).clamp_(0, 7)
+    expected = torch.zeros_like(histogram)
+    offsets = torch.arange(3, dtype=torch.int64) * 8
+    flat_indices = (expected_bins + offsets).reshape(-1)
+    expected.view(-1).scatter_add_(
+        0, flat_indices, torch.ones_like(flat_indices, dtype=torch.int32)
+    )
+
+    outputs = []
+    for _ in range(2):
+        outputs.append(
+            topk_routing_with_score_function(
+                logits,
+                topk=1,
+                score_function="sigmoid",
+                expert_bias=bias,
+                fused=False,
+                qb_histogram=histogram,
+                qb_bin_bounds=bounds,
+            )
+        )
+
+    torch.testing.assert_close(histogram, expected * 2)
+    selected = topk_result.indices[:, :1]
+    expected_map = torch.zeros_like(logits, dtype=torch.bool).scatter_(1, selected, True)
+    torch.testing.assert_close(outputs[0][1], expected_map)
+    torch.testing.assert_close(outputs[1][1], expected_map)
+
+
+@pytest.mark.skipif(
+    not torch.cuda.is_available() or not fused_topk_with_score_function_supports_qb,
+    reason="requires the Transformer Engine QB fused-router API",
+)
+def test_qb_fused_atomic_matches_unfused_histogram():
+    torch.manual_seed(1234)
+    logits = torch.randn(37, 64, device="cuda", dtype=torch.float32)
+    bias = torch.linspace(-0.1, 0.1, 64, device="cuda", dtype=torch.float32)
+    bounds = torch.tensor([-1.1, 1.1], device="cuda", dtype=torch.float32)
+    unfused_histogram = torch.zeros(64, 128, device="cuda", dtype=torch.int32)
+    fused_histogram = torch.zeros_like(unfused_histogram)
+
+    unfused_probs, unfused_map = topk_routing_with_score_function(
+        logits,
+        topk=8,
+        score_function="sigmoid",
+        expert_bias=bias,
+        fused=False,
+        qb_histogram=unfused_histogram,
+        qb_bin_bounds=bounds,
+    )
+    fused_probs, fused_map = topk_routing_with_score_function(
+        logits,
+        topk=8,
+        score_function="sigmoid",
+        expert_bias=bias,
+        fused=True,
+        qb_histogram=fused_histogram,
+        qb_bin_bounds=bounds,
+    )
+
+    torch.testing.assert_close(fused_probs, unfused_probs)
+    torch.testing.assert_close(fused_map, unfused_map)
+    torch.testing.assert_close(fused_histogram, unfused_histogram)
+
+
+def test_qb_finalize_updates_once_and_reset_preserves_buffers():
+    router = torch.nn.Module()
+    router.register_buffer("expert_bias", torch.zeros(2, dtype=torch.float32))
+    router.register_buffer(
+        "qb_histogram",
+        torch.tensor([[0, 2, 2, 0], [2, 2, 0, 0]], dtype=torch.int32),
+        persistent=False,
+    )
+    router.register_buffer("qb_bin_bounds", torch.tensor([-1.0, 1.0], dtype=torch.float32))
+    router.frozen_expert_bias = False
+    model = torch.nn.Module()
+    model.router = router
+    config = SimpleNamespace(
+        moe_router_topk=1,
+        moe_router_enable_expert_bias=False,
+        moe_router_load_balancing_type="quantile_balancing",
+    )
+    histogram_ptr = router.qb_histogram.data_ptr()
+    bounds_ptr = router.qb_bin_bounds.data_ptr()
+
+    _update_router_expert_bias_with_quantile([model], config)
+
+    torch.testing.assert_close(router.expert_bias, torch.tensor([0.25, -0.25]))
+    reset_model_temporary_tensors(config, [model])
+    assert torch.count_nonzero(router.qb_histogram) == 0
+    assert router.qb_histogram.data_ptr() == histogram_ptr
+    assert router.qb_bin_bounds.data_ptr() == bounds_ptr
+
+
+def test_qb_paged_stash_retry_discards_failed_attempt_histogram():
+    from megatron.core.transformer.moe.paged_stash import PagedStashRunner
+
+    histogram = torch.ones(3, 7, dtype=torch.int32)
+    histogram_ptr = histogram.data_ptr()
+    runner = PagedStashRunner.__new__(PagedStashRunner)
+    runner.moe_layers = [SimpleNamespace(router=SimpleNamespace(qb_histogram=histogram))]
+
+    runner._reset_qb_histograms()
+
+    assert torch.count_nonzero(histogram) == 0
+    assert histogram.data_ptr() == histogram_ptr
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="requires CUDA")
+@pytest.mark.internal
+@pytest.mark.parametrize(
+    "fused",
+    [
+        False,
+        pytest.param(
+            True,
+            marks=pytest.mark.skipif(
+                not fused_topk_with_score_function_supports_qb,
+                reason="requires the Transformer Engine QB fused-router API",
+            ),
+        ),
+    ],
+)
+def test_qb_mcore_router_accumulates_microbatches_and_finalizes(fused):
+    """Exercise unfused and fused QB through a real MoE router and finalizer."""
+    from megatron.core.models.gpt.gpt_layer_specs import get_gpt_layer_local_submodules
+    from megatron.core.transformer.moe.moe_layer import MoELayer, MoESubmodules
+    from megatron.core.transformer.spec_utils import get_submodules
+    from megatron.training.initialize import _set_random_seed
+    from tests.unit_tests.test_utilities import Utils
+
+    Utils.destroy_model_parallel()
+    Utils.initialize_model_parallel(1, 1)
+    _set_random_seed(seed_=123, data_parallel_random_init=False)
+    config = _config(
+        hidden_size=16,
+        ffn_hidden_size=32,
+        moe_router_topk=1,
+        moe_router_fusion=fused,
+        moe_token_dispatcher_type="alltoall",
+        moe_router_qb_num_bins=128,
+        params_dtype=torch.float32,
+        add_bias_linear=False,
+    )
+    submodules = get_submodules(
+        get_gpt_layer_local_submodules(config.num_moe_experts, moe_grouped_gemm=False).mlp
+    )
+    assert isinstance(submodules, MoESubmodules)
+    layer = MoELayer(config, submodules).cuda()
+    with torch.no_grad():
+        layer.router.weight.zero_()
+        layer.router.weight[:, 0].copy_(torch.tensor([3.0, 1.0, -0.5, -2.0], device="cuda"))
+
+    for direction in (1.0, -1.0):
+        hidden_states = torch.zeros(64, 1, config.hidden_size, device="cuda")
+        hidden_states[:, 0, 0] = direction * torch.linspace(-2.0, 2.0, 64, device="cuda")
+        output, _ = layer(hidden_states)
+        output.float().square().mean().backward()
+
+    router = layer.router
+    assert router.expert_bias is not None
+    assert "expert_bias" in router.state_dict()
+    assert not hasattr(router, "qb_bias")
+    torch.testing.assert_close(
+        router.qb_histogram.sum(dim=1),
+        torch.full((config.num_moe_experts,), 128, dtype=torch.int64, device="cuda"),
+    )
+    histogram_ptr = router.qb_histogram.data_ptr()
+    bounds_ptr = router.qb_bin_bounds.data_ptr()
+    old_bias = router.expert_bias.clone()
+    _update_router_expert_bias_with_quantile(
+        [layer], config, tp_dp_cp_group=torch.distributed.group.WORLD
+    )
+
+    assert not torch.equal(router.expert_bias, old_bias)
+    torch.testing.assert_close(
+        router.expert_bias.mean(), torch.zeros((), device="cuda"), atol=1e-7, rtol=0
+    )
+    reset_model_temporary_tensors(config, [layer])
+    assert torch.count_nonzero(router.qb_histogram) == 0
+    assert router.qb_histogram.data_ptr() == histogram_ptr
+    assert router.qb_bin_bounds.data_ptr() == bounds_ptr
+    Utils.destroy_model_parallel()
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="requires CUDA")
+@pytest.mark.internal
+def test_qb_router_histogram_gating_and_activation_recompute():
+    """Accumulate once under recompute, but not in eval, frozen, or no-grad forwards."""
+    from megatron.core.models.gpt.gpt_layer_specs import get_gpt_layer_local_submodules
+    from megatron.core.tensor_parallel.random import checkpoint
+    from megatron.core.transformer.moe.moe_layer import MoELayer, MoESubmodules
+    from megatron.core.transformer.spec_utils import get_submodules
+    from megatron.training.initialize import _set_random_seed
+    from tests.unit_tests.test_utilities import Utils
+
+    Utils.destroy_model_parallel()
+    try:
+        Utils.initialize_model_parallel(1, 1)
+        _set_random_seed(seed_=123, data_parallel_random_init=False)
+        config = _config(
+            hidden_size=16,
+            ffn_hidden_size=32,
+            moe_router_topk=2,
+            moe_router_fusion=False,
+            moe_token_dispatcher_type="alltoall",
+            moe_router_qb_num_bins=128,
+            params_dtype=torch.float32,
+            add_bias_linear=False,
+        )
+        submodules = get_submodules(
+            get_gpt_layer_local_submodules(config.num_moe_experts, moe_grouped_gemm=False).mlp
+        )
+        assert isinstance(submodules, MoESubmodules)
+        router = MoELayer(config, submodules).cuda().router
+        hidden_states = torch.randn(32, 1, config.hidden_size, device="cuda", requires_grad=True)
+
+        router.eval()
+        router(hidden_states)
+        assert torch.count_nonzero(router.qb_histogram) == 0
+
+        router.train()
+        router.frozen_expert_bias = True
+        router(hidden_states)
+        assert torch.count_nonzero(router.qb_histogram) == 0
+
+        router.frozen_expert_bias = False
+        with torch.no_grad():
+            router(hidden_states)
+        assert torch.count_nonzero(router.qb_histogram) == 0
+
+        routing_probs, _ = checkpoint(router, False, hidden_states)
+        assert torch.count_nonzero(router.qb_histogram) == 0
+        routing_probs.square().sum().backward()
+        torch.testing.assert_close(
+            router.qb_histogram.sum(dim=1),
+            torch.full(
+                (config.num_moe_experts,),
+                hidden_states.shape[0] * hidden_states.shape[1],
+                dtype=torch.int64,
+                device="cuda",
+            ),
+        )
+    finally:
+        Utils.destroy_model_parallel()
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="requires CUDA")
+@pytest.mark.internal
+@pytest.mark.parametrize(
+    "fused",
+    [
+        False,
+        pytest.param(
+            True,
+            marks=pytest.mark.skipif(
+                not fused_topk_with_score_function_supports_qb,
+                reason="requires the Transformer Engine QB fused-router API",
+            ),
+        ),
+    ],
+)
+@pytest.mark.parametrize(
+    "tp_size,ep_size,dense_dp_size,expert_dp_size", [(1, 4, 8, 2), (1, 8, 8, 1), (4, 2, 2, 1)]
+)
+def test_qb_world8_ep_topologies_finalize_model_grads(
+    fused, tp_size, ep_size, dense_dp_size, expert_dp_size
+):
+    """Validate QB reduction and finalization across eight-rank EP topologies."""
+    from megatron.core.models.gpt.gpt_layer_specs import get_gpt_layer_local_submodules
+    from megatron.core.transformer.moe.moe_layer import MoELayer, MoESubmodules
+    from megatron.core.transformer.spec_utils import get_submodules
+    from megatron.training.initialize import _set_random_seed
+    from tests.unit_tests.test_utilities import Utils
+
+    # torchrun exports WORLD_SIZE before MCore initializes the process group. Guard on the
+    # environment here so a normal single-GPU unit-test run skips before trying to construct an
+    # impossible EP4/EP8 topology, while the intended eight-rank launch reaches initialization.
+    if int(os.environ.get("WORLD_SIZE", "1")) != 8:
+        pytest.skip("requires a world size of 8")
+
+    Utils.destroy_model_parallel()
+    try:
+        Utils.initialize_model_parallel(
+            tensor_model_parallel_size=tp_size,
+            pipeline_model_parallel_size=1,
+            expert_model_parallel_size=ep_size,
+            expert_tensor_parallel_size=tp_size,
+        )
+        _set_random_seed(seed_=123, data_parallel_random_init=False)
+
+        # Establish that MCore generated exactly the topology under test. In particular, EP is
+        # folded into the router's dense-data-parallel group, while expert data parallelism is
+        # world_size / (TP * EP). The production QB reduction group spans TP x dense-DP x CP and
+        # must therefore contain all eight ranks for every parameterized topology.
+        assert dist.get_world_size() == 8
+        assert parallel_state.get_tensor_model_parallel_world_size() == tp_size
+        assert parallel_state.get_expert_model_parallel_world_size() == ep_size
+        assert parallel_state.get_data_parallel_world_size() == dense_dp_size
+        assert parallel_state.get_expert_data_parallel_world_size() == expert_dp_size
+        tp_dp_cp_group = parallel_state.get_tensor_and_data_parallel_group(
+            with_context_parallel=True
+        )
+        assert tp_dp_cp_group.size() == 8
+
+        config = _config(
+            hidden_size=16,
+            ffn_hidden_size=32,
+            num_moe_experts=8,
+            moe_router_topk=1,
+            moe_router_fusion=fused,
+            moe_token_dispatcher_type="alltoall",
+            moe_router_qb_num_bins=128,
+            tensor_model_parallel_size=tp_size,
+            expert_model_parallel_size=ep_size,
+            expert_tensor_parallel_size=tp_size,
+            # MCore requires sequence parallelism whenever training combines MoE with TP. The
+            # TP1 cases deliberately leave it off; TP4/EP2 turns it on to exercise the supported
+            # expert-tensor-parallel execution path rather than bypassing MoELayer's safety check.
+            sequence_parallel=tp_size > 1,
+            params_dtype=torch.float32,
+            add_bias_linear=False,
+        )
+        submodules = get_submodules(
+            get_gpt_layer_local_submodules(config.num_moe_experts, moe_grouped_gemm=False).mlp
+        )
+        assert isinstance(submodules, MoESubmodules)
+        layer = MoELayer(config, submodules).cuda()
+        with torch.no_grad():
+            layer.router.weight.zero_()
+            layer.router.weight[:, 0].copy_(
+                torch.tensor([3.0, 2.0, 1.0, 0.25, -0.25, -1.0, -2.0, -3.0], device="cuda")
+            )
+
+        dense_dp_rank = parallel_state.get_data_parallel_rank()
+        token_axis = torch.linspace(-2.0, 2.0, 32, device="cuda")
+        for microbatch in range(2):
+            hidden_states = torch.zeros(32, 1, config.hidden_size, device="cuda")
+            hidden_states[:, 0, 0] = (
+                (1.0 if microbatch == 0 else -1.0) * token_axis
+                + 0.35 * dense_dp_rank
+                + 0.2 * microbatch
+            )
+            output, _ = layer(hidden_states)
+            output.float().square().mean().backward()
+
+        # Every token contributes one margin sample to every expert's histogram, independent of
+        # top-k. Two 32-token microbatches must therefore leave exactly 64 samples per expert in
+        # each rank-local accumulator. This also proves accumulation happened across microbatches
+        # instead of replacing the first microbatch's statistics.
+        router = layer.router
+        local_histogram = router.qb_histogram.clone()
+        torch.testing.assert_close(
+            local_histogram.sum(dim=1),
+            torch.full((config.num_moe_experts,), 64, dtype=torch.int64, device="cuda"),
+        )
+        gathered_histograms = [torch.empty_like(local_histogram) for _ in range(8)]
+        dist.all_gather(gathered_histograms, local_histogram)
+
+        # Rank-dependent inputs deliberately produce at least two distinct local histograms. This
+        # prevents a false-positive where final biases agree merely because every rank started from
+        # identical statistics, without exercising the distributed histogram reduction.
+        assert any(
+            not torch.equal(gathered_histograms[0], histogram)
+            for histogram in gathered_histograms[1:]
+        )
+
+        # Build an independent oracle from an explicit all-reduce of the pre-finalization local
+        # histogram, then apply the same pure quantile update math. finalize_model_grads must
+        # reproduce both the resulting expert bias and the adaptively updated bin bounds.
+        expected_histogram = local_histogram.clone()
+        dist.all_reduce(expected_histogram, group=tp_dp_cp_group)
+        expected_bias, expected_bounds = get_updated_expert_bias_with_quantile(
+            expected_histogram,
+            router.qb_bin_bounds.clone(),
+            router.expert_bias.clone(),
+            config.moe_router_topk,
+        )
+        histogram_ptr = router.qb_histogram.data_ptr()
+        bounds_ptr = router.qb_bin_bounds.data_ptr()
+        model = _QBFinalizeModel(router, config)
+
+        finalize_model_grads([model])
+
+        torch.testing.assert_close(router.expert_bias, expected_bias)
+        torch.testing.assert_close(router.qb_bin_bounds, expected_bounds)
+
+        # Verify global agreement explicitly on WORLD, rather than checking only each rank against
+        # its locally computed oracle. This catches a wrong reduction group that could otherwise
+        # leave different EP or TP subgroups internally self-consistent.
+        gathered_biases = [torch.empty_like(router.expert_bias) for _ in range(8)]
+        gathered_bounds = [torch.empty_like(router.qb_bin_bounds) for _ in range(8)]
+        dist.all_gather(gathered_biases, router.expert_bias)
+        dist.all_gather(gathered_bounds, router.qb_bin_bounds)
+        for bias, bounds in zip(gathered_biases[1:], gathered_bounds[1:]):
+            torch.testing.assert_close(bias, gathered_biases[0])
+            torch.testing.assert_close(bounds, gathered_bounds[0])
+
+        # Finalization consumes the global-batch histogram exactly once and clears it in place.
+        # Stable storage addresses are required by full-iteration CUDA graph capture; the persistent
+        # bin-bound buffer must likewise be updated without replacement. The wrapper also proves the
+        # normal gradient synchronization stage still executes exactly once.
+        assert torch.count_nonzero(router.qb_histogram) == 0
+        assert router.qb_histogram.data_ptr() == histogram_ptr
+        assert router.qb_bin_bounds.data_ptr() == bounds_ptr
+        assert model.finish_grad_sync_calls == 1
+    finally:
+        Utils.destroy_model_parallel()