@@ -106,7 +106,29 @@ for file in "${sharded_files[@]}"; do
                 _ "$file"
             )
             ;;
-        op_tests/multigpu_tests/test_mega_moe_v2.py)
+        op_tests/multigpu_tests/test_mega_moe.py)
+            {
+                echo "Running gfx1250 MegaMoE fused-scatter accuracy on 8 GPUs when supported"
+            } | tee -a latest_test.log
+            test_cmd=(
+                timeout 60m
+                bash -c '
+                    set -euo pipefail
+                    test_file=$1
+                    arch=$(python3 -c \
+                        "from aiter.jit.utils.chip_info import get_gfx; print(get_gfx())")
+                    if [[ "$arch" != "gfx1250" ]]; then
+                        echo "Skipping $test_file: requires gfx1250, got $arch"
+                        exit 0
+                    fi
+                    exec env MORI_SHMEM_HEAP_SIZE=40G \
+                        torchrun --standalone --nproc_per_node=8 "$test_file" \
+                        --combine scatter_fused --layers 2 --acc_verify 1
+                '
+                _ "$file"
+            )
+            ;;
+        op_tests/multigpu_tests/test_mega_moe_v2_gfx1250.py)
             {
                 echo "Running MegaMoEV2 v4_pro fixed-slot and compact coverage on 8 GPUs"
             } | tee -a latest_test.log
 
@@ -19,7 +19,10 @@
 import flydsl.expr as fx
 from flydsl._mlir import ir
 from flydsl._mlir.dialects import llvm as _llvm_d
+from flydsl._mlir.dialects import rocdl as _rocdl_d
+from flydsl._mlir.dialects import scf
 from flydsl.expr import arith
+from flydsl.expr.typing import T
 
 __all__ = [
     "GeometryTuningTable",
@@ -32,9 +35,17 @@
     "fence_release",
     "fence_system_acquire",
     "fence_system_release",
+    "load_i32_acquire",
+    "load_i32_nt",
+    "load_i64_acquire",
     "load_i64_global",
+    "load_v4i32_nt",
+    "spin_until_eq_i32",
+    "spin_until_ge_i64",
+    "spin_until_gt_i32",
     "store_i32_system",
     "store_i64_global_system",
+    "waitcnt_all",
 ]
 
 
@@ -45,6 +56,95 @@ def _to_ptr_global(v):
     ).result
 
 
+def _ptr_plus(base_i64, offset, elem_bytes):
+    """Global pointer for base + offset*elem_bytes."""
+    addr = (
+        fx.Int64(arith.unwrap(base_i64)) + fx.Int64(arith.unwrap(offset)) * elem_bytes
+    )
+    return _to_ptr_global(addr)
+
+
+def _unwrap(v):
+    return v.ir_value() if hasattr(v, "ir_value") else v
+
+
+def waitcnt_all():
+    """Drain outstanding gfx12 load/store counters before a grid barrier."""
+    _rocdl_d.s_wait_storecnt(0)
+    _rocdl_d.s_wait_loadcnt(0)
+
+
+def load_i32_acquire(addr_i64):
+    """Volatile monotonic i32 load suitable for a spin-wait."""
+    return _llvm_d.LoadOp(
+        T.i32,
+        _to_ptr_global(addr_i64),
+        alignment=4,
+        volatile_=True,
+        ordering=_llvm_d.AtomicOrdering.monotonic,
+        syncscope="one-as",
+    ).res
+
+
+def load_i64_acquire(addr_i64):
+    """Volatile monotonic i64 load suitable for a spin-wait."""
+    return _llvm_d.LoadOp(
+        T.i64,
+        _to_ptr_global(addr_i64),
+        alignment=8,
+        volatile_=True,
+        ordering=_llvm_d.AtomicOrdering.monotonic,
+        syncscope="one-as",
+    ).res
+
+
+def load_i32_nt(base_i64, offset):
+    """Non-temporal global i32 load at base + offset*4."""
+    return _llvm_d.LoadOp(
+        T.i32, _ptr_plus(base_i64, offset, 4), alignment=4, nontemporal=True
+    ).res
+
+
+def load_v4i32_nt(base_i64, offset):
+    """Non-temporal global vector<4xi32> load at base + offset*4."""
+    return _llvm_d.LoadOp(
+        T.i32x4, _ptr_plus(base_i64, offset, 4), alignment=4, nontemporal=True
+    ).res
+
+
+def _spin(addr_i64, keep_waiting, *, width=32):
+    """Spin on a volatile load until ``keep_waiting`` becomes false."""
+    if width == 64:
+        ty, load, wrap = T.i64, load_i64_acquire, fx.Int64
+    else:
+        ty, load, wrap = T.i32, load_i32_acquire, fx.Int32
+    loop = scf.WhileOp([ty], [_unwrap(load(addr_i64))])
+    cond = ir.Block.create_at_start(loop.before, [ty])
+    body = ir.Block.create_at_start(loop.after, [ty])
+    with ir.InsertionPoint(cond):
+        scf.ConditionOp(
+            _unwrap(keep_waiting(wrap(cond.arguments[0]))), [cond.arguments[0]]
+        )
+    with ir.InsertionPoint(body):
+        scf.YieldOp([_unwrap(load(addr_i64))])
+    return wrap(loop.results[0])
+
+
+def spin_until_ge_i64(addr_i64, val):
+    """Spin until a monotonic cross-device i64 flag is at least ``val``."""
+    return _spin(addr_i64, lambda cur: cur < fx.Int64(val), width=64)
+
+
+def spin_until_eq_i32(addr_i64, val):
+    """Spin until an i32 flag equals ``val``."""
+    return _spin(addr_i64, lambda cur: cur != fx.Int32(val))
+
+
+def spin_until_gt_i32(addr_i64, val):
+    """Spin until an i32 flag exceeds ``val`` and return the observed value."""
+    return _spin(addr_i64, lambda cur: cur <= fx.Int32(val))
+
+
 def store_i32_system(addr_i64, offset, val):
     """System-scope release i32 store at ``addr_i64 + offset*4``."""
     base = arith.unwrap(addr_i64)
 
