diff --git a/benchmarks/ops/_moe_bench_utils.py b/benchmarks/ops/_moe_bench_utils.py
new file mode 100644
index 000000000..b91ebb8d3
--- /dev/null
+++ b/benchmarks/ops/_moe_bench_utils.py
@@ -0,0 +1,167 @@
+"""Small shared primitives for MoE benchmark entry points."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from typing import Any
+
+import torch
+import torch.nn.functional as F
+
+
+def cuda_time_ms(fn: Callable[[], Any], *, warmup: int, iters: int) -> float:
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    begin = torch.cuda.Event(enable_timing=True)
+    end = torch.cuda.Event(enable_timing=True)
+    begin.record()
+    for _ in range(iters):
+        fn()
+    end.record()
+    torch.cuda.synchronize()
+    return begin.elapsed_time(end) / iters
+
+
+def host_time_ms(fn: Callable[[], Any], *, warmup: int, iters: int) -> float:
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    begin = time.perf_counter()
+    for _ in range(iters):
+        fn()
+    torch.cuda.synchronize()
+    return (time.perf_counter() - begin) * 1e3 / iters
+
+
+def measure_dispatch(
+    run_fresh: Callable[[], Any],
+    *,
+    warmup: int,
+    iters: int,
+    make_cached: Callable[[Any], Callable[[], Any]] | None = None,
+    validate_cached: Callable[[Any, Any], None] | None = None,
+) -> tuple[Any, float, float | None]:
+    """Measure fresh dispatch and an optional cached-handle path."""
+    fresh_result = run_fresh()
+    fresh_ms = cuda_time_ms(run_fresh, warmup=warmup, iters=iters)
+    if make_cached is None:
+        return fresh_result, fresh_ms, None
+    run_cached = make_cached(fresh_result)
+    cached_result = run_cached()
+    cached_ms = cuda_time_ms(run_cached, warmup=warmup, iters=iters)
+    if validate_cached is not None:
+        validate_cached(fresh_result, cached_result)
+    return fresh_result, fresh_ms, cached_ms
+
+
+def make_routing_inputs(
+    num_tokens: int,
+    hidden_size: int,
+    top_k: int,
+    num_experts: int,
+    *,
+    distribution: str = "uniform",
+    topk_dtype: torch.dtype = torch.int32,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+    noise = torch.rand(num_tokens, num_experts, device="cuda")
+    if distribution == "uniform":
+        scores = noise
+    elif distribution == "hotspot":
+        scores = torch.linspace(4.0, -4.0, num_experts, device="cuda").unsqueeze(0) + noise
+    elif distribution == "longtail":
+        ranks = torch.arange(1, num_experts + 1, device="cuda")
+        scores = -torch.log(ranks).unsqueeze(0) + noise
+    else:
+        raise ValueError(f"unknown distribution {distribution!r}")
+    hidden = torch.randn(num_tokens, hidden_size, dtype=torch.bfloat16, device="cuda")
+    topk_ids = scores.topk(top_k, dim=-1).indices.to(topk_dtype)
+    weights = torch.softmax(torch.randn(num_tokens, top_k, device="cuda"), dim=-1)
+    return hidden, topk_ids, weights
+
+
+def make_expert_sizes(
+    num_pairs: int,
+    num_experts: int,
+    distribution: str,
+) -> list[int]:
+    if distribution == "uniform":
+        return [
+            num_pairs // num_experts + (expert < num_pairs % num_experts)
+            for expert in range(num_experts)
+        ]
+    generator = torch.Generator(device="cpu").manual_seed(42)
+    if distribution == "longtail":
+        probabilities = 1 / torch.arange(1, num_experts + 1, dtype=torch.float64).pow(1.2)
+    elif distribution == "hotspot":
+        hot_experts = max(1, num_experts // 8)
+        probabilities = torch.full((num_experts,), 0.2 / num_experts)
+        probabilities[:hot_experts] += 0.8 / hot_experts
+    elif distribution == "router-like":
+        probabilities = torch.softmax(torch.randn(num_experts, generator=generator) * 1.5, dim=0)
+    else:
+        raise ValueError(f"unknown distribution {distribution!r}")
+    assignments = torch.multinomial(
+        probabilities,
+        num_samples=num_pairs,
+        replacement=True,
+        generator=generator,
+    )
+    return torch.bincount(assignments, minlength=num_experts).tolist()
+
+
+def expert_mlp_reference(
+    hidden: torch.Tensor,
+    w_gate_up: torch.Tensor,
+    w_down: torch.Tensor,
+    sizes: list[int],
+) -> torch.Tensor:
+    output = torch.empty(
+        hidden.shape[0],
+        hidden.shape[1],
+        device=hidden.device,
+        dtype=torch.float32,
+    )
+    ffn_size = w_down.shape[-1]
+    start = 0
+    for expert, size in enumerate(sizes):
+        if size == 0:
+            continue
+        rows = hidden[start : start + size].float()
+        gate_up = rows @ w_gate_up[expert].float().t()
+        activated = F.silu(gate_up[:, :ffn_size]) * gate_up[:, ffn_size:]
+        output[start : start + size] = activated @ w_down[expert].float().t()
+        start += size
+    return output
+
+
+def check_output(
+    name: str,
+    actual: torch.Tensor,
+    reference: torch.Tensor,
+    *,
+    max_range_relative_tolerance: float,
+    relative_l2_tolerance: float,
+) -> tuple[float, float, float, float]:
+    actual = actual.float()
+    if not torch.isfinite(actual).all():
+        raise AssertionError(f"{name}: output contains NaN or Inf")
+    error = actual - reference
+    max_abs = error.abs().max().item()
+    rmse = error.square().mean().sqrt().item()
+    relative_l2 = (
+        torch.linalg.vector_norm(error) / torch.linalg.vector_norm(reference).clamp_min(1e-12)
+    ).item()
+    max_range_relative = max_abs / reference.abs().max().clamp_min(1e-12).item()
+    if max_range_relative > max_range_relative_tolerance or relative_l2 > relative_l2_tolerance:
+        raise AssertionError(
+            f"{name}: max_abs={max_abs}, rmse={rmse}, "
+            f"relative_l2={relative_l2}, "
+            f"max_range_relative={max_range_relative}"
+        )
+    return max_abs, rmse, relative_l2, max_range_relative
+
+
+def effective_tflops(logical_flops: int, elapsed_ms: float) -> float:
+    return logical_flops / (elapsed_ms / 1e3) / 1e12
diff --git a/benchmarks/ops/bench_deepep_dispatch.py b/benchmarks/ops/bench_deepep_dispatch.py
new file mode 100644
index 000000000..3335eea38
--- /dev/null
+++ b/benchmarks/ops/bench_deepep_dispatch.py
@@ -0,0 +1,153 @@
+"""Single-node multi-GPU DeepEP V2 dispatch benchmark for the M6 adapter.
+
+Example:
+    torchrun --standalone --nproc-per-node=8 \
+        benchmarks/ops/bench_deepep_dispatch.py \
+        --tokens 1 8 32 128 --hidden-size 7168 --num-experts 256 --top-k 8
+
+Only dispatch is timed.  Expert compute and combine are deliberately excluded.
+The script requires DeepEP V2, but TileOps itself does not depend on DeepEP.
+"""
+
+import argparse
+import json
+import os
+
+import torch
+import torch.distributed as dist
+from _moe_bench_utils import (
+    make_routing_inputs,
+    measure_dispatch,
+)
+
+from tileops.ops.moe import DeepEPDispatchAdapter
+
+WARMUP = 10
+ITERS = 50
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--hidden-size", type=int, default=7168)
+    parser.add_argument("--num-experts", type=int, default=256)
+    parser.add_argument("--top-k", type=int, default=8)
+    parser.add_argument("--tokens", type=int, nargs="+", default=[1, 8, 32, 128, 512])
+    parser.add_argument("--num-sms", type=int, default=0)
+    args = parser.parse_args()
+
+    try:
+        from deep_ep import ElasticBuffer
+    except ImportError as exc:
+        raise RuntimeError("bench_deepep_dispatch.py requires a DeepEP V2 installation") from exc
+
+    dist.init_process_group("nccl")
+    local_rank = int(os.environ["LOCAL_RANK"])
+    torch.cuda.set_device(local_rank)
+    group = dist.group.WORLD
+    world_size = dist.get_world_size(group)
+    rank = dist.get_rank(group)
+    if args.num_experts % world_size != 0:
+        raise ValueError(
+            f"num_experts must be divisible by world_size; got {args.num_experts} and {world_size}"
+        )
+    if args.top_k > args.num_experts:
+        raise ValueError("top_k cannot exceed num_experts")
+
+    max_tokens = max(args.tokens)
+    buffer = ElasticBuffer(
+        group,
+        num_max_tokens_per_rank=max_tokens,
+        hidden=args.hidden_size,
+        num_topk=args.top_k,
+        use_fp8_dispatch=False,
+    )
+    adapter = DeepEPDispatchAdapter(
+        buffer,
+        num_experts=args.num_experts,
+        num_local_experts=args.num_experts // world_size,
+        num_max_tokens_per_rank=max_tokens,
+        num_sms=args.num_sms,
+    )
+
+    for num_tokens in args.tokens:
+        torch.manual_seed(2026 + rank)
+        hidden, topk_ids, topk_weights = make_routing_inputs(
+            num_tokens,
+            args.hidden_size,
+            args.top_k,
+            args.num_experts,
+            topk_dtype=torch.int64,
+        )
+        offsets = torch.empty(
+            args.num_experts // world_size + 1,
+            dtype=torch.int32,
+            device="cuda",
+        )
+
+        def run_fresh(
+            hidden=hidden,
+            topk_ids=topk_ids,
+            topk_weights=topk_weights,
+            offsets=offsets,
+        ):
+            return adapter.dispatch(
+                hidden,
+                topk_ids,
+                topk_weights,
+                expert_offsets=offsets,
+            )
+
+        def make_cached(
+            result,
+            offsets=offsets,
+            cached_hidden=hidden,
+            cached_weights=topk_weights,
+        ):
+            cached_offsets = torch.empty_like(offsets)
+
+            def run_cached():
+                return adapter.dispatch(
+                    cached_hidden,
+                    None,
+                    cached_weights,
+                    expert_offsets=cached_offsets,
+                    cached_handle=result.combine_handle,
+                )
+
+            return run_cached
+
+        def validate_cached(result, cached_result):
+            torch.testing.assert_close(
+                cached_result.batch.expert_offsets,
+                result.batch.expert_offsets,
+            )
+
+        result, fresh_ms, cached_ms = measure_dispatch(
+            run_fresh,
+            warmup=WARMUP,
+            iters=ITERS,
+            make_cached=make_cached,
+            validate_cached=validate_cached,
+        )
+        assert cached_ms is not None
+        record = {
+            "rank": rank,
+            "world_size": world_size,
+            "tokens": num_tokens,
+            "top_k": args.top_k,
+            "num_experts": args.num_experts,
+            "num_local_experts": args.num_experts // world_size,
+            "hidden_size": args.hidden_size,
+            "sent_pairs": num_tokens * args.top_k,
+            "received_pairs": int(result.batch.valid_rows.item()),
+            "physical_rows": result.batch.capacity,
+            "dispatch_fresh_allocating_ms": round(fresh_ms, 4),
+            "dispatch_cached_allocating_ms": round(cached_ms, 4),
+        }
+        print(json.dumps(record), flush=True)
+        dist.barrier(group)
+    dist.destroy_process_group()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmarks/ops/bench_dispatched_expert.py b/benchmarks/ops/bench_dispatched_expert.py
new file mode 100644
index 000000000..070339a6c
--- /dev/null
+++ b/benchmarks/ops/bench_dispatched_expert.py
@@ -0,0 +1,62 @@
+"""Dynamic-capacity benchmark for the dispatched expert MLP."""
+
+import torch
+from _moe_bench_utils import cuda_time_ms, effective_tflops
+
+from tileops.ops.moe import DispatchedExpertMLPFwdOp, ExpertBatch
+
+DTYPE = torch.bfloat16
+E, H, F = 128, 2048, 1024
+WARMUP, ITERS = 20, 100
+
+
+def _run_capacity_sweep(capacity: int = 16_384) -> None:
+    hidden = torch.randn(capacity, H, device="cuda", dtype=DTYPE)
+    w_gate_up = torch.randn(E, 2 * F, H, device="cuda", dtype=DTYPE) * 0.02
+    w_down = torch.randn(E, H, F, device="cuda", dtype=DTYPE) * 0.02
+    unfused = DispatchedExpertMLPFwdOp(capacity, E, H, F, DTYPE, use_fused_activation=False)
+    fused = DispatchedExpertMLPFwdOp(capacity, E, H, F, DTYPE, use_fused_activation=True)
+    print(
+        "capacity,valid_rows,utilization,"
+        "capacity_unfused_ms,capacity_fused_ms,"
+        "unfused_effective_TFLOPS,fused_effective_TFLOPS"
+    )
+    for valid_rows in (0, capacity // 64, capacity // 8, capacity):
+        sizes = [valid_rows // E + (expert < valid_rows % E) for expert in range(E)]
+        offsets = [0]
+        for size in sizes:
+            offsets.append(offsets[-1] + size)
+        batch = ExpertBatch(
+            hidden=hidden,
+            expert_offsets=torch.tensor(offsets, device="cuda", dtype=torch.int32),
+        )
+        unfused_ms = cuda_time_ms(
+            lambda batch=batch: unfused.forward_batch(batch, w_gate_up, w_down),
+            warmup=WARMUP,
+            iters=ITERS,
+        )
+        fused_ms = cuda_time_ms(
+            lambda batch=batch: fused.forward_batch(batch, w_gate_up, w_down),
+            warmup=WARMUP,
+            iters=ITERS,
+        )
+        logical_flops = 6 * valid_rows * H * F
+        print(
+            f"{capacity},{valid_rows},{valid_rows / capacity:.6f},"
+            f"{unfused_ms:.4f},{fused_ms:.4f},"
+            f"{effective_tflops(logical_flops, unfused_ms):.2f},"
+            f"{effective_tflops(logical_flops, fused_ms):.2f}",
+            flush=True,
+        )
+
+
+def main() -> None:
+    assert torch.cuda.is_available()
+    torch.manual_seed(42)
+    torch.set_grad_enabled(False)
+    print(f"GPU={torch.cuda.get_device_name()} dtype={DTYPE} E={E} H={H} F={F}")
+    _run_capacity_sweep()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmarks/ops/bench_dispatched_expert_deepgemm.py b/benchmarks/ops/bench_dispatched_expert_deepgemm.py
new file mode 100644
index 000000000..56e994319
--- /dev/null
+++ b/benchmarks/ops/bench_dispatched_expert_deepgemm.py
@@ -0,0 +1,304 @@
+"""Compute-only BF16 Expert MLP comparison with DeepGEMM ordinary MoE."""
+
+import argparse
+
+import deep_gemm
+import torch
+import torch.nn.functional as F
+from _moe_bench_utils import (
+    check_output,
+    cuda_time_ms,
+    effective_tflops,
+    expert_mlp_reference,
+    host_time_ms,
+    make_expert_sizes,
+)
+
+from tileops.ops.moe import DispatchedExpertMLPFwdOp
+
+DTYPE = torch.bfloat16
+WARMUP, ITERS = 10, 50
+MAX_RANGE_REL_TOL = 0.02
+RELATIVE_L2_TOL = 0.01
+MODELS = {
+    "synthetic": (128, 8, 2048, 1024, [128, 1024]),
+    "glm5": (256, 8, 6144, 2048, [1, 32, 256, 512, 1024, 2048]),
+    "deepseek-v3": (256, 8, 7168, 2048, [1, 32, 256, 512, 1024, 2048]),
+    "qwen3-235b": (128, 8, 7168, 2048, [1, 16, 128, 256, 512, 1024]),
+    "qwen35-397b": (512, 10, 4096, 1024, [1, 52, 410, 820, 1639, 3277]),
+}
+
+
+def _run(
+    case_name: str,
+    num_pairs: int,
+    num_experts: int,
+    hidden_size: int,
+    ffn_size: int,
+    distribution: str,
+) -> str:
+    E, H, F_DIM = num_experts, hidden_size, ffn_size
+    sizes = make_expert_sizes(num_pairs, E, distribution)
+    alignment = deep_gemm.get_theoretical_mk_alignment_for_contiguous_layout(max(1, num_pairs // E))
+    deep_gemm.set_mk_alignment_for_contiguous_layout(alignment)
+    aligned_sizes = [(size + alignment - 1) // alignment * alignment for size in sizes]
+    physical_rows = sum(aligned_sizes)
+
+    tight = torch.randn(num_pairs, H, device="cuda", dtype=DTYPE)
+    w_gate_up = torch.randn(E, 2 * F_DIM, H, device="cuda", dtype=DTYPE) * 0.02
+    w_down = torch.randn(E, H, F_DIM, device="cuda", dtype=DTYPE) * 0.02
+    true_sizes = torch.tensor(sizes, device="cuda", dtype=torch.int32)
+    true_offsets = torch.tensor(
+        [sum(sizes[:expert]) for expert in range(E)],
+        device="cuda",
+        dtype=torch.int32,
+    )
+
+    aligned = torch.empty(physical_rows, H, device="cuda", dtype=DTYPE)
+    psum = torch.empty(E, device="cuda", dtype=torch.int32)
+
+    def pack_aligned():
+        aligned.zero_()
+        tight_start = physical_start = 0
+        for expert, (size, aligned_size) in enumerate(zip(sizes, aligned_sizes, strict=True)):
+            aligned[physical_start : physical_start + size].copy_(
+                tight[tight_start : tight_start + size]
+            )
+            psum[expert] = physical_start + size
+            tight_start += size
+            physical_start += aligned_size
+
+    pack_aligned()
+
+    gate_up = torch.empty(physical_rows, 2 * F_DIM, device="cuda", dtype=DTYPE)
+    activated = torch.empty(physical_rows, F_DIM, device="cuda", dtype=DTYPE)
+    output = torch.empty(physical_rows, H, device="cuda", dtype=DTYPE)
+
+    tileops_unfused = DispatchedExpertMLPFwdOp(
+        num_pairs, E, H, F_DIM, DTYPE, use_fused_activation=False
+    )
+    tileops_fused = DispatchedExpertMLPFwdOp(
+        num_pairs, E, H, F_DIM, DTYPE, use_fused_activation=True
+    )
+    if not tileops_fused.use_fused_activation:
+        raise RuntimeError(f"{case_name}: requested TileOps fused activation is not eligible")
+
+    def run_tileops_unfused():
+        return tileops_unfused(tight, w_gate_up, w_down, true_sizes, true_offsets)
+
+    def run_tileops_fused():
+        return tileops_fused(tight, w_gate_up, w_down, true_sizes, true_offsets)
+
+    def run_gemm1_into(destination):
+        deep_gemm.m_grouped_bf16_gemm_nt_contiguous(
+            aligned,
+            w_gate_up,
+            destination,
+            psum,
+            use_psum_layout=True,
+            expected_m_for_psum_layout=max(1, num_pairs // E),
+        )
+
+    def run_activation_into(gate_up_input, destination):
+        torch.mul(
+            F.silu(gate_up_input[:, :F_DIM]),
+            gate_up_input[:, F_DIM:],
+            out=destination,
+        )
+
+    def run_gemm2_into(activation_input, destination):
+        deep_gemm.m_grouped_bf16_gemm_nt_contiguous(
+            activation_input,
+            w_down,
+            destination,
+            psum,
+            use_psum_layout=True,
+            expected_m_for_psum_layout=max(1, num_pairs // E),
+        )
+
+    def run_gemm1():
+        run_gemm1_into(gate_up)
+
+    def run_activation():
+        run_activation_into(gate_up, activated)
+
+    def run_gemm2():
+        run_gemm2_into(activated, output)
+
+    def run_deepgemm_pipeline():
+        run_gemm1()
+        run_activation()
+        run_gemm2()
+
+    def run_deepgemm_allocating_pipeline():
+        local_gate_up = torch.empty_like(gate_up)
+        local_activated = torch.empty_like(activated)
+        local_output = torch.empty_like(output)
+        run_gemm1_into(local_gate_up)
+        run_activation_into(local_gate_up, local_activated)
+        run_gemm2_into(local_activated, local_output)
+        return local_output
+
+    # Validate every reported backend against an independent FP32 reference.
+    reference = expert_mlp_reference(tight, w_gate_up, w_down, sizes)
+    tileops_unfused_output = run_tileops_unfused()
+    tileops_fused_output = run_tileops_fused()
+    run_deepgemm_pipeline()
+    deepgemm_valid = torch.empty_like(tileops_unfused_output)
+    tight_start = physical_start = 0
+    for size, aligned_size in zip(sizes, aligned_sizes, strict=True):
+        deepgemm_valid[tight_start : tight_start + size].copy_(
+            output[physical_start : physical_start + size]
+        )
+        tight_start += size
+        physical_start += aligned_size
+    torch.cuda.synchronize()
+    check_kwargs = {
+        "max_range_relative_tolerance": MAX_RANGE_REL_TOL,
+        "relative_l2_tolerance": RELATIVE_L2_TOL,
+    }
+    tileops_unfused_error = check_output(
+        "TileOps unfused",
+        tileops_unfused_output,
+        reference,
+        **check_kwargs,
+    )
+    tileops_fused_error = check_output(
+        "TileOps fused",
+        tileops_fused_output,
+        reference,
+        **check_kwargs,
+    )
+    deepgemm_error = check_output(
+        "DeepGEMM",
+        deepgemm_valid,
+        reference,
+        **check_kwargs,
+    )
+
+    pack_ms = host_time_ms(pack_aligned, warmup=2, iters=10)
+    tileops_unfused_ms = cuda_time_ms(run_tileops_unfused, warmup=WARMUP, iters=ITERS)
+    tileops_fused_ms = cuda_time_ms(run_tileops_fused, warmup=WARMUP, iters=ITERS)
+    gemm1_ms = cuda_time_ms(run_gemm1, warmup=WARMUP, iters=ITERS)
+    activation_ms = cuda_time_ms(run_activation, warmup=WARMUP, iters=ITERS)
+    gemm2_ms = cuda_time_ms(run_gemm2, warmup=WARMUP, iters=ITERS)
+    deepgemm_preallocated_ms = cuda_time_ms(run_deepgemm_pipeline, warmup=WARMUP, iters=ITERS)
+    deepgemm_allocating_ms = cuda_time_ms(
+        run_deepgemm_allocating_pipeline, warmup=WARMUP, iters=ITERS
+    )
+
+    logical_flops = 6 * num_pairs * H * F_DIM
+    physical_flops = 6 * physical_rows * H * F_DIM
+    empty_experts = sum(size == 0 for size in sizes)
+    sizes_tensor = torch.tensor(sizes, dtype=torch.float32)
+    return (
+        f"{case_name},{distribution},{num_pairs / E:.4f},"
+        f"{min(sizes)},{max(sizes)},{sizes_tensor.std(unbiased=False).item():.4f},"
+        f"{num_pairs},{physical_rows},{empty_experts},"
+        f"{physical_rows / num_pairs - 1:.4f},"
+        f"{pack_ms:.4f},"
+        f"{tileops_unfused_ms:.4f},{tileops_fused_ms:.4f},"
+        f"{gemm1_ms:.4f},{activation_ms:.4f},{gemm2_ms:.4f},"
+        f"{deepgemm_preallocated_ms:.4f},{deepgemm_allocating_ms:.4f},"
+        f"{effective_tflops(logical_flops, tileops_unfused_ms):.2f},"
+        f"{effective_tflops(logical_flops, tileops_fused_ms):.2f},"
+        f"{effective_tflops(logical_flops, deepgemm_preallocated_ms):.2f},"
+        f"{effective_tflops(logical_flops, deepgemm_allocating_ms):.2f},"
+        f"{effective_tflops(physical_flops, deepgemm_preallocated_ms):.2f},"
+        f"{tileops_unfused_error[0]:.6f},"
+        f"{tileops_unfused_error[1]:.6f},"
+        f"{tileops_unfused_error[2]:.6f},"
+        f"{tileops_unfused_error[3]:.6f},"
+        f"{tileops_fused_error[0]:.6f},"
+        f"{tileops_fused_error[1]:.6f},"
+        f"{tileops_fused_error[2]:.6f},"
+        f"{tileops_fused_error[3]:.6f},"
+        f"{deepgemm_error[0]:.6f},{deepgemm_error[1]:.6f},"
+        f"{deepgemm_error[2]:.6f},{deepgemm_error[3]:.6f},"
+        f"{num_pairs}"
+    )
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--model",
+        choices=tuple(MODELS),
+        default="synthetic",
+    )
+    parser.add_argument(
+        "--tokens",
+        type=int,
+        nargs="+",
+        help="Token counts; defaults depend on --model.",
+    )
+    parser.add_argument(
+        "--m",
+        type=int,
+        nargs="+",
+        help="Exact logical rows per expert; takes precedence over --tokens.",
+    )
+    parser.add_argument(
+        "--distribution",
+        choices=("uniform", "longtail", "hotspot", "router-like"),
+        nargs="+",
+        default=("uniform",),
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    assert torch.cuda.is_available()
+    torch.manual_seed(42)
+    torch.set_float32_matmul_precision("highest")
+    torch.set_grad_enabled(False)
+    args = _parse_args()
+    num_experts, top_k, hidden_size, ffn_size, default_tokens = MODELS[args.model]
+    print(
+        f"GPU={torch.cuda.get_device_name()} dtype={DTYPE} "
+        f"model={args.model} E={num_experts} K={top_k} "
+        f"H={hidden_size} F={ffn_size}"
+    )
+    print(
+        "case,distribution,mean_M,min_M,max_M,std_M,"
+        "logical_rows,physical_rows,empty_experts,padding_ratio,pack_ms,"
+        "tileops_unfused_ms,tileops_fused_ms,"
+        "deepgemm_gemm1_ms,activation_ms,deepgemm_gemm2_ms,"
+        "deepgemm_preallocated_ms,deepgemm_allocating_ms,"
+        "tileops_unfused_effective_TFLOPS,"
+        "tileops_fused_effective_TFLOPS,"
+        "deepgemm_preallocated_effective_TFLOPS,"
+        "deepgemm_allocating_effective_TFLOPS,"
+        "deepgemm_physical_TFLOPS,"
+        "tileops_unfused_max_abs,tileops_unfused_rmse,"
+        "tileops_unfused_relative_l2,tileops_unfused_max_range_relative,"
+        "tileops_fused_max_abs,tileops_fused_rmse,"
+        "tileops_fused_relative_l2,tileops_fused_max_range_relative,"
+        "deepgemm_max_abs,deepgemm_rmse,deepgemm_relative_l2,"
+        "deepgemm_max_range_relative,"
+        "valid_rows_checked"
+    )
+    token_counts = args.tokens or default_tokens
+    cases = (
+        [(f"M={rows}", rows * num_experts) for rows in args.m]
+        if args.m
+        else [(f"T={tokens}", tokens * top_k) for tokens in token_counts]
+    )
+    for distribution in args.distribution:
+        for case_name, num_pairs in cases:
+            print(
+                _run(
+                    case_name,
+                    num_pairs,
+                    num_experts,
+                    hidden_size,
+                    ffn_size,
+                    distribution,
+                ),
+                flush=True,
+            )
+            torch.cuda.empty_cache()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmarks/ops/bench_expert_dispatch.py b/benchmarks/ops/bench_expert_dispatch.py
new file mode 100644
index 000000000..5196c7fd2
--- /dev/null
+++ b/benchmarks/ops/bench_expert_dispatch.py
@@ -0,0 +1,92 @@
+"""Dispatch-only benchmark for the tight local M6 reference path.
+
+DeepEP communication is intentionally not emulated here.  A real DeepEP run
+must report communication and adapter normalization separately; this benchmark
+measures the local count/prefix-sum/scatter/gather primitive used as the
+world-size-one reference.
+"""
+
+import argparse
+
+import torch
+from _moe_bench_utils import (
+    make_routing_inputs,
+    measure_dispatch,
+)
+
+from tileops.ops.moe import LocalExpertDispatcher
+
+WARMUP = 20
+ITERS = 100
+FIELDS = (
+    "tokens",
+    "top_k",
+    "num_experts",
+    "hidden_size",
+    "distribution",
+    "effective_rows",
+    "physical_rows",
+    "dispatch_allocating_ms",
+    "effective_GBps",
+)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--hidden-size", type=int, default=7168)
+    parser.add_argument("--num-experts", type=int, default=256)
+    parser.add_argument("--top-k", type=int, default=8)
+    parser.add_argument("--tokens", type=int, nargs="+", default=[1, 8, 32, 128, 512, 2048])
+    parser.add_argument(
+        "--distributions",
+        nargs="+",
+        choices=["uniform", "longtail", "hotspot"],
+        default=["uniform", "longtail", "hotspot"],
+    )
+    args = parser.parse_args()
+
+    print(",".join(FIELDS))
+    for num_tokens in args.tokens:
+        for distribution in args.distributions:
+            hidden, topk_ids, weights = make_routing_inputs(
+                num_tokens,
+                args.hidden_size,
+                args.top_k,
+                args.num_experts,
+                distribution=distribution,
+            )
+            dispatcher = LocalExpertDispatcher(
+                args.num_experts,
+                total_tokens=num_tokens,
+                top_k=args.top_k,
+                hidden_size=args.hidden_size,
+                dtype=torch.bfloat16,
+            )
+
+            def run(
+                dispatcher=dispatcher,
+                hidden=hidden,
+                topk_ids=topk_ids,
+                weights=weights,
+            ):
+                return dispatcher.dispatch(hidden, topk_ids, weights)
+
+            result, fresh_ms, _ = measure_dispatch(run, warmup=WARMUP, iters=ITERS)
+            effective_rows = num_tokens * args.top_k
+            logical_bytes = 2 * effective_rows * args.hidden_size * hidden.element_size()
+            values = (
+                num_tokens,
+                args.top_k,
+                args.num_experts,
+                args.hidden_size,
+                distribution,
+                effective_rows,
+                result.batch.capacity,
+                f"{fresh_ms:.4f}",
+                f"{logical_bytes / (fresh_ms / 1e3) / 1e9:.2f}",
+            )
+            print(",".join(map(str, values)), flush=True)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/ops/test_dispatched_expert.py b/tests/ops/test_dispatched_expert.py
new file mode 100644
index 000000000..c665fd650
--- /dev/null
+++ b/tests/ops/test_dispatched_expert.py
@@ -0,0 +1,252 @@
+"""Tests for the communication-independent dispatched expert MLP."""
+
+import pytest
+import torch
+import torch.nn.functional as F
+
+from tileops.kernels.grouped_gemm import GroupedGemmPersistent3WGKernel
+from tileops.ops.moe import (
+    DispatchedExpertMLPFwdOp,
+    ExpertBatch,
+)
+
+
+def _reference(hidden, w_gate_up, w_down, sizes):
+    outputs = []
+    start = 0
+    ffn_size = w_down.shape[-1]
+    for expert, size in enumerate(sizes):
+        rows = hidden[start : start + size].float()
+        gate_up = rows @ w_gate_up[expert].float().t()
+        act = F.silu(gate_up[:, :ffn_size]) * gate_up[:, ffn_size:]
+        outputs.append(act @ w_down[expert].float().t())
+        start += size
+    return torch.cat(outputs).to(hidden.dtype)
+
+
+@pytest.mark.smoke
+def test_expert_batch_contract_rejects_non_tight_layout():
+    with pytest.raises(ValueError, match="layout='tight'"):
+        ExpertBatch(
+            hidden=torch.empty(4, 8),
+            expert_offsets=torch.tensor([0, 2, 4], dtype=torch.int32),
+            layout="aligned",
+        )
+
+
+@pytest.mark.smoke
+def test_expert_batch_valid_rows_is_offsets_view():
+    hidden = torch.empty(4, 8)
+    offsets = torch.tensor([0, 2, 4], dtype=torch.int32)
+    batch = ExpertBatch(hidden=hidden, expert_offsets=offsets)
+
+    assert batch.valid_rows.shape == (1,)
+    assert batch.valid_rows.data_ptr() == offsets[-1:].data_ptr()
+    offsets[-1] = 3
+    assert batch.valid_rows.item() == 3
+
+
+@pytest.mark.parametrize("dtype", [torch.bfloat16, torch.float16])
+@pytest.mark.parametrize("sizes", [[4, 4, 4, 4], [0, 2, 13, 1]])
+@pytest.mark.parametrize("explicit_3wg_override", [False, True])
+@pytest.mark.smoke
+def test_dispatched_expert_matches_reference_and_preserves_rows(
+    dtype,
+    sizes,
+    explicit_3wg_override,
+):
+    if not torch.cuda.is_available():
+        pytest.skip("CUDA is required")
+
+    torch.manual_seed(0)
+    num_experts = len(sizes)
+    num_pairs = sum(sizes)
+    hidden_size, ffn_size = 128, 96
+    hidden = torch.randn(num_pairs, hidden_size, device="cuda", dtype=dtype) * 0.1
+    w_gate_up = (
+        torch.randn(num_experts, 2 * ffn_size, hidden_size, device="cuda", dtype=dtype) * 0.02
+    )
+    w_down = torch.randn(num_experts, hidden_size, ffn_size, device="cuda", dtype=dtype) * 0.02
+    true_sizes = torch.tensor(sizes, device="cuda", dtype=torch.int32)
+    true_offsets = torch.tensor(
+        [sum(sizes[:expert]) for expert in range(num_experts)],
+        device="cuda",
+        dtype=torch.int32,
+    )
+
+    op = DispatchedExpertMLPFwdOp(
+        num_pairs=num_pairs,
+        num_experts=num_experts,
+        hidden_size=hidden_size,
+        ffn_size=ffn_size,
+        dtype=dtype,
+        kernel_map=(
+            {"moe_grouped_gemm_kernel": GroupedGemmPersistent3WGKernel}
+            if explicit_3wg_override
+            else None
+        ),
+    )
+    output = op(hidden, w_gate_up, w_down, true_sizes, true_offsets)
+    reference = _reference(hidden, w_gate_up, w_down, sizes)
+
+    assert output.shape == hidden.shape
+    assert output.dtype == dtype
+    assert torch.allclose(output.float(), reference.float(), atol=1e-2, rtol=1e-2)
+
+
+@pytest.mark.smoke
+@pytest.mark.parametrize("use_fused_activation", [False, True])
+@pytest.mark.parametrize("sizes", [[0, 3, 5, 2], [0, 0, 0, 0]])
+def test_expert_batch_capacity_processes_only_device_valid_rows(
+    use_fused_activation,
+    sizes,
+):
+    if not torch.cuda.is_available():
+        pytest.skip("CUDA is required")
+    if use_fused_activation and torch.cuda.get_device_capability()[0] < 9:
+        pytest.skip("fused activation requires SM90")
+
+    torch.manual_seed(2)
+    capacity, hidden_size, ffn_size, num_experts = 24, 256, 128, 4
+    valid_count = sum(sizes)
+    dtype = torch.bfloat16
+    hidden = torch.randn(capacity, hidden_size, device="cuda", dtype=dtype) * 0.1
+    # An invalid tail must not affect valid rows.
+    hidden[valid_count:].fill_(float("nan"))
+    w_gate_up = (
+        torch.randn(
+            num_experts,
+            2 * ffn_size,
+            hidden_size,
+            device="cuda",
+            dtype=dtype,
+        )
+        * 0.02
+    )
+    w_down = (
+        torch.randn(
+            num_experts,
+            hidden_size,
+            ffn_size,
+            device="cuda",
+            dtype=dtype,
+        )
+        * 0.02
+    )
+    host_offsets = [0]
+    for size in sizes:
+        host_offsets.append(host_offsets[-1] + size)
+    offsets = torch.tensor(host_offsets, device="cuda", dtype=torch.int32)
+    batch = ExpertBatch(hidden, offsets)
+    op = DispatchedExpertMLPFwdOp(
+        num_pairs=capacity,
+        num_experts=num_experts,
+        hidden_size=hidden_size,
+        ffn_size=ffn_size,
+        dtype=dtype,
+        use_fused_activation=use_fused_activation,
+    )
+
+    output = op.forward_batch(batch, w_gate_up, w_down)
+    reference = _reference(hidden[:valid_count], w_gate_up, w_down, sizes)
+
+    assert output.hidden.shape == (capacity, hidden_size)
+    assert output.valid_rows.data_ptr() == offsets[-1:].data_ptr()
+    assert torch.isfinite(output.hidden[:valid_count]).all()
+    assert torch.allclose(
+        output.hidden[:valid_count].float(),
+        reference.float(),
+        atol=1e-2,
+        rtol=1e-2,
+    )
+
+
+@pytest.mark.smoke
+@pytest.mark.parametrize("use_fused_activation", [False, True])
+def test_expert_batch_cuda_graph_replays_different_valid_rows(
+    use_fused_activation,
+):
+    if not torch.cuda.is_available():
+        pytest.skip("CUDA is required")
+    if use_fused_activation and torch.cuda.get_device_capability()[0] < 9:
+        pytest.skip("fused activation requires SM90")
+
+    torch.manual_seed(3)
+    capacity, hidden_size, ffn_size, num_experts = 24, 256, 128, 4
+    dtype = torch.bfloat16
+    hidden = torch.empty(capacity, hidden_size, device="cuda", dtype=dtype)
+    offsets = torch.empty(num_experts + 1, device="cuda", dtype=torch.int32)
+    w_gate_up = (
+        torch.randn(
+            num_experts,
+            2 * ffn_size,
+            hidden_size,
+            device="cuda",
+            dtype=dtype,
+        )
+        * 0.02
+    )
+    w_down = (
+        torch.randn(
+            num_experts,
+            hidden_size,
+            ffn_size,
+            device="cuda",
+            dtype=dtype,
+        )
+        * 0.02
+    )
+    batch = ExpertBatch(hidden, offsets)
+    op = DispatchedExpertMLPFwdOp(
+        num_pairs=capacity,
+        num_experts=num_experts,
+        hidden_size=hidden_size,
+        ffn_size=ffn_size,
+        dtype=dtype,
+        use_fused_activation=use_fused_activation,
+    )
+
+    def set_batch(sizes):
+        count = sum(sizes)
+        hidden[:count].copy_(torch.randn(count, hidden_size, device="cuda", dtype=dtype) * 0.1)
+        hidden[count:].fill_(float("nan"))
+        host_offsets = [0]
+        for size in sizes:
+            host_offsets.append(host_offsets[-1] + size)
+        offsets.copy_(torch.tensor(host_offsets, device="cuda", dtype=torch.int32))
+        return count
+
+    first_sizes = [2, 0, 3, 2]
+    first_count = set_batch(first_sizes)
+    # Compile and warm allocator state before capture.
+    op.forward_batch(batch, w_gate_up, w_down)
+    side_stream = torch.cuda.Stream()
+    side_stream.wait_stream(torch.cuda.current_stream())
+    with torch.cuda.stream(side_stream):
+        for _ in range(2):
+            op.forward_batch(batch, w_gate_up, w_down)
+    torch.cuda.current_stream().wait_stream(side_stream)
+
+    graph = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(graph):
+        captured = op.forward_batch(batch, w_gate_up, w_down).hidden
+    graph.replay()
+    first_reference = _reference(hidden[:first_count], w_gate_up, w_down, first_sizes)
+    assert torch.allclose(
+        captured[:first_count].float(),
+        first_reference.float(),
+        atol=1e-2,
+        rtol=1e-2,
+    )
+
+    second_sizes = [0, 4, 1, 6]
+    second_count = set_batch(second_sizes)
+    graph.replay()
+    second_reference = _reference(hidden[:second_count], w_gate_up, w_down, second_sizes)
+    assert torch.isfinite(captured[:second_count]).all()
+    assert torch.allclose(
+        captured[:second_count].float(),
+        second_reference.float(),
+        atol=1e-2,
+        rtol=1e-2,
+    )
diff --git a/tests/ops/test_expert_dispatch.py b/tests/ops/test_expert_dispatch.py
new file mode 100644
index 000000000..8fad4d6d1
--- /dev/null
+++ b/tests/ops/test_expert_dispatch.py
@@ -0,0 +1,304 @@
+"""Correctness tests for tight local and DeepEP dispatch adapters."""
+
+from types import SimpleNamespace
+
+import pytest
+import torch
+import torch.nn.functional as F
+
+from tileops.ops.moe import (
+    DeepEPDispatchAdapter,
+    DispatchedExpertMLPFwdOp,
+    ExpertDispatchResult,
+    LocalDispatchHandle,
+    LocalExpertDispatcher,
+)
+
+
+def _routing(
+    num_tokens: int,
+    top_k: int,
+    num_experts: int,
+    *,
+    hidden_size: int = 16,
+    dtype: torch.dtype = torch.bfloat16,
+) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+    hidden = (
+        torch.arange(num_tokens * hidden_size, device="cuda", dtype=torch.float32)
+        .reshape(num_tokens, hidden_size)
+        .to(dtype)
+    )
+    scores = torch.rand(num_tokens, num_experts, device="cuda")
+    topk_ids = scores.topk(top_k, dim=-1).indices.to(torch.int32)
+    weights = torch.softmax(torch.randn(num_tokens, top_k, device="cuda"), dim=-1)
+    return hidden, topk_ids, weights
+
+
+@pytest.mark.smoke
+@pytest.mark.parametrize("num_tokens,top_k,num_experts", [(5, 2, 4), (3, 1, 8)])
+def test_local_dispatch_is_tight_and_reversible(
+    num_tokens: int,
+    top_k: int,
+    num_experts: int,
+) -> None:
+    hidden, topk_ids, weights = _routing(num_tokens, top_k, num_experts)
+    result = LocalExpertDispatcher(num_experts).dispatch(hidden, topk_ids, weights)
+
+    assert isinstance(result, ExpertDispatchResult)
+    assert isinstance(result.combine_handle, LocalDispatchHandle)
+    assert result.batch.layout == "tight"
+    assert result.batch.capacity == num_tokens * top_k
+    assert result.batch.expert_offsets.dtype == torch.int32
+    assert result.routing_weights.dtype == torch.float32
+
+    flat_ids = topk_ids.flatten()
+    counts = torch.bincount(flat_ids.to(torch.int64), minlength=num_experts)
+    expected_offsets = torch.cat(
+        (
+            torch.zeros(1, dtype=torch.int64, device="cuda"),
+            counts.cumsum(0),
+        )
+    ).to(torch.int32)
+    torch.testing.assert_close(result.batch.expert_offsets, expected_offsets)
+
+    mapping = result.combine_handle.forward_mapping.to(torch.int64)
+    source_rows = torch.arange(num_tokens, device="cuda").unsqueeze(1).expand(-1, top_k).flatten()
+    torch.testing.assert_close(result.batch.hidden[mapping], hidden[source_rows])
+    torch.testing.assert_close(result.routing_weights[mapping], weights.flatten())
+
+
+@pytest.mark.smoke
+def test_local_dispatch_reference_combine_applies_weights_once() -> None:
+    hidden, topk_ids, weights = _routing(7, 3, 6)
+    result = LocalExpertDispatcher(6).dispatch(hidden, topk_ids, weights)
+    handle = result.combine_handle
+    assert isinstance(handle, LocalDispatchHandle)
+
+    # Stand in for an expert function that preserves row order.
+    expert_output = result.batch.hidden.float() * 2.0
+    pair_output = expert_output[handle.forward_mapping.to(torch.int64)]
+    combined = (
+        pair_output.reshape(handle.num_tokens, handle.top_k, -1) * weights.unsqueeze(-1)
+    ).sum(dim=1)
+    reference = hidden.float() * 2.0 * weights.sum(dim=1, keepdim=True)
+    torch.testing.assert_close(combined, reference)
+
+
+@pytest.mark.smoke
+def test_local_dispatch_output_is_consumed_directly_by_m5() -> None:
+    num_tokens, top_k, num_experts = 4, 2, 4
+    hidden_size, ffn_size = 128, 128
+    hidden, topk_ids, weights = _routing(num_tokens, top_k, num_experts, hidden_size=hidden_size)
+    hidden = torch.randn_like(hidden) * 0.1
+    result = LocalExpertDispatcher(num_experts).dispatch(hidden, topk_ids, weights)
+    w_gate_up = (
+        torch.randn(
+            num_experts,
+            2 * ffn_size,
+            hidden_size,
+            dtype=torch.bfloat16,
+            device="cuda",
+        )
+        * 0.02
+    )
+    w_down = (
+        torch.randn(
+            num_experts,
+            hidden_size,
+            ffn_size,
+            dtype=torch.bfloat16,
+            device="cuda",
+        )
+        * 0.02
+    )
+    op = DispatchedExpertMLPFwdOp(
+        num_pairs=result.batch.capacity,
+        num_experts=num_experts,
+        hidden_size=hidden_size,
+        ffn_size=ffn_size,
+        dtype=torch.bfloat16,
+    )
+    output = op.forward_batch(result.batch, w_gate_up, w_down)
+
+    reference = torch.empty_like(output.hidden)
+    offsets = result.batch.expert_offsets.cpu().tolist()
+    for expert in range(num_experts):
+        start, end = offsets[expert], offsets[expert + 1]
+        gate_up = result.batch.hidden[start:end].float() @ w_gate_up[expert].float().T
+        activated = F.silu(gate_up[:, :ffn_size]) * gate_up[:, ffn_size:]
+        reference[start:end] = (activated @ w_down[expert].float().T).to(reference.dtype)
+
+    assert output.valid_rows is not None
+    torch.testing.assert_close(output.valid_rows, result.batch.valid_rows)
+    torch.testing.assert_close(
+        output.hidden,
+        reference,
+        rtol=2e-2,
+        atol=2e-2,
+    )
+
+
+class _FakeEvent:
+    def __init__(self) -> None:
+        self.waited = False
+
+    def current_stream_wait(self) -> None:
+        self.waited = True
+
+
+class _FakeDeepEPBuffer:
+    def __init__(self, num_local_experts: int, capacity_tail: int = 3) -> None:
+        self.num_local_experts = num_local_experts
+        self.capacity_tail = capacity_tail
+        self.kwargs = None
+        self.event = _FakeEvent()
+        self.recv_x = None
+
+    def dispatch(self, hidden_states, **kwargs):
+        self.kwargs = kwargs
+        cached_handle = kwargs["handle"]
+        topk_ids = kwargs["topk_idx"] if cached_handle is None else cached_handle.topk_idx
+        weights = kwargs["topk_weights"]
+        flat_ids = topk_ids.flatten()
+        order = torch.argsort(flat_ids, stable=True)
+        source_rows = (
+            torch.arange(hidden_states.shape[0], device=hidden_states.device)
+            .unsqueeze(1)
+            .expand_as(topk_ids)
+            .flatten()
+        )
+        valid_x = hidden_states[source_rows[order]]
+        valid_weights = weights.flatten()[order]
+        self.recv_x = torch.empty(
+            valid_x.shape[0] + self.capacity_tail,
+            hidden_states.shape[1],
+            dtype=hidden_states.dtype,
+            device=hidden_states.device,
+        )
+        self.recv_x[: valid_x.shape[0]].copy_(valid_x)
+        recv_weights = torch.empty(
+            valid_weights.shape[0] + self.capacity_tail,
+            dtype=torch.float32,
+            device=hidden_states.device,
+        )
+        recv_weights[: valid_weights.shape[0]].copy_(valid_weights)
+        counts = torch.bincount(flat_ids, minlength=self.num_local_experts).to(torch.int32)
+        handle = cached_handle or SimpleNamespace(
+            do_expand=True,
+            expert_alignment=1,
+            topk_idx=topk_ids,
+            psum_num_recv_tokens_per_expert=counts.cumsum(0, dtype=torch.int32),
+        )
+        return self.recv_x, None, recv_weights, handle, self.event
+
+
+@pytest.mark.smoke
+def test_deepep_adapter_uses_unpadded_expanded_layout_without_copy() -> None:
+    num_experts = 4
+    hidden, topk_ids, weights = _routing(5, 2, num_experts)
+    topk_ids = topk_ids.to(torch.int64)
+    buffer = _FakeDeepEPBuffer(num_experts)
+    offsets_buffer = torch.empty(num_experts + 1, dtype=torch.int32, device="cuda")
+    result = DeepEPDispatchAdapter(
+        buffer,
+        num_experts=num_experts,
+        num_local_experts=num_experts,
+        num_max_tokens_per_rank=32,
+        num_sms=12,
+    ).dispatch(
+        hidden,
+        topk_ids,
+        weights,
+        expert_offsets=offsets_buffer,
+    )
+
+    assert buffer.kwargs is not None
+    assert buffer.kwargs["do_expand"] is True
+    assert buffer.kwargs["expert_alignment"] == 1
+    assert buffer.kwargs["do_cpu_sync"] is False
+    assert buffer.kwargs["async_with_compute_stream"] is True
+    assert buffer.kwargs["num_sms"] == 12
+    assert buffer.kwargs["topk_idx"] is topk_ids
+    assert buffer.kwargs["handle"] is None
+    assert buffer.event.waited
+    assert result.batch.hidden is buffer.recv_x
+    assert result.batch.expert_offsets is offsets_buffer
+    assert result.batch.capacity == topk_ids.numel() + buffer.capacity_tail
+    assert result.batch.valid_rows.item() == topk_ids.numel()
+
+    counts = torch.bincount(topk_ids.flatten().to(torch.int64), minlength=num_experts)
+    expected_offsets = torch.cat(
+        (
+            torch.zeros(1, dtype=torch.int64, device="cuda"),
+            counts.cumsum(0),
+        )
+    ).to(torch.int32)
+    torch.testing.assert_close(result.batch.expert_offsets, expected_offsets)
+
+
+@pytest.mark.smoke
+def test_deepep_adapter_reuses_expanded_decode_handle() -> None:
+    num_experts = 4
+    hidden, topk_ids, weights = _routing(5, 2, num_experts)
+    topk_ids = topk_ids.to(torch.int64)
+    buffer = _FakeDeepEPBuffer(num_experts)
+    adapter = DeepEPDispatchAdapter(
+        buffer,
+        num_experts=num_experts,
+        num_local_experts=num_experts,
+        num_max_tokens_per_rank=32,
+    )
+    first = adapter.dispatch(hidden, topk_ids, weights)
+    cached_offsets = torch.empty(num_experts + 1, dtype=torch.int32, device="cuda")
+    second = adapter.dispatch(
+        hidden + 1,
+        None,
+        weights,
+        expert_offsets=cached_offsets,
+        cached_handle=first.combine_handle,
+    )
+
+    assert buffer.kwargs["topk_idx"] is None
+    assert buffer.kwargs["handle"] is first.combine_handle
+    assert second.combine_handle is first.combine_handle
+    assert second.batch.expert_offsets is cached_offsets
+    torch.testing.assert_close(second.batch.expert_offsets, first.batch.expert_offsets)
+
+
+@pytest.mark.smoke
+def test_deepep_adapter_does_not_hide_topk_dtype_conversion() -> None:
+    hidden, topk_ids, weights = _routing(4, 2, 4)
+    adapter = DeepEPDispatchAdapter(
+        _FakeDeepEPBuffer(4),
+        num_experts=4,
+        num_local_experts=4,
+        num_max_tokens_per_rank=16,
+    )
+    with pytest.raises(ValueError, match="topk_ids_dtype"):
+        adapter.dispatch(hidden, topk_ids, weights)
+
+
+@pytest.mark.smoke
+def test_deepep_adapter_rejects_non_device_prefix_sum() -> None:
+    hidden, topk_ids, weights = _routing(4, 2, 4)
+    topk_ids = topk_ids.to(torch.int64)
+
+    class BadBuffer:
+        def dispatch(self, hidden_states, **kwargs):
+            event = _FakeEvent()
+            handle = SimpleNamespace(psum_num_recv_tokens_per_expert=[1, 2, 3, 4])
+            recv_weights = torch.empty(
+                hidden_states.shape[0] * kwargs["topk_idx"].shape[1],
+                dtype=torch.float32,
+                device=hidden_states.device,
+            )
+            return hidden_states.repeat_interleave(2, 0), None, recv_weights, handle, event
+
+    adapter = DeepEPDispatchAdapter(
+        BadBuffer(),
+        num_experts=4,
+        num_local_experts=4,
+        num_max_tokens_per_rank=16,
+    )
+    with pytest.raises(TypeError, match="psum_num_recv_tokens_per_expert"):
+        adapter.dispatch(hidden, topk_ids, weights)
diff --git a/tests/ops/test_fused_gated.py b/tests/ops/test_fused_gated.py
index 7a6c11ff8..f4bf86721 100644
--- a/tests/ops/test_fused_gated.py
+++ b/tests/ops/test_fused_gated.py
@@ -88,6 +88,41 @@ def test_silu_and_mul_lazy_op_rebinds_shape() -> None:
         assert (op.M, op.N, op.dtype) == (m, n, torch.float16)
 
 
+@pytest.mark.smoke
+@pytest.mark.parametrize("strategy", ["direct", "explicit_parallel"])
+def test_silu_and_mul_supports_more_than_65535_rows(strategy: str) -> None:
+    """Large dispatched batches must not exceed CUDA's grid.y limit."""
+    m, n, dtype = 65_536, 128, torch.bfloat16
+    test = SiluAndMulTest(m, n, dtype)
+    op = SiluAndMulFwdOp(M=m, N=n, dtype=dtype, strategy=strategy)
+    atol, rtol = _get_tolerances(dtype)
+    test.check(op, *test.gen_inputs(), atol=atol, rtol=rtol)
+
+
+@pytest.mark.smoke
+@pytest.mark.parametrize("strategy", ["direct", "explicit_parallel"])
+def test_silu_and_mul_device_bounded_rows(strategy: str) -> None:
+    """A CUDA scalar bounds work while the compiled capacity stays fixed."""
+    m, n, valid = 32, 128, 11
+    x = torch.randn(m, 2 * n, device="cuda", dtype=torch.bfloat16)
+    x[valid:].fill_(float("nan"))
+    valid_rows = torch.tensor(valid, device="cuda", dtype=torch.int32)
+    kernel = SiluAndMulFwdKernel(
+        M=m,
+        N=n,
+        dtype=torch.bfloat16,
+        strategy=strategy,
+    )
+
+    output = kernel.forward_rows(x, valid_rows)
+    reference = F.silu(x[:valid, :n].float()) * x[:valid, n:].float()
+
+    assert torch.isfinite(output[:valid]).all()
+    assert torch.allclose(
+        output[:valid].float(), reference, atol=1.6e-2, rtol=1.6e-2
+    )
+
+
 # ---------------------------------------------------------------------------
 # GeluAndMul
 # ---------------------------------------------------------------------------
