diff --git a/python/sglang/srt/models/qwen2_moe.py b/python/sglang/srt/models/qwen2_moe.py index c699ec897..a259053ad 100644 --- a/python/sglang/srt/models/qwen2_moe.py +++ b/python/sglang/srt/models/qwen2_moe.py @@ -154,6 +154,7 @@ def can_fuse_shared_expert( or getattr(config, "shared_expert_intermediate_size", 0) <= 0 or config.shared_expert_intermediate_size != config.moe_intermediate_size or get_moe_a2a_backend().is_deepep() + or get_moe_a2a_backend().is_mori() ): return False diff --git a/test/registered/unit/models/test_shared_experts_fusion_gates.py b/test/registered/unit/models/test_shared_experts_fusion_gates.py index 38fc544a2..65d06092a 100644 --- a/test/registered/unit/models/test_shared_experts_fusion_gates.py +++ b/test/registered/unit/models/test_shared_experts_fusion_gates.py @@ -518,6 +518,45 @@ class TestWrapperEntryClassGates(_FusionGateCase): ) +class TestA2ABackendGate(_FusionGateCase): + """`can_fuse_shared_expert` must refuse for every DeepEP-class backend it + is wired for. MoRI runs the same per-rank EP expert layout as DeepEP, so a + fused shared expert would occupy a global slot the layers never allocate — + the routed experts then read the wrong rows and accuracy collapses.""" + + def _config(self): + return SimpleNamespace( + model_type="qwen3_5_moe_text", + shared_expert_intermediate_size=1024, + moe_intermediate_size=1024, + ) + + def _use_backend(self, name: str): + from sglang.srt.layers.moe.utils import MoeA2ABackend + from sglang.srt.runtime_context import get_flags + + moe = get_flags().moe + previous = moe.a2a_backend + moe.a2a_backend = MoeA2ABackend(name) + self.addCleanup(setattr, moe, "a2a_backend", previous) + + def test_the_a2a_backends_refuse_fusion(self): + from sglang.srt.models.qwen2_moe import can_fuse_shared_expert + + self._seed() + for backend in ("deepep", "mori"): + with self.subTest(backend=backend): + self._use_backend(backend) + self.assertFalse(can_fuse_shared_expert(self._config(), None)) + + def test_a_plain_tp_deployment_still_fuses(self): + from sglang.srt.models.qwen2_moe import can_fuse_shared_expert + + self._seed() + self._use_backend("none") + self.assertTrue(can_fuse_shared_expert(self._config(), None)) + + class TestFamiliesWithoutAGate(_FusionGateCase): def test_qwen2_moe_style_families_follow_the_intent(self): """A family with no gate must not grow one by accident: the installer