diff --git a/src/paddlefleet/transformer/moe/moe_router.py b/src/paddlefleet/transformer/moe/moe_router.py
index e51bdad612..0a63a5de3e 100644
--- a/src/paddlefleet/transformer/moe/moe_router.py
+++ b/src/paddlefleet/transformer/moe/moe_router.py
@@ -452,13 +452,6 @@ def __init__(
             self.expert_usage.stop_gradient = True
 
         if self.topk_method == "quantile_balancing":
-            if getattr(self.config, "moe_topk_fusion", False):
-                raise ValueError(
-                    "quantile_balancing is incompatible with moe_topk_fusion. "
-                    "The MoETopkFusion kernel does not support QB's histogram-based "
-                    "bias update, and enabling both causes incorrect gate normalization. "
-                    "Please set moe_topk_fusion=False when using quantile_balancing."
-                )
             if self.routing_type != "none":
                 raise ValueError(
                     "quantile_balancing is a self-contained load balancing method, "
@@ -1197,6 +1190,7 @@ def _accumulate_qb_histogram(
         biased_scores: paddle.Tensor,
         k: int,
         valid_mask: paddle.Tensor | None = None,
+        alpha: paddle.Tensor | None = None,
     ):
         """Accumulate required_bias into the QB histogram.
 
@@ -1219,17 +1213,25 @@ def _accumulate_qb_histogram(
             k: top-k value
             valid_mask: [N, 1] mask marking non-padding tokens, or None when no
                 padding information is available (all rows count).
+            alpha: [N] or [N, 1] pre-computed cutoff values from MoETopkFusion
+                kernel (optional). When provided, skips the internal topk
+                computation for cutoff.
         """
         N, E = raw_scores.shape
         B = self.qb_n_bins
 
         # Compute alpha: the (k+1)-th largest biased score per token
         # This is the "cutoff" -- the highest biased score NOT selected
-        # Clamp k+1 to at most E (in case of degenerate config)
-        topk_val = min(k + 1, int(E))
-        alpha = paddle.topk(biased_scores, k=topk_val, axis=-1, sorted=True)[0][
-            :, -1:
-        ]  # [N, 1] -- the smallest of top-(k+1), i.e., cutoff
+        if alpha is not None:
+            # Alpha provided externally (e.g., from MoETopkFusion kernel)
+            if alpha.ndim == 1:
+                alpha = alpha.unsqueeze(-1)  # [N] -> [N, 1]
+        else:
+            # Clamp k+1 to at most E (in case of degenerate config)
+            topk_val = min(k + 1, int(E))
+            alpha = paddle.topk(
+                biased_scores, k=topk_val, axis=-1, sorted=True
+            )[0][:, -1:]  # [N, 1] -- the smallest of top-(k+1), i.e., cutoff
 
         # required_bias[t, e] = alpha[t] - raw_scores[t, e]
         required_bias = alpha - raw_scores  # [N, E]
@@ -1710,14 +1712,8 @@ def forward(self, input, input_ids=None, origin_input_ids=None):
                     gates_ori.sum(-1, keepdim=True), min=1e-12
                 )
 
-        if (
-            getattr(self.config, "moe_topk_fusion", False)
-            and self.topk_method != "quantile_balancing"
-        ):
-            # Use MoETopkFusion Triton kernel for bit-exact alignment.
-            # This ensures the topk selection + normalization uses the exact same
-            # GPU kernel, avoiding FP32 rounding differences between
-            # Triton's scalar loop and Paddle's tensor ops.
+        if getattr(self.config, "moe_topk_fusion", False):
+            # Use MoETopkFusion Triton kernel for fused topk selection.
             MoETopkFusion = _get_moe_topk_fusion()
             use_node_limit = self.n_group > 1
             if not self.config.gpt_model_use_experimental_version:
@@ -1735,16 +1731,42 @@ def forward(self, input, input_ids=None, origin_input_ids=None):
                 _log_moe_md5(
                     probs_for_choice, "probs_for_choice", self._layer_number
                 )
-            top_gate, top_idx = MoETopkFusion.apply(
-                gates,  # gate_probs (original sigmoid scores)
-                probs_for_choice,  # probs_for_choice (with correction bias)
-                self.num_experts_per_tok,
-                use_node_limit,
-                self.n_group,
-                self.topk_group,
-                self.norm_topk_prob,  # norm_gate_logits
-            )
-            # top_gate is already normalized by the Triton kernel when norm_topk_prob=True
+
+            if self.topk_method == "quantile_balancing":
+                # QB path: kernel does NOT normalize (normalization stays eager
+                # for bit-exact alignment with original QB). Additionally returns
+                # alpha (the k+1-th largest choice value) for histogram accumulation.
+                top_gate, top_idx, alpha = MoETopkFusion.apply(
+                    gates,  # gate_probs (original sigmoid scores)
+                    probs_for_choice,  # probs_for_choice (with correction bias)
+                    self.num_experts_per_tok,
+                    use_node_limit,
+                    self.n_group,
+                    self.topk_group,
+                    False,  # norm_gate_logits=False (normalization stays eager)
+                    True,  # return_alpha=True
+                )
+                # Accumulate QB histogram with pre-computed alpha (skips internal topk)
+                if framework._dygraph_tracer()._has_grad:
+                    self._accumulate_qb_histogram(
+                        gates,
+                        probs_for_choice,
+                        self.num_experts_per_tok,
+                        valid_mask=input_ids_none_zero_mask,
+                        alpha=alpha,
+                    )
+            else:
+                # noaux_tc / other paths: kernel handles normalization
+                top_gate, top_idx = MoETopkFusion.apply(
+                    gates,  # gate_probs (original sigmoid scores)
+                    probs_for_choice,  # probs_for_choice (with correction bias)
+                    self.num_experts_per_tok,
+                    use_node_limit,
+                    self.n_group,
+                    self.topk_group,
+                    self.norm_topk_prob,  # norm_gate_logits
+                )
+            # top_gate is already normalized by the Triton kernel when norm_topk_prob=True (non-QB)
 
             _log_moe_md5(
                 top_idx.cast("float32"), "topk_indices", self._layer_number
@@ -1812,7 +1834,13 @@ def forward(self, input, input_ids=None, origin_input_ids=None):
 
         # norm
         if self.norm_topk_prob:
-            if not getattr(self.config, "moe_topk_fusion", False):
+            if (
+                not getattr(self.config, "moe_topk_fusion", False)
+                or self.topk_method == "quantile_balancing"
+            ):
+                # QB fusion path passes norm_gate_logits=False to the kernel,
+                # so normalization must happen here in eager (for bit-exact alignment).
+                # Non-fusion paths also normalize here.
                 if self.use_accuracy_compatible:
                     _sum_f64 = top_gate.cast(paddle.float64).sum(
                         axis=-1, keepdim=True
@@ -1821,7 +1849,7 @@ def forward(self, input, input_ids=None, origin_input_ids=None):
                 else:
                     denominator = top_gate.sum(axis=-1, keepdim=True) + 1e-20
                 top_gate = top_gate / denominator
-            # When gpt_model_use_experimental_version is True, top_gate is already normalized by MoETopkFusion
+            # When moe_topk_fusion=True and not QB, top_gate is already normalized by MoETopkFusion
 
         if self.routed_scaling_factor_learnable:
             top_gate = apply_learnable_routed_scaling(
diff --git a/src/paddlefleet/triton_ops/moe_topk_fusion.py b/src/paddlefleet/triton_ops/moe_topk_fusion.py
index 3e001997b8..3d8a86c7fa 100644
--- a/src/paddlefleet/triton_ops/moe_topk_fusion.py
+++ b/src/paddlefleet/triton_ops/moe_topk_fusion.py
@@ -37,12 +37,13 @@
 
 @enable_compat_on_triton_kernel
 @triton.jit
-def _fwd_kernel(
+def _fwd_kernel(  # pragma: no cover - triton kernel body compiles to PTX, not python-instrumentable
     ptr_gate,
     ptr_choice,
     ptr_out_probs,
     ptr_out_idx,
     ptr_out_sum,
+    ptr_out_alpha,
     stride_gate_s,
     stride_gate_e,
     stride_choice_s,
@@ -55,6 +56,7 @@ def _fwd_kernel(
     n_group: tl.constexpr,
     topk_group: tl.constexpr,
     norm_gate_logits: tl.constexpr,
+    return_alpha: tl.constexpr,
     BLOCK_SIZE: tl.constexpr,
 ):
     """
@@ -158,6 +160,13 @@ def _fwd_kernel(
         # Mask out this index so we don't pick it again
         choice_vals = tl.where(off_e != k_idx, choice_vals, float("-inf"))
 
+    # --- Alpha (cutoff) output ---
+    # After the top-k loop, choice_vals has the top-k positions masked to -inf.
+    # The max of the remaining values is the (k+1)-th largest = cutoff alpha.
+    if return_alpha:
+        alpha_val = tl.max(choice_vals, axis=0)
+        tl.store(ptr_out_alpha + pid, alpha_val)
+
     # --- Normalization ---
     if norm_gate_logits:
         # Sum the collected probs
@@ -263,6 +272,7 @@ def forward(
         n_group,
         topk_group,
         norm_gate_logits,
+        return_alpha=False,
     ):
         """
         Forward pass: select topk experts.
@@ -276,10 +286,14 @@ def forward(
             n_group: number of expert groups.
             topk_group: number of selected topk groups.
             norm_gate_logits: whether to normalize gate logits.
+            return_alpha: if True, additionally return the (k+1)-th largest
+                choice value per token (cutoff alpha) as a no-grad tensor.
 
         Returns:
             topk_probs: normalized topk probabilities, shape [seq_len, moe_k].
             topk_indices: topk expert indices, shape [seq_len, moe_k].
+            alpha (optional): cutoff values, shape [seq_len], only when
+                return_alpha=True.
         """
         seq_len, n_experts = gate_probs.shape
 
@@ -290,6 +304,9 @@ def forward(
             if norm_gate_logits
             else None
         )
+        alpha = (
+            paddle.empty((seq_len,), dtype="float32") if return_alpha else None
+        )
 
         # Block size must cover n_experts for the single-block reduction logic
         BLOCK_SIZE = triton.next_power_of_2(n_experts)
@@ -298,6 +315,8 @@ def forward(
 
         # Use topk_probs as dummy pointer for sum if not needed, as it is writable
         ptr_sum_arg = topk_sum if norm_gate_logits else topk_probs
+        # Use topk_probs as dummy pointer for alpha if not needed
+        ptr_alpha_arg = alpha if return_alpha else topk_probs
 
         _fwd_kernel[(seq_len,)](
             gate_probs,
@@ -305,6 +324,7 @@ def forward(
             topk_probs,
             topk_indices,
             ptr_sum_arg,
+            ptr_alpha_arg,
             int(gate_probs.stride(0)),
             int(gate_probs.stride(1)),
             int(probs_for_choice.stride(0)),
@@ -317,6 +337,7 @@ def forward(
             n_group if use_node_limit else 1,
             topk_group if use_node_limit else 1,
             norm_gate_logits,
+            return_alpha,
             BLOCK_SIZE,
         )
 
@@ -324,13 +345,19 @@ def forward(
         ctx.input_shape = gate_probs.shape
         ctx.norm_gate_logits = norm_gate_logits
         ctx.moe_k = moe_k
+        ctx.return_alpha = return_alpha
 
+        if return_alpha:
+            return topk_probs, topk_indices.to(paddle.int64), alpha
         return topk_probs, topk_indices.to(paddle.int64)
 
     @staticmethod
-    def backward(ctx, grad_output_probs, grad_output_indices):
+    def backward(ctx, grad_output_probs, grad_output_indices, grad_alpha=None):
         """
         Backward: compute the gradient with respect to gate_probs.
+
+        When return_alpha=True in forward, backward receives an extra
+        grad_alpha argument which is ignored (alpha has no gradient).
         """
         topk_indices, topk_normed_probs, topk_sum = ctx.saved_tensor()
 
diff --git a/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_3.py b/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_3.py
index 4cfe74d0b9..57256a0ec9 100644
--- a/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_3.py
+++ b/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_3.py
@@ -82,9 +82,11 @@ def test_forward_launches_kernel_and_saves_context_without_norm(self):
         self.assertIs(recorder.args[0], gate_probs)
         self.assertIs(recorder.args[1], probs_for_choice)
         self.assertIs(recorder.args[4], topk_probs)
-        self.assertEqual(recorder.args[12:16], (2, False, 1, 1))
-        self.assertEqual(recorder.args[16], False)
-        self.assertEqual(recorder.args[17], 32)
+        self.assertEqual(recorder.args[12], 5)
+        self.assertEqual(recorder.args[13:17], (2, False, 1, 1))
+        self.assertEqual(recorder.args[17], False)
+        self.assertFalse(recorder.args[18])
+        self.assertEqual(recorder.args[19], 32)
         self.assertEqual(topk_probs.shape, [2, 2])
         self.assertEqual(topk_indices.dtype, paddle.int64)
         saved_indices, saved_probs, saved_sum = ctx.saved
@@ -115,9 +117,11 @@ def test_forward_launches_kernel_with_node_limit_and_norm_sum(self):
         finally:
             moe_topk_fusion._fwd_kernel = old_kernel
 
-        self.assertEqual(recorder.args[12:16], (3, True, 8, 2))
-        self.assertTrue(recorder.args[16])
-        self.assertEqual(recorder.args[17], 64)
+        self.assertEqual(recorder.args[12], 64)
+        self.assertEqual(recorder.args[13:17], (3, True, 8, 2))
+        self.assertTrue(recorder.args[17])
+        self.assertFalse(recorder.args[18])
+        self.assertEqual(recorder.args[19], 64)
         self.assertEqual(ctx.saved[2].shape, [1])
         self.assertEqual(topk_probs.shape, [1, 3])
         self.assertEqual(topk_indices.shape, [1, 3])
diff --git a/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_4.py b/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_4.py
index 3deb028f9b..1a648eb891 100644
--- a/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_4.py
+++ b/tests/single_card_tests/ai_edited_test/ops/test_ai_moe_topk_fusion_4.py
@@ -239,6 +239,7 @@ def test_forward_kernel_python_body_exercises_group_topk_and_norm(self):
             FakePtr(),
             FakePtr(),
             FakePtr(),
+            FakePtr(),
             4,
             1,
             4,
@@ -251,6 +252,7 @@ def test_forward_kernel_python_body_exercises_group_topk_and_norm(self):
             2,
             2,
             True,
+            True,
             4,
         )
 
diff --git a/tests/single_card_tests/test_quantile_balancing.py b/tests/single_card_tests/test_quantile_balancing.py
index 9233a7db2e..a0e2177d8e 100644
--- a/tests/single_card_tests/test_quantile_balancing.py
+++ b/tests/single_card_tests/test_quantile_balancing.py
@@ -1377,10 +1377,10 @@ def test_bin_range_is_restored_from_state_dict(self):
         self.assertEqual(float(restored.qb_bin_min.item()), -1.75)
         self.assertEqual(float(restored.qb_bin_max.item()), 2.25)
 
-    def test_moe_topk_fusion_is_rejected(self):
-        with self.assertRaises(ValueError) as ctx:
-            _build_qb_router(moe_topk_fusion=True)
-        self.assertIn("incompatible with moe_topk_fusion", str(ctx.exception))
+    def test_moe_topk_fusion_is_accepted(self):
+        """QB + moe_topk_fusion is now supported (Plan A fusion)."""
+        router = _build_qb_router(moe_topk_fusion=True)
+        self.assertEqual(router.topk_method, "quantile_balancing")
 
     def test_non_qb_router_has_no_qb_state(self):
         router = _build_qb_router(topk_method="greedy")
@@ -1859,14 +1859,379 @@ def test_explicitly_disabled_balancing_is_accepted(self):
         self.assertEqual(router.topk_method, "quantile_balancing")
         self.assertEqual(router.qb_histogram.shape, [8, 1000])
 
-    def test_topk_fusion_is_rejected(self):
-        with self.assertRaises(ValueError) as ctx:
-            self._build_router(
-                moe_router_load_balancing_type="none",
-                router_aux_loss_coef=0.0,
-                moe_topk_fusion=True,
+    def test_topk_fusion_is_accepted(self):
+        """QB + moe_topk_fusion is now supported (Plan A fusion)."""
+        router = self._build_router(
+            moe_router_load_balancing_type="none",
+            router_aux_loss_coef=0.0,
+            moe_topk_fusion=True,
+        )
+        self.assertEqual(router.topk_method, "quantile_balancing")
+
+
+# =============================================================================
+# Test: MoETopkFusion return_alpha functionality
+# =============================================================================
+
+
+class TestMoETopkFusionAlpha(unittest.TestCase):
+    """Test that the MoETopkFusion kernel correctly returns the cutoff alpha."""
+
+    def _get_fusion_class(self):
+        from paddlefleet.triton_ops.moe_topk_fusion import MoETopkFusion
+
+        return MoETopkFusion
+
+    def test_return_alpha_matches_eager_topk(self):
+        """Alpha from kernel must match the (k+1)-th largest value from paddle.topk."""
+        paddle.seed(42)
+        N, E, k = 32, 16, 4
+
+        gate_probs = paddle.rand([N, E], dtype="float32")
+        probs_for_choice = gate_probs + paddle.randn([E]) * 0.1
+
+        MoETopkFusion = self._get_fusion_class()
+        topk_probs, topk_idx, alpha = MoETopkFusion.apply(
+            gate_probs,
+            probs_for_choice,
+            k,
+            False,  # use_node_limit
+            1,  # n_group
+            1,  # topk_group
+            False,  # norm_gate_logits
+            True,  # return_alpha
+        )
+
+        # Eager reference: (k+1)-th largest of probs_for_choice
+        topk_val = min(k + 1, E)
+        eager_alpha = paddle.topk(
+            probs_for_choice, k=topk_val, axis=-1, sorted=True
+        )[0][:, -1]  # [N]
+
+        np.testing.assert_array_equal(
+            alpha.numpy(),
+            eager_alpha.numpy(),
+            err_msg="Kernel alpha must be bit-exact with eager topk cutoff",
+        )
+
+    def test_return_alpha_no_regression_on_topk_probs(self):
+        """return_alpha=True must not change topk_probs or topk_indices."""
+        paddle.seed(123)
+        N, E, k = 64, 32, 8
+
+        gate_probs = paddle.rand([N, E], dtype="float32")
+        probs_for_choice = gate_probs + paddle.randn([E]) * 0.05
+
+        MoETopkFusion = self._get_fusion_class()
+
+        # Without alpha
+        probs_no_alpha, idx_no_alpha = MoETopkFusion.apply(
+            gate_probs,
+            probs_for_choice,
+            k,
+            False,
+            1,
+            1,
+            False,
+            False,
+        )
+
+        # With alpha
+        probs_with_alpha, idx_with_alpha, _ = MoETopkFusion.apply(
+            gate_probs,
+            probs_for_choice,
+            k,
+            False,
+            1,
+            1,
+            False,
+            True,
+        )
+
+        np.testing.assert_array_equal(
+            probs_no_alpha.numpy(),
+            probs_with_alpha.numpy(),
+            err_msg="topk_probs must be identical with or without return_alpha",
+        )
+        np.testing.assert_array_equal(
+            idx_no_alpha.numpy(),
+            idx_with_alpha.numpy(),
+            err_msg="topk_indices must be identical with or without return_alpha",
+        )
+
+    def test_qb_fusion_vs_eager_histogram(self):
+        """QB via fusion path must produce identical histogram as eager path."""
+        paddle.seed(7)
+        N, E, k = 128, 16, 4
+        B = 256
+
+        scores = paddle.sigmoid(paddle.randn([N, E], dtype="float32"))
+        bias = paddle.randn([E], dtype="float32") * 0.1
+        biased_scores = scores + bias.unsqueeze(0)
+        b_min = bias.min() - 1.0
+        b_max = bias.max() + 1.0
+
+        # --- Eager path: compute alpha via paddle.topk ---
+        topk_val = min(k + 1, E)
+        eager_alpha = paddle.topk(
+            biased_scores, k=topk_val, axis=-1, sorted=True
+        )[0][:, -1:]  # [N, 1]
+
+        # --- Fusion path: compute alpha via kernel ---
+        MoETopkFusion = self._get_fusion_class()
+        _, _, fusion_alpha = MoETopkFusion.apply(
+            scores,
+            biased_scores,
+            k,
+            False,
+            1,
+            1,
+            False,
+            True,
+        )
+        fusion_alpha_2d = fusion_alpha.unsqueeze(-1)  # [N, 1]
+
+        # Alpha must match
+        np.testing.assert_array_equal(
+            fusion_alpha_2d.numpy(),
+            eager_alpha.numpy(),
+            err_msg="Fusion alpha must be bit-exact with eager alpha",
+        )
+
+        # Now compute histograms with both alphas and compare
+        total_range = b_max - b_min
+        if total_range < 1e-8:
+            total_range = paddle.to_tensor(2.0)
+
+        def _compute_histogram(alpha_tensor):
+            required_bias = alpha_tensor - scores
+            bin_idx = ((required_bias - b_min) / total_range * B).cast(
+                paddle.int64
+            )
+            bin_idx = paddle.clip(bin_idx, min=0, max=B - 1)
+            offsets = paddle.arange(E, dtype=paddle.int64).unsqueeze(0) * B
+            flat_bins = (bin_idx + offsets).reshape([-1])
+            counts = paddle.zeros([E * B], dtype=paddle.int32)
+            weights = paddle.ones([N * E], dtype=paddle.int32)
+            counts.put_along_axis_(flat_bins, weights, axis=0, reduce="add")
+            return counts.reshape([E, B])
+
+        hist_eager = _compute_histogram(eager_alpha)
+        hist_fusion = _compute_histogram(fusion_alpha_2d)
+
+        np.testing.assert_array_equal(
+            hist_fusion.numpy(),
+            hist_eager.numpy(),
+            err_msg="Histogram from fusion alpha must be identical to eager",
+        )
+
+
+class TestAccumulateQBHistogramAlphaEquivalence(unittest.TestCase):
+    """Production `_accumulate_qb_histogram` must agree on the histogram
+    whether alpha comes from the fusion kernel or from its internal topk.
+
+    `TestMoETopkFusionAlpha.test_qb_fusion_vs_eager_histogram` only proves the
+    mathematical claim "bit-exact alpha implies bit-exact histogram" against a
+    re-implementation of the binning. These cases drive the real router method.
+    """
+
+    def setUp(self):
+        paddle.seed(11)
+        np.random.seed(11)
+
+    def _kernel_alpha(self, scores, biased, k):
+        from paddlefleet.triton_ops.moe_topk_fusion import MoETopkFusion
+
+        _, _, alpha = MoETopkFusion.apply(
+            scores,
+            biased,
+            k,
+            False,  # use_node_limit
+            1,  # n_group
+            1,  # topk_group
+            False,  # norm_gate_logits
+            True,  # return_alpha
+        )
+        return alpha
+
+    def _histogram_for(self, router, scores, biased, k, alpha, valid_mask):
+        """Run the accumulator from a zeroed histogram and return the result."""
+        router.qb_histogram = paddle.zeros_like(router.qb_histogram)
+        router._accumulate_qb_histogram(
+            scores, biased, k, valid_mask=valid_mask, alpha=alpha
+        )
+        return router.qb_histogram.numpy().copy()
+
+    def _run_equivalence(self, valid_mask):
+        router = _build_qb_router()
+        E = router.num_experts
+        k = router.num_experts_per_tok
+        N = 24
+
+        bias_np = np.linspace(-0.2, 0.2, E).astype(np.float32)
+        router.e_score_correction_bias.set_value(paddle.to_tensor(bias_np))
+
+        scores = paddle.to_tensor(np.random.rand(N, E).astype(np.float32))
+        biased = scores + paddle.to_tensor(bias_np).unsqueeze(0)
+
+        # alpha=None -> internal paddle.topk(k+1)
+        hist_internal = self._histogram_for(
+            router, scores, biased, k, None, valid_mask
+        )
+        # alpha from the fusion kernel -> internal topk skipped
+        alpha = self._kernel_alpha(scores, biased, k)
+        hist_kernel = self._histogram_for(
+            router, scores, biased, k, alpha, valid_mask
+        )
+
+        # Guard against a vacuous comparison: both paths must have counted.
+        self.assertGreater(int(hist_internal.sum()), 0)
+        np.testing.assert_array_equal(
+            hist_internal,
+            hist_kernel,
+            err_msg=(
+                "_accumulate_qb_histogram must produce a bit-identical "
+                "histogram whether alpha is supplied by the kernel or "
+                "computed internally"
+            ),
+        )
+        return N, E
+
+    def test_alpha_equivalence(self):
+        self._run_equivalence(valid_mask=None)
+
+    def test_alpha_equivalence_with_valid_mask(self):
+        # Exercises the padding-weight branch of the accumulator: masked rows
+        # must contribute zero counts on both paths.
+        N = 24
+        mask_np = np.ones((N, 1), dtype=np.int64)
+        mask_np[::3] = 0
+        valid_mask = paddle.to_tensor(mask_np)
+
+        _, E = self._run_equivalence(valid_mask=valid_mask)
+        self.assertEqual(E, 8)
+
+
+class TestQBFusionEndToEndBitExact(unittest.TestCase):
+    """Flipping `moe_topk_fusion` must not change a single bit of the QB
+    router's forward outputs or of the state it accumulates.
+
+    The other fusion cases check individual pieces (alpha, the accumulator).
+    This one drives the whole `TopKRouter.forward` both ways on one router
+    instance, so weights and inputs are guaranteed identical.
+    """
+
+    def setUp(self):
+        paddle.seed(2027)
+        np.random.seed(2027)
+
+    def _forward_once(self, router, hidden, input_ids, use_fusion):
+        # `moe_topk_fusion` is read off the config at forward time, so the same
+        # instance can serve both paths -- no weight copying needed.
+        router.config.moe_topk_fusion = use_fusion
+        router.qb_histogram = paddle.zeros_like(router.qb_histogram)
+        router.expert_usage = paddle.zeros_like(router.expert_usage)
+
+        kwargs = {} if input_ids is None else {"input_ids": input_ids}
+        cls = type(router)
+        with patch.object(
+            cls,
+            "_topk_quantile_balancing",
+            side_effect=cls._topk_quantile_balancing,
+            autospec=True,
+        ) as eager_topk:
+            _, top_gate, top_idx, probs, mask, _, _, _ = router(
+                hidden, **kwargs
             )
-        self.assertIn("moe_topk_fusion", str(ctx.exception))
+
+        # Keeps the comparison honest: fusion must really bypass the eager
+        # selection, otherwise both runs would trivially be the same code.
+        self.assertEqual(eager_topk.call_count, 0 if use_fusion else 1)
+
+        return {
+            "top_gate": top_gate.detach().numpy().copy(),
+            "top_idx": top_idx.detach().numpy().copy(),
+            "probs": probs.detach().numpy().copy(),
+            "mask": mask.detach().numpy().copy(),
+            "qb_histogram": router.qb_histogram.numpy().copy(),
+            "expert_usage": router.expert_usage.numpy().copy(),
+        }
+
+    def _run(self, input_ids):
+        router = _build_qb_router()
+        # A non-uniform, non-zero bias makes selection genuinely depend on it,
+        # which is what the kernel and the eager path must agree on.
+        bias = np.linspace(-0.15, 0.15, router.num_experts).astype(np.float32)
+        router.e_score_correction_bias.set_value(paddle.to_tensor(bias))
+        hidden = paddle.randn([4, 6, 32])
+
+        eager = self._forward_once(router, hidden, input_ids, False)
+        fused = self._forward_once(router, hidden, input_ids, True)
+
+        # Guard against a vacuous pass on all-zero state.
+        self.assertGreater(int(eager["qb_histogram"].sum()), 0)
+        self.assertGreater(int(eager["expert_usage"].sum()), 0)
+
+        for name in eager:
+            np.testing.assert_array_equal(
+                eager[name],
+                fused[name],
+                err_msg=f"{name} diverges once moe_topk_fusion is enabled",
+            )
+
+    def test_bit_exact_without_padding(self):
+        self._run(None)
+
+    def test_bit_exact_with_padding(self):
+        # Trailing padding exercises the valid_mask branches on both paths.
+        ids = np.ones((4, 6), dtype="int64")
+        ids[:, -2:] = 0
+        self._run(paddle.to_tensor(ids))
+
+
+class TestNonQBFusionForward(unittest.TestCase):
+    """Covers the non-QB branch of the fused router path.
+
+    When ``moe_topk_fusion=True`` and ``topk_method != "quantile_balancing"``
+    (e.g. ``noaux_tc``), ``TopKRouter.forward`` takes the ``else`` branch that
+    lets the Triton kernel own normalization (``norm_gate_logits`` is passed
+    through). The QB fusion tests never reach it because they always run with
+    ``topk_method == "quantile_balancing"``.
+    """
+
+    def setUp(self):
+        paddle.seed(2027)
+        np.random.seed(2027)
+
+    def test_noaux_tc_fusion_forward_runs(self):
+        router = _build_qb_router(
+            topk_method="noaux_tc",
+            moe_topk_fusion=True,
+        )
+        # A non-uniform, non-zero bias makes the selection genuinely depend on
+        # it, so the kernel path is not trivially equivalent to an unbiased one.
+        bias = np.linspace(-0.1, 0.1, router.num_experts).astype(np.float32)
+        router.e_score_correction_bias.set_value(paddle.to_tensor(bias))
+
+        hidden = paddle.randn([4, 6, 32])
+        _, top_gate, top_idx, _, _, _, _, _ = router(hidden)
+
+        # Shapes agree and select num_experts_per_tok experts per token.
+        self.assertEqual(top_gate.shape, top_idx.shape)
+        self.assertEqual(top_idx.shape[-1], router.num_experts_per_tok)
+        # Selected indices are valid experts.
+        idx_np = top_idx.numpy()
+        self.assertTrue((idx_np >= 0).all())
+        self.assertTrue((idx_np < router.num_experts).all())
+        # norm_topk_prob=True -> the kernel normalizes the selected gates to 1.
+        np.testing.assert_allclose(
+            top_gate.sum(axis=-1).numpy().reshape(-1),
+            np.ones(
+                top_idx.size // router.num_experts_per_tok, dtype=np.float32
+            ),
+            rtol=1e-5,
+            atol=1e-5,
+        )
 
 
 if __name__ == "__main__":