diff --git a/tests/core/kernels/npu/CMakeLists.txt b/tests/core/kernels/npu/CMakeLists.txt
index 29f9e9a85b..f1de4769bb 100644
--- a/tests/core/kernels/npu/CMakeLists.txt
+++ b/tests/core/kernels/npu/CMakeLists.txt
@@ -16,6 +16,16 @@ cc_test(
     pybind11::embed
 )
 
+# Keep the embedded-Python NPU probes isolated from the parallel CTest pool.
+# This limits shared-device contention but is not a substitute for validating
+# each operator's supported parameter combinations.
+set_tests_properties(
+  NpuXllmOpsTest.Dsv4QuantLightningIndexerPythonWrapperRunsOnA3
+  NpuXllmOpsTest.Dsv4QuantLightningIndexerProductionShapeRunsOnA3
+  NpuXllmOpsTest.Dsv4SparseAttentionPythonWrapperRunsOnA3
+  PROPERTIES RUN_SERIAL TRUE
+)
+
 # Temporarily disabled: Dsv4ScatterCache tests are unstable in the current
 # NPU test environment.
 # cc_test(
diff --git a/tests/core/kernels/npu/npu_xllm_ops_test.cpp b/tests/core/kernels/npu/npu_xllm_ops_test.cpp
index 3a47ab700d..cb4a1fbc5a 100644
--- a/tests/core/kernels/npu/npu_xllm_ops_test.cpp
+++ b/tests/core/kernels/npu/npu_xllm_ops_test.cpp
@@ -76,6 +76,12 @@ bool is_ascend950_device() {
          std::string(soc_name).find("Ascend950") != std::string::npos;
 }
 