diff --git a/tileops/kernels/elementwise.py b/tileops/kernels/elementwise.py
index f715b4f64..483b61230 100644
--- a/tileops/kernels/elementwise.py
+++ b/tileops/kernels/elementwise.py
@@ -173,9 +173,7 @@
     torch.bfloat16,
     torch.float32,
 )
-_BINARY_NO_BOOL_DTYPES = tuple(
-    dt for dt in _BINARY_FULL_DTYPES if dt is not torch.bool
-)
+_BINARY_NO_BOOL_DTYPES = tuple(dt for dt in _BINARY_FULL_DTYPES if dt is not torch.bool)
 
 
 def _is_fp8(dtype: torch.dtype) -> bool:
@@ -260,7 +258,12 @@ def _clamp_to_dtype_range(value, dtype: torch.dtype):
         if isinstance(value, float) and math.isinf(value):
             iinfo = torch.iinfo(dtype)
             return iinfo.max if value > 0 else iinfo.min
-        if dtype == torch.uint8 and isinstance(value, int) and not isinstance(value, bool) and value < 0:
+        if (
+            dtype == torch.uint8
+            and isinstance(value, int)
+            and not isinstance(value, bool)
+            and value < 0
+        ):
             return value & 0xFF
         return int(value)
     fvalue = float(value)
