[AMD][Fix] Route MoRI through the Qwen MoE all-to-all path (#32039)

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
 

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