+bool is_ascend910_93_device() {
+  const char* soc_name = aclrtGetSocName();
+  return soc_name != nullptr &&
+         std::string(soc_name).find("Ascend910_93") != std::string::npos;
+}
+
 torch::Tensor expand_kv_heads_reference(const torch::Tensor& tensor,
                                         int64_t num_heads) {
   const int64_t num_kv_heads = tensor.size(1);
@@ -209,6 +215,520 @@ TEST_F(NpuXllmOpsTest, EmbeddedInterpreterSeesOps) {
              .item<float>();
 }
 
+TEST_F(NpuXllmOpsTest, Dsv4OpsUseNpuDispatchKeys) {
+  py::gil_scoped_acquire gil;
+
+  py::exec(R"PY(
+import torch
+
+device_ops = (
+    "moe_gating_top_k_hash",
+    "dequant_swiglu_quant",
+    "hc_pre",
+    "hc_post",
+    "compressor",
+    "sparse_attn_sharedkv",
+    "quant_lightning_indexer",
+)
+for op_name in device_ops:
+    qualname = f"xllm_ops::{op_name}"
+    assert torch._C._dispatch_has_kernel_for_dispatch_key(
+        qualname, "PrivateUse1"
+    ), qualname
+    assert not torch._C._dispatch_has_kernel_for_dispatch_key(
+        qualname, "CompositeExplicitAutograd"
+    ), qualname
+
+for op_name in (
+    "sparse_attn_sharedkv_metadata",
+    "quant_lightning_indexer_metadata",
+):
+    assert torch._C._dispatch_has_kernel_for_dispatch_key(
+        f"xllm_ops::{op_name}", "CompositeExplicitAutograd"
+    ), op_name
+)PY");
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4GroupGemmMatchesInt32Reference) {
+  py::gil_scoped_acquire gil;
+
+  py::exec(R"PY(
+import torch
+from xllm.python.kernels_npu.moe import _group_gemm
+
+device = torch.device("privateuseone:0")
+torch.manual_seed(20260814)
+tokens, experts, input_dim, output_dim = 8, 2, 128, 256
+x_cpu = torch.randint(-4, 5, (tokens, input_dim), dtype=torch.int8)
+w_cpu = torch.randint(-4, 5, (experts, input_dim, output_dim), dtype=torch.int8)
+group_list_cpu = torch.tensor([4, 4], dtype=torch.int64)
+
+x = x_cpu.to(device)
+w = w_cpu.to(device)
+group_list = group_list_cpu.to(device)
+out = _group_gemm(
+    x=x,
+    weight=w,
+    scale=None,
+    per_token_scale=None,
+    group_list=group_list,
+    split_item=2,
+    group_type=0,
+    group_list_type=1,
+    output_dtype=torch.int32,
+)
+torch.npu.synchronize()
+
+expected = torch.cat((
+    x_cpu[:4].to(torch.int32) @ w_cpu[0].to(torch.int32),
+    x_cpu[4:].to(torch.int32) @ w_cpu[1].to(torch.int32),
+), dim=0)
+assert out.shape == (tokens, output_dim)
+assert out.dtype == torch.int32
+torch.testing.assert_close(out.cpu(), expected, rtol=0, atol=0)
+)PY");
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4GroupGemmAcceptsScaleAndPerTokenScale) {
+  py::gil_scoped_acquire gil;
+
+  py::exec(R"PY(
+import torch
+from xllm.python.kernels_npu.moe import _group_gemm
+
+device = torch.device("privateuseone:0")
+tokens, experts, input_dim, output_dim = 8, 2, 128, 128
+x = torch.randint(-4, 5, (tokens, input_dim), dtype=torch.int8, device=device)
+w = torch.randint(-4, 5, (experts, input_dim, output_dim), dtype=torch.int8, device=device)
+scale = torch.ones((experts, output_dim), dtype=torch.bfloat16, device=device)
+per_token_scale = torch.ones((tokens,), dtype=torch.float32, device=device)
+group_list = torch.tensor([4, 4], dtype=torch.int64, device=device)
+
+out = _group_gemm(
+    x=x,
+    weight=w,
+    scale=scale,
+    per_token_scale=per_token_scale,
+    group_list=group_list,
+    split_item=2,
+    group_type=0,
+    group_list_type=1,
+    output_dtype=torch.bfloat16,
+)
+torch.npu.synchronize()
+assert out.shape == (tokens, output_dim)
+assert out.dtype == torch.bfloat16
+)PY");
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4PartialRotaryPythonWrapperRunsOnNpu) {
+  py::gil_scoped_acquire gil;
+
+  py::exec(R"PY(
+import torch
+from xllm.python.kernels_npu.rotary_embedding import (
+    npu_inplace_partial_rotary_mul,
+)
+
+torch.manual_seed(2026)
+x_cpu = torch.randn((8, 2, 128), dtype=torch.float32).to(torch.bfloat16)
+cos_cpu = torch.randn((8, 64), dtype=torch.float32).to(torch.bfloat16)
+sin_cpu = torch.randn((8, 64), dtype=torch.float32).to(torch.bfloat16)
+
+expected = x_cpu.float().clone()
+segment = x_cpu[..., 64:128].float()
+swapped = torch.empty_like(segment)
+swapped[..., 0::2] = segment[..., 1::2]
+swapped[..., 1::2] = segment[..., 0::2]
+sign = torch.ones_like(cos_cpu.float())
+sign[..., 0::2] = -1
+expected[..., 64:128] = (
+    segment * cos_cpu.float().unsqueeze(1)
+    + swapped * sin_cpu.float().unsqueeze(1) * sign.unsqueeze(1)
+)
+expected = expected.to(torch.bfloat16).float()
+
+x = x_cpu.to("privateuseone:0")
+cos = cos_cpu.to(x.device)
+sin = sin_cpu.to(x.device)
+result = npu_inplace_partial_rotary_mul(x, cos, sin, 64, 64)
+torch.npu.synchronize()
+
+assert result.data_ptr() == x.data_ptr()
+torch.testing.assert_close(
+    x.cpu().float(), expected, atol=2e-2, rtol=2e-2
+)
+)PY");
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4CompressorPythonWrapperRunsOnNpu) {
+  py::gil_scoped_acquire gil;
+
+  py::exec(R"PY(
+import torch
+from xllm.python.kernels_npu.dsa import compressor
+
+device = torch.device("privateuseone:0")
+torch.manual_seed(2025)
+batch, tokens, hidden = 1, 128, 1024
+ratio, head_dim, coff, rope_dim = 128, 512, 1, 64
+compressed_tokens = tokens // ratio
+
+x_cpu = (torch.randn(batch, tokens, hidden) * 0.1).to(torch.float16)
+wkv_cpu = (torch.randn(coff * head_dim, hidden) * 0.05).to(torch.float16)
+wgate_cpu = (torch.randn(coff * head_dim, hidden) * 0.05).to(torch.float16)
+ape_cpu = (torch.randn(ratio, coff * head_dim) * 0.1).float()
+norm_cpu = (torch.randn(head_dim) * 0.1 + 1).to(torch.float16)
+rope_cos_cpu = (
+    torch.randn(batch, compressed_tokens, rope_dim) * 0.1
+).to(torch.float16)
+rope_sin_cpu = (
+    torch.randn(batch, compressed_tokens, rope_dim) * 0.1
+).to(torch.float16)
+
+projected_kv = x_cpu.float()[0] @ wkv_cpu.float().T
+scores = x_cpu.float()[0] @ wgate_cpu.float().T + ape_cpu
+pooled = (torch.softmax(scores, dim=0) * projected_kv).sum(0, keepdim=True)
+variance = pooled.square().mean(-1, keepdim=True)
+expected = pooled * torch.rsqrt(variance + 1e-6) * norm_cpu.float()
+rope_segment = expected[:, -rope_dim:].clone()
+half = rope_dim // 2
+rotated = torch.cat((-rope_segment[:, half:], rope_segment[:, :half]), dim=-1)
+expected[:, -rope_dim:] = (
+    rope_segment * rope_cos_cpu.float()[0]
+    + rotated * rope_sin_cpu.float()[0]
+)
+expected = expected.view(batch, compressed_tokens, head_dim).half().float()
+
+x = x_cpu.to(device)
+wkv = wkv_cpu.to(device)
+wgate = wgate_cpu.to(device)
+ape = ape_cpu.to(device)
+norm_weight = norm_cpu.to(device)
+rope_sin = rope_sin_cpu.to(device)
+rope_cos = rope_cos_cpu.to(device)
+kv_state = torch.zeros((1, 128, head_dim), dtype=torch.float32, device=device)
+score_state = torch.zeros_like(kv_state)
+kv_block_table = torch.tensor([[0]], dtype=torch.int32, device=device)
+score_block_table = torch.tensor([[0]], dtype=torch.int32, device=device)
+
+out, wkv_proj, softmax_res, norm_x, norm_rstd = compressor(
+    x,
+    wkv,
+    wgate,
+    kv_state,
+    score_state,
+    ape,
+    norm_weight,
+    rope_sin,
+    rope_cos,
+    kv_block_table,
+    score_block_table,
+    None,
+    None,
+    None,
+    rope_dim,
+    ratio,
+    coff,
+    1e-6,
+    1,
+    False,
+)
+torch.npu.synchronize()
+
+assert out.shape == (batch, compressed_tokens, head_dim)
+assert out.dtype == torch.float16
+assert wkv_proj.numel() == 0
+assert softmax_res.numel() == 0
+assert norm_x.numel() == 0
+assert norm_rstd.numel() == 0
+torch.testing.assert_close(
+    out.cpu().float(), expected, atol=2e-2, rtol=2e-2
+)
+)PY");
+}
+
+void run_dsv4_quant_lightning_indexer_probe(bool production_shape) {
+  py::gil_scoped_acquire gil;
+  py::dict locals;
+  locals["production_shape"] = production_shape;
+
+  py::exec(R"PY(
+import torch
+from xllm.python.kernels_npu.dsa import (
+    quant_lightning_indexer,
+    quant_lightning_indexer_metadata,
+)
+
+device = torch.device("privateuseone:0")
+torch.manual_seed(2026)
+heads, head_dim, page_size = 64, 128, 128
+batch = 1
+if production_shape:
+    q_tokens, kv_tokens = 84, 84
+    sparse_count, cmp_ratio = 512, 4
+    query_layout = "TND"
+    query_shape = (q_tokens, heads, head_dim)
+    weights_shape = (q_tokens, heads)
+    expected_indices_shape = (q_tokens, 1, sparse_count)
+else:
+    q_tokens, kv_tokens = 4, 128
+    sparse_count, cmp_ratio = 8, 1
+    query_layout = "BSND"
+    query_shape = (batch, q_tokens, heads, head_dim)
+    weights_shape = (batch, q_tokens, heads)
+    expected_indices_shape = (batch, q_tokens, 1, sparse_count)
+
+query_cpu = torch.randint(-8, 8, query_shape, dtype=torch.int8)
+key_cpu = torch.randint(-8, 8, (1, page_size, 1, head_dim), dtype=torch.int8)
+query = query_cpu.to(device)
+key = key_cpu.to(device)
+weights = torch.ones(weights_shape, dtype=torch.float16, device=device)
+query_scale = torch.ones_like(weights)
+key_scale = torch.ones((1, page_size, 1), dtype=torch.float16, device=device)
+query_lens = torch.tensor([q_tokens], dtype=torch.int32, device=device)
+key_lens = torch.tensor([kv_tokens], dtype=torch.int32, device=device)
+block_table = torch.tensor([[0]], dtype=torch.int32, device=device)
+metadata = quant_lightning_indexer_metadata(
+    heads,
+    1,
+    head_dim,
+    0,
+    0,
+    query_lens,
+    key_lens,
+    batch,
+    q_tokens,
+    kv_tokens,
+    query_layout,
+    "PA_BSND",
+    sparse_count,
+    3,
+    2**63 - 1,
+    2**63 - 1,
+    cmp_ratio,
+    "npu",
+)
+metadata_again = quant_lightning_indexer_metadata(
+    heads,
+    1,
+    head_dim,
+    0,
+    0,
+    query_lens,
+    key_lens,
+    batch,
+    q_tokens,
+    kv_tokens,
+    query_layout,
+    "PA_BSND",
+    sparse_count,
+    3,
+    2**63 - 1,
+    2**63 - 1,
+    cmp_ratio,
+    "npu",
+)
+indices, values = quant_lightning_indexer(
+    query,
+    key,
+    weights,
+    query_scale,
+    key_scale,
+    0,
+    0,
+    query_lens,
+    key_lens,
+    block_table,
+    metadata,
+    query_layout,
+    "PA_BSND",
+    sparse_count,
+    3,
+    2**63 - 1,
+    2**63 - 1,
+    cmp_ratio,
+    False,
+)
+torch.npu.synchronize()
+
+assert indices.shape == expected_indices_shape
+assert indices.dtype == torch.int32
+assert values.numel() == 0
+assert values.dtype == torch.float32
+assert torch.equal(metadata.cpu(), metadata_again.cpu())
+
+valid_key_count = kv_tokens // cmp_ratio
+indices_cpu = indices.cpu()
+assert torch.all(
+    (indices_cpu == -1)
+    | ((indices_cpu >= 0) & (indices_cpu < valid_key_count))
+)
+keys = key_cpu[0, :valid_key_count, 0].float()
+token_idx = q_tokens - 1
+if production_shape:
+    query_token = query_cpu[token_idx]
+    actual_indices = indices_cpu[token_idx, 0]
+else:
+    query_token = query_cpu[0, token_idx]
+    actual_indices = indices_cpu[0, token_idx, 0]
+dots = query_token.float() @ keys.T
+expected_top8 = set(torch.topk(dots.clamp_min(0).sum(0), 8).indices.tolist())
+actual_top8 = set(actual_indices[:8].tolist())
+assert len(expected_top8 & actual_top8) >= 4, (
+    sorted(expected_top8),
+    sorted(actual_top8),
+)
+)PY",
+           py::globals(),
+           locals);
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4QuantLightningIndexerPythonWrapperRunsOnA3) {
+  if (!is_ascend910_93_device()) {
+    GTEST_SKIP() << "Atlas A3 is required for this DSV4 QLI operator probe.";
+  }
+  run_dsv4_quant_lightning_indexer_probe(/*production_shape=*/false);
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4QuantLightningIndexerProductionShapeRunsOnA3) {
+  if (!is_ascend910_93_device()) {
+    GTEST_SKIP() << "Atlas A3 is required for the production-shape QLI probe.";
+  }
+  run_dsv4_quant_lightning_indexer_probe(/*production_shape=*/true);
+}
+
+TEST_F(NpuXllmOpsTest, Dsv4SparseAttentionPythonWrapperRunsOnA3) {
+  if (!is_ascend910_93_device()) {
+    GTEST_SKIP() << "Atlas A3 is required for this DSV4 sparse-attention "
+                    "operator probe.";
+  }
+  py::gil_scoped_acquire gil;
+
+  py::exec(R"PY(
+import torch
+from xllm.python.kernels_npu.dsa import (
+    sparse_attn_sharedkv,
+    sparse_attn_sharedkv_metadata,
+)
+
+device = torch.device("privateuseone:0")
+torch.manual_seed(1234)
+batch, q_tokens, kv_tokens = 1, 4, 16
+heads, head_dim, page_size = 64, 512, 16
+query_cpu = (torch.randn(batch, q_tokens, heads, head_dim) * 0.1).half()
+kv_cpu = (torch.randn(batch, kv_tokens, 1, head_dim) * 0.1).half()
+sinks_cpu = (torch.randn(heads) * 0.1).float()
+query = query_cpu.to(device)
+ori_kv = kv_cpu.view(1, page_size, 1, head_dim).to(device)
+block_table = torch.tensor([[0]], dtype=torch.int32, device=device)
+cu_q = torch.tensor([0, q_tokens], dtype=torch.int32, device=device)
+cu_kv = torch.tensor([0, kv_tokens], dtype=torch.int32, device=device)
+seq_q = torch.tensor([q_tokens], dtype=torch.int32, device=device)
+seq_kv = torch.tensor([kv_tokens], dtype=torch.int32, device=device)
+sinks = sinks_cpu.to(device)
+metadata = sparse_attn_sharedkv_metadata(
+    heads,
+    1,
+    head_dim,
+    cu_q,
+    cu_kv,
+    None,
+    seq_q,
+    seq_kv,
+    batch,
+    q_tokens,
+    kv_tokens,
+    0,
+    0,
+    4,
+    4,
+    3,
+    127,
+    0,
+    "BSND",
+    "PA_ND",
+    True,
+    False,
+)
+metadata_again = sparse_attn_sharedkv_metadata(
+    heads,
+    1,
+    head_dim,
+    cu_q,
+    cu_kv,
+    None,
+    seq_q,
+    seq_kv,
+    batch,
+    q_tokens,
+    kv_tokens,
+    0,
+    0,
+    4,
+    4,
+    3,
+    127,
+    0,
+    "BSND",
+    "PA_ND",
+    True,
+    False,
+)
+out, lse = sparse_attn_sharedkv(
+    query,
+    ori_kv,
+    None,
+    None,
+    None,
+    block_table,
+    None,
+    None,
+    None,
+    None,
+    None,
+    seq_kv,
+    sinks,
+    metadata,
+    head_dim**-0.5,
+    4,
+    4,
+    3,
+    127,
+    0,
+    "BSND",
+    "PA_ND",
+    False,
+)
+torch.npu.synchronize()
+
+assert out.shape == query.shape
+assert out.dtype == query.dtype
+assert lse.numel() == 0
+assert torch.equal(metadata.cpu(), metadata_again.cpu())
+
+expected = torch.zeros_like(query_cpu.float())
+keys = kv_cpu[0, :, 0].float()
+scale = head_dim**-0.5
+for q_idx in range(q_tokens):
+    diagonal = kv_tokens - q_tokens + q_idx
+    left = max(diagonal - 127, 0)
+    right = diagonal
+    selected_keys = keys[left:right + 1]
+    logits = query_cpu[0, q_idx].float() @ selected_keys.T * scale
+    sink_logits = sinks_cpu[:, None]
+    normalizer = torch.logsumexp(
+        torch.cat((logits, sink_logits), dim=1), dim=1
+    )
+    probabilities = torch.exp(logits - normalizer[:, None])
+    expected[0, q_idx] = probabilities @ selected_keys
+expected = expected.half().float()
+torch.testing.assert_close(
+    out.cpu().float(), expected, atol=2e-2, rtol=2e-2
+)
+)PY");
+}
+
 TEST_F(NpuXllmOpsTest, Qwen35_27B_TP4_FullAttentionMatchesReference) {
   py::gil_scoped_acquire gil;
   if (!is_ascend950_device()) {
diff --git a/tests/python/test_grouped_moe.py b/tests/python/test_grouped_moe.py
new file mode 100644
index 0000000000..08c6a923d7
--- /dev/null
+++ b/tests/python/test_grouped_moe.py
@@ -0,0 +1,149 @@
+# Copyright 2026 The xLLM Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://github.com/xLLM-AI/xllm/blob/main/LICENSE
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Contracts for the NPU pre-selected grouped MoE path."""
+
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+import torch
+
+_REPO_ROOT = Path(__file__).parents[2]
+
+
+def _load_npu_moe_module():
+    path = _REPO_ROOT / "xllm/python/kernels_npu/moe.py"
+    spec = importlib.util.spec_from_file_location("pr5_npu_moe", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_selected_expert_moe_matches_native_call_contract(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from xllm.python import kernels
+
+    moe = _load_npu_moe_module()
+
+    hidden = torch.empty(3, 16, dtype=torch.bfloat16)
+    topk_weights = torch.ones(3, 2, dtype=torch.bfloat16)
+    topk_ids = torch.tensor([[4, 0], [5, 9], [7, 6]], dtype=torch.int32)
+    expanded = torch.empty(6, 16, dtype=torch.bfloat16)
+    row_ids = torch.arange(6, dtype=torch.int32)
+    expert_tokens = torch.tensor([1, 3, 5, 6, 7, 8], dtype=torch.int64)
+    quantized = torch.empty(6, 16, dtype=torch.int8)
+    input_scale = torch.empty(6, dtype=torch.float32)
+    gemm1 = torch.empty(6, 32, dtype=torch.int32)
+    activated = torch.empty(6, 16, dtype=torch.int8)
+    activation_scale = torch.empty(6, dtype=torch.float32)
+    gemm2 = torch.empty(6, 16, dtype=torch.bfloat16)
+    calls: list[tuple[str, object]] = []
+
+    def init_routing(*args, **kwargs):
+        calls.append(("routing", kwargs))
+        return expanded, row_ids, expert_tokens, torch.empty(0)
+
+    def dynamic_quant(value):
+        assert value is expanded
+        calls.append(("dynamic_quant", value))
+        return quantized, input_scale
+
+    def dequant_swiglu_quant(**kwargs):
+        calls.append(("dequant_swiglu_quant", kwargs))
+        return activated, activation_scale
+
+    gemm_calls: list[dict[str, object]] = []
+
+    def group_gemm(**kwargs):
+        gemm_calls.append(kwargs)
+        return gemm1 if len(gemm_calls) == 1 else gemm2
+
+    def token_unpermute(**kwargs):
+        calls.append(("unpermute", kwargs))
+        return hidden
+
+    monkeypatch.setattr(moe, "_group_gemm", group_gemm)
+    monkeypatch.setattr(moe.torch_npu, "npu_moe_init_routing_v2", init_routing)
+    monkeypatch.setattr(moe.torch_npu, "npu_moe_token_unpermute", token_unpermute)
+    monkeypatch.setattr(kernels, "dynamic_quant", dynamic_quant, raising=False)
+    monkeypatch.setattr(kernels, "dequant_swiglu_quant", dequant_swiglu_quant, raising=False)
+
+    result = moe._grouped_moe_with_selected_experts_impl(
+        hidden,
+        topk_weights,
+        topk_ids,
+        torch.empty(4, 16, 32, dtype=torch.int8),
+        torch.empty(4, 16, 16, dtype=torch.int8),
+        torch.empty(4, 32),
+        torch.empty(4, 16),
+        num_total_experts=16,
+        start_expert_id=4,
+        num_experts_per_rank=4,
+        swiglu_limit=7.0,
+    )
+
+    assert result is hidden
+    routing = dict(calls)["routing"]
+    assert isinstance(routing, dict)
+    assert routing["active_expert_range"] == [4, 8]
+    assert routing["expert_num"] == 16
+    assert routing["quant_mode"] == -1
+
+    assert len(gemm_calls) == 2
+    assert gemm_calls[0]["scale"] is None
+    assert gemm_calls[0]["per_token_scale"] is None
+    assert gemm_calls[0]["output_dtype"] == torch.int32
+    assert gemm_calls[1]["scale"].dtype == torch.bfloat16
+    assert gemm_calls[1]["per_token_scale"] is activation_scale
+    assert gemm_calls[1]["output_dtype"] == torch.bfloat16
+    assert all(torch.equal(call["group_list"], expert_tokens[:4]) for call in gemm_calls)
+    assert all(call["group_list"].numel() == 4 for call in gemm_calls)
+    assert all(call["group_list_type"] == 1 for call in gemm_calls)
+
+    dequant = dict(calls)["dequant_swiglu_quant"]
+    assert isinstance(dequant, dict)
+    assert dequant["x"] is gemm1
+    assert dequant["activation_scale"] is input_scale
+    assert torch.equal(dequant["group_index"], expert_tokens[:4])
+    assert dequant["clamp_limit"] == 7.0
+
+    unpermute = dict(calls)["unpermute"]
+    assert isinstance(unpermute, dict)
+    torch.testing.assert_close(
+        unpermute["probs"],
+        torch.tensor([[1, 0], [1, 0], [1, 1]], dtype=torch.bfloat16),
+    )
+
+
+def test_selected_expert_moe_rejects_an_invalid_active_range() -> None:
+    moe = _load_npu_moe_module()
+
+    with pytest.raises(ValueError, match="active expert range"):
+        moe._grouped_moe_with_selected_experts_impl(
+            torch.empty(1, 16, dtype=torch.bfloat16),
+            torch.ones(1, 1, dtype=torch.bfloat16),
+            torch.zeros(1, 1, dtype=torch.int32),
+            torch.empty(4, 16, 32, dtype=torch.int8),
+            torch.empty(4, 16, 16, dtype=torch.int8),
+            torch.empty(4, 32),
+            torch.empty(4, 16),
+            num_total_experts=16,
+            start_expert_id=14,
+            num_experts_per_rank=4,
+        )
diff --git a/xllm/core/kernels/npu/npu_ops_api.h b/xllm/core/kernels/npu/npu_ops_api.h
index 8b5889ad7f..8af91086b4 100644
--- a/xllm/core/kernels/npu/npu_ops_api.h
+++ b/xllm/core/kernels/npu/npu_ops_api.h
@@ -446,4 +446,5 @@ std::tuple<torch::Tensor, torch::Tensor> apply_npu_mega_moe(
     int64_t dispatch_quant_out_dtype = 0,
     int64_t topo_type = 0,
     int64_t rank_num_per_server = 2);
+
 }  // namespace xllm::kernel::npu
diff --git a/xllm/core/kernels/npu/npu_ops_library.cpp b/xllm/core/kernels/npu/npu_ops_library.cpp
index 1c9ae56305..92c8a9f017 100644
--- a/xllm/core/kernels/npu/npu_ops_library.cpp
+++ b/xllm/core/kernels/npu/npu_ops_library.cpp
@@ -283,6 +283,19 @@ TORCH_LIBRARY(xllm_ops, m) {
       "fused_add_rms_norm(Tensor(a!) input, Tensor(b!) residual, Tensor "
       "weight, "
       "float eps) -> (Tensor, Tensor)");
+  // Fused RMSNorm + dynamic per-token int8 quant (W8A8 query preprocess).
+  // Returns (qr_int8, qr_pertoken_scale) matching C++ rms_norm_dynamic_quant
+  // (npu_ops_api.h:122), used by the DSV4 indexer build_query path.
+  m.def(
+      "rms_norm_dynamic_quant(Tensor input, Tensor weight, float eps) -> "
+      "(Tensor, Tensor)");
+  // In-place partial rotary embedding (interleaved). x is 4D [B,N,S,D], r1/r2
+  // are cos/sin [B,1,1,rope_head_dim]; partial_slice=[rope_start,
+  // rope_head_dim]. Mirrors C++ apply_partial_rope
+  // (deepseek_sparse_attention.cpp:151) used by the DSV4 indexer build_query.
+  m.def(
+      "npu_inplace_partial_rotary_mul(Tensor(a!) x, Tensor r1, Tensor r2, "
+      "str rotary_mode, int[] partial_slice) -> ()");
   m.def("silu_and_mul(Tensor input) -> Tensor");
   m.def(
       "fused_qk_norm_rope(Tensor(a!) qkv, int num_heads_q, int num_heads_k, "
@@ -349,11 +362,84 @@ TORCH_LIBRARY(xllm_ops, m) {
       "shard_valid_mask, Tensor restore_index, Tensor query_index, Tensor "
       "kv_gather_index, int[] q_cu_seqlens, int[] kv_cu_seqlens, int "
       "total_local)");
+  // ---- DeepSeek-V4 DSA kernels ----
+  // MoE hash routing gate (returns routed output, expert_idx, token_unpermute).
+  m.def(
+      "moe_gating_top_k_hash(Tensor x, int k, Tensor? bias, Tensor? input_ids, "
+      "Tensor? tid2eid, int k_group, int group_count, float "
+      "routed_scaling_factor, "
+      "float eps, int group_select_mode, int renorm, int norm_type, bool "
+      "out_flag) -> (Tensor, Tensor, Tensor)");
+  // Dequant + SwiGLU + quant (fused, replaces manual dequant loop).
+  m.def(
+      "dequant_swiglu_quant(Tensor x, Tensor? weight_scale, Tensor? "
+      "activation_scale, Tensor? bias, Tensor? quant_scale, Tensor? "
+      "quant_offset, Tensor? group_index, bool activate_left, int quant_mode, "
+      "int swiglu_mode, float clamp_limit, float glu_alpha, float glu_bias) "
+      "-> (Tensor, Tensor)");
+  // HyperConnection pre/post (hc_pre returns attn_input, post, comb).
+  m.def(
+      "hc_pre(Tensor x, Tensor hc_fn, Tensor hc_scale, Tensor hc_base, "
+      "int hc_mult, int hc_sinkhorn_iters, float norm_eps, float hc_eps) "
+      "-> (Tensor, Tensor, Tensor)");
+  m.def(
+      "hc_post(Tensor x, Tensor residual, Tensor post, Tensor comb) -> "
+      "Tensor");
+  // Compressor: NSA-style KV pooling. kv_state/score_state are in-place (Ref).
+  // Returns (cmp_kv, wkv_proj, softmax_res, norm_x, norm_rstd).
+  m.def(
+      "compressor(Tensor x, Tensor wkv, Tensor wgate, Tensor(a!) kv_state, "
+      "Tensor(b!) score_state, Tensor ape, Tensor norm_weight, Tensor "
+      "rope_sin, Tensor rope_cos, Tensor? kv_block_table, Tensor? "
+      "score_block_table, Tensor? cu_seqlens, Tensor? seqused, Tensor? "
+      "start_pos, int rope_head_dim, int cmp_ratio, int coff, float "
+      "norm_eps, int rotary_mode, bool enable_grad) -> (Tensor, Tensor, "
+      "Tensor, Tensor, Tensor)");
+  // Two-stage sparse attention over original + compressed KV.
+  m.def(
+      "sparse_attn_sharedkv(Tensor q, Tensor? ori_kv, Tensor? cmp_kv, "
+      "Tensor? ori_sparse_indices, Tensor? cmp_sparse_indices, Tensor? "
+      "ori_block_table, Tensor? cmp_block_table, Tensor? cu_seqlens_q, "
+      "Tensor? cu_seqlens_ori_kv, Tensor? cu_seqlens_cmp_kv, Tensor? "
+      "seqused_q, Tensor? seqused_kv, Tensor? sinks, Tensor? metadata, "
+      "float softmax_scale, int cmp_ratio, int ori_mask_mode, int "
+      "cmp_mask_mode, int ori_win_left, int ori_win_right, str layout_q, "
+      "str layout_kv, bool return_softmax_lse) -> (Tensor, Tensor)");
+  // AICPU tiling metadata builder for sparse_attn_sharedkv.
+  m.def(
+      "sparse_attn_sharedkv_metadata(int num_heads_q, int num_heads_kv, int "
+      "head_dim, Tensor? cu_seqlens_q, Tensor? cu_seqlens_ori_kv, Tensor? "
+      "cu_seqlens_cmp_kv, Tensor? seqused_q, Tensor? seqused_kv, int "
+      "batch_size, int max_seqlen_q, int max_seqlen_kv, int ori_topk, int "
+      "cmp_topk, int cmp_ratio, int ori_mask_mode, int cmp_mask_mode, int "
+      "ori_win_left, int ori_win_right, str layout_q, str layout_kv, bool "
+      "has_ori_kv, bool has_cmp_kv) -> Tensor");
+  // Quantized lightning indexer: int8 q/k top-k selection with cmp_ratio.
+  m.def(
+      "quant_lightning_indexer(Tensor query, Tensor key, Tensor weights, "
+      "Tensor query_dequant_scale, Tensor key_dequant_scale, int "
+      "query_quant_mode, int key_quant_mode, Tensor? actual_seq_lengths_query, "
+      "Tensor? actual_seq_lengths_key, Tensor? block_table, Tensor? metadata, "
+      "str layout_query, str layout_key, int sparse_count, int sparse_mode, "
+      "int pre_tokens, int next_tokens, int cmp_ratio, bool return_value) -> "
+      "(Tensor, Tensor)");
+  // AICPU tiling metadata builder for quant_lightning_indexer.
+  m.def(
+      "quant_lightning_indexer_metadata(int num_heads_q, int num_heads_k, int "
+      "head_dim, int query_quant_mode, int key_quant_mode, Tensor? "
+      "actual_seq_lengths_query, Tensor? actual_seq_lengths_key, int "
+      "batch_size, int max_seqlen_q, int max_seqlen_k, str layout_query, str "
+      "layout_key, int sparse_count, int sparse_mode, int pre_tokens, int "
+      "next_tokens, int cmp_ratio, str device) -> Tensor");
 }
 
 TORCH_LIBRARY_IMPL(xllm_ops, PrivateUse1, m) {
   m.impl("rms_norm", TORCH_FN(xllm::rms_norm_npu));
   m.impl("fused_add_rms_norm", TORCH_FN(xllm::fused_add_rms_norm_npu));
+  m.impl("rms_norm_dynamic_quant",
+         TORCH_FN(xllm::kernel::npu::rms_norm_dynamic_quant));
+  m.impl("npu_inplace_partial_rotary_mul",
+         TORCH_FN(xllm::kernel::npu::npu_inplace_partial_rotary_mul));
   m.impl("silu_and_mul", TORCH_FN(xllm::silu_and_mul_npu));
   m.impl("reshape_paged_cache", TORCH_FN(xllm::reshape_paged_cache_npu));
   m.impl("apply_rotary_embedding", TORCH_FN(xllm::apply_rotary_embedding_npu));
@@ -371,6 +457,17 @@ TORCH_LIBRARY_IMPL(xllm_ops, PrivateUse1, m) {
          TORCH_FN(xllm::kernel::npu::sparse_flash_attention));
   m.impl("sparse_flash_attention_out",
          TORCH_FN(xllm::kernel::npu::sparse_flash_attention_out));
+  m.impl("moe_gating_top_k_hash",
+         TORCH_FN(xllm::kernel::npu::moe_gating_top_k_hash));
+  m.impl("dequant_swiglu_quant",
+         TORCH_FN(xllm::kernel::npu::dequant_swiglu_quant));
+  m.impl("hc_pre", TORCH_FN(xllm::kernel::npu::hc_pre));
+  m.impl("hc_post", TORCH_FN(xllm::kernel::npu::hc_post));
+  m.impl("compressor", TORCH_FN(xllm::kernel::npu::compressor));
+  m.impl("sparse_attn_sharedkv",
+         TORCH_FN(xllm::kernel::npu::sparse_attn_sharedkv));
+  m.impl("quant_lightning_indexer",
+         TORCH_FN(xllm::kernel::npu::quant_lightning_indexer));
 }
 
 // build_cp_context is pure host index math with no Tensor input, so the
@@ -379,4 +476,11 @@ TORCH_LIBRARY_IMPL(xllm_ops, PrivateUse1, m) {
 // graph capture), so it needs no fake/meta registration.
 TORCH_LIBRARY_IMPL(xllm_ops, CompositeExplicitAutograd, m) {
   m.impl("build_cp_context", TORCH_FN(xllm::build_cp_context_npu));
+  // These metadata factories allow every Tensor argument to be omitted, so
+  // there may be no device key to dispatch on. Their implementations select
+  // the output NPU device explicitly (or inherit it from an optional Tensor).
+  m.impl("sparse_attn_sharedkv_metadata",
+         TORCH_FN(xllm::kernel::npu::sparse_attn_sharedkv_metadata));
+  m.impl("quant_lightning_indexer_metadata",
+         TORCH_FN(xllm::kernel::npu::quant_lightning_indexer_metadata));
 }
diff --git a/xllm/core/runtime/py_executor_impl.cpp b/xllm/core/runtime/py_executor_impl.cpp
index 7dac2df748..ab50d5b00f 100644
--- a/xllm/core/runtime/py_executor_impl.cpp
+++ b/xllm/core/runtime/py_executor_impl.cpp
@@ -17,7 +17,7 @@ limitations under the License.
 
 #include <Python.h>
 #include <glog/logging.h>
-#include <pybind11/embed.h>
+#include <pybind11/pybind11.h>
 #include <torch/extension.h>
 
 #include <memory>
@@ -57,9 +57,7 @@ void clear_python_object(py::object& object) {
   object = py::object();
 }
 
-}  // namespace
-
-PYBIND11_EMBEDDED_MODULE(xllm_runtime, m) {
+void register_xllm_runtime_module(py::module_& m) {
   register_attention_metadata_views(m);
 
 #if defined(USE_NPU)
@@ -74,6 +72,23 @@ PYBIND11_EMBEDDED_MODULE(xllm_runtime, m) {
 #endif
 }
 
+void ensure_xllm_runtime_module() {
+  py::module_ sys = py::module_::import("sys");
+  py::dict modules = py::reinterpret_borrow<py::dict>(sys.attr("modules"));
+  const py::str module_name("xllm_runtime");
+  if (modules.contains(module_name)) {
+    return;
+  }
+
+  py::object module_object =
+      py::module_::import("types").attr("ModuleType")(module_name);
+  py::module_ module = py::reinterpret_borrow<py::module_>(module_object);
+  register_xllm_runtime_module(module);
+  modules[module_name] = module;
+}
+
+}  // namespace
+
 PyExecutorImpl::PyExecutorImpl(CausalLM* model,
                                const ModelArgs& args,
                                const torch::Device& device,
@@ -86,7 +101,7 @@ PyExecutorImpl::PyExecutorImpl(CausalLM* model,
   CHECK(py_causal_lm_ != nullptr) << "PyExecutorImpl requires PyCausalLM";
 
   py::gil_scoped_acquire gil;
-  py::module_::import("xllm_runtime");
+  ensure_xllm_runtime_module();
   py::module_ executor_module =
       py::module_::import("xllm.python.model_executor.executor");
   py_executor_ =
diff --git a/xllm/models/llm/py_causal_lm.cpp b/xllm/models/llm/py_causal_lm.cpp
index 370774f9ac..77680fd8eb 100644
--- a/xllm/models/llm/py_causal_lm.cpp
+++ b/xllm/models/llm/py_causal_lm.cpp
@@ -216,7 +216,7 @@ py::dict PyCausalLM::build_config_dict(
 void PyCausalLM::load_model(std::unique_ptr<ModelLoader> loader) {
   py::gil_scoped_acquire gil;
   auto& state_dicts = loader->get_state_dicts();
-  py::module_::import("xllm_weight_loader");
+  ensure_xllm_weight_loader_module();
 
   py::list py_state_dicts;
   for (const auto& sd : state_dicts) {
diff --git a/xllm/models/py_model_helper.cpp b/xllm/models/py_model_helper.cpp
index 2a375124a8..53ec1e056e 100644
--- a/xllm/models/py_model_helper.cpp
+++ b/xllm/models/py_model_helper.cpp
@@ -15,7 +15,7 @@ limitations under the License.
 
 // Infrastructure for the embedded Python model executor:
 // - Interpreter lifecycle (ensure_python_interpreter)
-// - Weight loading (PyStateDict + PYBIND11_EMBEDDED_MODULE)
+// - Weight loading (PyStateDict Python binding)
 // - Config serialization (dtype_to_string, PyDictVisitor)
 
 #include "models/py_model_helper.h"
@@ -147,11 +147,22 @@ py::list PyStateDict::keys() const {
   return result;
 }
 
-PYBIND11_EMBEDDED_MODULE(xllm_weight_loader, m) {
-  py::class_<PyStateDict>(m, "StateDict")
+void ensure_xllm_weight_loader_module() {
+  py::module_ sys = py::module_::import("sys");
+  py::dict modules = py::reinterpret_borrow<py::dict>(sys.attr("modules"));
+  const py::str module_name("xllm_weight_loader");
+  if (modules.contains(module_name)) {
+    return;
+  }
+
+  py::object module_object =
+      py::module_::import("types").attr("ModuleType")(module_name);
+  py::module_ module = py::reinterpret_borrow<py::module_>(module_object);
+  py::class_<PyStateDict>(module, "StateDict")
       .def("get_tensor", &PyStateDict::get_tensor, py::arg("name"))
       .def("has", &PyStateDict::has, py::arg("name"))
       .def("keys", &PyStateDict::keys);
+  modules[module_name] = module;
 }
 
 }  // namespace xllm
diff --git a/xllm/models/py_model_helper.h b/xllm/models/py_model_helper.h
index 4b90fa86a5..f9331bf208 100644
--- a/xllm/models/py_model_helper.h
+++ b/xllm/models/py_model_helper.h
@@ -30,6 +30,10 @@ namespace xllm {
 // Initializes the embedded CPython interpreter (idempotent, process-wide).
 void ensure_python_interpreter();
 
+// Makes the internal StateDict binding importable as xllm_weight_loader.
+// The caller must hold the Python GIL.
+void ensure_xllm_weight_loader_module();
+
 // Convert torch dtype to the string form used by Python model config.
 std::string dtype_to_string(const torch::TensorOptions& options);
 
diff --git a/xllm/python/kernels_npu/__init__.py b/xllm/python/kernels_npu/__init__.py
index 42083cc209..cf25369a68 100644
--- a/xllm/python/kernels_npu/__init__.py
+++ b/xllm/python/kernels_npu/__init__.py
@@ -42,6 +42,15 @@
     causal_conv1d_decode,
     causal_conv1d_prefill,
 )
+from .dsa import (
+    compressor,
+    hc_post,
+    hc_pre,
+    quant_lightning_indexer,
+    quant_lightning_indexer_metadata,
+    sparse_attn_sharedkv,
+    sparse_attn_sharedkv_metadata,
+)
 from .gated_delta_net import (
     chunk_gated_delta_rule,
     fused_gdn_prefill_post_conv,
@@ -51,9 +60,12 @@
 from .linear import prepare_row_parallel_weight
 from .moe import (
     cutlass_fused_moe,
+    dequant_swiglu_quant,
     fused_moe,
     grouped_moe,
+    grouped_moe_with_selected_experts,
     moe_fused_topk,
+    moe_gating_top_k_hash,
     prepare_grouped_moe_weights,
     supports_cutlass_moe,
 )
@@ -61,6 +73,7 @@
     fused_add_rms_norm,
     l2_norm,
     rms_norm,
+    rms_norm_dynamic_quant,
     rms_norm_gated,
 )
 from .quantization import (
@@ -72,6 +85,7 @@
     fused_qk_norm_rope,
     interleaved_rotary_embedding,
     mrope,
+    npu_inplace_partial_rotary_mul,
     vision_rotary_mul,
 )
 from .sparse_attention import (
@@ -85,6 +99,7 @@
 __all__ = [
     "rms_norm",
     "fused_add_rms_norm",
+    "rms_norm_dynamic_quant",
     "l2_norm",
     "rms_norm_gated",
     "silu_and_mul",
@@ -93,12 +108,14 @@
     "vision_fusion_attention",
     "fused_qk_norm_rope",
     "interleaved_rotary_embedding",
+    "npu_inplace_partial_rotary_mul",
     "mrope",
     "vision_rotary_mul",
     "moe_fused_topk",
     "cutlass_fused_moe",
     "fused_moe",
     "grouped_moe",
+    "grouped_moe_with_selected_experts",
     "prepare_grouped_moe_weights",
     "supports_cutlass_moe",
     "prepare_row_parallel_weight",
@@ -112,6 +129,15 @@
     "sparse_flash_attention_out",
     "causal_conv1d_prefill",
     "causal_conv1d_decode",
+    "compressor",
+    "dequant_swiglu_quant",
+    "hc_pre",
+    "hc_post",
+    "moe_gating_top_k_hash",
+    "quant_lightning_indexer",
+    "quant_lightning_indexer_metadata",
+    "sparse_attn_sharedkv",
+    "sparse_attn_sharedkv_metadata",
     "resolve_gdn_prefill_backend",
     "fused_gdn_prefill_post_conv",
     "fused_recurrent_gated_delta_rule_packed_decode",
diff --git a/xllm/python/kernels_npu/_custom_op.py b/xllm/python/kernels_npu/_custom_op.py
index 6b9d91183b..6167f4c091 100644
--- a/xllm/python/kernels_npu/_custom_op.py
+++ b/xllm/python/kernels_npu/_custom_op.py
@@ -326,6 +326,374 @@ def _sparse_flash_attention_out_fake(
     return output
 
 
+# ---------------------------------------------------------------------------
+# DeepSeek-V4 DSA kernel fakes
+# ---------------------------------------------------------------------------
+
+# Matches kDsaMetadataBufferElements in xllm_ops_api.h.
+_DSA_METADATA_BUFFER_ELEMENTS = 1024
+
+
+def _rms_norm_dynamic_quant_fake(
+    input: torch.Tensor, weight: torch.Tensor, eps: float
+) -> tuple[torch.Tensor, torch.Tensor]:
+    del weight, eps
+    return input.new_empty(input.shape, dtype=torch.int8), input.new_empty(input.shape[:-1], dtype=torch.float32)
+
+
+def _npu_inplace_partial_rotary_mul_fake(
+    x: torch.Tensor,
+    r1: torch.Tensor,
+    r2: torch.Tensor,
+    rotary_mode: str,
+    partial_slice: list[int],
+) -> None:
+    del x, r1, r2, rotary_mode, partial_slice
+
+
+def _hc_pre_fake(
+    x: torch.Tensor,
+    hc_fn: torch.Tensor,
+    hc_scale: torch.Tensor,
+    hc_base: torch.Tensor,
+    hc_mult: int,
+    hc_sinkhorn_iters: int,
+    norm_eps: float,
+    hc_eps: float,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+    del hc_fn, hc_scale, hc_base, hc_sinkhorn_iters, norm_eps, hc_eps
+    if x.dim() == 4:
+        y_shape = (x.size(0), x.size(1), x.size(3))
+        post_shape = (x.size(0), x.size(1), hc_mult)
+        comb_shape = (x.size(0), x.size(1), hc_mult, hc_mult)
+    else:
+        y_shape = (x.size(0), x.size(2))
+        post_shape = (x.size(0), hc_mult)
+        comb_shape = (x.size(0), hc_mult, hc_mult)
+    attn_input = x.new_empty(y_shape, dtype=x.dtype)
+    post = x.new_empty(post_shape, dtype=torch.float32)
+    comb = x.new_empty(comb_shape, dtype=torch.float32)
+    return attn_input, post, comb
+
+
+def _hc_post_fake(
+    x: torch.Tensor,
+    residual: torch.Tensor,
+    post: torch.Tensor,
+    comb: torch.Tensor,
+) -> torch.Tensor:
+    del post, comb
+    # hc_post returns [T, hc_mult, hidden] (the merged residual streams).
+    return residual.new_empty(residual.shape, dtype=residual.dtype)
+
+
+def _compressor_fake(
+    x: torch.Tensor,
+    wkv: torch.Tensor,
+    wgate: torch.Tensor,
+    kv_state: torch.Tensor,
+    score_state: torch.Tensor,
+    ape: torch.Tensor,
+    norm_weight: torch.Tensor,
+    rope_sin: torch.Tensor,
+    rope_cos: torch.Tensor,
+    kv_block_table: torch.Tensor | None,
+    score_block_table: torch.Tensor | None,
+    cu_seqlens: torch.Tensor | None,
+    seqused: torch.Tensor | None,
+    start_pos: torch.Tensor | None,
+    rope_head_dim: int,
+    cmp_ratio: int,
+    coff: int,
+    norm_eps: float,
+    rotary_mode: int,
+    enable_grad: bool,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
+    del (
+        wkv,
+        wgate,
+        kv_state,
+        score_state,
+        ape,
+        rope_cos,
+        kv_block_table,
+        score_block_table,
+        cu_seqlens,
+        seqused,
+        start_pos,
+        rope_head_dim,
+        norm_eps,
+        rotary_mode,
+    )
+    head_dim = norm_weight.size(0)
+    if x.dim() == 3:
+        compressed_seq = (x.size(1) + cmp_ratio - 1) // cmp_ratio
+        cmp_kv_shape = (x.size(0), compressed_seq, head_dim)
+        grad_shapes = (
+            (x.size(0), x.size(1), coff * head_dim),
+            (x.size(0), compressed_seq, coff * cmp_ratio, head_dim),
+            (x.size(0), compressed_seq, head_dim),
+            (x.size(0), compressed_seq),
+        )
+    else:
+        compressed_seq = rope_sin.size(0)
+        cmp_kv_shape = (compressed_seq, head_dim)
+        grad_shapes = (
+            (x.size(0), coff * head_dim),
+            (compressed_seq, coff * cmp_ratio, head_dim),
+            (compressed_seq, head_dim),
+            (compressed_seq,),
+        )
+    outputs = [x.new_empty(cmp_kv_shape, dtype=x.dtype)]
+    if enable_grad:
+        outputs.extend(x.new_empty(shape, dtype=x.dtype) for shape in grad_shapes)
+    else:
+        outputs.extend(x.new_empty((0,), dtype=x.dtype) for _ in grad_shapes)
+    return tuple(outputs)  # type: ignore[return-value]
+
+
+def _sparse_attn_sharedkv_fake(
+    q: torch.Tensor,
+    ori_kv: torch.Tensor | None,
+    cmp_kv: torch.Tensor | None,
+    ori_sparse_indices: torch.Tensor | None,
+    cmp_sparse_indices: torch.Tensor | None,
+    ori_block_table: torch.Tensor | None,
+    cmp_block_table: torch.Tensor | None,
+    cu_seqlens_q: torch.Tensor | None,
+    cu_seqlens_ori_kv: torch.Tensor | None,
+    cu_seqlens_cmp_kv: torch.Tensor | None,
+    seqused_q: torch.Tensor | None,
+    seqused_kv: torch.Tensor | None,
+    sinks: torch.Tensor | None,
+    metadata: torch.Tensor | None,
+    softmax_scale: float,
+    cmp_ratio: int,
+    ori_mask_mode: int,
+    cmp_mask_mode: int,
+    ori_win_left: int,
+    ori_win_right: int,
+    layout_q: str,
+    layout_kv: str,
+    return_softmax_lse: bool,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    del (
+        ori_kv,
+        cmp_kv,
+        ori_sparse_indices,
+        cmp_sparse_indices,
+        ori_block_table,
+        cmp_block_table,
+        sinks,
+        metadata,
+        softmax_scale,
+        cmp_ratio,
+        ori_mask_mode,
+        cmp_mask_mode,
+        ori_win_left,
+        ori_win_right,
+        layout_q,
+        layout_kv,
+    )
+    out = q.new_empty(q.shape, dtype=q.dtype)
+    lse_shape = (*q.shape[:-1], 1) if return_softmax_lse else (0,)
+    lse = q.new_empty(lse_shape, dtype=torch.float32)
+    return out, lse
+
+
+def _sparse_attn_sharedkv_metadata_fake(
+    num_heads_q: int,
+    num_heads_kv: int,
+    head_dim: int,
+    cu_seqlens_q: torch.Tensor | None,
+    cu_seqlens_ori_kv: torch.Tensor | None,
+    cu_seqlens_cmp_kv: torch.Tensor | None,
+    seqused_q: torch.Tensor | None,
+    seqused_kv: torch.Tensor | None,
+    batch_size: int,
+    max_seqlen_q: int,
+    max_seqlen_kv: int,
+    ori_topk: int,
+    cmp_topk: int,
+    cmp_ratio: int,
+    ori_mask_mode: int,
+    cmp_mask_mode: int,
+    ori_win_left: int,
+    ori_win_right: int,
+    layout_q: str,
+    layout_kv: str,
+    has_ori_kv: bool,
+    has_cmp_kv: bool,
+) -> torch.Tensor:
+    del (
+        num_heads_q,
+        num_heads_kv,
+        head_dim,
+        batch_size,
+        max_seqlen_q,
+        max_seqlen_kv,
+        ori_topk,
+        cmp_topk,
+        cmp_ratio,
+        ori_mask_mode,
+        cmp_mask_mode,
+        ori_win_left,
+        ori_win_right,
+        layout_q,
+        layout_kv,
+        has_ori_kv,
+        has_cmp_kv,
+    )
+    for tensor in (
+        cu_seqlens_q,
+        cu_seqlens_ori_kv,
+        cu_seqlens_cmp_kv,
+        seqused_q,
+        seqused_kv,
+    ):
+        if tensor is not None:
+            return tensor.new_empty((_DSA_METADATA_BUFFER_ELEMENTS,), dtype=torch.int32)
+    return torch.empty((_DSA_METADATA_BUFFER_ELEMENTS,), dtype=torch.int32, device="npu")
+
+
+def _quant_lightning_indexer_fake(
+    query: torch.Tensor,
+    key: torch.Tensor,
+    weights: torch.Tensor,
+    query_dequant_scale: torch.Tensor,
+    key_dequant_scale: torch.Tensor,
+    query_quant_mode: int,
+    key_quant_mode: int,
+    actual_seq_lengths_query: torch.Tensor | None,
+    actual_seq_lengths_key: torch.Tensor | None,
+    block_table: torch.Tensor | None,
+    metadata: torch.Tensor | None,
+    layout_query: str,
+    layout_key: str,
+    sparse_count: int,
+    sparse_mode: int,
+    pre_tokens: int,
+    next_tokens: int,
+    cmp_ratio: int,
+    return_value: bool,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    del (
+        weights,
+        query_dequant_scale,
+        key_dequant_scale,
+        query_quant_mode,
+        key_quant_mode,
+        block_table,
+        metadata,
+        sparse_mode,
+        pre_tokens,
+        next_tokens,
+        cmp_ratio,
+    )
+    key_head_num = key.size(1) if layout_key == "TND" else key.size(2)
+    if layout_query == "BSND":
+        out_shape = (query.size(0), query.size(1), key_head_num, sparse_count)
+    else:
+        out_shape = (query.size(0), key_head_num, sparse_count)
+    out = query.new_zeros(out_shape, dtype=torch.int32)
+    val = (
+        query.new_empty(out_shape, dtype=torch.float32) if return_value else query.new_empty((0,), dtype=torch.float32)
+    )
+    return out, val
+
+
+def _quant_lightning_indexer_metadata_fake(
+    num_heads_q: int,
+    num_heads_k: int,
+    head_dim: int,
+    query_quant_mode: int,
+    key_quant_mode: int,
+    actual_seq_lengths_query: torch.Tensor | None,
+    actual_seq_lengths_key: torch.Tensor | None,
+    batch_size: int,
+    max_seqlen_q: int,
+    max_seqlen_k: int,
+    layout_query: str,
+    layout_key: str,
+    sparse_count: int,
+    sparse_mode: int,
+    pre_tokens: int,
+    next_tokens: int,
+    cmp_ratio: int,
+    device: str,
+) -> torch.Tensor:
+    del (
+        num_heads_q,
+        num_heads_k,
+        head_dim,
+        query_quant_mode,
+        key_quant_mode,
+        batch_size,
+        max_seqlen_q,
+        max_seqlen_k,
+        layout_query,
+        layout_key,
+        sparse_count,
+        sparse_mode,
+        pre_tokens,
+        next_tokens,
+        cmp_ratio,
+    )
+    for tensor in (actual_seq_lengths_query, actual_seq_lengths_key):
+        if tensor is not None:
+            return tensor.new_empty((_DSA_METADATA_BUFFER_ELEMENTS,), dtype=torch.int32)
+    return torch.empty((_DSA_METADATA_BUFFER_ELEMENTS,), dtype=torch.int32, device=device)
+
+
+def _moe_gating_top_k_hash_fake(
+    x: torch.Tensor,
+    k: int,
+    bias: torch.Tensor | None,
+    input_ids: torch.Tensor | None,
+    tid2eid: torch.Tensor | None,
+    k_group: int,
+    group_count: int,
+    routed_scaling_factor: float,
+    eps: float,
+    group_select_mode: int,
+    renorm: int,
+    norm_type: int,
+    out_flag: bool,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+    del bias, input_ids, tid2eid, k_group, group_count, routed_scaling_factor
+    del eps, group_select_mode, renorm, norm_type, out_flag
+    y_shape = (*x.shape[:-1], k)
+    y = x.new_empty(y_shape, dtype=x.dtype)
+    expert_idx = x.new_empty(y_shape, dtype=torch.int32)
+    out = x.new_empty(x.shape, dtype=torch.float32)
+    return y, expert_idx, out
+
+
+def _dequant_swiglu_quant_fake(
+    x: torch.Tensor,
+    weight_scale: torch.Tensor | None,
+    activation_scale: torch.Tensor | None,
+    bias: torch.Tensor | None,
+    quant_scale: torch.Tensor | None,
+    quant_offset: torch.Tensor | None,
+    group_index: torch.Tensor | None,
+    activate_left: bool,
+    quant_mode: int,
+    swiglu_mode: int,
+    clamp_limit: float,
+    glu_alpha: float,
+    glu_bias: float,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    del weight_scale, activation_scale, bias, quant_scale, quant_offset
+    del group_index, activate_left, quant_mode, swiglu_mode
+    del clamp_limit, glu_alpha, glu_bias
+    # Output is half of input's last dim (SwiGLU splits gate/up).
+    out_dim = x.size(-1) // 2
+    act_quantized = x.new_empty((*x.shape[:-1], out_dim), dtype=torch.int8)
+    act_scale = x.new_empty(x.shape[:-1], dtype=torch.float32)
+    return act_quantized, act_scale
+
+
 register_fake("xllm_ops::rms_norm", _rms_norm_fake)
 register_fake("xllm_ops::fused_add_rms_norm", _fused_add_rms_norm_fake)
 register_fake("xllm_ops::silu_and_mul", _silu_and_mul_fake)
@@ -339,3 +707,20 @@ def _sparse_flash_attention_out_fake(
 register_fake("xllm_ops::scatter_nd_update", _scatter_nd_update_fake)
 register_fake("xllm_ops::sparse_flash_attention", _sparse_flash_attention_fake)
 register_fake("xllm_ops::sparse_flash_attention_out", _sparse_flash_attention_out_fake)
+register_fake("xllm_ops::rms_norm_dynamic_quant", _rms_norm_dynamic_quant_fake)
+register_fake(
+    "xllm_ops::npu_inplace_partial_rotary_mul",
+    _npu_inplace_partial_rotary_mul_fake,
+)
+register_fake("xllm_ops::compressor", _compressor_fake)
+register_fake("xllm_ops::moe_gating_top_k_hash", _moe_gating_top_k_hash_fake)
+register_fake("xllm_ops::dequant_swiglu_quant", _dequant_swiglu_quant_fake)
+register_fake("xllm_ops::hc_pre", _hc_pre_fake)
+register_fake("xllm_ops::hc_post", _hc_post_fake)
+register_fake("xllm_ops::sparse_attn_sharedkv", _sparse_attn_sharedkv_fake)
+register_fake("xllm_ops::sparse_attn_sharedkv_metadata", _sparse_attn_sharedkv_metadata_fake)
+register_fake("xllm_ops::quant_lightning_indexer", _quant_lightning_indexer_fake)
+register_fake(
+    "xllm_ops::quant_lightning_indexer_metadata",
+    _quant_lightning_indexer_metadata_fake,
+)
diff --git a/xllm/python/kernels_npu/dsa.py b/xllm/python/kernels_npu/dsa.py
new file mode 100644
index 0000000000..4e977636b5
--- /dev/null
+++ b/xllm/python/kernels_npu/dsa.py
@@ -0,0 +1,306 @@
+# Copyright 2026 The xLLM Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://github.com/xLLM-AI/xllm/blob/main/LICENSE
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""NPU DeepSeek-V4 DSA kernels.
+
+These wrap the AscendC operators registered as ``torch.ops.xllm_ops.*`` by
+``core/kernels/npu/npu_ops_library.cpp``. They drive the two-stage sparse
+attention (original + compressed KV), the KV compressor, the quantized
+lightning indexer, and the HyperConnection pre/post used by DeepSeek-V4's DSA
+attention path.
+"""
+
+from __future__ import annotations
+
+import torch
+
+
+def hc_pre(
+    x: torch.Tensor,
+    hc_fn: torch.Tensor,
+    hc_scale: torch.Tensor,
+    hc_base: torch.Tensor,
+    hc_mult: int,
+    hc_sinkhorn_iters: int,
+    norm_eps: float,
+    hc_eps: float,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+    """HyperConnection pre: mix hc_mult streams into one sub-block input.
+
+    Returns ``(attn_input, post, comb)`` where post/comb feed ``hc_post``.
+    """
+    return torch.ops.xllm_ops.hc_pre(x, hc_fn, hc_scale, hc_base, hc_mult, hc_sinkhorn_iters, norm_eps, hc_eps)
+
+
+def hc_post(
+    x: torch.Tensor,
+    residual: torch.Tensor,
+    post: torch.Tensor,
+    comb: torch.Tensor,
+) -> torch.Tensor:
+    """HyperConnection post: combine sub-block output with the residual streams."""
+    return torch.ops.xllm_ops.hc_post(x, residual, post, comb)
+
+
+def compressor(
+    x: torch.Tensor,
+    wkv: torch.Tensor,
+    wgate: torch.Tensor,
+    kv_state: torch.Tensor,
+    score_state: torch.Tensor,
+    ape: torch.Tensor,
+    norm_weight: torch.Tensor,
+    rope_sin: torch.Tensor,
+    rope_cos: torch.Tensor,
+    kv_block_table: torch.Tensor | None,
+    score_block_table: torch.Tensor | None,
+    cu_seqlens: torch.Tensor | None,
+    seqused: torch.Tensor | None,
+    start_pos: torch.Tensor | None,
+    rope_head_dim: int,
+    cmp_ratio: int,
+    coff: int,
+    norm_eps: float,
+    rotary_mode: int,
+    enable_grad: bool,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
+    """Pool KV along the token axis by ``cmp_ratio`` (NSA-style compressor).
+
+    ``kv_state`` and ``score_state`` are updated in place.
+
+    Returns ``(cmp_kv, wkv_proj, softmax_res, norm_x, norm_rstd)``; only
+    ``cmp_kv`` is consumed by the DSA path.
+    """
+    # C++ moves DSA metadata to the active device before dispatch. Keep this
+    # adapter deterministic; experimental clone/noalias paths do not belong in
+    # the public binding.
+    kv_block_table = kv_block_table.to(x.device) if kv_block_table is not None else None
+    score_block_table = score_block_table.to(x.device) if score_block_table is not None else None
+    cu_seqlens = cu_seqlens.to(x.device) if cu_seqlens is not None else None
+    seqused = seqused.to(x.device) if seqused is not None else None
+    start_pos = start_pos.to(x.device) if start_pos is not None else None
+    return torch.ops.xllm_ops.compressor(
+        x,
+        wkv,
+        wgate,
+        kv_state,
+        score_state,
+        ape,
+        norm_weight,
+        rope_sin,
+        rope_cos,
+        kv_block_table,
+        score_block_table,
+        cu_seqlens,
+        seqused,
+        start_pos,
+        rope_head_dim,
+        cmp_ratio,
+        coff,
+        norm_eps,
+        rotary_mode,
+        enable_grad,
+    )
+
+
+def sparse_attn_sharedkv(
+    q: torch.Tensor,
+    ori_kv: torch.Tensor | None,
+    cmp_kv: torch.Tensor | None,
+    ori_sparse_indices: torch.Tensor | None,
+    cmp_sparse_indices: torch.Tensor | None,
+    ori_block_table: torch.Tensor | None,
+    cmp_block_table: torch.Tensor | None,
+    cu_seqlens_q: torch.Tensor | None,
+    cu_seqlens_ori_kv: torch.Tensor | None,
+    cu_seqlens_cmp_kv: torch.Tensor | None,
+    seqused_q: torch.Tensor | None,
+    seqused_kv: torch.Tensor | None,
+    sinks: torch.Tensor | None,
+    metadata: torch.Tensor | None,
+    softmax_scale: float,
+    cmp_ratio: int,
+    ori_mask_mode: int,
+    cmp_mask_mode: int,
+    ori_win_left: int,
+    ori_win_right: int,
+    layout_q: str,
+    layout_kv: str,
+    return_softmax_lse: bool,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    """Two-stage sparse attention over original and compressed KV."""
+    return torch.ops.xllm_ops.sparse_attn_sharedkv(
+        q,
+        ori_kv,
+        cmp_kv,
+        ori_sparse_indices,
+        cmp_sparse_indices,
+        ori_block_table,
+        cmp_block_table,
+        cu_seqlens_q,
+        cu_seqlens_ori_kv,
+        cu_seqlens_cmp_kv,
+        seqused_q,
+        seqused_kv,
+        sinks,
+        metadata,
+        softmax_scale,
+        cmp_ratio,
+        ori_mask_mode,
+        cmp_mask_mode,
+        ori_win_left,
+        ori_win_right,
+        layout_q,
+        layout_kv,
+        return_softmax_lse,
+    )
+
+
+def sparse_attn_sharedkv_metadata(
+    num_heads_q: int,
+    num_heads_kv: int,
+    head_dim: int,
+    cu_seqlens_q: torch.Tensor | None,
+    cu_seqlens_ori_kv: torch.Tensor | None,
+    cu_seqlens_cmp_kv: torch.Tensor | None,
+    seqused_q: torch.Tensor | None,
+    seqused_kv: torch.Tensor | None,
+    batch_size: int,
+    max_seqlen_q: int,
+    max_seqlen_kv: int,
+    ori_topk: int,
+    cmp_topk: int,
+    cmp_ratio: int,
+    ori_mask_mode: int,
+    cmp_mask_mode: int,
+    ori_win_left: int,
+    ori_win_right: int,
+    layout_q: str,
+    layout_kv: str,
+    has_ori_kv: bool,
+    has_cmp_kv: bool,
+) -> torch.Tensor:
+    """Build the AICPU tiling metadata for :func:`sparse_attn_sharedkv`."""
+    return torch.ops.xllm_ops.sparse_attn_sharedkv_metadata(
+        num_heads_q,
+        num_heads_kv,
+        head_dim,
+        cu_seqlens_q,
+        cu_seqlens_ori_kv,
+        cu_seqlens_cmp_kv,
+        seqused_q,
+        seqused_kv,
+        batch_size,
+        max_seqlen_q,
+        max_seqlen_kv,
+        ori_topk,
+        cmp_topk,
+        cmp_ratio,
+        ori_mask_mode,
+        cmp_mask_mode,
+        ori_win_left,
+        ori_win_right,
+        layout_q,
+        layout_kv,
+        has_ori_kv,
+        has_cmp_kv,
+    )
+
+
+def quant_lightning_indexer(
+    query: torch.Tensor,
+    key: torch.Tensor,
+    weights: torch.Tensor,
+    query_dequant_scale: torch.Tensor,
+    key_dequant_scale: torch.Tensor,
+    query_quant_mode: int,
+    key_quant_mode: int,
+    actual_seq_lengths_query: torch.Tensor | None,
+    actual_seq_lengths_key: torch.Tensor | None,
+    block_table: torch.Tensor | None,
+    metadata: torch.Tensor | None,
+    layout_query: str,
+    layout_key: str,
+    sparse_count: int,
+    sparse_mode: int,
+    pre_tokens: int,
+    next_tokens: int,
+    cmp_ratio: int,
+    return_value: bool,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    """Select the compressed key blocks each query attends to (int8 q/k)."""
+    return torch.ops.xllm_ops.quant_lightning_indexer(
+        query,
+        key,
+        weights,
+        query_dequant_scale,
+        key_dequant_scale,
+        query_quant_mode,
+        key_quant_mode,
+        actual_seq_lengths_query,
+        actual_seq_lengths_key,
+        block_table,
+        metadata,
+        layout_query,
+        layout_key,
+        sparse_count,
+        sparse_mode,
+        pre_tokens,
+        next_tokens,
+        cmp_ratio,
+        return_value,
+    )
+
+
+def quant_lightning_indexer_metadata(
+    num_heads_q: int,
+    num_heads_k: int,
+    head_dim: int,
+    query_quant_mode: int,
+    key_quant_mode: int,
+    actual_seq_lengths_query: torch.Tensor | None,
+    actual_seq_lengths_key: torch.Tensor | None,
+    batch_size: int,
+    max_seqlen_q: int,
+    max_seqlen_k: int,
+    layout_query: str,
+    layout_key: str,
+    sparse_count: int,
+    sparse_mode: int,
+    pre_tokens: int,
+    next_tokens: int,
+    cmp_ratio: int,
+    device: str,
+) -> torch.Tensor:
+    """Build the AICPU tiling metadata for :func:`quant_lightning_indexer`."""
+    return torch.ops.xllm_ops.quant_lightning_indexer_metadata(
+        num_heads_q,
+        num_heads_k,
+        head_dim,
+        query_quant_mode,
+        key_quant_mode,
+        actual_seq_lengths_query,
+        actual_seq_lengths_key,
+        batch_size,
+        max_seqlen_q,
+        max_seqlen_k,
+        layout_query,
+        layout_key,
+        sparse_count,
+        sparse_mode,
+        pre_tokens,
+        next_tokens,
+        cmp_ratio,
+        device,
+    )
diff --git a/xllm/python/kernels_npu/moe.py b/xllm/python/kernels_npu/moe.py
index 3786206702..b949b64eff 100644
--- a/xllm/python/kernels_npu/moe.py
+++ b/xllm/python/kernels_npu/moe.py
@@ -22,6 +22,72 @@
 _FRACTAL_NZ_FORMAT = 29
 
 
+def dequant_swiglu_quant(
+    x: torch.Tensor,
+    weight_scale: torch.Tensor | None,
+    activation_scale: torch.Tensor | None,
+    bias: torch.Tensor | None = None,
+    quant_scale: torch.Tensor | None = None,
+    quant_offset: torch.Tensor | None = None,
+    group_index: torch.Tensor | None = None,
+    activate_left: bool = True,
+    quant_mode: int = 1,
+    swiglu_mode: int = 1,
+    clamp_limit: float = 0.0,
+    glu_alpha: float = 1.0,
+    glu_bias: float = 0.0,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    """Apply the fused dequantization, SwiGLU, and dynamic quantization."""
+    return torch.ops.xllm_ops.dequant_swiglu_quant(
+        x,
+        weight_scale,
+        activation_scale,
+        bias,
+        quant_scale,
+        quant_offset,
+        group_index,
+        activate_left,
+        quant_mode,
+        swiglu_mode,
+        clamp_limit,
+        glu_alpha,
+        glu_bias,
+    )
+
+
+def moe_gating_top_k_hash(
+    x: torch.Tensor,
+    k: int,
+    bias: torch.Tensor | None,
+    input_ids: torch.Tensor | None,
+    tid2eid: torch.Tensor | None,
+    k_group: int,
+    group_count: int,
+    routed_scaling_factor: float,
+    eps: float,
+    group_select_mode: int,
+    renorm: int,
+    norm_type: int,
+    out_flag: bool,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+    """Select experts with the DeepSeek-V4 hash-routing gate."""
+    return torch.ops.xllm_ops.moe_gating_top_k_hash(
+        x,
+        k,
+        bias,
+        input_ids,
+        tid2eid,
+        k_group,
+        group_count,
+        routed_scaling_factor,
+        eps,
+        group_select_mode,
+        renorm,
+        norm_type,
+        out_flag,
+    )
+
+
 def supports_cutlass_moe(device: torch.device) -> bool:
     """Return whether ``device`` has the native expert GEMMs.
 
@@ -148,6 +214,159 @@ def grouped_moe(
     )
 
 
+def _group_gemm(
+    *,
+    x: torch.Tensor,
+    weight: torch.Tensor,
+    scale: torch.Tensor | None,
+    per_token_scale: torch.Tensor | None,
+    group_list: torch.Tensor,
+    split_item: int,
+    group_type: int,
+    group_list_type: int,
+    output_dtype: torch.dtype | None,
+) -> torch.Tensor:
+    outputs = torch.ops.npu.npu_grouped_matmul(
+        x=[x],
+        weight=[weight],
+        scale=None if scale is None else [scale],
+        per_token_scale=None if per_token_scale is None else [per_token_scale],
+        group_list=group_list,
+        split_item=split_item,
+        group_type=group_type,
+        group_list_type=group_list_type,
+        output_dtype=output_dtype,
+    )
+    return outputs[0]
+
+
+def _grouped_moe_with_selected_experts_impl(
+    hidden_states: torch.Tensor,
+    topk_weights: torch.Tensor,
+    topk_ids: torch.Tensor,
+    w13: torch.Tensor,
+    w2: torch.Tensor,
+    w13_scale: torch.Tensor,
+    w2_scale: torch.Tensor,
+    w13_offset: torch.Tensor | None = None,
+    w2_offset: torch.Tensor | None = None,
+    num_total_experts: int = -1,
+    start_expert_id: int = 0,
+    num_experts_per_rank: int = -1,
+    swiglu_limit: float = 0.0,
+) -> torch.Tensor:
+    """Run grouped quantized experts with pre-computed routing (no gate).
+
+    The routing and W8A8 grouped-matmul sequence mirrors the native NPU
+    ``FusedMoEImpl::select_experts`` and ``forward_expert`` paths.
+    """
+    num_tokens = hidden_states.shape[0]
+    expert_num = num_total_experts if num_total_experts > 0 else w13.shape[0]
+    local_expert_count = num_experts_per_rank if num_experts_per_rank > 0 else w13.shape[0]
+    active_range = [start_expert_id, start_expert_id + local_expert_count]
+    if start_expert_id < 0 or active_range[1] > expert_num:
+        raise ValueError(f"active expert range {active_range} is outside [0, {expert_num})")
+    if w13.shape[0] != local_expert_count or w2.shape[0] != local_expert_count:
+        raise ValueError("local expert count must match the first dimension of w13 and w2")
+    expanded_hidden, expanded_row_idx, expert_tokens, _ = torch_npu.npu_moe_init_routing_v2(
+        hidden_states,
+        topk_ids.to(torch.int32),
+        scale=None,
+        active_num=num_tokens * topk_ids.size(-1),
+        expert_num=expert_num,
+        expert_tokens_num_type=1,
+        expert_tokens_num_flag=True,
+        active_expert_range=active_range,
+        quant_mode=-1,
+    )
+    from xllm.python import kernels as _kernels
+
+    sorted_hidden_i8, pertoken_scale = _kernels.dynamic_quant(expanded_hidden)
+    if pertoken_scale is None:
+        raise RuntimeError("dynamic_quant did not return a per-token scale")
+    if expert_tokens.numel() < local_expert_count:
+        raise RuntimeError("npu_moe_init_routing_v2 returned fewer groups than local experts")
+    group_list = expert_tokens[:local_expert_count].to(torch.int64)
+    gemm1_out = _group_gemm(
+        x=sorted_hidden_i8,
+        weight=w13,
+        scale=None,
+        per_token_scale=None,
+        group_list=group_list,
+        split_item=2,
+        group_type=0,
+        group_list_type=1,
+        output_dtype=torch.int32,
+    )
+    act_i8, act_pt = _kernels.dequant_swiglu_quant(
+        x=gemm1_out,
+        weight_scale=w13_scale,
+        activation_scale=pertoken_scale,
+        bias=None,
+        quant_scale=None,
+        quant_offset=None,
+        group_index=group_list,
+        activate_left=True,
+        quant_mode=1,
+        swiglu_mode=1,
+        clamp_limit=swiglu_limit,
+        glu_alpha=1.0,
+        glu_bias=0.0,
+    )
+    del w13_offset, w2_offset
+    output = _group_gemm(
+        x=act_i8,
+        weight=w2,
+        scale=w2_scale.to(hidden_states.dtype),
+        per_token_scale=act_pt,
+        group_list=group_list,
+        split_item=2,
+        group_type=0,
+        group_list_type=1,
+        output_dtype=hidden_states.dtype,
+    )
+    local_mask = (topk_ids >= active_range[0]) & (topk_ids < active_range[1])
+    local_topk_weights = topk_weights * local_mask.to(topk_weights.dtype)
+    return torch_npu.npu_moe_token_unpermute(
+        permuted_tokens=output,
+        sorted_indices=expanded_row_idx.abs(),
+        probs=local_topk_weights.to(output.dtype),
+    )
+
+
+@torch.library.custom_op("xllm_python::grouped_moe_with_selected_experts", mutates_args=())
+def grouped_moe_with_selected_experts(
+    hidden_states: torch.Tensor,
+    topk_weights: torch.Tensor,
+    topk_ids: torch.Tensor,
+    w13: torch.Tensor,
+    w2: torch.Tensor,
+    w13_scale: torch.Tensor,
+    w2_scale: torch.Tensor,
+    w13_offset: torch.Tensor | None = None,
+    w2_offset: torch.Tensor | None = None,
+    num_total_experts: int = -1,
+    start_expert_id: int = 0,
+    num_experts_per_rank: int = -1,
+    swiglu_limit: float = 0.0,
+) -> torch.Tensor:
+    return _grouped_moe_with_selected_experts_impl(
+        hidden_states,
+        topk_weights,
+        topk_ids,
+        w13,
+        w2,
+        w13_scale,
+        w2_scale,
+        w13_offset,
+        w2_offset,
+        num_total_experts,
+        start_expert_id,
+        num_experts_per_rank,
+        swiglu_limit,
+    )
+
+
 @grouped_moe.register_fake
 def _grouped_moe_fake(
     hidden_states: torch.Tensor,
@@ -179,6 +398,27 @@ def _grouped_moe_fake(
     return torch.empty_like(hidden_states)
 
 
+@grouped_moe_with_selected_experts.register_fake
+def _grouped_moe_with_selected_experts_fake(
+    hidden_states: torch.Tensor,
+    topk_weights: torch.Tensor,
+    topk_ids: torch.Tensor,
+    w13: torch.Tensor,
+    w2: torch.Tensor,
+    w13_scale: torch.Tensor,
+    w2_scale: torch.Tensor,
+    w13_offset: torch.Tensor | None = None,
+    w2_offset: torch.Tensor | None = None,
+    num_total_experts: int = -1,
+    start_expert_id: int = 0,
+    num_experts_per_rank: int = -1,
+    swiglu_limit: float = 0.0,
+) -> torch.Tensor:
+    del topk_weights, topk_ids, w13, w2, w13_scale, w2_scale, w13_offset, w2_offset
+    del num_total_experts, start_expert_id, num_experts_per_rank, swiglu_limit
+    return torch.empty_like(hidden_states)
+
+
 def moe_fused_topk(
     gating_output: torch.Tensor,
     topk: int,
@@ -269,9 +509,12 @@ def fused_moe(
 
 
 __all__ = [
+    "dequant_swiglu_quant",
+    "moe_gating_top_k_hash",
     "supports_cutlass_moe",
     "prepare_grouped_moe_weights",
     "grouped_moe",
+    "grouped_moe_with_selected_experts",
     "moe_fused_topk",
     "cutlass_fused_moe",
     "fused_moe",
diff --git a/xllm/python/kernels_npu/normalization.py b/xllm/python/kernels_npu/normalization.py
index 75a0aa98f2..899c97cb15 100644
--- a/xllm/python/kernels_npu/normalization.py
+++ b/xllm/python/kernels_npu/normalization.py
@@ -20,6 +20,7 @@
 
 rms_norm = torch.ops.xllm_ops.rms_norm
 fused_add_rms_norm = torch.ops.xllm_ops.fused_add_rms_norm
+rms_norm_dynamic_quant = torch.ops.xllm_ops.rms_norm_dynamic_quant
 
 
 def l2_norm(value: torch.Tensor, eps: float = 1e-6) -> torch.Tensor:
@@ -61,4 +62,10 @@ def rms_norm_gated(
     )
 
 
-__all__ = ["rms_norm", "fused_add_rms_norm", "l2_norm", "rms_norm_gated"]
+__all__ = [
+    "rms_norm",
+    "fused_add_rms_norm",
+    "rms_norm_dynamic_quant",
+    "l2_norm",
+    "rms_norm_gated",
+]
diff --git a/xllm/python/kernels_npu/rotary_embedding.py b/xllm/python/kernels_npu/rotary_embedding.py
index 6874a54aa9..4d793fcac1 100644
--- a/xllm/python/kernels_npu/rotary_embedding.py
+++ b/xllm/python/kernels_npu/rotary_embedding.py
@@ -169,9 +169,41 @@ def vision_rotary_mul(
     return torch_npu.npu_rotary_mul(value.unsqueeze(0).contiguous(), cos_full, sin_full).squeeze(0)
 
 
+def npu_inplace_partial_rotary_mul(
+    x: torch.Tensor,
+    cos: torch.Tensor,
+    sin: torch.Tensor,
+    rope_start_dim: int,
+    rope_head_dim: int,
+    inverse: bool = False,
+) -> torch.Tensor:
+    """In-place partial interleaved RoPE on the ``[rope_start_dim:]`` slice.
+
+    Mirrors C++ ``apply_partial_rope`` (deepseek_sparse_attention.cpp:151-190):
+    x is 3D ``[M, n_head, head_dim]``; cos/sin are 2D ``[M, rope_head_dim]``
+    (per-token, no head dim). Reshaped to 4D for the NPU kernel
+    (``aclnnInplacePartialRotaryMul``, rotary_mode="interleave",
+    partial_slice=[rope_start_dim, rope_start_dim+rope_head_dim] -- a half-open
+    range, NOT [start, length]). Modifies x in place.
+    """
+    x4d = x.unsqueeze(2)  # [M, n_head, 1, head_dim]
+    cos4d = cos.view(cos.size(0), 1, 1, cos.size(1))
+    sin_cache = -sin if inverse else sin
+    sin4d = sin_cache.view(sin.size(0), 1, 1, sin.size(1))
+    torch.ops.xllm_ops.npu_inplace_partial_rotary_mul(
+        x4d,
+        cos4d,
+        sin4d,
+        "interleave",
+        [int(rope_start_dim), int(rope_start_dim + rope_head_dim)],
+    )
+    return x
+
+
 __all__ = [
     "fused_qk_norm_rope",
     "interleaved_rotary_embedding",
     "mrope",
     "vision_rotary_mul",
+    "npu_inplace_partial_rotary_mul",
 ]