@@ -307,20 +310,26 @@ def _wrap_fp8_accumulation(base_op, dtype, dtype_str, arity=1):
     if _fp8_needs_nonsaturating_cast(dtype):
         # e5m2: compute in fp16, leave result as fp16
         if arity == 1:
+
             def fp8_accum_op(x):
                 return base_op(T.cast(x, accum))
         else:
+
             def fp8_accum_op(a, b):
                 return base_op(T.cast(a, accum), T.cast(b, accum))
+
         return fp8_accum_op
 
     # e4m3fn: compute in fp16, saturating cast back
     if arity == 1:
+
         def fp8_accum_op(x):
             return T.Cast(dtype_str, base_op(T.cast(x, accum)))
     else:
+
         def fp8_accum_op(a, b):
             return T.Cast(dtype_str, base_op(T.cast(a, accum), T.cast(b, accum)))
+
     return fp8_accum_op
 
 
@@ -426,7 +435,12 @@ def _is_contiguous_same_shape(coalesced_shape, a_strides, b_strides):
 
 @functools.lru_cache(maxsize=32)
 def _make_binary_register_copy(
-    N_total, dtype, op_func, output_dtype=None, threads=256, num_per_thread=8,
+    N_total,
+    dtype,
+    op_func,
+    output_dtype=None,
+    threads=256,
+    num_per_thread=8,
 ):
     """Binary register_copy: fragment load -> compute -> fragment store.
 
@@ -451,12 +465,12 @@ def main(
                 a_reg = T.alloc_fragment((block_size,), dtype)
                 b_reg = T.alloc_fragment((block_size,), dtype)
                 y_reg = T.alloc_fragment((block_size,), out_dtype)
-                T.copy(a[bx * block_size:(bx + 1) * block_size], a_reg)
-                T.copy(b[bx * block_size:(bx + 1) * block_size], b_reg)
+                T.copy(a[bx * block_size : (bx + 1) * block_size], a_reg)
+                T.copy(b[bx * block_size : (bx + 1) * block_size], b_reg)
                 for i, j in T.Parallel(threads, num_per_thread):
                     idx = i * num_per_thread + j
                     y_reg[idx] = op_func(a_reg[idx], b_reg[idx])
-                T.copy(y_reg, y[bx * block_size:(bx + 1) * block_size])
+                T.copy(y_reg, y[bx * block_size : (bx + 1) * block_size])
 
         return main
 
@@ -465,14 +479,23 @@ def main(
 
 @functools.lru_cache(maxsize=32)
 def _make_binary_direct(
-    N_total, dtype, op_func, coalesced_shape, a_strides, b_strides,
-    a_numel, b_numel, output_dtype=None, threads=256,
+    N_total,
+    dtype,
+    op_func,
+    coalesced_shape,
+    a_strides,
+    b_strides,
+    a_numel,
+    b_numel,
+    output_dtype=None,
+    threads=256,
 ):
     """Binary direct: 1 element per thread with stride-based broadcast."""
     out_dtype = output_dtype or dtype
 
     # Fast path: same-shape contiguous inputs -- skip broadcast machinery
     if _is_contiguous_same_shape(coalesced_shape, a_strides, b_strides):
+
         @tilelang.jit(out_idx=[2])
         def kernel(threads):
             @T.prim_func
@@ -507,7 +530,11 @@ def main(
                 for i in T.Parallel(threads):
                     flat_idx = bx * threads + i
                     a_off, b_off = _compute_broadcast_offsets(
-                        flat_idx, ndim, divisors, a_strides, b_strides,
+                        flat_idx,
+                        ndim,
+                        divisors,
+                        a_strides,
+                        b_strides,
                     )
                     y[flat_idx] = op_func(a[a_off], b[b_off])
 
@@ -518,14 +545,24 @@ def main(
 
 @functools.lru_cache(maxsize=32)
 def _make_binary_explicit(
-    N_total, dtype, op_func, coalesced_shape, a_strides, b_strides,
-    a_numel, b_numel, output_dtype=None, threads=256, num_per_thread=8,
+    N_total,
+    dtype,
+    op_func,
+    coalesced_shape,
+    a_strides,
+    b_strides,
+    a_numel,
+    b_numel,
+    output_dtype=None,
+    threads=256,
+    num_per_thread=8,
 ):
     """Binary explicit_parallel: N elements per thread with stride-based broadcast."""
     out_dtype = output_dtype or dtype
 
     # Fast path: same-shape contiguous inputs -- skip broadcast machinery
     if _is_contiguous_same_shape(coalesced_shape, a_strides, b_strides):
+
         @tilelang.jit(out_idx=[2])
         def kernel(threads, num_per_thread):
             block_size = threads * num_per_thread
@@ -564,7 +601,11 @@ def main(
                 for i, j in T.Parallel(threads, num_per_thread):
                     flat_idx = (bx * threads + i) * num_per_thread + j
                     a_off, b_off = _compute_broadcast_offsets(
-                        flat_idx, ndim, divisors, a_strides, b_strides,
+                        flat_idx,
+                        ndim,
+                        divisors,
+                        a_strides,
+                        b_strides,
                     )
                     y[flat_idx] = op_func(a[a_off], b[b_off])
 
@@ -578,30 +619,61 @@ def main(
 # ---------------------------------------------------------------------------
 
 
-@functools.lru_cache(maxsize=32)
-def _make_fused_gated_direct(M, N, dtype, op_func, threads=256, output_dtype=None):
-    """FusedGated direct: 1 element per thread. x[:, :N] is gate, x[:, N:] is value.
+def _fused_gated_col(col_block, thread, item, threads, num_per_thread):
+    """Shared compile-time column mapping for all fused-gated row policies."""
+    return (col_block * threads + thread) * num_per_thread + item
 
