diff --git a/megatron/core/distributed/finalize_model_grads.py b/megatron/core/distributed/finalize_model_grads.py
index d5be7607714..4902b0f9b1f 100644
--- a/megatron/core/distributed/finalize_model_grads.py
+++ b/megatron/core/distributed/finalize_model_grads.py
@@ -13,6 +13,7 @@
 except ImportError:
     HAVE_DTENSOR = False
 
+from megatron.core.extensions.transformer_engine import mark_qb_bin_bounds_validated
 from megatron.core.pipeline_parallel.utils import (
     get_pp_last_rank,
     is_pp_first_stage,
@@ -21,7 +22,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,
@@ -329,6 +333,8 @@ def reset_model_temporary_tensors(config: TransformerConfig, model: List[torch.n
             if getattr(module, 'qb_beta_accum', None) is not None:
                 module.qb_beta_accum.zero_()
                 module.qb_beta_count.zero_()
+            if getattr(module, 'qb_histogram', None) is not None:
+                module.qb_histogram.zero_()
 
 
 def _update_router_expert_bias(
@@ -413,6 +419,47 @@ def _update_router_qb_beta(
         qb_beta.copy_(new_beta)
 
 
+def _update_router_expert_bias_with_quantile(
+    model: List[torch.nn.Module],
+    config: TransformerConfig,
+    reduction_groups: tuple[Optional[torch.distributed.ProcessGroup], ...] = (),
+):
+    """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)
+    for group in reduction_groups:
+        if get_pg_size(group) > 1:
+            torch.distributed.all_reduce(stacked_histogram, group=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)
+        if mark_qb_bin_bounds_validated is not None and bounds.is_cuda:
+            mark_qb_bin_bounds_validated(bounds)
+
+
 def _allreduce_non_tensor_model_parallel_grads(
     model: List[torch.nn.Module],
     config: TransformerConfig,
@@ -589,10 +636,14 @@ 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"
+            and config.moe_router_quantile_balancing_estimation_scope == "global_batch"
+            and config.gtp_weight_remat_size <= 1
+        ):
+            assert (
+                hasattr(pg_collection, 'tp_dp_cp') and pg_collection.tp_dp_cp is not None
+            ), "pg_collection must have tp_dp_cp when global router bias updates are enabled."
             tp_dp_cp_group = pg_collection.tp_dp_cp
         tp_group = pg_collection.tp
         pp_group = pg_collection.pp
@@ -687,7 +738,21 @@ 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":
-        _update_router_qb_beta(model, config, dp_cp_group=dp_cp_group)
+        if config.moe_router_quantile_balancing_estimation_scope == "micro_batch":
+            _update_router_qb_beta(model, config, dp_cp_group=dp_cp_group)
+        else:
+            if config.gtp_weight_remat_size > 1:
+                qb_reduction_groups = (tp_group, dp_cp_group)
+            else:
+                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
+                    )
+                qb_reduction_groups = (tp_dp_cp_group,)
+            _update_router_expert_bias_with_quantile(
+                model, config, reduction_groups=qb_reduction_groups
+            )
 
     reset_model_temporary_tensors(config, model)
 
diff --git a/megatron/core/extensions/transformer_engine.py b/megatron/core/extensions/transformer_engine.py
index 48f8ff2498b..c0beb2b88a4 100644
--- a/megatron/core/extensions/transformer_engine.py
+++ b/megatron/core/extensions/transformer_engine.py
@@ -3673,10 +3673,30 @@ def te_general_gemm(
         fused_topk_with_score_function,
     )
 
+    try:
+        from transformer_engine.pytorch.router import (  # pylint: disable=unused-import
+            mark_qb_bin_bounds_validated,
+        )
+    except ImportError:
+        mark_qb_bin_bounds_validated = None
+
+    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
+    mark_qb_bin_bounds_validated = None
 
 
 def set_save_original_input(module):
diff --git a/megatron/core/transformer/moe/README.md b/megatron/core/transformer/moe/README.md
index f3c35fe4f6e..8dd16929450 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** | Exact micro-batch or Kimi K3 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` |
 
@@ -526,7 +527,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: micro_batch or global_batch | micro_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 8d88d6a690f..0bbb3a5d022 100644
--- a/megatron/core/transformer/moe/moe_utils.py
+++ b/megatron/core/transformer/moe/moe_utils.py
@@ -42,6 +42,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,
     )
@@ -58,6 +59,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(
@@ -776,6 +778,8 @@ def topk_routing_with_score_function(
     router_replay: Optional['RouterReplay'] = None,
     dense_output: bool = False,
     precomputed_indices: Optional[torch.Tensor] = None,
+    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.
 
@@ -804,6 +808,10 @@ def topk_routing_with_score_function(
                                        selected by the caller. When given, the score function's
                                        own top-k is bypassed and probs are computed at these
                                        indices (e.g. for quantile balancing). Defaults to None.
+        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]:
@@ -822,6 +830,33 @@ 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 precomputed_indices is not None:
+            raise ValueError("Histogram Quantile Balancing computes its own top-(k+1).")
+        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].")
     assert not (
         fused and precomputed_indices is not None
     ), "precomputed_indices is not supported with the fused top-k score function."
@@ -835,16 +870,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,
@@ -922,7 +970,27 @@ def compute_topk(scores, topk, num_groups=None, group_topk=None):
             scores = torch.gather(scores, dim=1, index=top_indices)
         elif 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)
@@ -1229,6 +1297,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 a3a3fff76c1..0c935daf6f0 100644
--- a/megatron/core/transformer/moe/paged_stash.py
+++ b/megatron/core/transformer/moe/paged_stash.py
@@ -1053,6 +1053,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 = []
@@ -1168,6 +1176,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 6ce93c76837..a22f5ed7506 100644
--- a/megatron/core/transformer/moe/router.py
+++ b/megatron/core/transformer/moe/router.py
@@ -180,6 +180,7 @@ def __init__(
         self.score_function = self.config.moe_router_score_function
         self.input_jitter = None
         self.frozen_expert_bias = False
+        self.qb_estimation_scope = self.config.moe_router_quantile_balancing_estimation_scope
 
         self.enable_expert_bias = self.config.moe_router_enable_expert_bias
         if self.enable_expert_bias:
@@ -192,6 +193,12 @@ def __init__(
                 ),
                 persistent=False,
             )
+        else:
+            self.local_tokens_per_expert = None
+
+        if self.enable_expert_bias or (
+            self.routing_type == "quantile_balancing" and self.qb_estimation_scope == "global_batch"
+        ):
             self.register_buffer(
                 'expert_bias',
                 torch.zeros(
@@ -201,7 +208,6 @@ def __init__(
                 ),
             )
         else:
-            self.local_tokens_per_expert = None
             self.expert_bias = None
 
         # Initialize global tokens per expert for global aux loss
@@ -224,10 +230,32 @@ def __init__(
             self.global_tokens_per_expert = None
             self.ga_steps = None
 
+        if self.routing_type == "quantile_balancing" and self.qb_estimation_scope == "global_batch":
+            assert not self.is_aux_loss_enabled(), (
+                "Quantile balancing handles load balance via the bias update; "
+                "aux losses must be disabled (set moe_aux_loss_coeff to 0)."
+            )
+            self.qb_beta = None
+            self.qb_beta_accum = None
+            self.qb_beta_count = None
+            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()),
+            )
         # Quantile balancing replaces the aux loss with a per-expert bias `qb_beta`.
         # `qb_beta_accum`/`qb_beta_count` collect the per-microbatch quantile, reduced
         # and reset each global batch.
-        if self.routing_type == "quantile_balancing":
+        elif self.routing_type == "quantile_balancing":
             assert not self.is_aux_loss_enabled(), (
                 "Quantile balancing handles load balance via the bias update; "
                 "aux losses must be disabled (set moe_aux_loss_coeff to 0)."
@@ -254,10 +282,14 @@ def __init__(
                 torch.zeros((), dtype=torch.long, device=torch.cuda.current_device()),
                 persistent=False,
             )
+            self.qb_histogram = None
+            self.qb_bin_bounds = None
         else:
             self.qb_beta = None
             self.qb_beta_accum = None
             self.qb_beta_count = None
+            self.qb_histogram = None
+            self.qb_bin_bounds = None
 
         self.router_replay = None
         if self.config.moe_enable_routing_replay:
@@ -280,6 +312,9 @@ def _maintain_float32_expert_bias(self):
         if hasattr(self, 'qb_beta_accum') and self.qb_beta_accum is not None:
             if self.qb_beta_accum.dtype != torch.float32:
                 self.qb_beta_accum.data = self.qb_beta_accum.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.
@@ -317,9 +352,10 @@ def _sinkhorn_activation(logits):
     def quantile_balancing(self, logits: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
         """Apply quantile-balancing (QB) routing to the logits tensor.
 
-        Selects top-k experts per token using a dual coordinate-descent update on
-        a per-expert bias ``qb_beta``. Load balance is handled entirely by the bias
-        update; auxiliary losses must be disabled when QB is active.
+        ``micro_batch`` uses the existing exact estimator and ``qb_beta`` state.
+        ``global_batch`` accumulates Kimi K3 margin histograms and updates the shared
+        router ``expert_bias`` at the global-batch boundary. Auxiliary losses must be
+        disabled for either estimator.
 
         Args:
             logits (torch.Tensor): The logits tensor, shape ``[num_tokens, num_experts]``.
@@ -328,6 +364,24 @@ def quantile_balancing(self, logits: torch.Tensor) -> tuple[torch.Tensor, torch.
             Tuple[torch.Tensor, torch.Tensor]: Sparse routing probs and boolean
             routing map, each shaped ``[num_tokens, num_experts]``.
         """
+        if self.qb_estimation_scope == "global_batch":
+            # Activation checkpointing runs the original forward under no-grad and its
+            # recompute under enable-grad, so this gate records each token exactly once.
+            accumulate_histogram = (
+                self.training and torch.is_grad_enabled() and not self.frozen_expert_bias
+            )
+            return topk_routing_with_score_function(
+                logits,
+                self.topk,
+                use_pre_softmax=self.config.moe_router_pre_softmax,
+                scaling_factor=self.config.moe_router_topk_scaling_factor,
+                score_function=self.score_function,
+                expert_bias=self.expert_bias,
+                fused=self.config.moe_router_fusion,
+                qb_histogram=self.qb_histogram if accumulate_histogram else None,
+                qb_bin_bounds=self.qb_bin_bounds if accumulate_histogram else None,
+            )
+
         assert (
             not self.config.moe_router_fusion
         ), "Quantile balancing routing does not support moe_router_fusion."
@@ -957,7 +1011,8 @@ def _compiled_topk_routing(
     def _forward(self, input: torch.Tensor, padding_mask: Optional[torch.Tensor] = None):
         logits = self.gating(input).squeeze(1)  # [num_tokens, num_experts]
 
-        # QB selects on (logits - qb_beta); at inference qb_beta is fixed, so it's per-token.
+        # Micro-batch QB selects on (logits - qb_beta). Global-batch QB passes its
+        # fixed additive bias through the normal score-function route.
         precomputed_indices = None
         if self.qb_beta is not None:
             precomputed_indices = (logits - self.qb_beta).topk(self.topk, dim=1).indices
diff --git a/megatron/core/transformer/transformer_config.py b/megatron/core/transformer/transformer_config.py
index 245a164fb63..9186a765dde 100644
--- a/megatron/core/transformer/transformer_config.py
+++ b/megatron/core/transformer/transformer_config.py
@@ -750,9 +750,9 @@ 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": Dual coordinate-descent quantile balancing (QB). Load balance is
-    handled entirely by an internal per-expert bias update; auxiliary losses must be disabled
-    (`moe_aux_loss_coeff` = 0) when QB is selected.
+    - "quantile_balancing": Quantile balancing (QB). ``micro_batch`` preserves the exact
+    per-microbatch estimator, while ``global_batch`` uses Kimi K3's pooled histogram estimator.
+    Load balance is handled internally and auxiliary losses must be disabled.
     - "none": No load balancing.
     A list of strings can be provided to combine multiple aux-loss load balancing types.
     The default is "aux_loss".
@@ -829,10 +829,21 @@ class TransformerConfig(ModelParallelConfig):
     The default value 1e-3 is same as that used in DeepSeekV3."""
 
     moe_router_quantile_balancing_ema: float = 0.0
-    """EMA coefficient for the quantile-balancing per-expert bias (`qb_beta`), used only when
-    `moe_router_load_balancing_type` is "quantile_balancing". At each global batch the bias is
-    updated as `qb_beta = ema * qb_beta + (1 - ema) * local_quantile`. The default 0.0 means
-    no memory: the bias is replaced by the latest global-batch quantile estimate each step."""
+    """EMA coefficient for the micro-batch quantile-balancing bias (`qb_beta`). At each global
+    batch the bias is updated as `qb_beta = ema * qb_beta + (1 - ema) * local_quantile`. The
+    default 0.0 means no memory: the bias is replaced by the latest estimate each step."""
+
+    moe_router_quantile_balancing_estimation_scope: Literal['micro_batch', 'global_batch'] = (
+        "micro_batch"
+    )
+    """Population used to estimate the Quantile Balancing bias.
+
+    ``micro_batch`` preserves the pre-existing exact estimator. ``global_batch`` accumulates
+    Kimi K3 histograms across gradient-accumulation microbatches and estimates one pooled quantile.
+    """
+
+    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 
@@ -1886,6 +1897,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
@@ -1894,6 +1915,77 @@ 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."
+                )
+            scope = self.moe_router_quantile_balancing_estimation_scope
+            if scope not in ("micro_batch", "global_batch"):
+                raise ValueError(
+                    "moe_router_quantile_balancing_estimation_scope must be "
+                    "'micro_batch' or 'global_batch'."
+                )
+            if scope == "global_batch":
+                if self.moe_router_quantile_balancing_ema != 0.0:
+                    raise ValueError(
+                        "global_batch quantile_balancing derives the K3 bias directly and "
+                        "requires moe_router_quantile_balancing_ema=0."
+                    )
+                if self.moe_router_score_function != "sigmoid":
+                    raise ValueError(
+                        "global_batch quantile_balancing requires "
+                        "moe_router_score_function='sigmoid'."
+                    )
+                if self.moe_router_pre_softmax:
+                    raise ValueError(
+                        "global_batch quantile_balancing does not use pre-softmax routing."
+                    )
+                if self.moe_router_enable_expert_bias:
+                    raise ValueError(
+                        "global_batch 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("global_batch quantile_balancing does not support groups.")
+                if self.moe_enable_routing_replay:
+                    raise ValueError(
+                        "global_batch 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(
+                        "global_batch 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(
+                        "global_batch quantile_balancing with expert-rank capacity requires "
+                        "moe_paged_stash for the dropless retry."
+                    )
+                if self.num_moe_experts is None or not (
+                    0 < self.moe_router_topk < self.num_moe_experts
+                ):
+                    raise ValueError(
+                        "global_batch 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
@@ -1914,7 +2006,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 12a9067be7e..7a9c906b4e3 100644
--- a/megatron/training/arguments.py
+++ b/megatron/training/arguments.py
@@ -2304,6 +2304,8 @@ def _add_network_size_args(parser):
         "linear_attention_freq",
         "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
@@ -3540,9 +3542,14 @@ def _add_moe_args(parser):
     group.add_argument('--moe-router-load-balancing-type', nargs='+', type=str,
                        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; "quantile_balancing" (QB) uses dual coordinate descent on a per-expert bias to handle load balance internally; "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" selects either the exact micro-batch estimator or Kimi K3 global-batch histogram estimator via --moe-router-quantile-balancing-estimation-scope; "none" implies no load balancing. The default is "aux_loss".')
     group.add_argument('--moe-aux-loss-coeff', type=float, nargs='+', 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=['micro_batch', 'global_batch'], default='micro_batch',
+                       help='Population used to estimate quantile-balancing biases. "micro_batch" preserves the existing exact estimator; "global_batch" uses Kimi K3 histogram accumulation.')
+    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 9bbc2a7d8ad..6b20df08645 100644
--- a/megatron/training/checkpointing.py
+++ b/megatron/training/checkpointing.py
@@ -2310,6 +2310,10 @@ def _set_arg(arg_name, old_arg_name=None, force=False):
     else:
         setattr(args, 'moe_ffn_hidden_size', None)
     _set_arg('moe_router_topk', 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_token_dispatcher_type', force=False)
     _set_arg('moe_router_pre_softmax', force=True)
     _set_arg('moe_grouped_gemm', 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 51a0780e65d..ac9d9c3e3c9 100644
--- a/tests/unit_tests/models/test_hybrid_moe_model.py
+++ b/tests/unit_tests/models/test_hybrid_moe_model.py
@@ -229,7 +229,9 @@
     "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_ema": 0.0,
+    "moe_router_quantile_balancing_estimation_scope": "micro_batch",
     "moe_router_score_function": "sigmoid",
     "moe_router_skip_muon": True,
     "moe_router_topk": 6,
diff --git a/tests/unit_tests/test_checkpointing.py b/tests/unit_tests/test_checkpointing.py
index d5c625ca8ca..7a8755777cf 100644
--- a/tests/unit_tests/test_checkpointing.py
+++ b/tests/unit_tests/test_checkpointing.py
@@ -200,6 +200,38 @@ def test_load_args_restores_gdp_num_householder_from_checkpoint(
     assert restored_args.gdp_num_householder == expected_num_householder
 
 
+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..2227f5541d6
--- /dev/null
+++ b/tests/unit_tests/transformer/moe/test_quantile_balancing.py
@@ -0,0 +1,665 @@
+# 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_main_accepts_micro_batch_scope():
+    config = _config(moe_router_quantile_balancing_estimation_scope="micro_batch")
+    assert config.moe_router_quantile_balancing_estimation_scope == "micro_batch"
+
+
+def test_qb_main_defaults_to_micro_batch_scope():
+    config = TransformerConfig(
+        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_aux_loss_coeff=0.0,
+    )
+    assert 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_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)
+
+
+@pytest.mark.parametrize(
+    "overrides, error_match",
+    [
+        ({"moe_router_score_function": "softmax"}, "requires moe_router_score_function"),
+        ({"moe_router_enable_expert_bias": True}, "do not also enable"),
+        ({"moe_router_num_groups": 2, "moe_router_group_topk": 1}, "does not support group"),
+        ({"moe_enable_routing_replay": True}, "does not support routing replay"),
+        ({"moe_expert_rank_capacity_factor": 1.0}, "expert-rank capacity requires"),
+        ({"moe_router_topk": 4}, "requires.*0 < moe_router_topk"),
+        ({"moe_router_qb_num_bins": 1}, "must be greater than one"),
+    ],
+)
+def test_qb_rejects_incompatible_config(overrides, error_match):
+    """Reject QB combinations that cannot produce the required K3 histogram."""
+    with pytest.raises(ValueError, match=error_match):
+        _config(**overrides)
+
+
+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]))
+
+
+@pytest.mark.parametrize(
+    "overrides, error_match",
+    [
+        ({"qb_bin_bounds": None}, "must be provided together"),
+        ({"expert_bias": None}, "requires an expert bias"),
+        ({"score_function": "softmax"}, "requires score_function='sigmoid'"),
+        ({"use_pre_softmax": True}, "does not use pre-softmax"),
+        ({"topk": 4}, "requires topk < num_experts"),
+        ({"num_groups": 2}, "does not support group-limited routing"),
+        ({"router_replay": object()}, "does not support router replay"),
+        ({"qb_histogram": torch.zeros(3, 8, dtype=torch.int32)}, "must have shape"),
+        ({"qb_histogram": torch.zeros(4, 8, dtype=torch.int64)}, "dtype torch.int32"),
+        ({"qb_bin_bounds": torch.zeros(2, dtype=torch.float64)}, "must be an FP32 tensor"),
+    ],
+)
+def test_qb_routing_rejects_invalid_histogram_inputs(overrides, error_match):
+    """Validate QB-specific routing inputs before either fused or unfused dispatch."""
+    kwargs = dict(
+        logits=torch.zeros(2, 4),
+        topk=1,
+        score_function="sigmoid",
+        expert_bias=torch.zeros(4),
+        qb_histogram=torch.zeros(4, 8, dtype=torch.int32),
+        qb_bin_bounds=torch.tensor([-1.0, 1.0], dtype=torch.float32),
+    )
+    kwargs.update(overrides)
+    with pytest.raises(ValueError, match=error_match):
+        topk_routing_with_score_function(**kwargs)
+
+
+def test_qb_empty_histogram_preserves_bias_and_bounds():
+    histogram = torch.zeros(2, 4, dtype=torch.int32)
+    bin_bounds = torch.tensor([-2.0, 3.0], dtype=torch.float32)
+    expert_bias = torch.tensor([0.25, -0.25], 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, expert_bias)
+    torch.testing.assert_close(updated_bounds, bin_bounds)
+
+
+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, reduction_groups=())
+
+    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_finalize_without_active_router_is_a_noop():
+    model = torch.nn.Module()
+    config = SimpleNamespace(moe_router_topk=1)
+
+    _update_router_expert_bias_with_quantile([model], config, reduction_groups=())
+
+
+def test_qb_router_maintains_float32_bias_and_bounds():
+    from megatron.core.transformer.moe.router import TopKRouter
+
+    router = SimpleNamespace(
+        expert_bias=torch.zeros(4, dtype=torch.bfloat16),
+        qb_bin_bounds=torch.tensor([-1.0, 1.0], dtype=torch.bfloat16),
+    )
+
+    TopKRouter._maintain_float32_expert_bias(router)
+
+    assert router.expert_bias.dtype == torch.float32
+    assert router.qb_bin_bounds.dtype == torch.float32
+
+
+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, monkeypatch):
+    """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()
+    marked_bounds = []
+    monkeypatch.setitem(
+        _update_router_expert_bias_with_quantile.__globals__,
+        "mark_qb_bin_bounds_validated",
+        marked_bounds.append,
+    )
+    _update_router_expert_bias_with_quantile(
+        [layer], config, reduction_groups=(torch.distributed.group.WORLD,)
+    )
+
+    assert len(marked_bounds) == 1
+    assert marked_bounds[0] is router.qb_bin_bounds
+    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()