@@ -21,8 +21,6 @@
 # SOFTWARE.
 """Barrier-wait and top-k reduce kernel for gfx1250 Stage2-fused MegaMoE."""
 
-import os
-
 import flydsl.compiler as flyc
 import flydsl.expr as fx
 import mori.cco.device.flydsl as cco
@@ -37,7 +35,6 @@
     create_buffer_resource_from_addr,
 )
 
-from . import primitives as P
 from .config import (
     _LANE_MASK as LANE_MASK,
 )
@@ -97,15 +94,16 @@ def _make_combine_fused_sync(
 ):
     """Cross-device barrier kernel for the split fused-combine path.
 
-    A single 1-block, 1-wave kernel that does ONLY Stage A: wait until every
-    peer's gemm2 P2P writes into our comb_inp are globally visible. Launch this
-    first on the stream, then the ``include_sync=False`` reduce kernel (grid
-    fully free). Kernel retirement (stream-ordered) is the completion signal the
-    reduce waits on -- no in-kernel fence needed.
+    A single-block kernel that does ONLY Stage A: wait until every peer's gemm2
+    P2P writes into our comb_inp are globally visible. Launch this first on the
+    stream, then the ``include_sync=False`` reduce kernel (grid fully free).
+    Kernel retirement (stream-ordered) is the completion signal the reduce waits
+    on -- no in-kernel fence needed.
 
-    One wave suffices: only lanes [0, npes) do any work (push the phase to each
-    peer, then poll the npes local slots), there is no cross-warp dependency and
-    hence no fx.barrier(). The kernel's cost is the peer wait, not thread count.
+    One thread per peer pushes the phase and polls that peer's local slot. The
+    block is rounded up to whole waves, so rack-scale domains larger than one
+    wave are covered without introducing a cross-wave dependency or barrier.
+    The kernel's cost is the peer wait, not thread count.
 
     Because there is exactly one block, the per-block flag bookkeeping collapses
     to a single monotonic counter in xdb_flag[0]; the remaining _XDB_FLAG_SLOTS
@@ -114,7 +112,9 @@ def _make_combine_fused_sync(
     a stale counter from its own slot and clear its barrier without waiting.
     """
 