-    ``op_func(gate, value)`` is the compound operation that applies the
-    activation to *gate* and multiplies by *value*.  For fp8 dtypes the
-    caller wraps it via ``_wrap_fp8_accumulation`` so this factory stays
-    fp8-agnostic.
 
-    Args:
-        output_dtype: TileLang dtype string for the output tensor. Defaults to dtype.
+def _make_fused_gated_store(N, op_func):
+    """Create the single fused-gated pointwise epilogue implementation."""
+
+    @T.macro
+    def store(x, y, row, col):
+        gate = x[row, col]
+        value = x[row, N + col]
+        y[row, col] = op_func(gate, value)
+
+    return store
+
+
+@functools.lru_cache(maxsize=32)
+def _make_fused_gated(
+    M,
+    N,
+    dtype,
+    op_func,
+    threads=256,
+    num_per_thread=8,
+    output_dtype=None,
+):
+    """Build the normal-row FusedGated strategy.
+
+    ``num_per_thread=1`` selects the direct strategy; larger static values
+    select explicit parallelism without duplicating mapping or epilogue code.
     """
+    block_N = threads * num_per_thread
     out_dtype = output_dtype or dtype
+    store = _make_fused_gated_store(N, op_func)
 
     @tilelang.jit(out_idx=[1])
-    def kernel(threads_arg):
+    def kernel(threads_arg, npt_arg):
         @T.prim_func
         def main(x: T.Tensor((M, 2 * N), dtype), y: T.Tensor((M, N), out_dtype)):
-            with T.Kernel(T.ceildiv(N, threads_arg), M, threads=threads_arg) as (bx, by):
-                for i in T.Parallel(threads_arg):
-                    col = bx * threads_arg + i
-                    gate = x[by, col]
-                    value = x[by, N + col]
-                    y[by, col] = op_func(gate, value)
+            if M <= 65535:
+                with T.Kernel(T.ceildiv(N, block_N), M, threads=threads_arg) as (bx, by):
+                    for i, j in T.Parallel(threads_arg, npt_arg):
+                        col = _fused_gated_col(bx, i, j, threads_arg, npt_arg)
+                        store(x, y, by, col)
+            else:
+                col_blocks = T.ceildiv(N, block_N)
+                with T.Kernel(M * col_blocks, threads=threads_arg) as bx:
+                    row = bx // col_blocks
+                    col_block = bx % col_blocks
+                    for i, j in T.Parallel(threads_arg, npt_arg):
+                        col = _fused_gated_col(
+                            col_block, i, j, threads_arg, npt_arg
+                        )
+                        store(x, y, row, col)
 
         return main
 
@@ -609,31 +681,46 @@ def main(x: T.Tensor((M, 2 * N), dtype), y: T.Tensor((M, N), out_dtype)):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_fused_gated_explicit(M, N, dtype, op_func, threads=256, num_per_thread=8,
-                               output_dtype=None):
-    """FusedGated explicit_parallel: N elements per thread.
-
-    ``op_func(gate, value)`` is the compound operation (see
-    ``_make_fused_gated_direct``).  fp8 accumulation is handled by the
-    caller wrapping ``op_func`` via ``_wrap_fp8_accumulation``, so this
-    factory no longer needs an ``fp8_accum`` parameter.
+def _make_fused_gated_bounded(
+    M,
+    N,
+    dtype,
+    op_func,
+    grid_rows,
+    threads=256,
+    num_per_thread=8,
+    output_dtype=None,
+):
+    """Persistent FusedGated with a device row count.
 
-    Args:
-        output_dtype: TileLang dtype string for the output tensor. Defaults to dtype.
+    ``num_per_thread=1`` is the direct strategy. Keeping that choice static
+    shares the pointwise source without adding a runtime bounded/strategy
+    branch.
     """
     block_N = threads * num_per_thread
     out_dtype = output_dtype or dtype
+    store = _make_fused_gated_store(N, op_func)
 
-    @tilelang.jit(out_idx=[1])
+    @tilelang.jit(out_idx=[2])
     def kernel(threads_arg, npt_arg):
         @T.prim_func
-        def main(x: T.Tensor((M, 2 * N), dtype), y: T.Tensor((M, N), out_dtype)):
-            with T.Kernel(T.ceildiv(N, block_N), M, threads=threads_arg) as (bx, by):
-                for i, j in T.Parallel(threads_arg, npt_arg):
-                    col = (bx * threads_arg + i) * npt_arg + j
-                    gate = x[by, col]
-                    value = x[by, N + col]
-                    y[by, col] = op_func(gate, value)
+        def main(
+            x: T.Tensor((M, 2 * N), dtype),
+            valid_rows: T.Tensor((1,), "int32"),
+            y: T.Tensor((M, N), out_dtype),
+        ):
+            col_blocks = T.ceildiv(N, block_N)
+            with T.Kernel(grid_rows * col_blocks, threads=threads_arg) as bx:
+                worker_row = bx // col_blocks
+                col_block = bx % col_blocks
+                for row_iter in T.serial(T.ceildiv(M, grid_rows)):
+                    row = worker_row + row_iter * grid_rows
+                    for i, j in T.Parallel(threads_arg, npt_arg):
+                        col = _fused_gated_col(
+                            col_block, i, j, threads_arg, npt_arg
+                        )
+                        if row < valid_rows[0]:
+                            store(x, y, row, col)
 
         return main
 
@@ -695,7 +782,9 @@ def __init__(self, N_total, dtype, strategy=None, config=None, tune=False):
         # the CUDA codegen cannot lower. Keep bool inputs on the scalar path.
         bool_output = torch.bool == self.OUTPUT_DTYPE
         bool_output_needs_scalar = bool_output and dtype in (
-            torch.uint8, torch.int8, torch.int16,
+            torch.uint8,
+            torch.int8,
+            torch.int16,
         )
         if dtype == torch.bool:
             if strategy is not None and strategy != "direct":
@@ -747,20 +836,29 @@ def _build_kernel(self, strategy):
         effective_op = self._get_effective_op_func()
         if strategy == "direct":
             return _make_unary_direct(
-                self.N_total, self.dtype_str, effective_op,
-                output_dtype=self.output_dtype_str, threads=cfg["threads"],
+                self.N_total,
+                self.dtype_str,
+                effective_op,
+                output_dtype=self.output_dtype_str,
+                threads=cfg["threads"],
             )
         elif strategy == "explicit_parallel":
             return _make_unary_explicit(
-                self.N_total, self.dtype_str, effective_op,
+                self.N_total,
+                self.dtype_str,
+                effective_op,
                 output_dtype=self.output_dtype_str,
-                threads=cfg["threads"], num_per_thread=cfg["num_per_thread"],
+                threads=cfg["threads"],
+                num_per_thread=cfg["num_per_thread"],
             )
         elif strategy == "register_copy":
             return _make_unary_regcopy(
-                self.N_total, self.dtype_str, effective_op,
+                self.N_total,
+                self.dtype_str,
+                effective_op,
                 output_dtype=self.output_dtype_str,
-                threads=cfg["threads"], num_per_thread=cfg["num_per_thread"],
+                threads=cfg["threads"],
+                num_per_thread=cfg["num_per_thread"],
             )
         else:
             raise ValueError(f"Unknown strategy: {strategy}")
@@ -792,11 +890,7 @@ def autotune_configs(self) -> list[dict]:
             # fp16 / bf16 / fp32
             threads_opts = [128, 256, 512]
             npt_opts = [2, 4, 8]
-        return [
-            {"threads": t, "num_per_thread": n}
-            for t in threads_opts
-            for n in npt_opts
-        ]
+        return [{"threads": t, "num_per_thread": n} for t in threads_opts for n in npt_opts]
 
     def autotune(self, warmup: int = 10, rep: int = 10) -> None:
         """Override to handle serialization failures in the TileLang autotuner.
