[CI] Migrate nightly tests to test/registered/ (#15582)

test/registered/kernels/test_fp4_moe.py

@@ -0,0 +1,458 @@
+# SPDX-License-Identifier: Apache-2.0
+from typing import Callable
+
+import pytest
+import torch
+from flashinfer import fp4_quantize, scaled_fp4_grouped_quantize
+
+from sglang.test.ci.ci_register import register_cuda_ci
+
+register_cuda_ci(est_time=300, suite="nightly-4-gpu-b200", nightly=True)
+from flashinfer.fused_moe import cutlass_fused_moe as flashinfer_cutlass_fused_moe
+from sgl_kernel import scaled_fp4_quant, silu_and_mul
+from torch.nn import functional as F
+
+from sglang.srt.layers.moe.cutlass_moe import cutlass_moe_fp4
+from sglang.srt.layers.moe.cutlass_moe_params import CutlassMoEParams, CutlassMoEType
+from sglang.srt.layers.moe.topk import TopKConfig, select_experts
+
+if torch.cuda.get_device_capability() < (10, 0):
+    pytest.skip(
+        reason="Nvfp4 Requires compute capability of 10 or above.",
+        allow_module_level=True,
+    )
+
+kE2M1ToFloat = torch.tensor(
+    [0.0, 0.5, 1.0, 1.5, 2.0, 3.0, 4.0, 6.0], dtype=torch.float32
+)
+
+FLOAT8_E4M3_MAX = 448.0
+FLOAT4_E2M1_MAX = 6.0
+
+
+def convert_swizzled_to_linear(a_sf_swizzled: torch.Tensor, m, k, block_size):
+    m_tiles = (m + 128 - 1) // 128
+    f = block_size * 4
+    k_tiles = (k + f - 1) // f
+    tmp = torch.reshape(a_sf_swizzled, (1, m_tiles, k_tiles, 32, 4, 4))
+    tmp = torch.permute(tmp, (0, 1, 4, 3, 2, 5))
+    out = tmp.reshape(m_tiles * 128, k_tiles * f // block_size)
+    return out[0:m, 0:k]
+
+
+def dequantize_nvfp4_to_dtype(
+    tensor_fp4, tensor_sf, global_scale, dtype, device, block_size=16
+):
+    """Dequantize the fp4 tensor back to high precision."""
+    # Two fp4 values are packed into one uint8.
+    assert tensor_fp4.dtype == torch.uint8
+    m, packed_k = tensor_fp4.shape
+    k = packed_k * 2
+    tensor_f32 = break_fp4_bytes(tensor_fp4, dtype)
+    tensor_f32 = tensor_f32.reshape(m, k // block_size, block_size)
+    tensor_sf = tensor_sf.view(torch.float8_e4m3fn)
+    tensor_sf = convert_swizzled_to_linear(tensor_sf, m, k, block_size)
+    tensor_sf_dtype = tensor_sf.to(torch.float32) / global_scale
+
+    # scale the tensor
+    out = (tensor_f32 * tensor_sf_dtype.unsqueeze(-1)).reshape(m, k)
+    return out.to(dtype=dtype)
+
+
+def break_fp4_bytes(a, dtype):
+    assert a.dtype == torch.uint8
+    m, n = a.shape
+
+    # Vectorized nibble processing
+    a_flat = a.flatten()
+    high = (a_flat & 0xF0) >> 4  # Upper nibbles
+    low = a_flat & 0x0F  # Lower nibbles
+
+    # Combine nibbles for batch processing
+    combined = torch.stack((low, high), dim=1).flatten()
+
+    # Vectorized sign and magnitude extraction
+    signs = (combined & 0x08).to(torch.bool)  # Sign bits
+    abs_vals = (combined & 0x07).to(torch.long)  # Magnitude indices
+
+    # Device-aware lookup and sign application
+    kE2M1 = kE2M1ToFloat.to(device=a.device)
+    values = kE2M1[abs_vals] * torch.where(signs, -1.0, 1.0)
+
+    # Reshape to final form
+    return values.reshape(m, n * 2).to(dtype=dtype)
+
+
+def compute_routing(router_logits: torch.Tensor, top_k: int):
+    routing_weights = torch.softmax(router_logits, dim=1, dtype=torch.float)
+    routing_weights, selected_experts = torch.topk(routing_weights, top_k, dim=-1)
+    routing_weights /= routing_weights.sum(dim=-1, keepdim=True)
+    routing_weights = routing_weights.float()
+    return routing_weights, selected_experts
+
+
+def prepare_inputs(
+    hidden_states: torch.Tensor,
+    router_logits: torch.Tensor,
+    num_experts: int,
+    topk: int,
+):
+    routing_weights, topk_idx = compute_routing(router_logits, topk)
+
+    masked_m = []
+    for i in range(num_experts):
+        mask = topk_idx.view(-1) == i
+        masked_m.append(mask.sum())
+
+    masked_m = torch.tensor(masked_m, dtype=torch.int32)
+    hidden_states_3d = torch.empty(
+        (num_experts, max(masked_m), hidden_states.shape[1]), dtype=hidden_states.dtype
+    )
+    for i in range(num_experts):
+        hidden_states_3d[i, : masked_m[i], :] = hidden_states[topk_idx.view(-1) == i]
+
+    return hidden_states_3d, masked_m, topk_idx, routing_weights
+
+
+MNK_FACTORS = [
+    (2, 1024, 1024),
+    (2, 1024, 1536),
+    (2, 3072, 1024),
+    (2, 3072, 1536),
+    (64, 1024, 1024),
+    (64, 1024, 1536),
+    (64, 3072, 1024),
+    (64, 2048, 1024),
+    (224, 1024, 1024),
+    (224, 1024, 1536),
+]
+
+
+# Reference implementation of torch_moe
+def torch_moe(a, w1, w2, score, topk, expert_map):
+    B, D = a.shape
+    a = a.view(B, -1, D).repeat(1, topk, 1).reshape(-1, D)
+    out = torch.zeros(B * topk, w2.shape[1], dtype=a.dtype, device=a.device)
+    score = torch.softmax(score, dim=-1, dtype=torch.float32)
+    topk_weight, topk_ids = torch.topk(score, topk)
+    topk_weight = topk_weight.view(-1)
+    topk_ids = topk_ids.view(-1)
+    if expert_map is not None:
+        topk_ids = expert_map[topk_ids]
+    for i in range(w1.shape[0]):
+        mask = topk_ids == i
+        if mask.sum():
+            out[mask] = silu_and_mul(a[mask] @ w1[i].transpose(0, 1)) @ w2[i].transpose(
+                0, 1
+            )
+    return (
+        out.view(B, -1, w2.shape[1]) * topk_weight.view(B, -1, 1).to(out.dtype)
+    ).sum(dim=1)
+
+
+def torch_moe_nvfp4(a, w1, w2, topk, topk_weight, topk_ids):
+    B, D = a.shape
+    a = a.view(B, -1, D).repeat(1, topk, 1).reshape(-1, D)
+    out = torch.zeros(B * topk, w2.shape[1], dtype=a.dtype, device=a.device)
+
+    topk_weight = topk_weight.view(-1)
+    topk_ids = topk_ids.view(-1)
+
+    for i in range(w1.shape[0]):
+        mask = topk_ids == i
+        if mask.sum():
+            m = w1[i].shape[0]
+            assert m % 2 == 0
+            # Note: w1 and w3 are swapped!
+            w3_expert, w1_expert = w1[i][m // 2 :, :], w1[i][: m // 2, :]
+            inter = F.silu(a[mask] @ w1_expert.t()) * (a[mask] @ w3_expert.t())
+            inter_gs = torch.tensor(1.0).cuda()
+            inter_q, inter_blockscale = fp4_quantize(inter, inter_gs)
+            inter = dequantize_nvfp4_to_dtype(
+                inter_q,
+                inter_blockscale,
+                inter_gs,
+                dtype=inter.dtype,
+                device=inter.device,
+                block_size=16,
+            ).cuda()
+            out[mask] = inter @ w2[i].transpose(0, 1)
+    return (
+        out.view(B, -1, w2.shape[1]) * topk_weight.view(B, -1, 1).to(out.dtype)
+    ).sum(dim=1)
+
+
+def flashinfer_cutedsl_grouped_gemm_nt_masked(
+    hidden_states: torch.Tensor,  # 3d
+    input_global_scale: torch.Tensor,  # (l,)
+    weights: torch.Tensor,
+    w_global_scale: torch.Tensor,  # (l,)
+    masked_m: torch.Tensor,
+):
+    from flashinfer.cute_dsl.blockscaled_gemm import grouped_gemm_nt_masked
+
+    # hidden_states: [l, m, k]
+    # weights: [l, n, k]
+    aq, aq_sf = scaled_fp4_grouped_quantize(
+        hidden_states,
+        masked_m.to(hidden_states.device),
+        input_global_scale,
+    )
+    num_experts, n, k = weights.shape
+    bq, bq_sf = scaled_fp4_grouped_quantize(
+        weights,
+        torch.ones(num_experts, device=weights.device, dtype=torch.int32) * n,
+        w_global_scale,
+    )
+
+    out = torch.zeros(
+        (num_experts, max(masked_m), n), dtype=weights.dtype, device=aq.device
+    )
+    out = out.permute(1, 2, 0)  # requirement of kernel
+    sf_vec_size = 16
+    ab_dtype = "float4_e2m1fn"
+    sf_dtype = "float8_e4m3fn"
+    c_dtype = "bfloat16"
+    alpha = 1.0 / (input_global_scale * w_global_scale).to(out.dtype).view(
+        1, 1, num_experts
+    )
+
+    def get_cute_dtype(input: torch.Tensor) -> str:
+        if input.dtype == torch.bfloat16:
+            return "bfloat16"
+        elif input.dtype == torch.float16:
+            return "float16"
+        elif input.dtype == torch.float32:
+            return "float32"
+        else:
+            raise ValueError(f"Unsupported cute dtype {input.dtype}")
+
+    grouped_gemm_nt_masked(
+        (aq, aq_sf),
+        (bq, bq_sf),
+        out,
+        masked_m.to(aq.device),
+        ab_dtype=ab_dtype,
+        sf_dtype=sf_dtype,
+        c_dtype=c_dtype,
+        sf_vec_size=sf_vec_size,
+        alpha=alpha,
+        alpha_dtype=get_cute_dtype(alpha),
+    )
+
+    return out
+
+
+def check_moe(
+    m: int,
+    n: int,
+    k: int,
+    e: int,
+    topk: int,
+    dtype: torch.dtype,
+    moe_impl: Callable,
+    flip_w13: bool,
+):
+    torch.manual_seed(7)
+    a = torch.randn((m, k), device="cuda", dtype=dtype) / 10
+    w1 = torch.randn((e, 2 * n, k), device="cuda", dtype=dtype) / 10
+    quant_blocksize = 16
+    round_up = lambda x, y: (x + y - 1) // y * y
+    sf_w1_2n = round_up(2 * n, 128)
+    sf_w1_k = round_up(k // quant_blocksize, 4)
+    w1_blockscale = torch.empty(
+        (e, sf_w1_2n, sf_w1_k), device="cuda", dtype=torch.float8_e4m3fn
+    )
+
+    w2 = torch.randn((e, k, n), device="cuda", dtype=dtype) / 10
+    sf_w2_k = round_up(k, 128)
+    sf_w2_n = round_up(n // quant_blocksize, 4)
+    w2_blockscale = torch.empty(
+        (e, sf_w2_k, sf_w2_n), device="cuda", dtype=torch.float8_e4m3fn
+    )
+
+    w1_q = torch.empty((e, 2 * n, k // 2), device="cuda", dtype=torch.uint8)
+    w2_q = torch.empty((e, k, n // 2), device="cuda", dtype=torch.uint8)
+    w1_gs = torch.empty((e,), device="cuda", dtype=torch.float32)
+    w2_gs = torch.empty((e,), device="cuda", dtype=torch.float32)
+
+    for expert in range(e):
+        w1_amax = torch.abs(w1).max().to(torch.float32)
+        w2_amax = torch.abs(w2).max().to(torch.float32)
+        w1_gs[expert] = FLOAT8_E4M3_MAX * FLOAT4_E2M1_MAX / w1_amax
+        w2_gs[expert] = FLOAT8_E4M3_MAX * FLOAT4_E2M1_MAX / w2_amax
+
+        w1_q[expert], w1_blockscale[expert] = scaled_fp4_quant(
+            w1[expert], w1_gs[expert]
+        )
+
+        w2_q[expert], w2_blockscale[expert] = scaled_fp4_quant(
+            w2[expert], w2_gs[expert]
+        )
+
+    score = torch.randn((m, e), device="cuda", dtype=dtype)
+
+    topk_output = select_experts(
+        hidden_states=a,
+        router_logits=score,
+        topk_config=TopKConfig(top_k=topk, renormalize=False),
+    )
+    topk_weights, topk_ids, _ = topk_output
+
+    a1_gs = torch.ones((e,), device="cuda", dtype=torch.float32)
+    a2_gs = torch.ones((e,), device="cuda", dtype=torch.float32)
+    test_output = moe_impl(
+        a=a,
+        topk_weights=topk_weights,
+        topk_ids=topk_ids,
+        w1_q=w1_q,
+        w2_q=w2_q,
+        a1_gs=a1_gs,
+        w1_blockscale=w1_blockscale,
+        w1_alphas=(1 / w1_gs),
+        a2_gs=a2_gs,
+        w2_blockscale=w2_blockscale,
+        w2_alphas=(1 / w2_gs),
+    )
+
+    # Reference check:
+    a_global_scale = (
+        (FLOAT8_E4M3_MAX * FLOAT4_E2M1_MAX) / torch.amax(a.flatten(), dim=-1)
+    ).to(torch.float32)
+    a_fp4, a_scale_interleaved = scaled_fp4_quant(a, a_global_scale)
+    _, m_k = a_fp4.shape
+    a_in_dtype = dequantize_nvfp4_to_dtype(
+        a_fp4,
+        a_scale_interleaved,
+        a_global_scale,
+        dtype=a.dtype,
+        device=a.device,
+        block_size=quant_blocksize,
+    )
+
+    w1_d = torch.empty((e, 2 * n, k), device="cuda", dtype=dtype)
+    w2_d = torch.empty((e, k, n), device="cuda", dtype=dtype)
+
+    for idx in range(0, e):
+        w1_d[idx] = dequantize_nvfp4_to_dtype(
+            w1_q[idx],
+            w1_blockscale[idx],
+            w1_gs[idx],
+            dtype=w1.dtype,
+            device=w1.device,
+            block_size=quant_blocksize,
+        )
+        w2_d[idx] = dequantize_nvfp4_to_dtype(
+            w2_q[idx],
+            w2_blockscale[idx],
+            w2_gs[idx],
+            dtype=w2.dtype,
+            device=w2.device,
+            block_size=quant_blocksize,
+        )
+
+    if flip_w13:
+        dim = -2
+        size = w1_d.size(dim)
+        assert size % 2 == 0, f"Expected even size in dim {dim}, got {size}"
+        half = size // 2
+        # Reorder weight
+        w1, w3 = w1_d.split(half, dim=dim)
+        w1_d = torch.cat([w3, w1], dim=dim).contiguous()
+
+    torch_output = torch_moe(a_in_dtype, w1_d, w2_d, score, topk, None)
+
+    torch.testing.assert_close(torch_output, test_output, atol=1e-1, rtol=1e-1)
+
+
+@pytest.mark.parametrize("m,n,k", MNK_FACTORS)
+@pytest.mark.parametrize("e", [40, 64, 256])
+@pytest.mark.parametrize("topk", [1, 6, 8])
+@pytest.mark.parametrize("dtype", [torch.half, torch.bfloat16])
+@torch.inference_mode()
+def test_cutlass_fp4_moe_no_graph(
+    m: int, n: int, k: int, e: int, topk: int, dtype: torch.dtype
+):
+    def cutlass_moe_impl(
+        a,
+        topk_weights,
+        topk_ids,
+        w1_q,
+        w2_q,
+        a1_gs,
+        w1_blockscale,
+        w1_alphas,
+        a2_gs,
+        w2_blockscale,
+        w2_alphas,
+    ):
+        params = CutlassMoEParams(
+            CutlassMoEType.BlockscaledFP4,
+            device=a.device,
+            num_experts=e,
+            intermediate_size_per_partition=n,  # n
+            hidden_size=k,
+        )  # k
+        return cutlass_moe_fp4(
+            a=a,
+            a1_gscale=a1_gs,
+            w1_fp4=w1_q,
+            w1_blockscale=w1_blockscale,
+            w1_alphas=w1_alphas,
+            a2_gscale=a2_gs,
+            w2_fp4=w2_q,
+            w2_blockscale=w2_blockscale,
+            w2_alphas=w2_alphas,
+            topk_weights=topk_weights,
+            topk_ids=topk_ids,
+            params=params,
+            apply_router_weight_on_input=False,
+        )
+
+    check_moe(m, n, k, e, topk, dtype, cutlass_moe_impl, flip_w13=False)
+
+
+@pytest.mark.parametrize("m,n,k", MNK_FACTORS)
+@pytest.mark.parametrize("e", [40, 64, 256])
+@pytest.mark.parametrize("topk", [1, 6, 8])
+@pytest.mark.parametrize("dtype", [torch.half, torch.bfloat16])
+@torch.inference_mode()
+def test_flashinfer_fp4_moe_no_graph(
+    m: int, n: int, k: int, e: int, topk: int, dtype: torch.dtype
+):
+    def flashinfer_moe_impl(
+        a,
+        topk_weights,
+        topk_ids,
+        w1_q,
+        w2_q,
+        a1_gs,
+        w1_blockscale,
+        w1_alphas,
+        a2_gs,
+        w2_blockscale,
+        w2_alphas,
+    ):
+        return flashinfer_cutlass_fused_moe(
+            a,
+            topk_ids.to(torch.int),
+            topk_weights,
+            w1_q.view(torch.long),
+            w2_q.view(torch.long),
+            a.dtype,
+            quant_scales=[
+                a1_gs,
+                w1_blockscale.view(torch.int32),
+                w1_alphas,
+                a2_gs,
+                w2_blockscale.view(torch.int32),
+                w2_alphas,
+            ],
+        )[0]
+
+    check_moe(m, n, k, e, topk, dtype, flashinfer_moe_impl, flip_w13=True)
+
+
+if __name__ == "__main__":
+    test_cutlass_fp4_moe_no_graph(224, 1024, 1024, 256, 8, torch.half)
+    test_flashinfer_fp4_moe_no_graph(224, 1024, 1024, 256, 8, torch.half)

test/registered/kernels/test_nsa_indexer.py

@@ -0,0 +1,619 @@
+import unittest
+from typing import Optional
+from unittest.mock import MagicMock, patch
+
+import torch
+
+from sglang.test.ci.ci_register import register_cuda_ci
+
+register_cuda_ci(est_time=2, suite="nightly-1-gpu", nightly=True)
+
+from sglang.srt.layers import dp_attention as _dp_attn
+
+# Patch DP-attention globals before importing backends
+_dp_attn.get_attention_tp_size = lambda: 1  # TP size = 1 for unit test
+
+from sglang.srt.configs.model_config import AttentionArch
+from sglang.srt.layers.attention.nsa.nsa_indexer import (
+    BaseIndexerMetadata,
+    Indexer,
+    rotate_activation,
+)
+from sglang.srt.layers.attention.nsa_backend import NativeSparseAttnBackend
+from sglang.srt.layers.layernorm import LayerNorm
+from sglang.srt.layers.linear import LinearBase
+from sglang.srt.mem_cache.memory_pool import NSATokenToKVPool
+from sglang.srt.model_executor.forward_batch_info import ForwardBatch, ForwardMode
+from sglang.srt.server_args import ServerArgs, set_global_server_args_for_scheduler
+from sglang.test.test_utils import CustomTestCase
+
+# Global configuration for all indexer tests
+DEFAULT_CONFIG = {
+    "device": "cuda",
+    "dtype": torch.bfloat16,
+    "kv_cache_dtype": torch.float8_e4m3fn,
+    "context_len": 2048,
+    "max_bs": 64,
+    "hidden_size": 5120,
+    "index_n_heads": 1,
+    "index_head_dim": 128,
+    "rope_head_dim": 64,
+    "index_topk": 64,
+    "q_lora_rank": 1536,
+    "kv_lora_rank": 512,
+    "qk_rope_head_dim": 64,
+    "max_position_embeddings": 163840,
+    "rope_theta": 10000.0,
+    "layer_id": 0,
+    "page_size": 64,
+}
+
+
+class MockIndexerMetadata(BaseIndexerMetadata):
+    """Mock implementation of BaseIndexerMetadata for testing."""
+
+    def __init__(self, batch_size, seq_lens, page_table=None):
+        self.batch_size = batch_size
+        self.seq_lens = seq_lens
+        self.page_table = page_table
+        self.device = "cuda"
+
+    def get_seqlens_int32(self) -> torch.Tensor:
+        """Return: (batch_size,) int32 tensor"""
+        return torch.tensor(self.seq_lens, dtype=torch.int32, device=self.device)
+
+    def get_page_table_64(self) -> torch.Tensor:
+        """Return: (batch_size, num_blocks) int32, page table with page size 64."""
+        if self.page_table is not None:
+            return self.page_table
+        # Create a simple page table for testing
+        max_seq_len = max(self.seq_lens)
+        num_blocks = (max_seq_len + 63) // 64  # Round up to page size 64
+        page_table = torch.zeros(
+            (self.batch_size, num_blocks), dtype=torch.int32, device=self.device
+        )
+        for i in range(self.batch_size):
+            # Simple linear mapping: block i maps to page i
+            num_blocks_needed = (self.seq_lens[i] + 63) // 64
+            page_table[i, :num_blocks_needed] = torch.arange(
+                num_blocks_needed, device=self.device
+            )
+        return page_table
+
+    def get_seqlens_expanded(self) -> torch.Tensor:
+        """Return: (sum_extend_seq_len,) int32 tensor"""
+        # For extend mode, each new token attends to progressively more tokens
+        # For a sequence being extended from position 0 to seq_len, token i attends to i+1 tokens
+        result = []
+        for seq_len in self.seq_lens:
+            result.extend(range(1, seq_len + 1))
+        return torch.tensor(result, dtype=torch.int32, device=self.device)
+
+    def topk_transform(
+        self,
+        logits: torch.Tensor,
+        topk: int,
+        ks: Optional[torch.Tensor] = None,
+    ) -> torch.Tensor:
+        """
+        Perform topk selection on the logits.
+        For testing, just return the topk indices.
+        """
+        return torch.topk(logits, k=topk, dim=-1).indices
+
+
+class MockModelRunner:
+    def __init__(self, config=None):
+        self.device = "cuda"
+        self.config = {**DEFAULT_CONFIG, **(config or {})}
+        self.dtype = self.config["dtype"]
+        self.kv_cache_dtype = self.config["kv_cache_dtype"]
+        self.is_hybrid_swa = False
+
+        # Model configuration
+        attention_arch = AttentionArch.MLA
+        max_context_len = self.config["context_len"]
+        max_batch_size = self.config["max_bs"]
+
+        # Create mock hf_config for NSA - instantiate it as an object, not a type
+        hf_config = type(
+            "HfConfig",
+            (),
+            {
+                "architectures": ["DeepseekV3ForCausalLM"],
+                "index_topk": self.config["index_topk"],
+                "index_head_dim": self.config["index_head_dim"],
+                "index_n_heads": self.config["index_n_heads"],
+            },
+        )()
+
+        self.model_config = type(
+            "ModelConfig",
+            (),
+            {
+                "context_len": max_context_len,
+                "is_multimodal": False,
+                "attention_arch": attention_arch,
+                "num_attention_heads": 128,
+                "kv_lora_rank": self.config["kv_lora_rank"],
+                "qk_rope_head_dim": self.config["qk_rope_head_dim"],
+                "hf_config": hf_config,
+            },
+        )()
+
+        self.sliding_window_size = None
+        self.page_size = self.config["page_size"]
+
+        # Create req_to_token_pool
+        self.req_to_token_pool = type(
+            "TokenPool",
+            (),
+            {
+                "size": max_batch_size,
+                "req_to_token": torch.zeros(
+                    max_batch_size,
+                    max_context_len,
+                    dtype=torch.int32,
+                    device=self.device,
+                ),
+            },
+        )()
+
+        # Create NSATokenToKVPool
+        max_total_num_tokens = max_batch_size * max_context_len
+        self.token_to_kv_pool = NSATokenToKVPool(
+            size=max_total_num_tokens,
+            page_size=self.config["page_size"],
+            dtype=self.config["kv_cache_dtype"],
+            kv_lora_rank=self.config["kv_lora_rank"],
+            qk_rope_head_dim=self.config["qk_rope_head_dim"],
+            layer_num=1,
+            device=self.device,
+            index_head_dim=self.config["index_head_dim"],
+            enable_memory_saver=False,
+        )
+
+        # Required by backend with NSA-specific attributes
+        self.server_args = type(
+            "ServerArgs",
+            (),
+            {
+                "kv_cache_dtype": "auto",
+                "speculative_eagle_topk": None,
+                "speculative_num_draft_tokens": 0,
+                "enable_deterministic_inference": False,
+                "nsa_prefill_backend": "flashmla_sparse",
+                "nsa_decode_backend": "fa3",
+            },
+        )()
+
+
+@unittest.skipIf(not torch.cuda.is_available(), "Test requires CUDA")
+class TestNSAIndexer(CustomTestCase):
+    @classmethod
+    def setUpClass(cls):
+        """Set up global server args for testing."""
+        server_args = ServerArgs(model_path="dummy")
+        server_args.enable_dp_attention = False
+        server_args.nsa_prefill_backend = "flashmla_sparse"
+        server_args.nsa_decode_backend = "flashmla_sparse"
+        set_global_server_args_for_scheduler(server_args)
+
+        # Check GPU capability for FP8
+        if torch.cuda.is_available():
+            compute_capability = torch.cuda.get_device_capability()
+            cls.supports_fp8 = compute_capability[0] >= 9  # Hopper or newer
+
+    @classmethod
+    def tearDownClass(cls):
+        """Clean up after all tests."""
+        pass
+
+    def setUp(self):
+        # Test parameters
+        self.batch_size = 2
+        self.seq_len = 128
+        self.config = DEFAULT_CONFIG.copy()
+        self.device = "cuda"
+        self.dtype = torch.bfloat16
+
+    def _init_model_runner(self, config_override=None):
+        """Initialize model runner with optional config override."""
+        config = self.config.copy()
+        if config_override:
+            config.update(config_override)
+        self.model_runner = MockModelRunner(config)
+        self.backend = NativeSparseAttnBackend(self.model_runner)
+
+    def _create_indexer(self, **kwargs):
+        """Create an Indexer instance with default parameters."""
+        params = {
+            "hidden_size": self.config["hidden_size"],
+            "index_n_heads": self.config["index_n_heads"],
+            "index_head_dim": self.config["index_head_dim"],
+            "rope_head_dim": self.config["rope_head_dim"],
+            "index_topk": self.config["index_topk"],
+            "q_lora_rank": self.config["q_lora_rank"],
+            "max_position_embeddings": self.config["max_position_embeddings"],
+            "rope_theta": self.config["rope_theta"],
+            "layer_id": self.config["layer_id"],
+            "scale_fmt": "ue8m0",
+            "block_size": 128,
+            "quant_config": None,  # No quantization for testing
+        }
+        params.update(kwargs)
+
+        torch.set_default_dtype(self.dtype)
+        indexer = Indexer(**params)
+        # Move indexer to CUDA device
+        indexer = indexer.to(device=self.device)
+
+        # Convert linear layer weights to bfloat16 (but preserve LayerNorm's float32
+        # and weights_proj's float32 - it uses params_dtype=torch.float32 in production)
+        # Need to recursively convert LinearBase submodules (like ReplicatedLinear)
+        for name, module in indexer.named_modules():
+            # Check for LinearBase (parent of ReplicatedLinear) but exclude LayerNorm
+            # Also exclude weights_proj which uses float32 params in production
+            if isinstance(module, LinearBase) and not isinstance(module, LayerNorm):
+                if "weights_proj" not in name:
+                    module.to(dtype=self.dtype)
+
+        return indexer
+
+    def _create_forward_batch(
+        self, mode, batch_size=None, seq_len=None, extend_len=None
+    ):
+        """Create a forward batch for testing."""
+        batch_size = batch_size or self.batch_size
+        seq_len = seq_len or self.seq_len
+
+        if mode == ForwardMode.EXTEND:
+            q_len = extend_len or seq_len
+            total_len = seq_len
+
+            forward_batch = ForwardBatch(
+                batch_size=batch_size,
+                input_ids=torch.randint(
+                    0, 100, (batch_size, q_len), device=self.device
+                ),
+                out_cache_loc=torch.arange(
+                    batch_size * (total_len - q_len),
+                    batch_size * total_len,
+                    device=self.device,
+                ),
+                seq_lens_sum=batch_size * total_len,
+                forward_mode=mode,
+                req_pool_indices=torch.arange(batch_size, device=self.device),
+                seq_lens=torch.tensor([total_len] * batch_size, device=self.device),
+                seq_lens_cpu=torch.tensor([total_len] * batch_size, device="cpu"),
+                extend_prefix_lens=torch.tensor(
+                    [total_len - q_len] * batch_size, device=self.device
+                ),
+                extend_prefix_lens_cpu=torch.tensor(
+                    [total_len - q_len] * batch_size, device="cpu"
+                ),
+                extend_seq_lens=torch.tensor([q_len] * batch_size, device=self.device),
+                extend_seq_lens_cpu=torch.tensor([q_len] * batch_size, device="cpu"),
+                attn_backend=self.backend,
+            )
+        else:  # ForwardMode.DECODE
+            decode_len = 1
+            total_len = seq_len + decode_len
+
+            forward_batch = ForwardBatch(
+                batch_size=batch_size,
+                input_ids=torch.randint(
+                    0, 100, (batch_size, decode_len), device=self.device
+                ),
+                out_cache_loc=torch.arange(
+                    batch_size * seq_len, batch_size * total_len, device=self.device
+                ),
+                seq_lens_sum=batch_size * total_len,
+                forward_mode=mode,
+                req_pool_indices=torch.arange(batch_size, device=self.device),
+                seq_lens=torch.tensor([total_len] * batch_size, device=self.device),
+                seq_lens_cpu=torch.tensor([total_len] * batch_size, device="cpu"),
+                attn_backend=self.backend,
+            )
+
+        # Add token pools
+        forward_batch.req_to_token_pool = self.model_runner.req_to_token_pool
+        forward_batch.token_to_kv_pool = self.model_runner.token_to_kv_pool
+
+        # Mock write to req_to_token_pool
+        page_size = self.model_runner.page_size
+        for i in range(batch_size):
+            seq_length = total_len
+            for j in range(seq_length):
+                self.model_runner.req_to_token_pool.req_to_token[i, j] = (
+                    i * seq_length + j + page_size
+                )
+
+        return forward_batch
+
+    def _verify_topk_output(self, topk_indices, batch_size, q_len, topk):
+        """Verify the topk indices output shape and basic properties."""
+        self.assertIsNotNone(topk_indices)
+        self.assertEqual(topk_indices.device.type, "cuda")
+
+        # Check shape - should be (total_q_len, topk_padded)
+        # where topk_padded is aligned to 2048
+        self.assertEqual(len(topk_indices.shape), 2)
+        self.assertEqual(topk_indices.shape[0], batch_size * q_len)
+
+        # Check that topk is padded to at least topk
+        self.assertGreaterEqual(topk_indices.shape[1], topk)
+
+        # Check for padding values (-1)
+        has_padding = (topk_indices == -1).any()
+        self.assertTrue(
+            has_padding or topk_indices.shape[1] == topk,
+            "Output should have padding or exact topk size",
+        )
+
+    @patch("sglang.srt.layers.attention.nsa.nsa_indexer.deep_gemm")
+    def test_indexer_basic_creation(self, mock_deep_gemm):
+        """Test basic indexer creation and initialization."""
+        mock_deep_gemm.get_num_sms.return_value = 132
+
+        indexer = self._create_indexer()
+
+        self.assertEqual(indexer.hidden_size, self.config["hidden_size"])
+        self.assertEqual(indexer.n_heads, self.config["index_n_heads"])
+        self.assertEqual(indexer.head_dim, self.config["index_head_dim"])
+        self.assertEqual(indexer.rope_head_dim, self.config["rope_head_dim"])
+        self.assertEqual(indexer.index_topk, self.config["index_topk"])
+        self.assertEqual(indexer.layer_id, self.config["layer_id"])
+
+    @patch("sglang.srt.layers.attention.nsa.nsa_indexer.deep_gemm")
+    @patch("sglang.srt.layers.attention.nsa.triton_kernel.act_quant")
+    def test_forward_extend_mode(self, mock_act_quant, mock_deep_gemm):
+        """Test indexer forward pass in extend mode."""
+        if not self.supports_fp8:
+            self.skipTest("FP8 requires Hopper GPU or newer")
+
+        # Setup mocks
+        mock_deep_gemm.get_num_sms.return_value = 132
+        mock_deep_gemm.get_paged_mqa_logits_metadata.return_value = MagicMock()
+
+        def mock_quant(x, *args, **kwargs):
+            # Return FP8 tensor and scale
+            return x.to(torch.float8_e4m3fn), torch.ones(
+                x.shape[0], dtype=torch.float32, device=x.device
+            )
+
+        mock_act_quant.side_effect = mock_quant
+
+        # Mock deep_gemm.fp8_mqa_logits to return logits (ragged path)
+        def mock_mqa_logits(q, kv, weights, ks, ke, *args, **kwargs):
+            # q shape: (sum_extend_seq_len, ...), return logits for each query token
+            num_queries = q.shape[0]
+            # kv is a tuple (k_fp8, k_scale), get total number of keys from k_fp8
+            k_fp8, k_scale = kv
+            max_kv_len = k_fp8.shape[0]  # Total keys across all batches (k_offset)
+            return torch.randn(
+                num_queries, max_kv_len, dtype=torch.float32, device="cuda"
+            )
+
+        mock_deep_gemm.fp8_mqa_logits.side_effect = mock_mqa_logits
+
+        # Also mock the paged version for completeness
+        def mock_paged_mqa_logits(q, kv, weights, *args, **kwargs):
+            batch_size = q.shape[0]
+            seq_len = 128
+            return torch.randn(batch_size, seq_len, dtype=torch.float32, device="cuda")
+
+        mock_deep_gemm.fp8_paged_mqa_logits.side_effect = mock_paged_mqa_logits
+
+        self._init_model_runner()
+
+        indexer = self._create_indexer()
+        forward_batch = self._create_forward_batch(ForwardMode.EXTEND)
+
+        # Create input tensors
+        total_tokens = self.batch_size * self.seq_len
+        hidden_states = torch.randn(
+            total_tokens,
+            self.config["hidden_size"],
+            dtype=self.dtype,
+            device=self.device,
+        )
+        q_lora = torch.randn(
+            total_tokens,
+            self.config["q_lora_rank"],
+            dtype=self.dtype,
+            device=self.device,
+        )
+        positions = torch.arange(total_tokens, device=self.device)
+
+        # Run forward pass
+        with patch.object(
+            self.backend,
+            "get_indexer_metadata",
+            return_value=MockIndexerMetadata(
+                self.batch_size, [self.seq_len] * self.batch_size
+            ),
+        ):
+            topk_indices = indexer(
+                x=hidden_states,
+                q_lora=q_lora,
+                positions=positions,
+                forward_batch=forward_batch,
+                layer_id=self.config["layer_id"],
+            )
+
+        # Verify output
+        self._verify_topk_output(
+            topk_indices, self.batch_size, self.seq_len, self.config["index_topk"]
+        )
+
+    @patch("sglang.srt.layers.attention.nsa.nsa_indexer.deep_gemm")
+    @patch("sglang.srt.layers.attention.nsa.triton_kernel.act_quant")
+    def test_forward_decode_mode(self, mock_act_quant, mock_deep_gemm):
+        """Test indexer forward pass in decode mode."""
+        if not self.supports_fp8:
+            self.skipTest("FP8 requires Hopper GPU or newer")
+
+        # Setup mocks
+        mock_deep_gemm.get_num_sms.return_value = 132
+        mock_deep_gemm.get_paged_mqa_logits_metadata.return_value = MagicMock()
+
+        def mock_quant(x, *args, **kwargs):
+            return x.to(torch.float8_e4m3fn), torch.ones(
+                x.shape[0], dtype=torch.float32, device=x.device
+            )
+
+        mock_act_quant.side_effect = mock_quant
+
+        def mock_paged_mqa_logits(q, kv, weights, *args, **kwargs):
+            batch_size = q.shape[0]
+            seq_len = 128
+            return torch.randn(batch_size, seq_len, dtype=torch.float32, device="cuda")
+
+        mock_deep_gemm.fp8_paged_mqa_logits.side_effect = mock_paged_mqa_logits
+
+        self._init_model_runner()
+
+        indexer = self._create_indexer()
+        forward_batch = self._create_forward_batch(ForwardMode.DECODE)
+
+        # Create input tensors for decode (batch_size tokens only)
+        hidden_states = torch.randn(
+            self.batch_size,
+            self.config["hidden_size"],
+            dtype=self.dtype,
+            device=self.device,
+        )
+        q_lora = torch.randn(
+            self.batch_size,
+            self.config["q_lora_rank"],
+            dtype=self.dtype,
+            device=self.device,
+        )
+        positions = torch.arange(self.batch_size, device=self.device)
+
+        # Run forward pass
+        with patch.object(
+            self.backend,
+            "get_indexer_metadata",
+            return_value=MockIndexerMetadata(
+                self.batch_size, [self.seq_len + 1] * self.batch_size
+            ),
+        ):
+            topk_indices = indexer(
+                x=hidden_states,
+                q_lora=q_lora,
+                positions=positions,
+                forward_batch=forward_batch,
+                layer_id=self.config["layer_id"],
+            )
+
+        # Verify output - decode mode has q_len=1
+        self._verify_topk_output(
+            topk_indices, self.batch_size, 1, self.config["index_topk"]
+        )
+
+    def test_rotate_activation(self):
+        """Test the Hadamard transform (rotate_activation) function."""
+        # Test with power-of-2 hidden size
+        hidden_size = 128
+        x = torch.randn(16, hidden_size, dtype=torch.bfloat16, device=self.device)
+
+        try:
+            output = rotate_activation(x)
+            self.assertEqual(output.shape, x.shape)
+            self.assertEqual(output.dtype, torch.bfloat16)
+        except ImportError:
+            self.skipTest("sgl_kernel not available for hadamard_transform")
+
+    def test_rotate_activation_invalid_size(self):
+        """Test that rotate_activation fails with non-power-of-2 size."""
+        # Test with non-power-of-2 hidden size
+        hidden_size = 129  # Not a power of 2
+        x = torch.randn(16, hidden_size, dtype=torch.bfloat16, device=self.device)
+
+        with self.assertRaises(AssertionError):
+            rotate_activation(x)
+
+    def test_indexer_metadata_interface(self):
+        """Test the BaseIndexerMetadata interface implementation."""
+        batch_size = 4
+        seq_lens = [64, 128, 96, 112]
+
+        metadata = MockIndexerMetadata(batch_size, seq_lens)
+
+        # Test get_seqlens_int32
+        seqlens = metadata.get_seqlens_int32()
+        self.assertEqual(seqlens.shape, (batch_size,))
+        self.assertEqual(seqlens.dtype, torch.int32)
+        self.assertTrue(torch.all(seqlens == torch.tensor(seq_lens, device="cuda")))
+
+        # Test get_page_table_64
+        page_table = metadata.get_page_table_64()
+        self.assertEqual(len(page_table.shape), 2)
+        self.assertEqual(page_table.shape[0], batch_size)
+        self.assertEqual(page_table.dtype, torch.int32)
+
+        # Test topk_transform
+        logits = torch.randn(batch_size, 128, device="cuda")
+        topk = 64
+        topk_indices = metadata.topk_transform(logits, topk)
+        self.assertEqual(topk_indices.shape, (batch_size, topk))
+
+    # TODO: enable this test after indexer accuracy aligned
+    # @patch("sglang.srt.layers.attention.nsa.nsa_indexer.deep_gemm")
+    # def test_indexer_with_different_topk(self, mock_deep_gemm):
+    #     """Test indexer with different topk values."""
+    #     mock_deep_gemm.get_num_sms.return_value = 132
+
+    #     for topk in [32, 64, 128]:
+    #         with self.subTest(topk=topk):
+    #             indexer = self._create_indexer(index_topk=topk)
+    #             self.assertEqual(indexer.index_topk, topk)
+
+    @patch("sglang.srt.layers.attention.nsa.nsa_indexer.deep_gemm")
+    def test_indexer_with_fused_wk(self, mock_deep_gemm):
+        """Test indexer creation with fused wk and weights projection."""
+        mock_deep_gemm.get_num_sms.return_value = 132
+
+        # Note: fuse_wk_and_weights_proj feature is not currently implemented
+        # This test verifies basic indexer creation still works
+        indexer = self._create_indexer()
+        self.assertIsNotNone(indexer)
+
+    @patch("sglang.srt.layers.attention.nsa.nsa_indexer.deep_gemm")
+    def test_indexer_with_alt_stream(self, mock_deep_gemm):
+        """Test indexer creation with alternative CUDA stream."""
+        mock_deep_gemm.get_num_sms.return_value = 132
+
+        alt_stream = torch.cuda.Stream()
+        indexer = self._create_indexer(alt_stream=alt_stream)
+        self.assertEqual(indexer.alt_stream, alt_stream)
+
+    def test_shape_sanity_checks(self):
+        """Test various shape combinations for consistency."""
+        test_configs = [
+            {"batch_size": 1, "seq_len": 64},
+            {"batch_size": 4, "seq_len": 128},
+            {"batch_size": 8, "seq_len": 256},
+        ]
+
+        for config in test_configs:
+            with self.subTest(**config):
+                batch_size = config["batch_size"]
+                seq_len = config["seq_len"]
+
+                # Test metadata shapes
+                metadata = MockIndexerMetadata(batch_size, [seq_len] * batch_size)
+
+                seqlens = metadata.get_seqlens_int32()
+                self.assertEqual(seqlens.shape, (batch_size,))
+
+                page_table = metadata.get_page_table_64()
+                expected_blocks = (seq_len + 63) // 64
+                self.assertEqual(page_table.shape[0], batch_size)
+                self.assertGreaterEqual(page_table.shape[1], expected_blocks)
+
+
+if __name__ == "__main__":
+    unittest.main()