-    @flyc.kernel(known_block_size=[WAVE, 1, 1])
+    sync_block_size = ((npes + WAVE - 1) // WAVE) * WAVE
+
+    @flyc.kernel(known_block_size=[sync_block_size, 1, 1])
     def ep_combine_fused_sync(
         arena: Int64,
         addr_xdb_flag: Int64,
@@ -139,7 +139,7 @@ def ep_combine_fused_sync(
             xdb_peer_slot = fx.Int64(
                 window.lsa_ptr(my_lsa_rank, off_xdb_mem)
             ) + fx.Int64(tid) * fx.Int64(8)
-            P.spin_until_ge_i64(xdb_peer_slot, phase)
+            comm_ops.spin_until_ge_i64(xdb_peer_slot, phase)
 
     @flyc.jit
     def run(
@@ -150,7 +150,7 @@ def run(
     ):
         ep_combine_fused_sync(arena, addr_xdb_flag, my_lsa_rank).launch(
             grid=(1, 1, 1),
-            block=[WAVE, 1, 1],
+            block=[sync_block_size, 1, 1],
             stream=stream,
         )
 
@@ -269,7 +269,7 @@ def ep_combine_fused(
                 xdb_peer_slot = fx.Int64(
                     window.lsa_ptr(my_lsa_rank, off_xdb_mem)
                 ) + fx.Int64(tid) * fx.Int64(8)
-                P.spin_until_ge_i64(xdb_peer_slot, phase)
+                comm_ops.spin_until_ge_i64(xdb_peer_slot, phase)
             fx.barrier()
 
         # ── Stage B: local read of comb_inp[tok*topk + k] + unweighted topk sum ──
@@ -303,20 +303,14 @@ def ep_combine_fused(
             def _one(off, expert_addrs=expert_addrs, out_base=out_base):
                 acc = zero_acc()
                 for k_slot in range_constexpr(topk):
-                    v = P.load_i32_nt(expert_addrs[k_slot], off)
+                    v = comm_ops.load_i32_nt(expert_addrs[k_slot], off)
                     acc = acc + to_acc(v)
                 buffer_store(from_acc(acc), rsrc_out, out_base + off)
 
-            # Inner-unrolled vec4: issue _UNROLL x topk v4 loads FIRST (load-first,
-            # high MLP to hide the 6 strided comb_inp streams), then reduce + vec4
-            # store. Tail units
-            # (< STEP_V4) fall back to the scalar _one path.
-            # _UNROLL=1 keeps VGPR low so the grid can fill all CUs (occupancy),
-            # which matters far more than per-wave MLP here: the reduce is HBM-BW
-            # bound, and a big grid (block_num ~= #CUs) is what saturates HBM. A
-            # deep unroll (4) blew VGPR to ~200 and capped occupancy, throttling
-            # load issue (s_clause). Overridable via SCATTER_COMB_UNROLL for tuning.
-            _UNROLL = int(os.environ.get("SCATTER_COMB_UNROLL", "1"))
+            # One vec4 group keeps VGPR low so the grid can fill all CUs. Deeper
+            # unrolling increases VGPR pressure and reduces occupancy on this
+            # HBM-bandwidth-bound reduce.
+            _UNROLL = 1
             VEC = 4
             STEP_CHUNK = WAVE * VEC  # 128 i32 elems/round across the wave
             STEP_V4 = _UNROLL * STEP_CHUNK  # _UNROLL * 128
@@ -328,7 +322,7 @@ def _one(off, expert_addrs=expert_addrs, out_base=out_base):
                     _off_r = base + _r * STEP_CHUNK
                     _pre.append(
                         [
-                            P.load_v4i32_nt(expert_addrs[k_slot], _off_r)
+                            comm_ops.load_v4i32_nt(expert_addrs[k_slot], _off_r)
                             for k_slot in range_constexpr(topk)
                         ]
                     )
 
@@ -7,13 +7,6 @@
 _LANE_MASK = _WAVE_SIZE - 1
 _LOG2_WAVE_SIZE = 5
 
-_FUSED_COMBINE_TIERS = (
-    (32, (8, 16)),
-    (256, (16, 16)),
-    (1024, (32, 16)),
-    (None, (64, 16)),
-)
-
 _DISPATCH_EP4 = (
     (256, 128, 16),
     (512, 192, 32),
 
@@ -43,7 +43,6 @@
     create_buffer_resource_from_addr,
 )
 
-from . import primitives as P
 from .config import (
     _LANE_MASK as LANE_MASK,
 )
@@ -258,22 +257,22 @@ def ep_dispatch(
         # token/count stores above are complete before the grid barrier makes
         # them visible to peers (unlike HIP __syncthreads, gpu.barrier has no
         # implicit s_waitcnt).
-        P.waitcnt_all()
+        comm_ops.waitcnt_all()
         fx.barrier()
         if tid == 0:
             comm_ops.atomic_add_system(fx.Int64(addr_disp_bar), arith.constant(1))
 
         local_recv_num = fx.Int64(window.lsa_ptr(my_lsa_rank, off_recv_num))
         for dest_pe in range(lane, npes, WAVE):
             if global_warp_id == 0:
-                P.spin_until_eq_i32(fx.Int64(addr_disp_bar), block_num)
+                comm_ops.spin_until_eq_i32(fx.Int64(addr_disp_bar), block_num)
                 buffer_store(arith.constant(0), rsrc_disp_bar, 0)
                 signal_value = (
                     buffer_load(rsrc_dest_ctr, dest_pe, vec_width=1, dtype=T.i32) + 1
                 )
                 peer_recv_num = fx.Int64(window.lsa_ptr(dest_pe, off_recv_num))
                 recv_num_remote_addr = peer_recv_num + fx.Int64(rank) * fx.Int64(4)
-                P.spin_until_eq_i32(recv_num_remote_addr, 0)
+                comm_ops.spin_until_eq_i32(recv_num_remote_addr, 0)
                 comm_ops.store_i32_system(
                     recv_num_remote_addr, arith.constant(0), signal_value
                 )
@@ -282,7 +281,7 @@ def ep_dispatch(
         for src_pe in range(lane, npes, WAVE):
             if global_warp_id == 0:
                 recv_num_src_addr = local_recv_num + fx.Int64(src_pe) * fx.Int64(4)
-                signal_value = P.spin_until_gt_i32(recv_num_src_addr, 0)
+                signal_value = comm_ops.spin_until_gt_i32(recv_num_src_addr, 0)
                 peer_recv_count = signal_value - 1
                 comm_ops.store_i32_system(
                     recv_num_src_addr, arith.constant(0), arith.constant(0)