@@ -872,8 +966,17 @@ def op_func(a, b):
         raise NotImplementedError
 
     def __init__(
-        self, N_total, dtype, coalesced_shape, a_strides, b_strides,
-        a_numel, b_numel, strategy=None, config=None, tune=False,
+        self,
+        N_total,
+        dtype,
+        coalesced_shape,
+        a_strides,
+        b_strides,
+        a_numel,
+        b_numel,
+        strategy=None,
+        config=None,
+        tune=False,
     ):
         super().__init__()
         if self.SUPPORTED_DTYPES is not None and dtype not in self.SUPPORTED_DTYPES:
@@ -895,22 +998,24 @@ def __init__(
         self.a_numel = a_numel
         self.b_numel = b_numel
         self._same_shape = _is_contiguous_same_shape(
-            coalesced_shape, a_strides, b_strides,
+            coalesced_shape,
+            a_strides,
+            b_strides,
         )
         # Validate a caller-provided strategy up front so typos raise the
         # same ValueError regardless of dtype (the bool override below
         # otherwise silently accepts an unknown strategy for bool inputs).
         if strategy is not None and strategy not in self.STRATEGIES:
-            raise ValueError(
-                f"Unknown strategy '{strategy}', expected one of {self.STRATEGIES}"
-            )
+            raise ValueError(f"Unknown strategy '{strategy}', expected one of {self.STRATEGIES}")
         # torch.bool maps to TileLang ``boolx<N>`` for vectorised loads /
         # stores, which the CUDA codegen cannot lower. Force the scalar
         # ``direct`` strategy for bool inputs regardless of caller request.
         bool_input = dtype == torch.bool
         bool_output = torch.bool == self.OUTPUT_DTYPE
         bool_output_needs_scalar = bool_output and dtype in (
-            torch.uint8, torch.int8, torch.int16,
+            torch.uint8,
+            torch.int8,
+            torch.int16,
         )
         if bool_input:
             if strategy is not None and strategy != "direct":
@@ -976,24 +1081,39 @@ def _build_kernel(self, strategy):
             kernel_output_dtype = _fp8_accum_dtype_str()
         if strategy == "direct":
             return _make_binary_direct(
-                self.N_total, self.dtype_str, effective_op,
-                self.coalesced_shape, self.a_strides, self.b_strides,
-                self.a_numel, self.b_numel,
-                output_dtype=kernel_output_dtype, threads=cfg["threads"],
+                self.N_total,
+                self.dtype_str,
+                effective_op,
+                self.coalesced_shape,
+                self.a_strides,
+                self.b_strides,
+                self.a_numel,
+                self.b_numel,
+                output_dtype=kernel_output_dtype,
+                threads=cfg["threads"],
             )
         elif strategy == "explicit_parallel":
             return _make_binary_explicit(
-                self.N_total, self.dtype_str, effective_op,
-                self.coalesced_shape, self.a_strides, self.b_strides,
-                self.a_numel, self.b_numel,
+                self.N_total,
+                self.dtype_str,
+                effective_op,
+                self.coalesced_shape,
+                self.a_strides,
+                self.b_strides,
+                self.a_numel,
+                self.b_numel,
                 output_dtype=kernel_output_dtype,
-                threads=cfg["threads"], num_per_thread=cfg["num_per_thread"],
+                threads=cfg["threads"],
+                num_per_thread=cfg["num_per_thread"],
             )
         elif strategy == "register_copy":
             return _make_binary_register_copy(
-                self.N_total, self.dtype_str, effective_op,
+                self.N_total,
+                self.dtype_str,
+                effective_op,
                 output_dtype=kernel_output_dtype,
-                threads=cfg["threads"], num_per_thread=cfg["num_per_thread"],
+                threads=cfg["threads"],
+                num_per_thread=cfg["num_per_thread"],
             )
         else:
             raise ValueError(f"Unknown strategy: {strategy}")
@@ -1020,11 +1140,7 @@ def autotune_configs(self) -> list[dict]:
             # fp16 / bf16 / fp32
             threads_opts = [128, 256, 512]
             npt_opts = [2, 4, 8]
-        return [
-            {"threads": t, "num_per_thread": n}
-            for t in threads_opts
-            for n in npt_opts
-        ]
+        return [{"threads": t, "num_per_thread": n} for t in threads_opts for n in npt_opts]
 
     def autotune(self, warmup: int = 10, rep: int = 10) -> None:
         """Override to handle known TileLang autotuner fallback failures.
@@ -1047,7 +1163,8 @@ def autotune(self, warmup: int = 10, rep: int = 10) -> None:
             ):
                 warnings.warn(  # noqa: B028
                     f"{self.__class__.__name__} autotuning failed "
-                    f"({message}); falling back to default_config.")
+                    f"({message}); falling back to default_config."
+                )
                 self.config = dict(self.default_config)
             else:
                 raise
@@ -1144,15 +1261,23 @@ def _build_kernel(self, strategy):
         cfg = self.default_config
         effective_op = self._get_effective_op_func()
         if strategy == "direct":
-            return _make_fused_gated_direct(
-                self.M, self.N, self.dtype_str, effective_op,
+            return _make_fused_gated(
+                self.M,
+                self.N,
+                self.dtype_str,
+                effective_op,
                 threads=cfg["threads"],
+                num_per_thread=1,
                 output_dtype=self._kernel_output_dtype,
             )
         elif strategy == "explicit_parallel":
-            return _make_fused_gated_explicit(
-                self.M, self.N, self.dtype_str, effective_op,
-                cfg["threads"], cfg["num_per_thread"],
+            return _make_fused_gated(
+                self.M,
+                self.N,
+                self.dtype_str,
+                effective_op,
+                cfg["threads"],
+                cfg["num_per_thread"],
                 output_dtype=self._kernel_output_dtype,
             )
         else:
@@ -1184,11 +1309,7 @@ def autotune_configs(self) -> list[dict]:
             # fp16 / bf16 / fp32
             threads_opts = [128, 256, 512]
             npt_opts = [2, 4, 8]
-        return [
-            {"threads": t, "num_per_thread": n}
-            for t in threads_opts
-            for n in npt_opts
-        ]
+        return [{"threads": t, "num_per_thread": n} for t in threads_opts for n in npt_opts]
 
     def autotune(self, warmup: int = 10, rep: int = 10) -> None:
         """Override to handle serialization failures in the TileLang autotuner.
@@ -1220,9 +1341,10 @@ def init_config(self, config=None, tune=False):
         # to avoid JIT lookup overhead on every forward() call.
         cfg = self.config
         if self.strategy == "direct":
-            self._compiled_fn = self.kernel(cfg["threads"])
+            self._compiled_fn = self.kernel(cfg["threads"], 1)
         else:
             self._compiled_fn = self.kernel(cfg["threads"], cfg["num_per_thread"])
+        self._bounded_compiled_fn = None
 
     def forward(self, x):
         result = self._compiled_fn(x)
@@ -1230,6 +1352,51 @@ def forward(self, x):
             result = result.to(self._fp8_output_dtype)
         return result
 
+    def forward_rows(self, x, valid_rows):
+        """Apply the gated activation only to ``valid_rows`` leading rows.
+
+        ``valid_rows`` remains a CUDA int32 tensor so one compiled capacity
+        can be replayed with different received row counts without host sync.
+        Rows in the returned capacity tail are unspecified.
+        """
+        if valid_rows.dtype != torch.int32 or valid_rows.numel() != 1:
+            raise ValueError("valid_rows must be a one-element torch.int32 tensor")
+        if valid_rows.device != x.device:
+            raise ValueError("valid_rows and x must be on the same device")
+        if self._bounded_compiled_fn is None:
+            cfg = self.config
+            effective_op = self._get_effective_op_func()
+            sm_count = torch.cuda.get_device_properties(x.device).multi_processor_count
+            grid_rows = min(self.M, 16 * sm_count)
+            if self.strategy == "direct":
+                bounded_kernel = _make_fused_gated_bounded(
+                    self.M,
+                    self.N,
+                    self.dtype_str,
+                    effective_op,
+                    grid_rows,
+                    threads=cfg["threads"],
+                    num_per_thread=1,
+                    output_dtype=self._kernel_output_dtype,
+                )
+                self._bounded_compiled_fn = bounded_kernel(cfg["threads"], 1)
+            else:
+                bounded_kernel = _make_fused_gated_bounded(
+                    self.M,
+                    self.N,
+                    self.dtype_str,
+                    effective_op,
+                    grid_rows,
+                    cfg["threads"],
+                    cfg["num_per_thread"],
+                    output_dtype=self._kernel_output_dtype,
+                )
+                self._bounded_compiled_fn = bounded_kernel(cfg["threads"], cfg["num_per_thread"])
+        result = self._bounded_compiled_fn(x, valid_rows.reshape(1))
+        if self._fp8_output_dtype is not None:
+            result = result.to(self._fp8_output_dtype)
+        return result
+
 
 # ---------------------------------------------------------------------------
 # Concrete kernel subclasses
@@ -1308,8 +1475,18 @@ def op_func(a, b):
         )
 
     def __init__(
-        self, N_total, dtype, coalesced_shape, a_strides, b_strides,
-        a_numel, b_numel, strategy=None, config=None, tune=False, alpha=1,
+        self,
+        N_total,
+        dtype,
+        coalesced_shape,
+        a_strides,
+        b_strides,
+        a_numel,
+        b_numel,
+        strategy=None,
+        config=None,
+        tune=False,
+        alpha=1,
     ):
         # PyTorch's torch.add / torch.sub reject a floating alpha when the
         # input tensor is integral (or bool). Mirror that contract here so
@@ -1319,13 +1496,19 @@ def __init__(
         # alpha=-1 → 255; bool alpha=2 → True via low-bit). The kernel's
         # T.cast(int(alpha), a.dtype) reproduces that wrap.
         if dtype in _BITWISE_DTYPES and float(alpha) != float(int(alpha)):
-            raise ValueError(
-                "alpha must be an integer when input dtype is integral"
-            )
+            raise ValueError("alpha must be an integer when input dtype is integral")
         self._alpha = alpha
         super().__init__(
-            N_total, dtype, coalesced_shape, a_strides, b_strides,
-            a_numel, b_numel, strategy=strategy, config=config, tune=tune,
+            N_total,
+            dtype,
+            coalesced_shape,
+            a_strides,
+            b_strides,
+            a_numel,
+            b_numel,
+            strategy=strategy,
+            config=config,
+            tune=tune,
         )
 
     def _alpha_op_func(self):
@@ -1517,13 +1700,31 @@ def op_func(a, b):
         raise NotImplementedError("Use _make_lerp_op_func(weight) instead")
 
     def __init__(
-        self, N_total, dtype, coalesced_shape, a_strides, b_strides,
-        a_numel, b_numel, strategy=None, config=None, tune=False, weight=0.5,
+        self,
+        N_total,
+        dtype,
+        coalesced_shape,
+        a_strides,
+        b_strides,
+        a_numel,
+        b_numel,
+        strategy=None,
+        config=None,
+        tune=False,
+        weight=0.5,
     ):
         self._weight = weight
         super().__init__(
-            N_total, dtype, coalesced_shape, a_strides, b_strides,
-            a_numel, b_numel, strategy=strategy, config=config, tune=tune,
+            N_total,
+            dtype,
+            coalesced_shape,
+            a_strides,
+            b_strides,
+            a_numel,
+            b_numel,
+            strategy=strategy,
+            config=config,
+            tune=tune,
         )
 
     def _build_kernel(self, strategy):
@@ -1535,7 +1736,10 @@ def lerp_func(a, b):
 
         # Wrap with fp8 accumulation via shared helper
         effective_op = _wrap_fp8_accumulation(
-            lerp_func, self.dtype, self.dtype_str, arity=2,
+            lerp_func,
+            self.dtype,
+            self.dtype_str,
+            arity=2,
         )
 
         # For e5m2: kernel output is fp16 (non-saturating path)
@@ -1548,24 +1752,39 @@ def lerp_func(a, b):
         cfg = self.default_config
         if strategy == "direct":
             return _make_binary_direct(
-                self.N_total, self.dtype_str, effective_op,
-                self.coalesced_shape, self.a_strides, self.b_strides,
-                self.a_numel, self.b_numel,
-                output_dtype=kernel_output_dtype, threads=cfg["threads"],
+                self.N_total,
+                self.dtype_str,
+                effective_op,
+                self.coalesced_shape,
+                self.a_strides,
+                self.b_strides,
+                self.a_numel,
+                self.b_numel,
+                output_dtype=kernel_output_dtype,
+                threads=cfg["threads"],
             )
         elif strategy == "explicit_parallel":
             return _make_binary_explicit(
-                self.N_total, self.dtype_str, effective_op,
-                self.coalesced_shape, self.a_strides, self.b_strides,
-                self.a_numel, self.b_numel,
+                self.N_total,
+                self.dtype_str,
+                effective_op,
+                self.coalesced_shape,
+                self.a_strides,
+                self.b_strides,
+                self.a_numel,
+                self.b_numel,
                 output_dtype=kernel_output_dtype,
-                threads=cfg["threads"], num_per_thread=cfg["num_per_thread"],
+                threads=cfg["threads"],
+                num_per_thread=cfg["num_per_thread"],
             )
         elif strategy == "register_copy":
             return _make_binary_register_copy(
-                self.N_total, self.dtype_str, effective_op,
+                self.N_total,
+                self.dtype_str,
+                effective_op,
                 output_dtype=kernel_output_dtype,
-                threads=cfg["threads"], num_per_thread=cfg["num_per_thread"],
+                threads=cfg["threads"],
+                num_per_thread=cfg["num_per_thread"],
             )
         else:
             raise ValueError(f"Unknown strategy: {strategy}")
@@ -2345,7 +2564,8 @@ def __init__(self, N_total, dtype, config=None, tune=False):
         if not self._skip_fp8_output:
             builder_kwargs["output_dtype"] = self.dtype_to_str(self.output_dtype)
         self.kernel = self._builder_fn()(
-            *self._builder_positional_args(), **builder_kwargs,
+            *self._builder_positional_args(),
+            **builder_kwargs,
         )
         self.init_config(config, tune)
 
@@ -2397,8 +2617,9 @@ def forward(self, x):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_leaky_relu_kernel(N, dtype, negative_slope, output_dtype=None,
-                            is_fp8=False, threads=256, npt=8):
+def _make_leaky_relu_kernel(
+    N, dtype, negative_slope, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build leaky_relu kernel: y = x if x > 0 else negative_slope * x.
 
     For non-fp8 dtypes, uses register_copy strategy: fragment load -> compute
@@ -2467,8 +2688,7 @@ def _builder_args(self):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_elu_kernel(N, dtype, alpha, output_dtype=None, is_fp8=False,
-                     threads=256, npt=8):
+def _make_elu_kernel(N, dtype, alpha, output_dtype=None, is_fp8=False, threads=256, npt=8):
     """Build ELU kernel: y = x if x > 0 else alpha * (exp(x) - 1).
 
     For non-fp8 dtypes, uses register_copy strategy: fragment load -> compute
@@ -2495,7 +2715,11 @@ def main(x: T.Tensor((N,), dtype), y: T.Tensor((N,), out_dtype)):
                             a = T.cast(alpha, "float32")
                             one = T.cast(1.0, "float32")
                             v32 = T.cast(val, "float32")
-                            y[idx] = T.if_then_else(v32 > zero, T.Cast(out_dtype, v32), T.Cast(out_dtype, a * (T.exp(v32) - one)))
+                            y[idx] = T.if_then_else(
+                                v32 > zero,
+                                T.Cast(out_dtype, v32),
+                                T.Cast(out_dtype, a * (T.exp(v32) - one)),
+                            )
 
             return main
     else:
@@ -2515,7 +2739,9 @@ def main(x: T.Tensor((N,), dtype), y: T.Tensor((N,), dtype)):
                         one = T.cast(1.0, "float32")
                         v32 = T.cast(val, "float32")
                         y_reg[i * npt_arg + j] = T.if_then_else(
-                            v32 > zero, val, T.Cast(val.dtype, a * (T.exp(v32) - one)),
+                            v32 > zero,
+                            val,
+                            T.Cast(val.dtype, a * (T.exp(v32) - one)),
                         )
                     T.copy(y_reg, y[bx * block_size : (bx + 1) * block_size])
 
@@ -2540,8 +2766,9 @@ def _builder_args(self):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_hardtanh_kernel(N, dtype, min_val, max_val, output_dtype=None,
-                          is_fp8=False, threads=256, npt=8):
+def _make_hardtanh_kernel(
+    N, dtype, min_val, max_val, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build hardtanh kernel: y = clamp(x, min_val, max_val).
 
     For non-fp8 dtypes, uses register_copy strategy: fragment load -> compute
@@ -2610,8 +2837,9 @@ def _builder_args(self):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_softplus_kernel(N, dtype, beta, threshold, output_dtype=None,
-                          is_fp8=False, threads=256, npt=8):
+def _make_softplus_kernel(
+    N, dtype, beta, threshold, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build softplus kernel: y = log(1 + exp(x*beta))/beta if x*beta <= threshold else x.
 
     For non-fp8 dtypes, uses register_copy strategy: fragment load -> compute
@@ -2640,7 +2868,9 @@ def main(x: T.Tensor((N,), dtype), y: T.Tensor((N,), out_dtype)):
                             one = T.cast(1.0, "float32")
                             scaled = v32 * b
                             sp = T.log(one + T.exp(scaled)) / b
-                            y[idx] = T.if_then_else(scaled > t, T.Cast(out_dtype, v32), T.Cast(out_dtype, sp))
+                            y[idx] = T.if_then_else(
+                                scaled > t, T.Cast(out_dtype, v32), T.Cast(out_dtype, sp)
+                            )
 
             return main
     else:
@@ -2662,7 +2892,9 @@ def main(x: T.Tensor((N,), dtype), y: T.Tensor((N,), dtype)):
                         scaled = v32 * b
                         sp = T.log(one + T.exp(scaled)) / b
                         y_reg[i * npt_arg + j] = T.if_then_else(
-                            scaled > t, val, T.Cast(val.dtype, sp),
+                            scaled > t,
+                            val,
+                            T.Cast(val.dtype, sp),
                         )
                     T.copy(y_reg, y[bx * block_size : (bx + 1) * block_size])
 
@@ -2688,8 +2920,9 @@ def _builder_args(self):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_prelu_kernel(N, C, inner_size, dtype, output_dtype=None,
-                       is_fp8=False, threads=256, npt=8):
+def _make_prelu_kernel(
+    N, C, inner_size, dtype, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build PReLU kernel: y = x if x > 0 else weight[channel] * x.
 
     Weight is per-channel. Channel index follows PyTorch convention:
@@ -2727,7 +2960,9 @@ def main(
                             v = T.cast(val, accum)
                             wf = T.cast(w, accum)
                             zero = T.cast(0, accum)
-                            y[idx] = T.if_then_else(v > zero, T.Cast(out_dtype, v), T.Cast(out_dtype, wf * v))
+                            y[idx] = T.if_then_else(
+                                v > zero, T.Cast(out_dtype, v), T.Cast(out_dtype, wf * v)
+                            )
 
             return main
     else:
@@ -2812,7 +3047,9 @@ def main(
                         idx = (bx * threads_arg + i) * npt_arg + j
                         if idx < N:
                             out[idx] = T.if_then_else(
-                                cond[idx] != T.cast(0, "uint8"), x[idx], y_in[idx],
+                                cond[idx] != T.cast(0, "uint8"),
+                                x[idx],
+                                y_in[idx],
                             )
 
             return main
@@ -2837,7 +3074,9 @@ def main(
                     for i, j in T.Parallel(threads_arg, npt_arg):
                         k = i * npt_arg + j
                         x_reg[k] = T.if_then_else(
-                            c_reg[k] != T.cast(0, "uint8"), x_reg[k], y_reg[k],
+                            c_reg[k] != T.cast(0, "uint8"),
+                            x_reg[k],
+                            y_reg[k],
                         )
                     T.copy(x_reg, out[bx * block_size : (bx + 1) * block_size])
 
@@ -2861,8 +3100,7 @@ def forward(self, cond, x, y):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_lerp_tensor_kernel(N, dtype, output_dtype=None, is_fp8=False,
-                             threads=256, npt=8):
+def _make_lerp_tensor_kernel(N, dtype, output_dtype=None, is_fp8=False, threads=256, npt=8):
     """Build Tensor-weight lerp kernel: out = a + weight * (b - a).
 
     The Op layer pre-broadcasts ``input`` / ``end`` / ``weight`` to the
@@ -2932,8 +3170,18 @@ def forward(self, a, b, w):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_clamp_kernel(N, dtype, has_min, has_max, min_val, max_val,
-                       output_dtype=None, is_fp8=False, threads=256, npt=8):
+def _make_clamp_kernel(
+    N,
+    dtype,
+    has_min,
+    has_max,
+    min_val,
+    max_val,
+    output_dtype=None,
+    is_fp8=False,
+    threads=256,
+    npt=8,
+):
     """Build clamp kernel: y = clamp(x, min_val, max_val) with optional bounds.
 
     For non-fp8 dtypes, uses register_copy strategy: fragment load -> compute
@@ -3015,9 +3263,9 @@ def _builder_args(self):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_clamp_tensor_kernel(N, dtype, has_min, has_max,
-                              output_dtype=None, is_fp8=False,
-                              threads=256, npt=8):
+def _make_clamp_tensor_kernel(
+    N, dtype, has_min, has_max, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build Tensor-bound clamp kernel.
 
     Inputs (all flat, length N, pre-broadcast/expanded by the Op layer):
@@ -3047,6 +3295,7 @@ def _make_clamp_tensor_kernel(N, dtype, has_min, has_max,
 
     if is_fp8:
         if has_min and has_max:
+
             @tilelang.jit(out_idx=[3])
             def kernel(threads_arg, npt_arg):
                 @T.prim_func
@@ -3076,6 +3325,7 @@ def main(
 
             return kernel
         if has_min:
+
             @tilelang.jit(out_idx=[2])
             def kernel(threads_arg, npt_arg):
                 @T.prim_func
@@ -3129,6 +3379,7 @@ def main(
 
     # non-fp8 path (register_copy)
     if has_min and has_max:
+
         @tilelang.jit(out_idx=[3])
         def kernel(threads_arg, npt_arg):
             @T.prim_func
@@ -3164,6 +3415,7 @@ def main(
 
         return kernel
     if has_min:
+
         @tilelang.jit(out_idx=[2])
         def kernel(threads_arg, npt_arg):
             @T.prim_func
@@ -3236,8 +3488,7 @@ class ClampTensorFwdKernel(ParametricUnaryKernel):
 
     _DEFAULT_THREADS = 512
 
-    def __init__(self, N_total, dtype, has_min, has_max,
-                 config=None, tune=False):
+    def __init__(self, N_total, dtype, has_min, has_max, config=None, tune=False):
         if not (has_min or has_max):
             raise ValueError(
                 "ClampTensorFwdKernel requires has_min or has_max to be True",
@@ -3266,8 +3517,9 @@ def forward(self, x, lo=None, hi=None):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_masked_fill_kernel(N, dtype, fill_value, output_dtype=None,
-                             is_fp8=False, threads=256, npt=8):
+def _make_masked_fill_kernel(
+    N, dtype, fill_value, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build masked_fill kernel: out = mask ? fill_value : x.
 
     The Op layer packs the bool mask as uint8 so that T.copy can
@@ -3303,7 +3555,9 @@ def main(
                             fv = T.cast(fill_value, out_dtype)
                             x_val = T.Cast(out_dtype, x[idx])
                             out[idx] = T.if_then_else(
-                                mask[idx] != T.cast(0, "uint8"), fv, x_val,
+                                mask[idx] != T.cast(0, "uint8"),
+                                fv,
+                                x_val,
                             )
 
             return main
@@ -3326,7 +3580,9 @@ def main(
                         k = i * npt_arg + j
                         fv = T.cast(fill_value, dtype)
                         x_reg[k] = T.if_then_else(
-                            m_reg[k] != T.cast(0, "uint8"), fv, x_reg[k],
+                            m_reg[k] != T.cast(0, "uint8"),
+                            fv,
+                            x_reg[k],
                         )
                     T.copy(x_reg, out[bx * block_size : (bx + 1) * block_size])
 
@@ -3367,8 +3623,9 @@ def forward(self, x, mask):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_masked_fill_tensor_value_kernel(N, dtype, output_dtype=None,
-                                          is_fp8=False, threads=256, npt=8):
+def _make_masked_fill_tensor_value_kernel(
+    N, dtype, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build masked_fill kernel with a 0-dim Tensor fill value.
 
     Inputs (all flat, length N, pre-broadcast/expanded by the Op layer):
@@ -3384,6 +3641,7 @@ def _make_masked_fill_tensor_value_kernel(N, dtype, output_dtype=None,
     block_size = threads * npt
 
     if is_fp8:
+
         @tilelang.jit(out_idx=[3])
         def kernel(threads_arg, npt_arg):
             @T.prim_func
@@ -3400,7 +3658,9 @@ def main(
                         if idx < N:
                             x_val = T.Cast(out_dtype, x[idx])
                             out[idx] = T.if_then_else(
-                                mask[idx] != T.cast(0, "uint8"), fv, x_val,
+                                mask[idx] != T.cast(0, "uint8"),
+                                fv,
+                                x_val,
                             )
 
             return main
@@ -3425,7 +3685,9 @@ def main(
                 for i, j in T.Parallel(threads_arg, npt_arg):
                     k = i * npt_arg + j
                     x_reg[k] = T.if_then_else(
-                        m_reg[k] != T.cast(0, "uint8"), fv, x_reg[k],
+                        m_reg[k] != T.cast(0, "uint8"),
+                        fv,
+                        x_reg[k],
                     )
                 T.copy(x_reg, out[bx * block_size : (bx + 1) * block_size])
 
@@ -3460,8 +3722,9 @@ def forward(self, x, mask, value):
 
 
 @functools.lru_cache(maxsize=32)
-def _make_nan_to_num_kernel(N, dtype, nan_val, posinf_val, neginf_val,
-                            output_dtype=None, is_fp8=False, threads=256, npt=8):
+def _make_nan_to_num_kernel(
+    N, dtype, nan_val, posinf_val, neginf_val, output_dtype=None, is_fp8=False, threads=256, npt=8
+):
     """Build nan_to_num kernel: replace NaN, +Inf, -Inf with given values.
 
     For non-fp8 dtypes, uses register_copy strategy: fragment load -> compute
@@ -3538,8 +3801,9 @@ def main(x: T.Tensor((N,), dtype), y: T.Tensor((N,), dtype)):
 class NanToNumFwdKernel(ParametricUnaryKernel):
     """NanToNum: replace NaN, +Inf, -Inf with specified values."""
 
-    def __init__(self, N_total, dtype, nan_val=0.0, posinf_val=1e4, neginf_val=-1e4,
-                 config=None, tune=False):
+    def __init__(
+        self, N_total, dtype, nan_val=0.0, posinf_val=1e4, neginf_val=-1e4, config=None, tune=False
+    ):
         self._raw_nan_val = nan_val
         self._raw_posinf_val = posinf_val
         self._raw_neginf_val = neginf_val
@@ -3583,7 +3847,11 @@ def main(out: T.Tensor((N_total,), dtype)):
                         row = rem // seq_len
                         col = rem % seq_len
                         # slope = 2^(-8 * (h+1) / num_heads)
-                        exp_val = T.cast(-8.0, "float32") * T.cast(h + 1, "float32") / T.cast(num_heads, "float32")
+                        exp_val = (
+                            T.cast(-8.0, "float32")
+                            * T.cast(h + 1, "float32")
+                            / T.cast(num_heads, "float32")
+                        )
                         slope = T.exp2(exp_val)
                         dist = T.cast(row - col, "float32")
                         # Use abs via if_then_else since T.abs may not handle int
@@ -3626,8 +3894,11 @@ def __init__(self, seq_len, num_heads, dtype, config=None, tune=False):
         self._fp8_output_dtype, self.output_dtype = _get_fp8_output_dtypes(dtype)
         cfg = self.default_config
         self.kernel = _make_alibi_kernel(
-            seq_len, num_heads, self.dtype_to_str(self.output_dtype),
-            cfg["threads"], cfg["num_per_thread"],
+            seq_len,
+            num_heads,
+            self.dtype_to_str(self.output_dtype),
+            cfg["threads"],
+            cfg["num_per_thread"],
         )
         self.init_config(config, tune)
 
@@ -3671,7 +3942,11 @@ def main(out: T.Tensor((N_total,), dtype)):
                         dim_pair = dim // 2
                         # angle = pos / 10000^(2*dim_pair / d_model)
                         base = T.cast(10000.0, "float32")
-                        exp_frac = T.cast(dim_pair, "float32") * T.cast(2.0, "float32") / T.cast(d_model, "float32")
+                        exp_frac = (
+                            T.cast(dim_pair, "float32")
+                            * T.cast(2.0, "float32")
+                            / T.cast(d_model, "float32")
+                        )
                         divisor = T.pow(base, exp_frac)
                         angle = T.cast(pos, "float32") / divisor
                         # Even dim -> sin, odd dim -> cos
@@ -3715,8 +3990,11 @@ def __init__(self, seq_len, d_model, dtype, config=None, tune=False):
         self._fp8_output_dtype, self.output_dtype = _get_fp8_output_dtypes(dtype)
         cfg = self.default_config
         self.kernel = _make_sinusoidal_kernel(
-            seq_len, d_model, self.dtype_to_str(self.output_dtype),
-            cfg["threads"], cfg["num_per_thread"],
+            seq_len,
+            d_model,
+            self.dtype_to_str(self.output_dtype),
+            cfg["threads"],
+            cfg["num_per_thread"],
         )
         self.init_config(config, tune)
 
diff --git a/tileops/ops/moe/__init__.py b/tileops/ops/moe/__init__.py
index 57f9c26f4..34069f0cd 100644
--- a/tileops/ops/moe/__init__.py
+++ b/tileops/ops/moe/__init__.py
@@ -5,9 +5,16 @@
 from .permute_align import MoePermuteAlignFwdOp
 from .prepare_finalize.no_dp_ep import MoEPrepareAndFinalizeNoDPEP
 from .routed_expert import (
+    DeepEPDispatchAdapter,
+    DispatchedExpertMLPFwdOp,
+    ExpertBatch,
+    ExpertBatchOutput,
+    ExpertDispatchResult,
     FusedMoEExperts,
     FusedMoEExpertsModular,
     FusedMoEExpertsNopadPersistent3WGFwdOp,
+    LocalDispatchHandle,
+    LocalExpertDispatcher,
     MoeGroupedGemmNopad3WGFusedActFwdOp,
     MoeGroupedGemmNopadFwdOp,
     MoePermuteNopadFwdOp,
@@ -20,6 +27,11 @@
 from .shared_fused_moe import SharedFusedMoE
 
 __all__ = [
+    "DeepEPDispatchAdapter",
+    "DispatchedExpertMLPFwdOp",
+    "ExpertBatch",
+    "ExpertBatchOutput",
+    "ExpertDispatchResult",
     "FusedMoEExperts",
     "FusedMoEExpertsModular",
     "FusedMoEExpertsNopadPersistent3WGFwdOp",
@@ -28,6 +40,8 @@
     "FusedMoeFwdCbFwdOp",
     "FusedMoeFwdOp",
     "FusedTopKOp",
+    "LocalDispatchHandle",
+    "LocalExpertDispatcher",
     "MoEPrepareAndFinalizeNoDPEP",
     "MoeGroupedGemmNopad3WGFusedActFwdOp",
     "MoeGroupedGemmNopadFwdOp",
diff --git a/tileops/ops/moe/routed_expert/__init__.py b/tileops/ops/moe/routed_expert/__init__.py
index 0e82b873f..d69246c25 100644
--- a/tileops/ops/moe/routed_expert/__init__.py
+++ b/tileops/ops/moe/routed_expert/__init__.py
@@ -1,12 +1,21 @@
 """Routed expert implementations and supporting operations."""
 
 from .abc import (
+    ExpertBatch,
+    ExpertBatchOutput,
     FusedMoEExperts,
     FusedMoEExpertsModular,
     PrepareResult,
     WeightedReduce,
     WeightedReduceNoOp,
 )
+from .dispatch import (
+    DeepEPDispatchAdapter,
+    ExpertDispatchResult,
+    LocalDispatchHandle,
+    LocalExpertDispatcher,
+)
+from .dispatched_expert import DispatchedExpertMLPFwdOp
 from .fused_routed_expert import (
     FusedMoEExpertsNopadPersistent3WGFwdOp,
 )
@@ -16,9 +25,16 @@
 from .unpermute import MoeUnpermuteFwdOp
 
 __all__ = [
+    "DeepEPDispatchAdapter",
+    "DispatchedExpertMLPFwdOp",
+    "ExpertBatch",
+    "ExpertBatchOutput",
+    "ExpertDispatchResult",
     "FusedMoEExperts",
     "FusedMoEExpertsModular",
     "FusedMoEExpertsNopadPersistent3WGFwdOp",
+    "LocalDispatchHandle",
+    "LocalExpertDispatcher",
     "MoeGroupedGemmNopad3WGFusedActFwdOp",
     "MoeGroupedGemmNopadFwdOp",
     "MoePermuteNopadFwdOp",
diff --git a/tileops/ops/moe/routed_expert/abc.py b/tileops/ops/moe/routed_expert/abc.py
index 685a88261..5f0d90d0c 100644
--- a/tileops/ops/moe/routed_expert/abc.py
+++ b/tileops/ops/moe/routed_expert/abc.py
@@ -11,6 +11,8 @@
 from tileops.ops.op_base import Op
 
 __all__ = [
+    "ExpertBatch",
+    "ExpertBatchOutput",
     "FusedMoEExperts",
     "FusedMoEExpertsModular",
     "FusedMoEPrepareAndFinalize",
@@ -20,6 +22,64 @@
 ]
 
 
+@dataclass(frozen=True)
+class ExpertBatch:
+    """Canonical communication-to-compute contract for routed expert rows.
+
+    ``hidden`` uses a tight expert-major layout. Expert ``e`` owns rows
+    ``expert_offsets[e]:expert_offsets[e + 1]``; adjacent equal offsets encode
+    an empty expert. ``valid_rows`` is the device-side
+    ``expert_offsets[-1:]`` view, making offsets the single source of truth.
+    Offsets may change between CUDA Graph replays while capacity and tensor
+    addresses stay fixed. TileOps borrows all input buffers for the duration
+    of the call and does not mutate them.
+    """
+
+    hidden: Tensor
+    expert_offsets: Tensor
+    layout: str = "tight"
+
+    def __post_init__(self) -> None:
+        if self.hidden.ndim != 2:
+            raise ValueError(
+                f"hidden must be rank 2 [capacity, H], got {self.hidden.shape}"
+            )
+        if self.expert_offsets.ndim != 1:
+            raise ValueError(
+                "expert_offsets must be rank 1 [E_local + 1], got "
+                f"{self.expert_offsets.shape}"
+            )
+        if self.expert_offsets.dtype != torch.int32:
+            raise ValueError(
+                "expert_offsets must use torch.int32, got "
+                f"{self.expert_offsets.dtype}"
+            )
+        if self.hidden.device != self.expert_offsets.device:
+            raise ValueError("hidden and expert_offsets must be on the same device")
+        if self.layout != "tight":
+            raise ValueError(
+                f"only layout='tight' is supported, got {self.layout!r}"
+            )
+    @property
+    def capacity(self) -> int:
+        return self.hidden.shape[0]
+
+    @property
+    def valid_rows(self) -> Tensor:
+        """One-element device view of the valid tight-row count."""
+        return self.expert_offsets[-1:]
+
+
+@dataclass(frozen=True)
+class ExpertBatchOutput:
+    """Expert MLP output; routing weights have not been applied."""
+
+    hidden: Tensor
+    valid_rows: Tensor | None = None
+    row_order_preserved: bool = True
+    routing_weights_applied: bool = False
+
+
 def _validate_fused_moe_experts_dtypes(
     op_dtype: torch.dtype,
     output: Tensor,
diff --git a/tileops/ops/moe/routed_expert/dispatch.py b/tileops/ops/moe/routed_expert/dispatch.py
new file mode 100644
index 000000000..79cafc5c7
--- /dev/null
+++ b/tileops/ops/moe/routed_expert/dispatch.py
@@ -0,0 +1,349 @@
+"""Dispatch adapters that normalize routed rows to a tight ``ExpertBatch``.
+
+The compute contract remains communication independent.  Communication
+backends are adapted here and their opaque combine handles never enter the
+expert MLP.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+import torch
+from torch import Tensor
+
+from .abc import ExpertBatch
+from .permute_nopad import MoePermuteNopadFwdOp
+
+__all__ = [
+    "DeepEPDispatchAdapter",
+    "ExpertDispatchResult",
+    "LocalDispatchHandle",
+    "LocalExpertDispatcher",
+]
+
+
+class _DeepEPEvent(Protocol):
+    def current_stream_wait(self) -> None: ...
+
+
+class _DeepEPBuffer(Protocol):
+    def dispatch(self, x: Tensor, **kwargs: Any) -> tuple[Any, ...]: ...
+
+
+@dataclass(frozen=True)
+class LocalDispatchHandle:
+    """Metadata required to invert a local dispatch.
+
+    ``forward_mapping`` maps each flattened source ``(token, top-k slot)`` pair
+    to its row in the tight expert-major batch.  Routing weights are deliberately
+    kept in :class:`ExpertDispatchResult` and are not applied by dispatch.
+    """
+
+    forward_mapping: Tensor
+    num_tokens: int
+    top_k: int
+
+
+@dataclass(frozen=True)
+class ExpertDispatchResult:
+    """Normalized output of a local or external expert dispatch.
+
+    ``routing_weights`` has one entry per physical row in ``batch.hidden``.
+    Only rows before ``batch.valid_rows`` are defined.  ``combine_handle`` is
+    owned by the dispatch backend and must be passed back to that backend; the
+    TileOps expert compute path never inspects it.
+    """
+
+    batch: ExpertBatch
+    routing_weights: Tensor
+    combine_handle: object
+    event: object | None = None
+
+    def __post_init__(self) -> None:
+        if self.routing_weights.ndim != 1:
+            raise ValueError(
+                f"routing_weights must be rank 1 [capacity], got {self.routing_weights.shape}"
+            )
+        if self.routing_weights.shape[0] != self.batch.capacity:
+            raise ValueError(
+                "routing_weights capacity must equal batch capacity; got "
+                f"{self.routing_weights.shape[0]} and {self.batch.capacity}"
+            )
+        if self.routing_weights.dtype != torch.float32:
+            raise ValueError(
+                f"routing_weights must use torch.float32, got {self.routing_weights.dtype}"
+            )
+        if self.routing_weights.device != self.batch.hidden.device:
+            raise ValueError("routing_weights and batch.hidden must be on the same device")
+
+
+class LocalExpertDispatcher:
+    """World-size-one reference dispatcher with tight expert-major output."""
+
+    def __init__(
+        self,
+        num_experts: int,
+        *,
+        total_tokens: int | None = None,
+        top_k: int | None = None,
+        hidden_size: int | None = None,
+        dtype: torch.dtype | None = None,
+    ) -> None:
+        if num_experts <= 0:
+            raise ValueError(f"num_experts must be positive, got {num_experts}")
+        self.num_experts = num_experts
+        self._permute = MoePermuteNopadFwdOp(
+            total_tokens=total_tokens,
+            top_k=top_k,
+            num_experts=num_experts,
+            hidden_size=hidden_size,
+            dtype=dtype,
+        )
+
+    def dispatch(
+        self,
+        hidden_states: Tensor,
+        topk_ids: Tensor,
+        topk_weights: Tensor,
+    ) -> ExpertDispatchResult:
+        """Expand every routed pair and sort it by local expert.
+
+        Duplicate expert selections are preserved as distinct routed pairs.
+        ``topk_ids`` must contain local expert IDs in ``[0, num_experts)``.
+        """
+        _validate_routing_inputs(hidden_states, topk_ids, topk_weights)
+        if topk_ids.dtype != torch.int32:
+            raise ValueError(
+                f"LocalExpertDispatcher requires topk_ids.dtype torch.int32, got {topk_ids.dtype}"
+            )
+
+        permuted, true_offsets, true_sizes, _, forward_mapping = self._permute(
+            hidden_states, topk_ids
+        )
+        expert_offsets = torch.empty(
+            self.num_experts + 1, dtype=torch.int32, device=hidden_states.device
+        )
+        expert_offsets[:-1].copy_(true_offsets)
+        expert_offsets[-1:].copy_(true_offsets[-1:] + true_sizes[-1:])
+
+        dispatched_weights = torch.empty(
+            topk_ids.numel(), dtype=torch.float32, device=hidden_states.device
+        )
+        dispatched_weights.scatter_(
+            0,
+            forward_mapping.to(torch.int64),
+            topk_weights.flatten(),
+        )
+        handle = LocalDispatchHandle(
+            forward_mapping=forward_mapping,
+            num_tokens=hidden_states.shape[0],
+            top_k=topk_ids.shape[1],
+        )
+        return ExpertDispatchResult(
+            batch=ExpertBatch(hidden=permuted, expert_offsets=expert_offsets),
+            routing_weights=dispatched_weights,
+            combine_handle=handle,
+        )
+
+
+class DeepEPDispatchAdapter:
+    """Adapt a DeepEP V2 ``ElasticBuffer`` dispatch to ``ExpertBatch``.
+
+    The adapter intentionally imports no DeepEP package.  The caller owns and
+    injects an initialized ``ElasticBuffer``.  Dispatch uses DeepEP's expanded
+    layout with ``expert_alignment=1``: one row per routed pair, grouped by
+    local expert, without inter-expert padding.  The received activation buffer
+    is therefore borrowed directly by ``ExpertBatch`` without a data copy.
+
+    ``do_cpu_sync=False`` keeps the received row count on the GPU.  DeepEP's
+    inclusive per-expert prefix sum is converted to TileOps' exclusive
+    ``[0, end_0, ..., end_E]`` convention on the current CUDA stream.
+    """
+
+    def __init__(
+        self,
+        buffer: _DeepEPBuffer,
+        *,
+        num_experts: int,
+        num_local_experts: int,
+        num_max_tokens_per_rank: int,
+        num_sms: int = 0,
+        topk_ids_dtype: torch.dtype = torch.int64,
+    ) -> None:
+        if num_experts <= 0:
+            raise ValueError(f"num_experts must be positive, got {num_experts}")
+        if num_local_experts <= 0:
+            raise ValueError(f"num_local_experts must be positive, got {num_local_experts}")
+        if num_local_experts > num_experts:
+            raise ValueError(
+                "num_local_experts cannot exceed num_experts; got "
+                f"{num_local_experts} and {num_experts}"
+            )
+        if num_max_tokens_per_rank <= 0:
+            raise ValueError(
+                f"num_max_tokens_per_rank must be positive, got {num_max_tokens_per_rank}"
+            )
+        if topk_ids_dtype not in (torch.int32, torch.int64):
+            raise ValueError(
+                f"topk_ids_dtype must be torch.int32 or torch.int64, got {topk_ids_dtype}"
+            )
+        self.buffer = buffer
+        self.num_experts = num_experts
+        self.num_local_experts = num_local_experts
+        self.num_max_tokens_per_rank = num_max_tokens_per_rank
+        self.num_sms = num_sms
+        self.topk_ids_dtype = topk_ids_dtype
+
+    def dispatch(
+        self,
+        hidden_states: Tensor,
+        topk_ids: Tensor | None,
+        topk_weights: Tensor,
+        *,
+        expert_offsets: Tensor | None = None,
+        cached_handle: object | None = None,
+    ) -> ExpertDispatchResult:
+        """Run asynchronous DeepEP dispatch and normalize its GPU metadata.
+
+        ``expert_offsets`` may be supplied by a graph-aware caller to keep its
+        address stable across replays.  Waiting uses a CUDA stream dependency;
+        it does not synchronize the host.  To reuse a decode layout, pass the
+        prior ``combine_handle`` as ``cached_handle`` and set ``topk_ids=None``.
+        """
+        if hidden_states.dtype != torch.bfloat16:
+            raise ValueError(
+                "DeepEP BF16 dispatch requires hidden_states.dtype "
+                f"torch.bfloat16, got {hidden_states.dtype}"
+            )
+        if cached_handle is None:
+            if topk_ids is None:
+                raise ValueError("topk_ids is required when cached_handle is not provided")
+            _validate_routing_inputs(hidden_states, topk_ids, topk_weights)
+            if topk_ids.dtype != self.topk_ids_dtype:
+                raise ValueError(
+                    "DeepEP topk_ids dtype must match the adapter's configured "
+                    f"topk_ids_dtype {self.topk_ids_dtype}, got {topk_ids.dtype}"
+                )
+        else:
+            if topk_ids is not None:
+                raise ValueError("topk_ids must be None when cached_handle is provided")
+            _validate_cached_routing_inputs(hidden_states, topk_weights, cached_handle)
+            if not getattr(cached_handle, "do_expand", False):
+                raise ValueError("cached_handle must come from an expanded DeepEP dispatch")
+            if getattr(cached_handle, "expert_alignment", None) != 1:
+                raise ValueError("cached_handle must use expert_alignment=1")
+
+        recv_x, recv_topk_ids, recv_weights, handle, event = self.buffer.dispatch(
+            hidden_states,
+            topk_idx=topk_ids,
+            topk_weights=topk_weights,
+            num_experts=self.num_experts,
+            num_max_tokens_per_rank=self.num_max_tokens_per_rank,
+            expert_alignment=1,
+            num_sms=self.num_sms,
+            async_with_compute_stream=True,
+            do_cpu_sync=False,
+            do_expand=True,
+            handle=cached_handle,
+        )
+
+        if isinstance(recv_x, tuple):
+            raise ValueError("DeepEP returned FP8 data/scales; this M6 adapter supports BF16 only")
+        if recv_topk_ids is not None:
+            raise RuntimeError("DeepEP expanded dispatch unexpectedly returned recv_topk_idx")
+        if recv_weights is None:
+            raise RuntimeError("DeepEP dispatch did not return routing weights required by combine")
+        if not hasattr(event, "current_stream_wait"):
+            raise TypeError("DeepEP dispatch event must provide current_stream_wait()")
+
+        event.current_stream_wait()
+        inclusive_offsets = getattr(handle, "psum_num_recv_tokens_per_expert", None)
+        if not isinstance(inclusive_offsets, Tensor):
+            raise TypeError("DeepEP handle must expose psum_num_recv_tokens_per_expert")
+        if inclusive_offsets.shape != (self.num_local_experts,):
+            raise ValueError(
+                "DeepEP per-expert prefix sum must have shape "
+                f"[{self.num_local_experts}], got {inclusive_offsets.shape}"
+            )
+        if inclusive_offsets.dtype != torch.int32:
+            raise ValueError(
+                f"DeepEP per-expert prefix sum must use torch.int32, got {inclusive_offsets.dtype}"
+            )
+        if inclusive_offsets.device != recv_x.device:
+            raise ValueError("DeepEP prefix sum and received activations must share a device")
+
+        if expert_offsets is None:
+            expert_offsets = torch.empty(
+                self.num_local_experts + 1,
+                dtype=torch.int32,
+                device=recv_x.device,
+            )
+        else:
+            _validate_offsets_buffer(expert_offsets, self.num_local_experts, recv_x.device)
+        expert_offsets[0].zero_()
+        expert_offsets[1:].copy_(inclusive_offsets)
+
+        return ExpertDispatchResult(
+            batch=ExpertBatch(hidden=recv_x, expert_offsets=expert_offsets),
+            routing_weights=recv_weights,
+            combine_handle=handle,
+            event=event,
+        )
+
+
+def _validate_routing_inputs(
+    hidden_states: Tensor,
+    topk_ids: Tensor,
+    topk_weights: Tensor,
+) -> None:
+    if not hidden_states.is_cuda:
+        raise ValueError("hidden_states must be a CUDA tensor")
+    if not topk_ids.is_cuda or not topk_weights.is_cuda:
+        raise ValueError("topk_ids and topk_weights must be CUDA tensors")
+    if not (hidden_states.device == topk_ids.device == topk_weights.device):
+        raise ValueError("hidden_states, topk_ids, and topk_weights must share a device")
+    if hidden_states.ndim != 2:
+        raise ValueError(f"hidden_states must be rank 2 [T, H], got {hidden_states.shape}")
+    if topk_ids.ndim != 2:
+        raise ValueError(f"topk_ids must be rank 2 [T, K], got {topk_ids.shape}")
+    if topk_weights.shape != topk_ids.shape:
+        raise ValueError(
+            "topk_weights must have the same shape as topk_ids; got "
+            f"{topk_weights.shape} and {topk_ids.shape}"
+        )
+    if topk_ids.shape[0] != hidden_states.shape[0]:
+        raise ValueError(
+            "routing token count must equal hidden_states token count; got "
+            f"{topk_ids.shape[0]} and {hidden_states.shape[0]}"
+        )
+    if topk_weights.dtype != torch.float32:
+        raise ValueError(f"topk_weights must use torch.float32, got {topk_weights.dtype}")
+
+
+def _validate_cached_routing_inputs(
+    hidden_states: Tensor,
+    topk_weights: Tensor,
+    cached_handle: object,
+) -> None:
+    cached_topk_ids = getattr(cached_handle, "topk_idx", None)
+    if not isinstance(cached_topk_ids, Tensor):
+        raise TypeError("cached_handle must expose its device topk_idx tensor")
+    _validate_routing_inputs(hidden_states, cached_topk_ids, topk_weights)
+
+
+def _validate_offsets_buffer(
+    expert_offsets: Tensor,
+    num_local_experts: int,
+    device: torch.device,
+) -> None:
+    if expert_offsets.shape != (num_local_experts + 1,):
+        raise ValueError(
+            "expert_offsets buffer must have shape "
+            f"[{num_local_experts + 1}], got {expert_offsets.shape}"
+        )
+    if expert_offsets.dtype != torch.int32:
+        raise ValueError(f"expert_offsets buffer must use torch.int32, got {expert_offsets.dtype}")
+    if expert_offsets.device != device:
+        raise ValueError("expert_offsets buffer and received activations must share a device")
diff --git a/tileops/ops/moe/routed_expert/dispatched_expert.py b/tileops/ops/moe/routed_expert/dispatched_expert.py
new file mode 100644
index 000000000..fc8049780
--- /dev/null
+++ b/tileops/ops/moe/routed_expert/dispatched_expert.py
@@ -0,0 +1,237 @@
+"""Communication-independent expert MLP for tight expert-major batches."""
+
+from __future__ import annotations
+
+import logging
+from typing import Dict, Optional
+
+import torch
+from torch import Tensor
+
+from tileops.kernels.grouped_gemm import GroupedGemmPersistent3WGKernel
+from tileops.kernels.grouped_gemm.grouped_gemm_persistent_3wg import (
+    _DEFAULT_CONFIG as _3WG_DEFAULT_CONFIG,
+)
+from tileops.kernels.kernel_base import Kernel
+from tileops.kernels.moe.moe_grouped_gemm_nopad import MoeGroupedGemmNopadKernel
+from tileops.kernels.moe.moe_grouped_gemm_persistent_3wg_fused_act import (
+    _DEFAULT_CONFIG as _FUSED_ACT_DEFAULT_CONFIG,
+)
+from tileops.ops.moe._activation import build_activation_op
+from tileops.ops.op_base import Op
+
+from .abc import ExpertBatch, ExpertBatchOutput
+from .moe_grouped_gemm_nopad import MoeGroupedGemmNopadFwdOp
+
+__all__ = ["DispatchedExpertMLPFwdOp"]
+
+_logger = logging.getLogger(__name__)
+
+
+class DispatchedExpertMLPFwdOp(Op):
+    """Run an expert MLP over a pre-dispatched tight expert-major batch.
+
+    ``expert_input`` is partitioned into contiguous expert segments described
+    by ``true_offsets`` and ``true_sizes``. Empty experts are represented by a
+    zero size. Output row ``i`` always corresponds to input row ``i``.
+
+    This computation boundary deliberately has no routing, communication, or
+    weighted-reduction inputs. In particular, routing weights are not applied.
+    """
+
+    def __init__(
+        self,
+        num_pairs: int,
+        num_experts: int,
+        hidden_size: int,
+        ffn_size: int,
+        dtype: torch.dtype = torch.bfloat16,
+        gemm_kernel: Optional[type] = None,
+        kernel_map: Optional[Dict[str, Kernel]] = None,
+        *,
+        activation: str = "silu_and_mul",
+        use_fused_activation: bool = False,
+    ):
+        self.dispatch_kernel(kernel_map)
+        self.num_pairs = num_pairs
+        self.num_experts = num_experts
+        self.hidden_size = hidden_size
+        self.ffn_size = ffn_size
+        self.dtype = dtype
+        self.activation = activation
+
+        requested_kernel_cls = (kernel_map or {}).get(
+            "moe_grouped_gemm_kernel",
+            gemm_kernel or GroupedGemmPersistent3WGKernel,
+        )
+        kernel_cls = requested_kernel_cls
+        block_n = _3WG_DEFAULT_CONFIG["block_n"]
+        block_k = _3WG_DEFAULT_CONFIG["block_k"]
+        gate_up_n = ffn_size * 2
+        if kernel_cls is GroupedGemmPersistent3WGKernel:
+            gate_up_ok = gate_up_n % block_n == 0 and hidden_size % block_k == 0
+            down_ok = hidden_size % block_n == 0 and ffn_size % block_k == 0
+            if not (gate_up_ok and down_ok):
+                _logger.warning(
+                    "DispatchedExpertMLPFwdOp: dims not aligned to 3WG block "
+                    "(gate_up_n=%d, hidden_size=%d, ffn_size=%d; block_n=%d, "
+                    "block_k=%d) — falling back to MoeGroupedGemmNopadKernel.",
+                    gate_up_n,
+                    hidden_size,
+                    ffn_size,
+                    block_n,
+                    block_k,
+                )
+                kernel_cls = MoeGroupedGemmNopadKernel
+
+        # Resolve the GEMM implementation exactly once. In particular, the
+        # caller's original override must not be merged after an alignment
+        # fallback and silently reinstate an ineligible kernel.
+        effective_gemm_kernel_map = {
+            **(kernel_map or {}),
+            "moe_grouped_gemm_kernel": kernel_cls,
+        }
+        self.use_fused_activation = use_fused_activation
+        if use_fused_activation:
+            fused_block_n = _FUSED_ACT_DEFAULT_CONFIG["block_n"]
+            eligible = (
+                torch.cuda.is_available()
+                and torch.cuda.get_device_capability()[0] >= 9
+                and kernel_cls is GroupedGemmPersistent3WGKernel
+                and activation in ("silu_and_mul", "gelu_and_mul")
+                and ffn_size % fused_block_n == 0
+            )
+            if not eligible:
+                _logger.warning(
+                    "use_fused_activation=True not eligible (requires CUDA + SM90 + "
+                    "GroupedGemmPersistent3WGKernel gate_up GEMM with no conflicting "
+                    "moe_grouped_gemm_kernel override + activation in {silu_and_mul, "
+                    "gelu_and_mul} + ffn_size %% %d == 0); falling back to unfused "
+                    "activation. ffn_size=%d, activation=%s.",
+                    fused_block_n,
+                    ffn_size,
+                    activation,
+                )
+                self.use_fused_activation = False
+
+        if self.use_fused_activation:
+            from .moe_grouped_gemm_nopad_fused_act import (
+                MoeGroupedGemmNopad3WGFusedActFwdOp,
+            )
+
+            self._gemm_gate_up = MoeGroupedGemmNopad3WGFusedActFwdOp(
+                numel=num_pairs,
+                num_experts=num_experts,
+                ffn=ffn_size,
+                k=hidden_size,
+                dtype=dtype,
+                activation=activation,
+                kernel_map=effective_gemm_kernel_map,
+            )
+            self._activation_op = None
+        else:
+            self._gemm_gate_up = MoeGroupedGemmNopadFwdOp(
+                numel=num_pairs,
+                num_experts=num_experts,
+                n=ffn_size * 2,
+                k=hidden_size,
+                dtype=dtype,
+                kernel_map=effective_gemm_kernel_map,
+            )
+            self._activation_op = build_activation_op(
+                activation,
+                M=num_pairs,
+                N=ffn_size,
+                dtype=dtype,
+                kernel_map=kernel_map,
+            )
+        self._gemm_down = MoeGroupedGemmNopadFwdOp(
+            numel=num_pairs,
+            num_experts=num_experts,
+            n=hidden_size,
+            k=ffn_size,
+            dtype=dtype,
+            kernel_map=effective_gemm_kernel_map,
+        )
+
+    @property
+    def default_kernel_map(self) -> dict:
+        return {}
+
+    def forward(
+        self,
+        expert_input: Tensor,
+        w_gate_up: Tensor,
+        w_down: Tensor,
+        true_sizes: Tensor,
+        true_offsets: Tensor,
+    ) -> Tensor:
+        """Return ``[num_pairs, hidden_size]`` without changing row order."""
+        return self._forward(
+            expert_input,
+            w_gate_up,
+            w_down,
+            true_sizes,
+            true_offsets,
+            valid_rows=None,
+        )
+
+    def _forward(
+        self,
+        expert_input: Tensor,
+        w_gate_up: Tensor,
+        w_down: Tensor,
+        true_sizes: Tensor,
+        true_offsets: Tensor,
+        valid_rows: Tensor | None,
+    ) -> Tensor:
+        gate_up = self._gemm_gate_up(expert_input, w_gate_up, true_sizes, true_offsets)
+        act = (
+            gate_up
+            if self.use_fused_activation
+            else (
+                self._activation_op(gate_up)
+                if valid_rows is None
+                else self._activation_op.kernel.forward_rows(gate_up, valid_rows)
+            )
+        )
+        return self._gemm_down(act, w_down, true_sizes, true_offsets)
+
+    def forward_batch(
+        self,
+        batch: ExpertBatch,
+        w_gate_up: Tensor,
+        w_down: Tensor,
+    ) -> ExpertBatchOutput:
+        """Run the MLP from canonical ``expert_offsets[E_local + 1]``.
+
+        Offset differencing and the received row count stay on the input
+        device. Grouped GEMMs schedule only expert tiles described by the
+        offsets; the standalone activation is bounded by ``valid_rows``.
+        """
+        if batch.capacity != self.num_pairs:
+            raise ValueError(
+                f"batch capacity must equal num_pairs={self.num_pairs}, got {batch.capacity}"
+            )
+        if batch.hidden.shape[1] != self.hidden_size:
+            raise ValueError(
+                f"batch hidden size must equal {self.hidden_size}, got {batch.hidden.shape[1]}"
+            )
+        if batch.expert_offsets.numel() != self.num_experts + 1:
+            raise ValueError(
+                "expert_offsets must contain num_experts + 1 entries; "
+                f"expected {self.num_experts + 1}, "
+                f"got {batch.expert_offsets.numel()}"
+            )
+        true_offsets = batch.expert_offsets[:-1]
+        true_sizes = batch.expert_offsets[1:] - true_offsets
+        valid_rows = batch.valid_rows
+        hidden = self._forward(
+            batch.hidden,
+            w_gate_up,
+            w_down,
+            true_sizes,
+            true_offsets,
+            valid_rows,
+        )
+        return ExpertBatchOutput(hidden=hidden, valid_rows=valid_rows)
diff --git a/tileops/ops/moe/routed_expert/fused_routed_expert.py b/tileops/ops/moe/routed_expert/fused_routed_expert.py
index 16a712fd1..e640578e3 100644
--- a/tileops/ops/moe/routed_expert/fused_routed_expert.py
+++ b/tileops/ops/moe/routed_expert/fused_routed_expert.py
@@ -2,28 +2,12 @@
 
 from __future__ import annotations
 
-import logging
 from typing import Dict, Optional
 
 import torch
 from torch import Tensor
 
-from tileops.kernels.grouped_gemm import (
-    GroupedGemmPersistent3WGKernel,
-)
-from tileops.kernels.grouped_gemm.grouped_gemm_persistent_3wg import (
-    _DEFAULT_CONFIG as _3WG_DEFAULT_CONFIG,
-)
 from tileops.kernels.kernel_base import Kernel
-from tileops.kernels.moe.moe_grouped_gemm_nopad import MoeGroupedGemmNopadKernel
-
-# Imported unconditionally: the eligibility check reads its block_n even when
-# use_fused_activation ends up False. The wrapper class itself is deferred to
-# the fused branch (imported lazily in __init__).
-from tileops.kernels.moe.moe_grouped_gemm_persistent_3wg_fused_act import (
-    _DEFAULT_CONFIG as _FUSED_ACT_DEFAULT_CONFIG,
-)
-from tileops.ops.moe._activation import build_activation_op
 
 from .abc import (
     FusedMoEExpertsModular,
@@ -31,7 +15,7 @@
     WeightedReduceNoOp,
     _validate_fused_moe_experts_dtypes,
 )
-from .moe_grouped_gemm_nopad import MoeGroupedGemmNopadFwdOp
+from .dispatched_expert import DispatchedExpertMLPFwdOp
 from .permute_nopad import MoePermuteNopadFwdOp
 from .unpermute import MoeUnpermuteFwdOp
 
@@ -39,9 +23,6 @@
     "FusedMoEExpertsNopadPersistent3WGFwdOp",
 ]
 
-_logger = logging.getLogger(__name__)
-
-
 class FusedMoEExpertsNopadPersistent3WGFwdOp(FusedMoEExpertsModular):
     """Expert GEMM using tight (T*K rows, no-pad) layout with 3WG persistent kernel.
 
@@ -129,87 +110,28 @@ def __init__(
             int((expert_map >= 0).sum().item()) if expert_map is not None else num_experts
         )
 
-        kernel_cls = gemm_kernel or GroupedGemmPersistent3WGKernel
-
-        # 3WG requires N and K aligned to its default block dimensions.
-        # Fall back to tile scheduler kernel for small/unaligned dimensions.
-        _3wg_block_n = _3WG_DEFAULT_CONFIG["block_n"]
-        _3wg_block_k = _3WG_DEFAULT_CONFIG["block_k"]
-        gate_up_n = ffn_size * 2
-        if kernel_cls is GroupedGemmPersistent3WGKernel:
-            gate_up_ok = (gate_up_n % _3wg_block_n == 0) and (hidden_size % _3wg_block_k == 0)
-            down_ok = (hidden_size % _3wg_block_n == 0) and (ffn_size % _3wg_block_k == 0)
-            if not (gate_up_ok and down_ok):
-                _logger.warning(
-                    "FusedMoEExpertsNopadPersistent3WGFwdOp: dims not aligned "
-                    "to 3WG block (gate_up_n=%d, hidden_size=%d, ffn_size=%d; "
-                    "block_n=%d, block_k=%d) — falling back to "
-                    "MoeGroupedGemmNopadKernel.",
-                    gate_up_n, hidden_size, ffn_size, _3wg_block_n, _3wg_block_k,
-                )
-                kernel_cls = MoeGroupedGemmNopadKernel
-
-        # A caller can steer the gate_up GEMM either via gemm_kernel (already
-        # folded into kernel_cls) or via kernel_map["moe_grouped_gemm_kernel"]
-        # (merged into the unfused gate_up / down ops below). The fused gate_up
-        # wrapper keys off "moe_grouped_gemm_fused_act_kernel" and cannot honor a
-        # "moe_grouped_gemm_kernel" override, so enabling fusion alongside a
-        # non-3WG override would silently produce a fused 3WG gate_up next to an
-        # overridden down GEMM. Disable fusion in that case so the override
-        # applies uniformly through the unfused path.
-        gemm_override = (kernel_map or {}).get("moe_grouped_gemm_kernel")
-        self.use_fused_activation = use_fused_activation
-        if use_fused_activation:
-            fused_block_n = _FUSED_ACT_DEFAULT_CONFIG["block_n"]
-            ok = (
-                torch.cuda.is_available()
-                and torch.cuda.get_device_capability()[0] >= 9
-                and kernel_cls is GroupedGemmPersistent3WGKernel
-                and (gemm_override is None
-                     or gemm_override is GroupedGemmPersistent3WGKernel)
-                and activation in ("silu_and_mul", "gelu_and_mul")
-                and (ffn_size % fused_block_n == 0)
-            )
-            if not ok:
-                _logger.warning(
-                    "use_fused_activation=True not eligible (requires CUDA + SM90 + "
-                    "GroupedGemmPersistent3WGKernel gate_up GEMM with no conflicting "
-                    "moe_grouped_gemm_kernel override + activation in {silu_and_mul, "
-                    "gelu_and_mul} + ffn_size %% %d == 0); falling back to unfused "
-                    "activation. ffn_size=%d, activation=%s.",
-                    fused_block_n, ffn_size, activation,
-                )
-                self.use_fused_activation = False
-
         self._permute = MoePermuteNopadFwdOp(
             num_experts=num_experts, dtype=dtype, expert_map=expert_map,
             kernel_map=kernel_map,
         )
         self.activation = activation
-        if self.use_fused_activation:
-            from .moe_grouped_gemm_nopad_fused_act import (
-                MoeGroupedGemmNopad3WGFusedActFwdOp,
-            )
-            self._gemm_gate_up = MoeGroupedGemmNopad3WGFusedActFwdOp(
-                numel=numel, num_experts=num_experts_local,
-                ffn=ffn_size, k=hidden_size, dtype=dtype, activation=activation,
-                kernel_map=kernel_map,
-            )
-            self._activation_op = None
-        else:
-            self._gemm_gate_up = MoeGroupedGemmNopadFwdOp(
-                numel=numel, num_experts=num_experts_local,
-                n=ffn_size * 2, k=hidden_size, dtype=dtype,
-                kernel_map={"moe_grouped_gemm_kernel": kernel_cls, **(kernel_map or {})},
-            )
-            self._activation_op = build_activation_op(
-                activation, M=numel, N=ffn_size, dtype=dtype, kernel_map=kernel_map,
-            )
-        self._gemm_down = MoeGroupedGemmNopadFwdOp(
-            numel=numel, num_experts=num_experts_local,
-            n=hidden_size, k=ffn_size, dtype=dtype,
-            kernel_map={"moe_grouped_gemm_kernel": kernel_cls, **(kernel_map or {})},
+        self._expert_mlp = DispatchedExpertMLPFwdOp(
+            num_pairs=numel,
+            num_experts=num_experts_local,
+            hidden_size=hidden_size,
+            ffn_size=ffn_size,
+            dtype=dtype,
+            gemm_kernel=gemm_kernel,
+            kernel_map=kernel_map,
+            activation=activation,
+            use_fused_activation=use_fused_activation,
         )
+        self.use_fused_activation = self._expert_mlp.use_fused_activation
+        # Keep these implementation attributes available for callers/tests
+        # that inspected the legacy composite op.
+        self._gemm_gate_up = self._expert_mlp._gemm_gate_up
+        self._activation_op = self._expert_mlp._activation_op
+        self._gemm_down = self._expert_mlp._gemm_down
         self._unpermute = MoeUnpermuteFwdOp(
             total_tokens=num_tokens, top_k=top_k,
             hidden_size=hidden_size, dtype=dtype, padded_batch_sum=numel,
@@ -279,9 +201,9 @@ def forward(
             topk_weights, topk_ids, expert_map, workspace1, workspace2,
         )
         perm_h, true_offsets, true_sizes, _, fwd_idx = self._permute(hidden_states, topk_ids)
-        gate_up = self._gemm_gate_up(perm_h, w_gate_up, true_sizes, true_offsets)
-        act = gate_up if self.use_fused_activation else self._activation_op(gate_up)
-        mm2 = self._gemm_down(act, w_down, true_sizes, true_offsets)
+        mm2 = self._expert_mlp(
+            perm_h, w_gate_up, w_down, true_sizes, true_offsets
+        )
         # Unpermute reduces into ``output`` directly and folds
         # ``routed_scaling_factor`` into its prim_func — no separate copy/scale.
         self._unpermute(mm2, fwd_idx, topk_weights, out=output)