diff --git a/docs/configuration/server.md b/docs/configuration/server.md
index 64f2082491..d6310444a5 100644
--- a/docs/configuration/server.md
+++ b/docs/configuration/server.md
@@ -194,6 +194,16 @@ startup rather than silently drafting a wrong-width block. A checkpoint with
 `block_size: 8` therefore wants `--speculative-num-draft-tokens 8
 --speculative-num-steps 7`.
 
+A checkpoint whose architecture is `DFlash2DraftModel` uses the same `DFLASH`
+launch method. TokenSpeed selects its grouped-convolution and candidate-selector
+runtime from the checkpoint architecture; no separate algorithm flag is needed.
+Greedy requests use the selector's best path. Stochastic requests cache the
+realized top-K selector distribution per request and use lossless `p/q`
+rejection plus `relu(p-q)` residual sampling. Select `--sampling-backend
+flashinfer` or `flashinfer_full` for that path; the Triton sampling backends
+currently fail fast for DFlash2 stochastic verification instead of silently
+falling back to target-only acceptance.
+
 A block drafter writes its KV at the target's cache locations, so it shares the
 target's page table: `--block-size` is a target-side choice and the draft
 follows it. Any sliding window the draft checkpoint declares is an attention
diff --git a/docs/recipes/models.md b/docs/recipes/models.md
index c862539aca..142d8b9310 100644
--- a/docs/recipes/models.md
+++ b/docs/recipes/models.md
@@ -210,6 +210,14 @@ attention. It does not yet expose an equivalent of SGLang's
 `--speculative-dflash-draft-window-size`; add such a flag before relying on
 bounded draft attention for long-context deployments.
 
+Official DFlash2 checkpoints that declare `DFlash2DraftModel` use the same
+`--speculative-algorithm DFLASH` launch. Their grouped dynamic convolutions and
+candidate selector are enabled automatically from the draft architecture. Add
+`--sampling-backend flashinfer` (or `flashinfer_full`) for nonzero-temperature
+sampling so the selector's sparse q distribution participates in lossless
+rejection verification. `greedy` remains the lower-overhead temperature-zero
+path.
+
 ## Kimi K3
 
 Kimi-K3 combines a MoonViT vision encoder with a hybrid KDA
diff --git a/python/tokenspeed/runtime/configs/model_config.py b/python/tokenspeed/runtime/configs/model_config.py
index e8008d7a1e..99a2ef1f41 100644
--- a/python/tokenspeed/runtime/configs/model_config.py
+++ b/python/tokenspeed/runtime/configs/model_config.py
@@ -284,6 +284,21 @@ def _resolve_attention_family(
     return None
 
 
+def _is_dflash2_mla(
+    hf_config: PretrainedConfig,
+    hf_text_config: PretrainedConfig,
+) -> bool:
+    architectures = _model_architectures(hf_config, hf_text_config)
+    dflash_config = getattr(hf_text_config, "dflash_config", None) or getattr(
+        hf_config, "dflash_config", None
+    )
+    return (
+        "DFlash2DraftModel" in architectures
+        and isinstance(dflash_config, dict)
+        and dflash_config.get("attention_mode") == "mla"
+    )
+
+
 def _apply_attention_family_defaults(
     server_args: ServerArgs,
     spec: _AttentionFamilySpec,
@@ -541,6 +556,8 @@ def __init__(
         if attention_family is not None:
             _apply_attention_family_defaults(server_args, attention_family)
             attention_family.configure(self)
+        elif _is_dflash2_mla(self.hf_config, self.hf_text_config):
+            configure_mla_attention(self)
         elif "MiniCPM3ForCausalLM" in self.hf_config.architectures:
             self.head_dim = 128
             self.attention_arch = AttentionArch.MLA
diff --git a/python/tokenspeed/runtime/execution/drafter/__init__.py b/python/tokenspeed/runtime/execution/drafter/__init__.py
index dacbd85b48..8437cce359 100644
--- a/python/tokenspeed/runtime/execution/drafter/__init__.py
+++ b/python/tokenspeed/runtime/execution/drafter/__init__.py
@@ -46,6 +46,7 @@ def get_drafter_impl(spec_algo: str, model: torch.nn.Module) -> type[BaseDrafter
     # Imports are local: drafter modules pull in kernel ops and model code,
     # and this package init must stay importable from lightweight contexts.
     from tokenspeed.runtime.execution.drafter.dflash import DFlash
+    from tokenspeed.runtime.execution.drafter.dflash2 import DFlash2
     from tokenspeed.runtime.execution.drafter.dspark import DSpark
     from tokenspeed.runtime.execution.drafter.eagle import Eagle
     from tokenspeed.runtime.models.inkling_nextn import (
@@ -59,6 +60,12 @@ def get_drafter_impl(spec_algo: str, model: torch.nn.Module) -> type[BaseDrafter
         "DSPARK": DSpark,
     }
 
+    if spec_algo == "DFLASH":
+        from tokenspeed.runtime.models.dflash2 import DFlash2DraftModel
+
+        if isinstance(model, DFlash2DraftModel):
+            return DFlash2
+
     # "MTP" covers two algorithms:
     # (1) Eagle-like MTP (e.g. DeepSeek) stays on Eagle in eagle.py;
     # (2) Vanilla MTP (e.g. Inkling) with multi-layer weights stays on Mtp in mtp.py.
diff --git a/python/tokenspeed/runtime/execution/drafter/dflash2.py b/python/tokenspeed/runtime/execution/drafter/dflash2.py
new file mode 100644
index 0000000000..38752b6ca8
--- /dev/null
+++ b/python/tokenspeed/runtime/execution/drafter/dflash2.py
@@ -0,0 +1,244 @@
+# Copyright (c) 2026 LightSeek Foundation
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""DFlash2 proposal path on top of TokenSpeed's native DFlash block runtime."""
+
+from __future__ import annotations
+
+import torch
+
+from tokenspeed.runtime.execution.drafter.dflash import DFlash
+from tokenspeed.runtime.execution.forward_batch_info import ForwardMode
+from tokenspeed.runtime.layers.logits_processor import LogitsMetadata, LogitsProcessor
+from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
+from tokenspeed.runtime.utils.nvtx import nvtx_range
+
+
+def _walk_best_path(
+    candidate_ids: torch.Tensor,
+    scores: torch.Tensor,
+    anchor_token_ids: torch.Tensor,
+    out: torch.Tensor,
+) -> torch.Tensor:
+    """Walk a fixed DFlash2 candidate lattice without host-side tensor reads."""
+    batch_size, num_steps, top_k = candidate_ids.shape
+    out[:, 0].copy_(anchor_token_ids)
+    previous = torch.zeros(batch_size, dtype=torch.int64, device=candidate_ids.device)
+    for step in range(num_steps):
+        transitions = torch.gather(
+            scores[:, step],
+            1,
+            previous[:, None, None].expand(-1, 1, top_k),
+        ).squeeze(1)
+        previous = torch.argmax(transitions, dim=-1)
+        token = torch.gather(candidate_ids[:, step], 1, previous[:, None]).squeeze(1)
+        out[:, step + 1].copy_(token)
+    return out
+
+
+def _walk_sampled_path(
+    candidate_ids: torch.Tensor,
+    scores: torch.Tensor,
+    anchor_token_ids: torch.Tensor,
+    temperatures: torch.Tensor,
+    coins: torch.Tensor,
+    out: torch.Tensor,
+) -> torch.Tensor:
+    """Sample the selector path and return each realized sparse q row."""
+    batch_size, num_steps, top_k = candidate_ids.shape
+    out[:, 0].copy_(anchor_token_ids)
+    previous = torch.zeros(batch_size, dtype=torch.int64, device=candidate_ids.device)
+    realized_probs = torch.empty(
+        (batch_size, num_steps, top_k),
+        dtype=torch.float32,
+        device=candidate_ids.device,
+    )
+    stochastic = temperatures > 0
+    safe_temperatures = torch.where(
+        stochastic, temperatures, torch.ones_like(temperatures)
+    )
+    for step in range(num_steps):
+        transitions = (
+            torch.gather(
+                scores[:, step],
+                1,
+                previous[:, None, None].expand(-1, 1, top_k),
+            )
+            .squeeze(1)
+            .float()
+        )
+        probabilities = torch.softmax(transitions / safe_temperatures[:, None], dim=-1)
+        realized_probs[:, step].copy_(probabilities)
+        sampled = (probabilities.cumsum(dim=-1) < coins[:, step, None]).sum(dim=-1)
+        sampled = sampled.clamp(max=top_k - 1)
+        greedy = torch.argmax(transitions, dim=-1)
+        previous = torch.where(stochastic, sampled, greedy)
+        token = torch.gather(candidate_ids[:, step], 1, previous[:, None]).squeeze(1)
+        out[:, step + 1].copy_(token)
+    return realized_probs
+
+
+class DFlash2(DFlash):
+    """DFlash block runtime with the DFlash2 top-k transition selector."""
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.candidate_selector = getattr(self.model, "candidate_selector", None)
+        if self.candidate_selector is None:
+            raise ValueError(
+                "DFlash2 requires a draft model with candidate_selector weights."
+            )
+        if self.draft_query_width != self.spec_num_tokens:
+            raise ValueError("DFlash2 requires the anchor-plus-mask DFlash layout.")
+
+        config = self.model.config
+        nested = getattr(config, "dflash_config", {}) or {}
+        self.selector_top_k = int(nested.get("selector_top_k"))
+        self.output_multiplier = float(nested.get("output_multiplier", 1.0))
+        self.final_logit_softcapping = float(
+            nested.get("final_logit_softcapping") or 0.0
+        )
+        self.candidate_logits_processor: LogitsProcessor | None = None
+        self.sampling_backend = None
+        self.dflash2_verify_mode = "unwired"
+        self._draft_candidate_ids_pool: torch.Tensor | None = None
+        self._draft_candidate_probs_pool: torch.Tensor | None = None
+
+    def wire_target(self, target_model) -> None:
+        super().wire_target(target_model)
+        self.candidate_logits_processor = LogitsProcessor(
+            self.model.config,
+            logit_scale=self.output_multiplier,
+            tp_rank=self.logits_processor.tp_rank,
+            tp_size=self.logits_processor.tp_size,
+            tp_group=self.logits_processor.tp_group,
+        )
+        self.candidate_logits_processor.final_logit_softcapping = (
+            self.final_logit_softcapping if self.final_logit_softcapping > 0 else None
+        )
+
+    def wire_sampling_backend(self, sampling_backend) -> None:
+        mode = str(getattr(sampling_backend, "dflash2_verify_mode", "unsupported"))
+        if mode not in ("greedy", "rejection"):
+            raise ValueError(
+                f"{type(sampling_backend).__name__} cannot verify DFlash2 q; use "
+                "--sampling-backend greedy, flashinfer, or flashinfer_full"
+            )
+        self.sampling_backend = sampling_backend
+        self.dflash2_verify_mode = mode
+        if mode == "rejection":
+            pool_rows = int(self.runtime_states.valid_cache_lengths.shape[0])
+            shape = (pool_rows, self.spec_num_tokens - 1, self.selector_top_k)
+            self._draft_candidate_ids_pool = torch.zeros(
+                shape, dtype=torch.int64, device=self.device
+            )
+            self._draft_candidate_probs_pool = torch.zeros(
+                shape, dtype=torch.float32, device=self.device
+            )
+
+    def get_draft_distribution(self, base_ctx) -> SparseDraftDistribution | None:
+        if (
+            self.dflash2_verify_mode != "rejection"
+            or base_ctx.bs == base_ctx.num_extends
+        ):
+            return None
+        if (
+            self._draft_candidate_ids_pool is None
+            or self._draft_candidate_probs_pool is None
+        ):
+            raise RuntimeError("DFlash2 rejection buffers were not initialized")
+        req_pool_indices = self.input_buffers.req_pool_indices_buf[
+            base_ctx.num_extends : base_ctx.bs
+        ]
+        return SparseDraftDistribution(
+            candidate_ids=self._draft_candidate_ids_pool.index_select(
+                0, req_pool_indices
+            ),
+            probabilities=self._draft_candidate_probs_pool.index_select(
+                0, req_pool_indices
+            ),
+        )
+
+    def _compute_candidates(
+        self, hidden_states: torch.Tensor
+    ) -> tuple[torch.Tensor, torch.Tensor]:
+        if self.candidate_logits_processor is None:
+            raise RuntimeError("DFlash2 must be wired to the target before drafting.")
+        metadata = LogitsMetadata(forward_mode=ForwardMode.DECODE)
+        logits = self.candidate_logits_processor._get_logits(
+            hidden_states, self.lm_head, metadata
+        )
+        unary_logits, candidate_ids = torch.topk(
+            logits, self.selector_top_k, dim=-1, sorted=False
+        )
+        return candidate_ids, unary_logits
+
+    @nvtx_range("dflash2_sample_block", color="purple")
+    def _sample_block(
+        self,
+        draft_hidden: torch.Tensor,
+        block_ids: torch.Tensor,
+        next_tokens: torch.Tensor,
+    ) -> torch.Tensor:
+        hidden_states = draft_hidden[:, 1:, :]
+        batch_size, num_steps, _ = hidden_states.shape
+        candidate_ids, unary_logits = self._compute_candidates(
+            hidden_states.reshape(-1, self.hidden_size)
+        )
+        candidate_ids = candidate_ids.view(batch_size, num_steps, self.selector_top_k)
+        unary_logits = unary_logits.view_as(candidate_ids)
+        anchor_token_ids = (
+            block_ids[:, 0]
+            .to(torch.int64)
+            .clamp(0, int(self.model.config.vocab_size) - 1)
+        )
+        scores = self.candidate_selector(
+            candidate_ids,
+            unary_logits,
+            hidden_states,
+            anchor_token_ids,
+        )
+        if self.dflash2_verify_mode == "rejection":
+            if self.sampling_backend is None:
+                raise RuntimeError("DFlash2 sampling backend is not wired")
+            temperatures, coins = self.sampling_backend.dflash2_proposal_state(
+                self.input_buffers.req_pool_indices_buf[:batch_size],
+                batch_size,
+                num_steps,
+            )
+            realized_probs = _walk_sampled_path(
+                candidate_ids,
+                scores,
+                anchor_token_ids,
+                temperatures,
+                coins,
+                next_tokens,
+            )
+            req_pool_indices = self.input_buffers.req_pool_indices_buf[:batch_size]
+            self._draft_candidate_ids_pool.index_copy_(
+                0, req_pool_indices, candidate_ids
+            )
+            self._draft_candidate_probs_pool.index_copy_(
+                0, req_pool_indices, realized_probs
+            )
+        else:
+            _walk_best_path(candidate_ids, scores, anchor_token_ids, next_tokens)
+        next_tokens.clamp_(min=0, max=int(self.model.config.vocab_size) - 1)
+        return next_tokens
diff --git a/python/tokenspeed/runtime/execution/model_executor.py b/python/tokenspeed/runtime/execution/model_executor.py
index 6058875749..c5d36de575 100644
--- a/python/tokenspeed/runtime/execution/model_executor.py
+++ b/python/tokenspeed/runtime/execution/model_executor.py
@@ -438,6 +438,9 @@ def __init__(
                 vocab_size=config.vocab_size,
             )
             self.drafter.wire_target(self.model_runner.model)
+            wire_sampling_backend = getattr(self.drafter, "wire_sampling_backend", None)
+            if wire_sampling_backend is not None:
+                wire_sampling_backend(self.sampling_backend)
             MultimodalRuntime.wire_drafter(
                 self.input_buffers, self.model_runner.model_config
             )
@@ -722,6 +725,7 @@ def _run_sampling(
         sampling_info: SamplingBatchInfo,
         ctx: ForwardContext,
         candidates: torch.Tensor | None = None,
+        draft_distribution=None,
     ):
         if self.drafter is None:
             return self.sampling_backend.sample(logits_output, sampling_info)
@@ -734,7 +738,7 @@ def _run_sampling(
 
         if num_extends == 0:
             output_tokens, accept_lengths = self.sampling_backend.verify(
-                logits_output, sampling_info, candidates
+                logits_output, sampling_info, candidates, draft_distribution
             )
             accept_lengths = self._apply_force_single_token_verify(
                 accept_lengths, 0, num_decodes, ctx.decode_input_ids
@@ -748,7 +752,10 @@ def _run_sampling(
         )
         decode_out = LogitsProcessorOutput(next_token_logits=logits[num_extends:])
         decode_tokens, decode_accept = self.sampling_backend.verify(
-            decode_out, sampling_info[num_extends:], candidates
+            decode_out,
+            sampling_info[num_extends:],
+            candidates,
+            draft_distribution,
         )
         decode_accept = self._apply_force_single_token_verify(
             decode_accept, num_extends, num_decodes, ctx.decode_input_ids
@@ -851,12 +858,22 @@ def _forward_step(
             if self.config.spec_algo is not None
             else None
         )
+        draft_distribution = (
+            self.drafter.get_draft_distribution(ctx)
+            if self.drafter is not None
+            and hasattr(self.drafter, "get_draft_distribution")
+            else None
+        )
 
         if self.capturable_grammar is not None:
             self.capturable_grammar.wait_bitmask()
 
         output_tokens, accept_lengths = self._run_sampling(
-            logits_output, sampling_info, ctx, candidates
+            logits_output,
+            sampling_info,
+            ctx,
+            candidates,
+            draft_distribution,
         )
 
         # Backstop: flag any request whose sampled id falls outside [0, vocab)
diff --git a/python/tokenspeed/runtime/layers/attention/backends/mla.py b/python/tokenspeed/runtime/layers/attention/backends/mla.py
index 8c9cffc0ba..c05a8b0bff 100644
--- a/python/tokenspeed/runtime/layers/attention/backends/mla.py
+++ b/python/tokenspeed/runtime/layers/attention/backends/mla.py
@@ -813,6 +813,8 @@ def forward_decode(
         value_weight = kwargs.get("value_weight")
         gate = kwargs.get("output_gate")
         projected_out = kwargs.get("projected_output")
+        window_left = int(getattr(layer, "sliding_window_size", -1) or -1)
+        noncausal_block_size = self.spec_num_tokens if self._block_decode_active else 1
         if value_weight is not None:
             # Fuse projection and gate into decode to avoid materializing latent output.
             result = mla_decode_with_kvcache(
@@ -829,6 +831,8 @@ def forward_decode(
                 gate=gate,
                 out=projected_out,
                 logit_cap=layer.logit_cap,
+                window_left=window_left,
+                noncausal_block_size=noncausal_block_size,
             )
         else:
             result = mla_decode_with_kvcache(
@@ -843,6 +847,8 @@ def forward_decode(
                 softmax_scale=softmax_scale,
                 logit_cap=layer.logit_cap,
                 solution=self.kernel_solution,
+                window_left=window_left,
+                noncausal_block_size=noncausal_block_size,
             )
         output = self._unwrap_output(result)
         if value_weight is not None:
diff --git a/python/tokenspeed/runtime/models/dflash.py b/python/tokenspeed/runtime/models/dflash.py
index 0222d80928..a312374cf3 100644
--- a/python/tokenspeed/runtime/models/dflash.py
+++ b/python/tokenspeed/runtime/models/dflash.py
@@ -33,6 +33,7 @@
 from tokenspeed.runtime.distributed.mapping import Mapping
 from tokenspeed.runtime.execution.context import ForwardContext
 from tokenspeed.runtime.layers.activation import SiluAndMul
+from tokenspeed.runtime.layers.attention.kv_cache.recipes.spec import FULL_ATTENTION
 from tokenspeed.runtime.layers.dense.unquant import UnquantizedLinearMethod
 from tokenspeed.runtime.layers.layernorm import RMSNorm
 from tokenspeed.runtime.layers.linear import (
@@ -51,6 +52,8 @@
 
 
 class DFlashAttention(nn.Module):
+    cache_group_id = FULL_ATTENTION
+
     def __init__(
         self,
         config,
@@ -132,6 +135,7 @@ def __init__(
             num_kv_heads=self.num_kv_heads,
             layer_id=layer_id,
             sliding_window_size=sliding_window,
+            group_id=self.cache_group_id,
         )
 
     def _apply_qk_norm(
@@ -355,6 +359,8 @@ def forward(
 
 
 class DFlashDraftModel(nn.Module):
+    decoder_layer_cls = DFlashDecoderLayer
+
     def __init__(
         self,
         config,
@@ -368,7 +374,7 @@ def __init__(
         eps = float(getattr(config, "rms_norm_eps", 1e-6))
         self.layers = nn.ModuleList(
             [
-                DFlashDecoderLayer(
+                self.decoder_layer_cls(
                     config=config,
                     mapping=mapping,
                     layer_id=i,
diff --git a/python/tokenspeed/runtime/models/dflash2.py b/python/tokenspeed/runtime/models/dflash2.py
new file mode 100644
index 0000000000..8c2d7296cc
--- /dev/null
+++ b/python/tokenspeed/runtime/models/dflash2.py
@@ -0,0 +1,497 @@
+# Copyright (c) 2026 LightSeek Foundation
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""DFlash2 draft model using the official grouped-conv and selector nodes."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+import torch
+import torch.nn.functional as F
+from torch import nn
+
+from tokenspeed.runtime.distributed.comm_manager import CommManager
+from tokenspeed.runtime.distributed.comm_ops import all_reduce
+from tokenspeed.runtime.distributed.mapping import Mapping
+from tokenspeed.runtime.execution.context import ForwardContext
+from tokenspeed.runtime.layers.attention.kv_cache.recipes.spec import FULL_ATTENTION
+from tokenspeed.runtime.layers.quantization.base_config import QuantizationConfig
+from tokenspeed.runtime.model_loader.weight_utils import default_weight_loader
+from tokenspeed.runtime.models.deepseek_v3 import _prepare_mla_kv_b_proj_weights
+from tokenspeed.runtime.models.dflash import (
+    DFlashDecoderLayer,
+    DFlashDraftModel,
+    _get_dflash_layer_sliding_window,
+)
+from tokenspeed.runtime.models.kimi_k3_dspark import K3DSparkAttention
+from tokenspeed.runtime.utils import add_prefix
+
+
+def _dflash2_config_value(config: Any, key: str, default: Any = None) -> Any:
+    nested = getattr(config, "dflash_config", {}) or {}
+    return nested.get(key, getattr(config, key, default))
+
+
+def _dflash2_uses_mla(config: Any) -> bool:
+    return str(_dflash2_config_value(config, "attention_mode", "gqa")).lower() == "mla"
+
+
+def _dflash2_mla_rope(config: Any) -> tuple[float, dict[str, Any] | None]:
+    parameters = dict(getattr(config, "rope_parameters", None) or {})
+    rope_theta = float(parameters.get("rope_theta", getattr(config, "rope_theta", 1e6)))
+    if str(parameters.get("rope_type", "default")).lower() not in (
+        "yarn",
+        "deepseek_yarn",
+    ):
+        return rope_theta, None
+    scaling = {
+        key: parameters[key]
+        for key in (
+            "factor",
+            "original_max_position_embeddings",
+            "beta_fast",
+            "beta_slow",
+            "mscale",
+            "mscale_all_dim",
+        )
+        if key in parameters
+    }
+    scaling["rope_type"] = "deepseek_yarn"
+    return rope_theta, scaling
+
+
+def _grouped_conv(
+    hidden_states: torch.Tensor,
+    delta: torch.Tensor,
+    base: torch.Tensor,
+    block_size: int,
+    num_groups: int,
+    group_size: int,
+    taps: int,
+) -> torch.Tensor:
+    """Apply DFlash2's grouped dynamic depthwise convolution to flat blocks."""
+    blocks = hidden_states.unflatten(-1, (num_groups, group_size))
+    coefficients = base.view(1, taps, num_groups, group_size) + delta.unsqueeze(-1)
+    output = coefficients[:, 0] * blocks
+    position = torch.arange(hidden_states.shape[0], device=hidden_states.device)
+    if block_size & (block_size - 1) == 0:
+        position = position & (block_size - 1)
+    else:
+        position = position % block_size
+    for tap in range(1, taps):
+        shifted = F.pad(blocks[:-tap], (0, 0, 0, 0, tap, 0))
+        output = output + coefficients[:, tap] * shifted * (position >= tap).view(
+            -1, 1, 1
+        )
+    return output.flatten(-2)
+
+
+class DFlashGroupedConv(nn.Module):
+    """Official DFlash2 grouped convolution, kept as ordinary PyTorch nodes."""
+
+    def __init__(
+        self,
+        hidden_size: int,
+        taps: int,
+        group_size: int,
+        block_size: int,
+        params_dtype: torch.dtype | None = None,
+    ) -> None:
+        super().__init__()
+        if hidden_size % group_size:
+            raise ValueError(
+                f"conv_group_size={group_size} must divide hidden_size={hidden_size}."
+            )
+        if taps < 1 or taps > block_size:
+            raise ValueError(
+                f"conv_kernel_size={taps} must be in [1, block_size={block_size}]."
+            )
+        self.block_size = int(block_size)
+        self.taps = int(taps)
+        self.group_size = int(group_size)
+        self.num_groups = int(hidden_size) // self.group_size
+        self.base_kernel = nn.Parameter(
+            torch.empty(2, self.taps, hidden_size, dtype=params_dtype)
+        )
+        self.kernel_projection = nn.Linear(
+            hidden_size,
+            2 * self.taps * self.num_groups,
+            bias=False,
+            dtype=params_dtype,
+        )
+
+    def _convolve(
+        self, hidden_states: torch.Tensor, delta: torch.Tensor, side: int
+    ) -> torch.Tensor:
+        return _grouped_conv(
+            hidden_states,
+            delta,
+            self.base_kernel[side],
+            self.block_size,
+            self.num_groups,
+            self.group_size,
+            self.taps,
+        )
+
+    def prepare(self, hidden_states: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
+        coefficients = self.kernel_projection(hidden_states).reshape(
+            hidden_states.shape[0], 2, self.taps, self.num_groups
+        )
+        return self._convolve(hidden_states, coefficients[:, 0], 0), coefficients[:, 1]
+
+    def finish(
+        self, hidden_states: torch.Tensor, coefficients: torch.Tensor
+    ) -> torch.Tensor:
+        return self._convolve(hidden_states, coefficients, 1)
+
+
+def _score_edges(
+    predecessor_table: torch.Tensor,
+    successor_table: torch.Tensor,
+    candidate_ids: torch.Tensor,
+    unary_logits: torch.Tensor,
+    hidden: torch.Tensor,
+    anchor_token_ids: torch.Tensor,
+    top_k: int,
+) -> torch.Tensor:
+    successors = successor_table[candidate_ids]
+    predecessor_ids = torch.cat(
+        (
+            anchor_token_ids[:, None, None].expand(-1, 1, top_k),
+            candidate_ids[:, :-1],
+        ),
+        dim=1,
+    )
+    predecessors = predecessor_table[predecessor_ids]
+    return unary_logits[:, :, None] + torch.einsum(
+        "blpr,blcr->blpc", predecessors * hidden[:, :, None], successors
+    )
+
+
+class CandidateSelector(nn.Module):
+    """Score top-k token lattices with DFlash2's low-rank transition model."""
+
+    def __init__(
+        self,
+        hidden_size: int,
+        vocab_size: int,
+        rank: int,
+        top_k: int,
+        params_dtype: torch.dtype | None = None,
+    ) -> None:
+        super().__init__()
+        if rank < 1:
+            raise ValueError(f"selector_rank must be positive, got {rank}.")
+        if not 2 <= top_k <= vocab_size:
+            raise ValueError(
+                f"selector_top_k must be in [2, {vocab_size}], got {top_k}."
+            )
+        self.top_k = int(top_k)
+        self.predecessor_codebook = nn.Parameter(
+            torch.empty(vocab_size, rank, dtype=params_dtype)
+        )
+        self.successor_codebook = nn.Parameter(
+            torch.empty(vocab_size, rank, dtype=params_dtype)
+        )
+        self.hidden_projection = nn.Linear(
+            hidden_size, rank, bias=False, dtype=params_dtype
+        )
+
+    def forward(
+        self,
+        candidate_ids: torch.Tensor,
+        unary_logits: torch.Tensor,
+        hidden_states: torch.Tensor,
+        anchor_token_ids: torch.Tensor,
+    ) -> torch.Tensor:
+        hidden = self.hidden_projection(hidden_states)
+        return _score_edges(
+            self.predecessor_codebook,
+            self.successor_codebook,
+            candidate_ids,
+            unary_logits,
+            hidden,
+            anchor_token_ids,
+            self.top_k,
+        )
+
+
+class DFlash2DecoderLayer(DFlashDecoderLayer):
+    """DFlash layer with dynamic convolutions around attention and the MLP."""
+
+    def __init__(
+        self,
+        config,
+        mapping: Mapping,
+        layer_id: int,
+        quant_config: QuantizationConfig | None = None,
+        prefix: str = "",
+    ) -> None:
+        super().__init__(
+            config=config,
+            mapping=mapping,
+            layer_id=layer_id,
+            quant_config=quant_config,
+            prefix=prefix,
+        )
+        self.layer_id = int(layer_id)
+        self._uses_mla = _dflash2_uses_mla(config)
+        self.comm_manager = None
+        if self._uses_mla:
+            rope_theta, rope_scaling = _dflash2_mla_rope(config)
+            self.self_attn = K3DSparkAttention(
+                config=config,
+                mapping=mapping,
+                hidden_size=int(config.hidden_size),
+                num_heads=int(config.num_attention_heads),
+                qk_nope_head_dim=int(config.qk_nope_head_dim),
+                qk_rope_head_dim=int(config.qk_rope_head_dim),
+                v_head_dim=int(config.v_head_dim),
+                q_lora_rank=int(config.q_lora_rank),
+                kv_lora_rank=int(config.kv_lora_rank),
+                rope_theta=rope_theta,
+                rope_scaling=rope_scaling,
+                # Match training: grow the YaRN cache on demand instead of
+                # materializing K3's one-million-token limit in every layer.
+                max_position_embeddings=min(
+                    int(getattr(config, "max_position_embeddings", 32768)), 32768
+                ),
+                quant_config=quant_config,
+                layer_id=layer_id,
+                prefix=add_prefix("self_attn", prefix),
+                reduce_attn_results=False,
+            )
+            sliding_window = _get_dflash_layer_sliding_window(config, layer_id)
+            for attention in (self.self_attn.attn_mqa, self.self_attn.attn_mha):
+                attention.cache_group_id = FULL_ATTENTION
+                attention.group_id = FULL_ATTENTION
+                # Storage remains in Kimi-K3's full-attention group. This field
+                # is only the compute visibility contract for the MLA backend.
+                attention.sliding_window_size = sliding_window
+            self.comm_manager = CommManager(
+                mapping=mapping,
+                layer_id=layer_id,
+                is_moe=False,
+                prev_is_moe=False,
+                input_layernorm=self.input_layernorm,
+                post_attn_layernorm=self.post_attention_layernorm,
+            )
+        conv_args = dict(
+            hidden_size=int(config.hidden_size),
+            taps=int(_dflash2_config_value(config, "conv_kernel_size")),
+            group_size=int(_dflash2_config_value(config, "conv_group_size")),
+            block_size=int(_dflash2_config_value(config, "block_size")),
+        )
+        self.attention_conv = DFlashGroupedConv(**conv_args)
+        self.mlp_conv = DFlashGroupedConv(**conv_args)
+
+    def forward(
+        self,
+        positions: torch.Tensor,
+        hidden_states: torch.Tensor,
+        ctx: ForwardContext,
+        out_cache_loc: torch.Tensor,
+        residual: torch.Tensor | None,
+    ) -> tuple[torch.Tensor, torch.Tensor]:
+        if ctx.forward_mode.is_idle():
+            return super().forward(
+                positions, hidden_states, ctx, out_cache_loc, residual
+            )
+
+        if residual is None:
+            residual = hidden_states
+            hidden_states = self.input_layernorm(hidden_states)
+        else:
+            # Borrowed vocab-parallel embeddings are shard-local partials. The
+            # grouped conv needs a complete hidden row, while every later layer
+            # already receives the all-reduced output of the preceding MLP
+            # conv. Reduce exactly once, at the first draft layer.
+            if self.layer_id == 0 and self.mapping.dense.tp_size > 1:
+                hidden_states = all_reduce(hidden_states, self.mapping.dense.tp_group)
+            hidden_states, residual = self.input_layernorm(hidden_states, residual)
+
+        hidden_states, coefficients = self.attention_conv.prepare(hidden_states)
+        attention_kwargs = dict(
+            positions=positions,
+            hidden_states=hidden_states,
+            ctx=ctx,
+            out_cache_loc=out_cache_loc,
+        )
+        if self.comm_manager is not None:
+            attention_kwargs["comm_manager"] = self.comm_manager
+        hidden_states = self.self_attn(**attention_kwargs)
+        if self.mapping.attn.tp_size > 1:
+            hidden_states = all_reduce(hidden_states, self.mapping.attn.tp_group)
+        hidden_states = self.attention_conv.finish(hidden_states, coefficients)
+
+        hidden_states, residual = self.post_attention_layernorm(hidden_states, residual)
+        hidden_states, coefficients = self.mlp_conv.prepare(hidden_states)
+        hidden_states = self.mlp(hidden_states)
+        if self.mapping.dense.tp_size > 1:
+            hidden_states = all_reduce(hidden_states, self.mapping.dense.tp_group)
+        hidden_states = self.mlp_conv.finish(hidden_states, coefficients)
+        return hidden_states, residual
+
+
+class DFlash2DraftModel(DFlashDraftModel):
+    """DFlash2 checkpoint entry class."""
+
+    decoder_layer_cls = DFlash2DecoderLayer
+
+    def __init__(
+        self,
+        config,
+        mapping: Mapping,
+        quant_config: QuantizationConfig | None = None,
+        prefix: str = "",
+    ) -> None:
+        super().__init__(
+            config=config,
+            mapping=mapping,
+            quant_config=quant_config,
+            prefix=prefix,
+        )
+        dtype = self.fc.weight.dtype
+        self.input_embedding_scale = float(
+            _dflash2_config_value(config, "input_embedding_scale", 1.0)
+        )
+        self.block_size = int(_dflash2_config_value(config, "block_size"))
+        self.candidate_selector = CandidateSelector(
+            hidden_size=int(config.hidden_size),
+            vocab_size=int(config.vocab_size),
+            rank=int(_dflash2_config_value(config, "selector_rank")),
+            top_k=int(_dflash2_config_value(config, "selector_top_k")),
+            params_dtype=dtype,
+        )
+
+    @property
+    def _uses_mla(self) -> bool:
+        return _dflash2_uses_mla(self.config)
+
+    @torch.no_grad()
+    def write_context_kv(
+        self,
+        ctx_hidden: torch.Tensor,
+        positions: torch.Tensor,
+        cache_locs: torch.Tensor,
+        token_to_kv_pool,
+    ) -> None:
+        if not self._uses_mla:
+            return super().write_context_kv(
+                ctx_hidden, positions, cache_locs, token_to_kv_pool
+            )
+        if ctx_hidden.shape[0] == 0:
+            return
+        for layer in self.layers:
+            attn = layer.self_attn
+            latent = attn.project_latent_kv(ctx_hidden)
+            latent = attn.apply_latent_rope(positions, latent)
+            token_to_kv_pool.set_mla_kv_buffer(
+                attn.attn_mqa,
+                cache_locs,
+                latent[..., : attn.kv_lora_rank].contiguous(),
+                latent[..., attn.kv_lora_rank :].contiguous(),
+            )
+
+    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]):
+        if not self._uses_mla:
+            return super().load_weights(weights)
+
+        params = dict(self.named_parameters())
+        loaded: set[str] = set()
+        unexpected: list[str] = []
+        stacked = (
+            ("gate_up_proj", "gate_proj", 0),
+            ("gate_up_proj", "up_proj", 1),
+        )
+        fused_qkv_a_offsets = {
+            "q_a_proj": 0,
+            "kv_a_proj_with_mqa": int(self.config.q_lora_rank),
+        }
+
+        for name, loaded_weight in weights:
+            name = name.removeprefix("model.")
+            if name == "embed_tokens.weight" or "rotary_emb.inv_freq" in name:
+                continue
+
+            for param_name, weight_name, shard_id in stacked:
+                if f".{weight_name}." not in name:
+                    continue
+                target = name.replace(weight_name, param_name)
+                param = params.get(target)
+                if param is None:
+                    unexpected.append(name)
+                    break
+                param.weight_loader(param, loaded_weight, shard_id)
+                loaded.add(target)
+                break
+            else:
+                fused_key = next(
+                    (key for key in fused_qkv_a_offsets if f".{key}." in name),
+                    None,
+                )
+                if fused_key is not None:
+                    target = name.replace(fused_key, "fused_qkv_a_proj_with_mqa")
+                    param = params.get(target)
+                    if param is None:
+                        unexpected.append(name)
+                        continue
+                    param.weight_loader(
+                        param,
+                        loaded_weight,
+                        begin_size=fused_qkv_a_offsets[fused_key],
+                    )
+                    loaded.add(target)
+                    continue
+
+                param = params.get(name)
+                if param is None:
+                    unexpected.append(name)
+                    continue
+                loader = getattr(param, "weight_loader", default_weight_loader)
+                loader(param, loaded_weight)
+                loaded.add(name)
+
+        if unexpected:
+            raise ValueError(
+                f"DFlash2 MLA checkpoint has {len(unexpected)} unexpected weights: "
+                f"{sorted(unexpected)[:8]}"
+            )
+        missing = sorted(set(params) - loaded)
+        if missing:
+            raise ValueError(
+                f"DFlash2 MLA checkpoint is missing {len(missing)} weights: {missing[:8]}"
+            )
+        for layer in self.layers:
+            self_attn = layer.self_attn
+            self_attn.w_kc, self_attn.w_vc = _prepare_mla_kv_b_proj_weights(
+                self_attn.kv_b_proj.weight, self_attn
+            )
+
+    @torch.no_grad()
+    def forward(self, *args, input_embeds: torch.Tensor | None = None, **kwargs):
+        if input_embeds is not None and self.input_embedding_scale != 1.0:
+            input_embeds = input_embeds * self.input_embedding_scale
+        return super().forward(*args, input_embeds=input_embeds, **kwargs)
+
+
+EntryClass = [DFlash2DraftModel]
diff --git a/python/tokenspeed/runtime/sampling/backends/base.py b/python/tokenspeed/runtime/sampling/backends/base.py
index f970325bee..52228d6c2a 100644
--- a/python/tokenspeed/runtime/sampling/backends/base.py
+++ b/python/tokenspeed/runtime/sampling/backends/base.py
@@ -30,6 +30,7 @@
 if TYPE_CHECKING:
     from tokenspeed.runtime.layers.logits_processor import LogitsProcessorOutput
     from tokenspeed.runtime.sampling.dp_sampling_config import DpSamplingRuntimeConfig
+    from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
     from tokenspeed.runtime.sampling.sampling_batch_info import SamplingBatchInfo
     from tokenspeed.runtime.sampling.sampling_params import SamplingParams
     from tokenspeed.runtime.utils.server_args import ServerArgs
@@ -118,6 +119,7 @@ class SamplingBackend(ABC):
     # a no-op.
     _HAS_POOL_STATE: bool = False
     _SUPPORTS_DP_VERIFY: bool = False
+    dflash2_verify_mode = "unsupported"
 
     def __init__(self, config: SamplingBackendConfig) -> None:
 
@@ -269,6 +271,17 @@ def get_packed_output_d2h(
         return None and let the caller fall back to two separate D2Hs."""
         return None
 
+    def dflash2_proposal_state(
+        self,
+        req_pool_indices: torch.Tensor,
+        batch_size: int,
+        num_steps: int,
+    ) -> tuple[torch.Tensor, torch.Tensor]:
+        """Return per-row temperatures and independent proposal coins."""
+        raise RuntimeError(
+            f"{type(self).__name__} does not support stochastic DFlash2 proposals"
+        )
+
     @abstractmethod
     def sample(
         self,
@@ -282,4 +295,5 @@ def verify(
         logits_output: LogitsProcessorOutput,
         sampling_info: SamplingBatchInfo,
         candidates: torch.Tensor,
+        draft_distribution: SparseDraftDistribution | None = None,
     ) -> tuple[torch.Tensor, torch.Tensor]: ...
diff --git a/python/tokenspeed/runtime/sampling/backends/flashinfer.py b/python/tokenspeed/runtime/sampling/backends/flashinfer.py
index 62caa8cda8..22c1b0a85c 100644
--- a/python/tokenspeed/runtime/sampling/backends/flashinfer.py
+++ b/python/tokenspeed/runtime/sampling/backends/flashinfer.py
@@ -55,6 +55,7 @@
     DpSamplingRuntimeConfig,
     slice_dp_vocab_mask,
 )
+from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
 from tokenspeed.runtime.sampling.registry import register_backend
 from tokenspeed.runtime.sampling.utils import (
     coin_eps,
@@ -82,6 +83,7 @@ class FlashInferSamplingBackend(SamplingBackend):
 
     _HAS_POOL_STATE = True
     _SUPPORTS_DP_VERIFY = True
+    dflash2_verify_mode = "rejection"
 
     def __init__(self, config: SamplingBackendConfig) -> None:
 
@@ -231,6 +233,11 @@ def _init_shared_buffers(self, config: SamplingBackendConfig) -> None:
         self._final_coins_buf = torch.zeros(
             (max_pad_bs,), dtype=torch.float32, device=config.device
         )
+        self._draft_coins_buf = torch.zeros(
+            (max_pad_bs, max(max_n - 1, 1)),
+            dtype=torch.float32,
+            device=config.device,
+        )
 
         # Stub generator used during CUDA-graph capture/warm-up (no requests yet).
         self._capture_gen = torch.Generator(device=config.device)
@@ -281,10 +288,16 @@ def _prepare_step_hook(
         if request_pool_indices is None:
             self._coins_buf[:bs, :n].uniform_(lo, 1.0, generator=self._capture_gen)
             self._final_coins_buf[:bs].uniform_(lo, 1.0, generator=self._capture_gen)
+            self._draft_coins_buf[:bs, : max(n - 1, 1)].uniform_(
+                lo, 1.0, generator=self._capture_gen
+            )
             return
 
         cpu_coins = torch.empty((bs, n), dtype=torch.float32, pin_memory=True)
         cpu_final = torch.empty((bs,), dtype=torch.float32, pin_memory=True)
+        cpu_draft = torch.empty(
+            (bs, max(n - 1, 1)), dtype=torch.float32, pin_memory=True
+        )
 
         for i, pool_idx in enumerate(request_pool_indices):
             gen = self._cpu_generator_per_slot[pool_idx]
@@ -295,9 +308,27 @@ def _prepare_step_hook(
                 )
             cpu_coins[i, :n].uniform_(lo, 1.0, generator=gen)
             cpu_final[i].uniform_(lo, 1.0, generator=gen)
+            cpu_draft[i].uniform_(lo, 1.0, generator=gen)
 
         self._coins_buf[:bs, :n].copy_(cpu_coins, non_blocking=True)
         self._final_coins_buf[:bs].copy_(cpu_final, non_blocking=True)
+        self._draft_coins_buf[:bs, : max(n - 1, 1)].copy_(cpu_draft, non_blocking=True)
+
+    def dflash2_proposal_state(
+        self,
+        req_pool_indices: torch.Tensor,
+        batch_size: int,
+        num_steps: int,
+    ) -> tuple[torch.Tensor, torch.Tensor]:
+        if num_steps > self._draft_coins_buf.shape[1]:
+            raise ValueError(
+                f"DFlash2 needs {num_steps} proposal coins but the sampler "
+                f"allocated {self._draft_coins_buf.shape[1]}"
+            )
+        temperatures = self._temperature_pool.index_select(
+            0, req_pool_indices[:batch_size]
+        )
+        return temperatures, self._draft_coins_buf[:batch_size, :num_steps]
 
     @nvtx_range("sampling:sample", color="yellow")
     def sample(
@@ -366,6 +397,7 @@ def verify(
         logits_output: LogitsProcessorOutput,
         sampling_info: SamplingBatchInfo,
         candidates: torch.Tensor,
+        draft_distribution: SparseDraftDistribution | None = None,
     ) -> tuple[torch.Tensor, torch.Tensor]:
 
         bs = candidates.shape[0]
@@ -410,9 +442,25 @@ def verify(
                 pool_indices = torch.nn.functional.pad(
                     sampling_info.req_pool_indices, (0, pad_bs - effective_bs)
                 )[shard]
+                if draft_distribution is not None:
+                    draft_distribution = SparseDraftDistribution(
+                        candidate_ids=torch.nn.functional.pad(
+                            draft_distribution.candidate_ids,
+                            (0, 0, 0, 0, 0, pad_bs - effective_bs),
+                        )[shard],
+                        probabilities=torch.nn.functional.pad(
+                            draft_distribution.probabilities,
+                            (0, 0, 0, 0, 0, pad_bs - effective_bs),
+                        )[shard],
+                    )
             else:
                 candidates = candidates[shard]
                 pool_indices = sampling_info.req_pool_indices[shard]
+                if draft_distribution is not None:
+                    draft_distribution = SparseDraftDistribution(
+                        candidate_ids=draft_distribution.candidate_ids[shard],
+                        probabilities=draft_distribution.probabilities[shard],
+                    )
             vocab_mask = slice_dp_vocab_mask(
                 vocab_mask,
                 full_bs=effective_bs,
@@ -517,20 +565,36 @@ def verify(
                 )
             target_probs = target_probs.reshape(bs, n, -1)
 
-            chain_speculative_sampling_target_only(
-                predicts=predict,
-                accept_index=accept_index,
-                accept_token_num=accept_length,
-                candidates=candidates,
-                uniform_samples=coins[:bs, :n],
-                uniform_samples_for_final_sampling=final_coins[:bs],
-                target_probs=target_probs,
-                draft_probs=None,
-                threshold_single=SPECULATIVE_ACCEPT_THRESHOLD_SINGLE,
-                threshold_acc=SPECULATIVE_ACCEPT_THRESHOLD_ACC,
-                deterministic=not dp_sampling,
-                enable_pdl=pdl_enabled(),
-            )
+            if draft_distribution is None:
+                chain_speculative_sampling_target_only(
+                    predicts=predict,
+                    accept_index=accept_index,
+                    accept_token_num=accept_length,
+                    candidates=candidates,
+                    uniform_samples=coins[:bs, :n],
+                    uniform_samples_for_final_sampling=final_coins[:bs],
+                    target_probs=target_probs,
+                    draft_probs=None,
+                    threshold_single=SPECULATIVE_ACCEPT_THRESHOLD_SINGLE,
+                    threshold_acc=SPECULATIVE_ACCEPT_THRESHOLD_ACC,
+                    deterministic=not dp_sampling,
+                    enable_pdl=pdl_enabled(),
+                )
+            else:
+                from tokenspeed.runtime.sampling.dflash2 import (
+                    verify_sparse_draft_distribution,
+                )
+
+                verify_sparse_draft_distribution(
+                    predicts=predict,
+                    accept_index=accept_index,
+                    accept_token_num=accept_length,
+                    candidates=candidates,
+                    target_probs=target_probs,
+                    draft_distribution=draft_distribution,
+                    acceptance_coins=coins[:bs, : n - 1],
+                    final_coins=final_coins[:bs],
+                )
 
         accept_length += 1
         logprobs_local = None
diff --git a/python/tokenspeed/runtime/sampling/backends/flashinfer_full.py b/python/tokenspeed/runtime/sampling/backends/flashinfer_full.py
index db8070eb82..26db6688b7 100644
--- a/python/tokenspeed/runtime/sampling/backends/flashinfer_full.py
+++ b/python/tokenspeed/runtime/sampling/backends/flashinfer_full.py
@@ -58,6 +58,7 @@
 
 if TYPE_CHECKING:
     from tokenspeed.runtime.layers.logits_processor import LogitsProcessorOutput
+    from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
     from tokenspeed.runtime.sampling.sampling_batch_info import SamplingBatchInfo
     from tokenspeed.runtime.sampling.sampling_params import SamplingParams
 
@@ -369,6 +370,7 @@ def verify(
         logits_output: LogitsProcessorOutput,
         sampling_info: SamplingBatchInfo,
         candidates: torch.Tensor,
+        draft_distribution: SparseDraftDistribution | None = None,
     ) -> tuple[torch.Tensor, torch.Tensor]:
 
         bs = candidates.shape[0]
@@ -445,20 +447,36 @@ def verify(
         coins = self._coins_buf[:bs, :num_tokens_per_req]
         coins_for_final_sampling = self._final_coins_buf[:bs]
 
-        chain_speculative_sampling_target_only(
-            predicts=predict,
-            accept_index=accept_index,
-            accept_token_num=accept_length,
-            candidates=candidates.to(torch.int32),
-            uniform_samples=coins,
-            uniform_samples_for_final_sampling=coins_for_final_sampling,
-            target_probs=target_probs,
-            draft_probs=None,
-            threshold_single=SPECULATIVE_ACCEPT_THRESHOLD_SINGLE,
-            threshold_acc=SPECULATIVE_ACCEPT_THRESHOLD_ACC,
-            deterministic=True,
-            enable_pdl=pdl_enabled(),
-        )
+        if draft_distribution is None:
+            chain_speculative_sampling_target_only(
+                predicts=predict,
+                accept_index=accept_index,
+                accept_token_num=accept_length,
+                candidates=candidates.to(torch.int32),
+                uniform_samples=coins,
+                uniform_samples_for_final_sampling=coins_for_final_sampling,
+                target_probs=target_probs,
+                draft_probs=None,
+                threshold_single=SPECULATIVE_ACCEPT_THRESHOLD_SINGLE,
+                threshold_acc=SPECULATIVE_ACCEPT_THRESHOLD_ACC,
+                deterministic=True,
+                enable_pdl=pdl_enabled(),
+            )
+        else:
+            from tokenspeed.runtime.sampling.dflash2 import (
+                verify_sparse_draft_distribution,
+            )
+
+            verify_sparse_draft_distribution(
+                predicts=predict,
+                accept_index=accept_index,
+                accept_token_num=accept_length,
+                candidates=candidates,
+                target_probs=target_probs,
+                draft_distribution=draft_distribution,
+                acceptance_coins=coins[:, : num_tokens_per_req - 1],
+                final_coins=coins_for_final_sampling,
+            )
 
         accept_length += 1
 
diff --git a/python/tokenspeed/runtime/sampling/backends/greedy.py b/python/tokenspeed/runtime/sampling/backends/greedy.py
index 97dc4b28e4..132e5c803f 100644
--- a/python/tokenspeed/runtime/sampling/backends/greedy.py
+++ b/python/tokenspeed/runtime/sampling/backends/greedy.py
@@ -41,6 +41,7 @@
 if TYPE_CHECKING:
 
     from tokenspeed.runtime.layers.logits_processor import LogitsProcessorOutput
+    from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
     from tokenspeed.runtime.sampling.sampling_batch_info import SamplingBatchInfo
 
 
@@ -121,6 +122,7 @@ def _verify_chain_greedy(
 
 
 class GreedySamplingBackend(SamplingBackend):
+    dflash2_verify_mode = "greedy"
     """Greedy-only backend: argmax for single-step, chain-greedy verify for
     multi-step verification. No flashinfer / min_p / penalty machinery, no
     coin buffers. Verify uses the fused CUDA kernel when available; falls
@@ -198,6 +200,7 @@ def verify(
         logits_output: LogitsProcessorOutput,
         sampling_info: SamplingBatchInfo,
         candidates: torch.Tensor,
+        draft_distribution: SparseDraftDistribution | None = None,
     ) -> tuple[torch.Tensor, torch.Tensor]:
 
         bs = candidates.shape[0]
diff --git a/python/tokenspeed/runtime/sampling/backends/triton.py b/python/tokenspeed/runtime/sampling/backends/triton.py
index 47698ffde7..d0ca9e2990 100644
--- a/python/tokenspeed/runtime/sampling/backends/triton.py
+++ b/python/tokenspeed/runtime/sampling/backends/triton.py
@@ -55,6 +55,7 @@
 
 if TYPE_CHECKING:
     from tokenspeed.runtime.layers.logits_processor import LogitsProcessorOutput
+    from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
     from tokenspeed.runtime.sampling.sampling_batch_info import SamplingBatchInfo
     from tokenspeed.runtime.sampling.sampling_params import SamplingParams
 
@@ -541,7 +542,13 @@ def verify(
         logits_output: LogitsProcessorOutput,
         sampling_info: SamplingBatchInfo,
         candidates: torch.Tensor,
+        draft_distribution: SparseDraftDistribution | None = None,
     ) -> tuple[torch.Tensor, torch.Tensor]:
+        if draft_distribution is not None:
+            raise RuntimeError(
+                "TritonSamplingBackend does not yet implement DFlash2 q rejection; "
+                "use --sampling-backend flashinfer or flashinfer_full"
+            )
         bs = candidates.shape[0]
         num_tokens_per_req = candidates.shape[1]
 
diff --git a/python/tokenspeed/runtime/sampling/backends/triton_full.py b/python/tokenspeed/runtime/sampling/backends/triton_full.py
index 292d102706..78edfd5045 100644
--- a/python/tokenspeed/runtime/sampling/backends/triton_full.py
+++ b/python/tokenspeed/runtime/sampling/backends/triton_full.py
@@ -69,6 +69,7 @@
 
 if TYPE_CHECKING:
     from tokenspeed.runtime.layers.logits_processor import LogitsProcessorOutput
+    from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
     from tokenspeed.runtime.sampling.sampling_batch_info import SamplingBatchInfo
     from tokenspeed.runtime.sampling.sampling_params import SamplingParams
 
@@ -485,7 +486,13 @@ def verify(
         logits_output: LogitsProcessorOutput,
         sampling_info: SamplingBatchInfo,
         candidates: torch.Tensor,
+        draft_distribution: SparseDraftDistribution | None = None,
     ) -> tuple[torch.Tensor, torch.Tensor]:
+        if draft_distribution is not None:
+            raise RuntimeError(
+                "TritonFullSamplingBackend does not yet implement DFlash2 q "
+                "rejection; use --sampling-backend flashinfer or flashinfer_full"
+            )
         bs = candidates.shape[0]
         num_tokens_per_req = candidates.shape[1]
 
diff --git a/python/tokenspeed/runtime/sampling/dflash2.py b/python/tokenspeed/runtime/sampling/dflash2.py
new file mode 100644
index 0000000000..d87f28035a
--- /dev/null
+++ b/python/tokenspeed/runtime/sampling/dflash2.py
@@ -0,0 +1,118 @@
+# Copyright (c) 2026 LightSeek Foundation
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""Lossless DFlash2 rejection sampling from a sparse q distribution."""
+
+from __future__ import annotations
+
+import torch
+
+from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
+
+
+def verify_sparse_draft_distribution(
+    *,
+    predicts: torch.Tensor,
+    accept_index: torch.Tensor,
+    accept_token_num: torch.Tensor,
+    candidates: torch.Tensor,
+    target_probs: torch.Tensor,
+    draft_distribution: SparseDraftDistribution,
+    acceptance_coins: torch.Tensor,
+    final_coins: torch.Tensor,
+) -> None:
+    """Losslessly verify a candidate chain against sparse DFlash2 q.
+
+    ``candidates`` and ``target_probs`` use TokenSpeed's anchor-plus-drafts
+    layout ``[B, N]`` and ``[B, N, V]``. The first candidate is the verified
+    anchor. ``draft_distribution`` holds q for candidates 1..N-1. Accepted
+    drafts use ``min(1, p(token) / q(token))``; the first rejection is sampled
+    from normalized ``relu(p - q)``. If every draft is accepted, the final
+    token is sampled from the target's bonus row.
+
+    All control flow is static in ``N`` and all random values arrive through
+    persistent buffers, so the operation is CUDA-graph capturable.
+    """
+
+    batch_size, num_tokens = candidates.shape
+    num_steps = num_tokens - 1
+    if draft_distribution.candidate_ids.shape[:2] != (batch_size, num_steps):
+        raise ValueError("DFlash2 sparse candidate shape does not match verify width")
+    if draft_distribution.probabilities.shape != draft_distribution.candidate_ids.shape:
+        raise ValueError("DFlash2 sparse candidate IDs and probabilities must align")
+    if target_probs.shape[:2] != (batch_size, num_tokens):
+        raise ValueError("DFlash2 target probability shape does not match candidates")
+
+    predict_rows = predicts.view(batch_size, num_tokens)
+    predict_rows.zero_()
+    accept_index.fill_(-1)
+    accept_token_num.zero_()
+
+    row_ids = torch.arange(batch_size, device=candidates.device, dtype=torch.int64)
+    flat_base = row_ids * num_tokens
+    accept_index[:, 0].copy_(flat_base.to(accept_index.dtype))
+    alive = torch.ones(batch_size, dtype=torch.bool, device=candidates.device)
+
+    sparse_ids = draft_distribution.candidate_ids.to(torch.int64)
+    sparse_probs = draft_distribution.probabilities.float()
+    eps = torch.finfo(torch.float32).tiny
+
+    for step in range(num_steps):
+        proposed = candidates[:, step + 1].to(torch.int64)
+        p_token = target_probs[:, step].gather(1, proposed[:, None]).squeeze(1)
+        q_token = (
+            sparse_probs[:, step]
+            * (sparse_ids[:, step] == proposed[:, None]).to(torch.float32)
+        ).sum(dim=-1)
+        ratio = (p_token / q_token.clamp_min(eps)).clamp(max=1.0)
+        accepted = alive & (acceptance_coins[:, step] <= ratio)
+        predict_rows[:, step].copy_(
+            torch.where(accepted, proposed, torch.zeros_like(proposed)).to(
+                predict_rows.dtype
+            )
+        )
+        accept_index[:, step + 1].copy_(
+            torch.where(
+                accepted,
+                flat_base + step + 1,
+                torch.full_like(flat_base, -1),
+            ).to(accept_index.dtype)
+        )
+        accept_token_num.add_(accepted.to(accept_token_num.dtype))
+        alive = alive & accepted
+
+    final_position = accept_token_num.to(torch.int64)
+    target_final = target_probs[row_ids, final_position]
+    rejected = ~alive
+    q_position = final_position.clamp(max=max(num_steps - 1, 0))
+    q_ids = sparse_ids[row_ids, q_position]
+    q_values = sparse_probs[row_ids, q_position] * rejected[:, None]
+
+    residual = target_final.clone()
+    residual.scatter_add_(1, q_ids, -q_values)
+    residual.clamp_min_(0.0)
+    residual_mass = residual.sum(dim=-1, keepdim=True)
+    residual = torch.where(residual_mass > 0, residual, target_final)
+    cumulative = residual.cumsum(dim=-1)
+    cutoff = final_coins[:, None] * cumulative[:, -1:]
+    sampled = (cumulative < cutoff).sum(dim=-1).clamp(max=target_probs.shape[-1] - 1)
+    predict_rows.scatter_(
+        1, final_position[:, None], sampled[:, None].to(predict_rows.dtype)
+    )
diff --git a/python/tokenspeed/runtime/sampling/draft_distribution.py b/python/tokenspeed/runtime/sampling/draft_distribution.py
new file mode 100644
index 0000000000..3362e09a2b
--- /dev/null
+++ b/python/tokenspeed/runtime/sampling/draft_distribution.py
@@ -0,0 +1,39 @@
+# Copyright (c) 2026 LightSeek Foundation
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import torch
+
+
+@dataclass(frozen=True)
+class SparseDraftDistribution:
+    """Per-step sparse draft probabilities for lossless verification.
+
+    Both tensors have shape ``[batch, draft_steps, top_k]``. Candidate IDs are
+    global vocabulary IDs and probabilities are normalized independently for
+    every batch row and draft step. The target's bonus-token row has no draft
+    distribution and is therefore not represented here.
+    """
+
+    candidate_ids: torch.Tensor
+    probabilities: torch.Tensor
diff --git a/test/runtime/layers/test_mla_verify_metadata.py b/test/runtime/layers/test_mla_verify_metadata.py
index feec84728e..37973380fa 100644
--- a/test/runtime/layers/test_mla_verify_metadata.py
+++ b/test/runtime/layers/test_mla_verify_metadata.py
@@ -15,25 +15,31 @@ def _run_mla_decode(
     q_len_per_req: int = 2,
     data_type: torch.dtype = torch.float32,
     page_size: int = 16,
+    draft_block_decode: bool = False,
+    sliding_window_size: int = -1,
 ) -> dict[str, torch.Tensor]:
     captured = {}
 
     def fake_mla_decode_with_kvcache(**kwargs):
-        captured["q"] = kwargs["q"]
-        captured["cache_seqlens"] = kwargs["cache_seqlens"]
+        captured.update(kwargs)
         return torch.zeros(*kwargs["q"].shape[:-1], 4)
 
     monkeypatch.setattr(
         mla_backend, "mla_decode_with_kvcache", fake_mla_decode_with_kvcache
     )
     backend = object.__new__(mla_backend.MLAAttnBackend)
+    metadata_rows = bs * q_len_per_req if draft_block_decode else bs
+    seq_lens = torch.tensor([64, 128], dtype=torch.int32)[:bs]
+    if draft_block_decode:
+        seq_lens = seq_lens.repeat_interleave(q_len_per_req)
     backend.forward_decode_metadata = SimpleNamespace(
         num_extends=0,
-        page_table=torch.zeros(bs, 1, dtype=torch.int32),
-        seq_lens=torch.tensor([64, 128], dtype=torch.int32)[:bs],
+        page_table=torch.zeros(metadata_rows, 1, dtype=torch.int32),
+        seq_lens=seq_lens,
     )
     backend.is_draft = is_draft
-    backend.draft_block_decode = False
+    backend.draft_block_decode = draft_block_decode
+    backend.spec_num_tokens = q_len_per_req if draft_block_decode else 1
     backend.max_context_len = 256
     backend.kernel_page_size = page_size
     backend.kv_lora_rank = 2
@@ -51,6 +57,7 @@ def fake_mla_decode_with_kvcache(**kwargs):
         logit_cap=0.0,
         k_scale_float=None,
         layer_id=0,
+        sliding_window_size=sliding_window_size,
     )
     token_to_kv_pool = SimpleNamespace(
         get_key_buffer=lambda layer_id: torch.zeros(page_size, 4).to(data_type)
@@ -92,3 +99,19 @@ def test_fp8_decode_dispatches_with_native_fp8_query(monkeypatch):
     )
 
     assert captured["q"].dtype == torch.float8_e4m3fn
+
+
+def test_dflash2_block_decode_passes_exact_sliding_window(monkeypatch):
+    captured = _run_mla_decode(
+        monkeypatch,
+        is_draft=True,
+        bs=2,
+        q_len_per_req=8,
+        draft_block_decode=True,
+        # PagedAttention stores HF's inclusive window as window_left.
+        sliding_window_size=4095,
+    )
+
+    assert captured["window_left"] == 4095
+    assert captured["noncausal_block_size"] == 8
+    assert captured["q"].shape[0] == 16
diff --git a/test/runtime/test_dflash2.py b/test/runtime/test_dflash2.py
new file mode 100644
index 0000000000..184291c8c7
--- /dev/null
+++ b/test/runtime/test_dflash2.py
@@ -0,0 +1,327 @@
+# Copyright (c) 2026 LightSeek Foundation
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+import torch
+
+from tokenspeed.runtime.configs.model_config import _is_dflash2_mla
+from tokenspeed.runtime.execution.drafter import get_drafter_impl
+from tokenspeed.runtime.execution.drafter.dflash2 import (
+    DFlash2,
+    _walk_best_path,
+    _walk_sampled_path,
+)
+from tokenspeed.runtime.layers.attention.kv_cache.recipes.spec import FULL_ATTENTION
+from tokenspeed.runtime.models.dflash import DFlashAttention
+from tokenspeed.runtime.models.dflash2 import (
+    CandidateSelector,
+    DFlash2DraftModel,
+    DFlashGroupedConv,
+    _dflash2_mla_rope,
+    _dflash2_uses_mla,
+    _grouped_conv,
+    _score_edges,
+)
+from tokenspeed.runtime.sampling.dflash2 import verify_sparse_draft_distribution
+from tokenspeed.runtime.sampling.draft_distribution import SparseDraftDistribution
+
+
+def test_dflash2_architecture_dispatches_to_its_selector_runtime() -> None:
+    model = DFlash2DraftModel.__new__(DFlash2DraftModel)
+    assert get_drafter_impl("DFLASH", model) is DFlash2
+
+
+@pytest.mark.parametrize(
+    ("architecture", "attention_mode", "expected"),
+    (
+        ("DFlash2DraftModel", "mla", True),
+        ("DFlash2DraftModel", "gqa", False),
+        ("DFlashDraftModel", "mla", False),
+    ),
+)
+def test_dflash2_mla_attention_family_detection(
+    architecture: str, attention_mode: str, expected: bool
+) -> None:
+    config = SimpleNamespace(
+        architectures=[architecture],
+        dflash_config={"attention_mode": attention_mode},
+    )
+    assert _is_dflash2_mla(config, config) is expected
+
+
+def test_dflash2_mla_model_mode_and_yarn_config() -> None:
+    config = SimpleNamespace(
+        dflash_config={"attention_mode": "mla"},
+        rope_theta=1_000_000.0,
+        rope_parameters={
+            "rope_type": "yarn",
+            "rope_theta": 50_000.0,
+            "factor": 32.0,
+            "original_max_position_embeddings": 32768,
+            "beta_fast": 32,
+            "beta_slow": 1,
+            "mscale": 1.0,
+            "mscale_all_dim": 1.0,
+        },
+    )
+
+    assert _dflash2_uses_mla(config)
+    rope_theta, scaling = _dflash2_mla_rope(config)
+    assert rope_theta == 50_000.0
+    assert scaling == {
+        "rope_type": "deepseek_yarn",
+        "factor": 32.0,
+        "original_max_position_embeddings": 32768,
+        "beta_fast": 32,
+        "beta_slow": 1,
+        "mscale": 1.0,
+        "mscale_all_dim": 1.0,
+    }
+
+
+def test_dflash_attention_uses_the_full_attention_cache_group() -> None:
+    assert DFlashAttention.cache_group_id == FULL_ATTENTION
+
+
+def test_candidate_logits_processor_is_created_after_target_wiring() -> None:
+    drafter = DFlash2.__new__(DFlash2)
+    drafter.model = SimpleNamespace(config=SimpleNamespace(vocab_size=32))
+    drafter.output_multiplier = 1.25
+    drafter.final_logit_softcapping = 8.0
+    drafter.candidate_logits_processor = None
+    drafter.logits_processor = SimpleNamespace(tp_rank=0, tp_size=1, tp_group=None)
+
+    with mock.patch("tokenspeed.runtime.execution.drafter.dflash2.DFlash.wire_target"):
+        drafter.wire_target(SimpleNamespace())
+
+    assert drafter.candidate_logits_processor is not None
+    assert drafter.candidate_logits_processor.logit_scale == 1.25
+    assert drafter.candidate_logits_processor.final_logit_softcapping == 8.0
+
+
+@pytest.mark.parametrize("block_size", (6, 8))
+def test_grouped_conv_matches_a_block_local_reference(block_size: int) -> None:
+    torch.manual_seed(0)
+    hidden = torch.randn(2 * block_size, 12)
+    delta = torch.randn(2 * block_size, 3, 3)
+    base = torch.randn(3, 12)
+    actual = _grouped_conv(hidden, delta, base, block_size, 3, 4, 3)
+
+    expected = torch.zeros_like(hidden)
+    for row in range(2 * block_size):
+        position = row % block_size
+        for tap in range(min(position + 1, 3)):
+            for group in range(3):
+                sl = slice(group * 4, (group + 1) * 4)
+                expected[row, sl] += (base[tap, sl] + delta[row, tap, group]) * hidden[
+                    row - tap, sl
+                ]
+    torch.testing.assert_close(actual, expected)
+
+
+def test_candidate_selector_edges_match_the_official_equation() -> None:
+    torch.manual_seed(1)
+    batch, steps, top_k, vocab, rank = 2, 3, 4, 19, 5
+    predecessor = torch.randn(vocab, rank)
+    successor = torch.randn(vocab, rank)
+    candidate_ids = torch.randint(vocab, (batch, steps, top_k))
+    unary = torch.randn(batch, steps, top_k)
+    hidden = torch.randn(batch, steps, rank)
+    anchors = torch.randint(vocab, (batch,))
+
+    actual = _score_edges(
+        predecessor, successor, candidate_ids, unary, hidden, anchors, top_k
+    )
+    expected = torch.empty_like(actual)
+    for b in range(batch):
+        for step in range(steps):
+            for previous in range(top_k):
+                predecessor_id = (
+                    anchors[b] if step == 0 else candidate_ids[b, step - 1, previous]
+                )
+                for candidate in range(top_k):
+                    successor_id = candidate_ids[b, step, candidate]
+                    expected[b, step, previous, candidate] = unary[
+                        b, step, candidate
+                    ] + torch.dot(
+                        predecessor[predecessor_id] * hidden[b, step],
+                        successor[successor_id],
+                    )
+    torch.testing.assert_close(actual, expected)
+
+
+def test_best_path_follows_the_selected_predecessor() -> None:
+    candidate_ids = torch.tensor([[[10, 11], [20, 21], [30, 31]]])
+    scores = torch.zeros(1, 3, 2, 2)
+    scores[0, 0, 0, 1] = 3
+    scores[0, 1, 1, 0] = 4
+    scores[0, 2, 0, 1] = 5
+    out = torch.empty(1, 4, dtype=torch.int32)
+    _walk_best_path(candidate_ids, scores, torch.tensor([7]), out)
+    assert out.tolist() == [[7, 11, 20, 31]]
+
+
+def test_sampled_path_returns_the_realized_q_rows() -> None:
+    candidate_ids = torch.tensor([[[10, 11], [20, 21]]])
+    scores = torch.tensor([[[[0.0, 1.0], [8.0, 9.0]], [[0.0, 2.0], [3.0, 0.0]]]])
+    out = torch.empty(1, 3, dtype=torch.int32)
+    probabilities = _walk_sampled_path(
+        candidate_ids,
+        scores,
+        torch.tensor([7]),
+        torch.tensor([1.0]),
+        torch.tensor([[0.8, 0.8]]),
+        out,
+    )
+    assert out.tolist() == [[7, 11, 20]]
+    torch.testing.assert_close(probabilities.sum(dim=-1), torch.ones(1, 2))
+    torch.testing.assert_close(probabilities[0, 1], torch.softmax(scores[0, 1, 1], 0))
+
+
+def test_sparse_q_changes_acceptance_ratio_and_residual_sample() -> None:
+    candidates = torch.tensor([[0, 1, 2, 3]], dtype=torch.int32)
+    target_probs = torch.zeros(1, 4, 5)
+    target_probs[0, 0, 1] = 0.2
+    target_probs[0, 0, 4] = 0.8
+    target_probs[0, 1, 2] = 0.2
+    target_probs[0, 1, 3] = 0.8
+    target_probs[0, 2, 3] = 1.0
+    target_probs[0, 3, 4] = 1.0
+    draft = SparseDraftDistribution(
+        candidate_ids=torch.tensor([[[1, 4], [2, 4], [3, 4]]]),
+        probabilities=torch.tensor([[[0.4, 0.6], [0.8, 0.2], [0.7, 0.3]]]),
+    )
+    predicts = torch.empty(4, dtype=torch.int32)
+    accept_index = torch.empty(1, 4, dtype=torch.int32)
+    accept_count = torch.empty(1, dtype=torch.int32)
+    verify_sparse_draft_distribution(
+        predicts=predicts,
+        accept_index=accept_index,
+        accept_token_num=accept_count,
+        candidates=candidates,
+        target_probs=target_probs,
+        draft_distribution=draft,
+        acceptance_coins=torch.tensor([[0.3, 0.3, 0.3]]),
+        final_coins=torch.tensor([0.5]),
+    )
+    # Step 0 accepts because 0.3 <= p/q = 0.5. Target-only verification
+    # would reject because 0.3 > p = 0.2. Step 1 rejects and relu(p-q)
+    # places all residual mass on token 3.
+    assert accept_count.tolist() == [1]
+    assert predicts[:2].tolist() == [1, 3]
+    assert accept_index.tolist() == [[0, 1, -1, -1]]
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA is required")
+def test_official_nodes_capture_and_replay_in_one_cuda_graph() -> None:
+    device = torch.device("cuda")
+    conv = DFlashGroupedConv(16, taps=2, group_size=4, block_size=8).to(device)
+    selector = CandidateSelector(16, vocab_size=32, rank=4, top_k=4).to(device)
+    static_hidden = torch.randn(8, 16, device=device)
+    static_candidates = torch.randint(32, (1, 7, 4), device=device)
+    static_unary = torch.randn(1, 7, 4, device=device)
+    static_anchor = torch.tensor([3], device=device)
+    static_out = torch.empty(1, 8, dtype=torch.int32, device=device)
+    static_target_probs = torch.softmax(torch.randn(1, 8, 32, device=device), -1)
+    static_q_probs = torch.softmax(torch.randn(1, 7, 4, device=device), -1)
+    static_accept_coins = torch.full((1, 7), 0.25, device=device)
+    static_final_coins = torch.full((1,), 0.5, device=device)
+    static_predicts = torch.empty(8, dtype=torch.int32, device=device)
+    static_accept_index = torch.empty(1, 8, dtype=torch.int32, device=device)
+    static_accept_count = torch.empty(1, dtype=torch.int32, device=device)
+
+    for _ in range(3):
+        prepared, coefficients = conv.prepare(static_hidden)
+        finished = conv.finish(prepared, coefficients)
+        scores = selector(
+            static_candidates, static_unary, finished[1:].unsqueeze(0), static_anchor
+        )
+        _walk_best_path(static_candidates, scores, static_anchor, static_out)
+        verify_sparse_draft_distribution(
+            predicts=static_predicts,
+            accept_index=static_accept_index,
+            accept_token_num=static_accept_count,
+            candidates=static_out,
+            target_probs=static_target_probs,
+            draft_distribution=SparseDraftDistribution(
+                static_candidates, static_q_probs
+            ),
+            acceptance_coins=static_accept_coins,
+            final_coins=static_final_coins,
+        )
+    torch.cuda.synchronize()
+
+    graph = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(graph):
+        prepared, coefficients = conv.prepare(static_hidden)
+        finished = conv.finish(prepared, coefficients)
+        scores = selector(
+            static_candidates, static_unary, finished[1:].unsqueeze(0), static_anchor
+        )
+        _walk_best_path(static_candidates, scores, static_anchor, static_out)
+        verify_sparse_draft_distribution(
+            predicts=static_predicts,
+            accept_index=static_accept_index,
+            accept_token_num=static_accept_count,
+            candidates=static_out,
+            target_probs=static_target_probs,
+            draft_distribution=SparseDraftDistribution(
+                static_candidates, static_q_probs
+            ),
+            acceptance_coins=static_accept_coins,
+            final_coins=static_final_coins,
+        )
+
+    new_hidden = torch.randn_like(static_hidden)
+    static_hidden.copy_(new_hidden)
+    graph.replay()
+    torch.cuda.synchronize()
+    replayed = static_out.clone()
+    replayed_predicts = static_predicts.clone()
+    replayed_accept_count = static_accept_count.clone()
+
+    prepared, coefficients = conv.prepare(new_hidden)
+    finished = conv.finish(prepared, coefficients)
+    eager_scores = selector(
+        static_candidates, static_unary, finished[1:].unsqueeze(0), static_anchor
+    )
+    expected = torch.empty_like(static_out)
+    _walk_best_path(static_candidates, eager_scores, static_anchor, expected)
+    torch.testing.assert_close(replayed, expected)
+    expected_predicts = torch.empty_like(static_predicts)
+    expected_accept_index = torch.empty_like(static_accept_index)
+    expected_accept_count = torch.empty_like(static_accept_count)
+    verify_sparse_draft_distribution(
+        predicts=expected_predicts,
+        accept_index=expected_accept_index,
+        accept_token_num=expected_accept_count,
+        candidates=expected,
+        target_probs=static_target_probs,
+        draft_distribution=SparseDraftDistribution(static_candidates, static_q_probs),
+        acceptance_coins=static_accept_coins,
+        final_coins=static_final_coins,
+    )
+    torch.testing.assert_close(replayed_predicts, expected_predicts)
+    torch.testing.assert_close(replayed_accept_count, expected_accept_count)
diff --git a/test/runtime/test_kimi_k3_cache_pool.py b/test/runtime/test_kimi_k3_cache_pool.py
index b66b281bc4..8c0a4242bd 100644
--- a/test/runtime/test_kimi_k3_cache_pool.py
+++ b/test/runtime/test_kimi_k3_cache_pool.py
@@ -16,6 +16,17 @@
 )
 
 
+def test_kimi_k3_draft_mla_cache_retains_full_history() -> None:
+    """DFlash2 SWA changes compute visibility, never draft KV retention."""
+    num_draft_layers = 6
+    recipe = kimi_recipe(draft_layers=num_draft_layers)
+
+    assert recipe.group_ids[-num_draft_layers:] == (FULL_ATTENTION,) * num_draft_layers
+    assert (
+        recipe.layer_types[-num_draft_layers:] == (FULL_ATTENTION,) * num_draft_layers
+    )
+
+
 @pytest.mark.skipif(not torch.cuda.is_available(), reason="requires CUDA")
 def test_kimi_k3_pool_binds_mla_and_kda_to_one_lcm_backing() -> None:
     text_config = KimiLinearConfig()
diff --git a/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/__init__.py b/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/__init__.py
index b442c13952..dcd88d1bd8 100644
--- a/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/__init__.py
+++ b/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/__init__.py
@@ -2942,6 +2942,8 @@ def mla_decode_with_kvcache(
     logit_cap: float = 0.0,
     return_lse: bool = False,
     out: torch.Tensor | None = None,
+    window_left: int = -1,
+    noncausal_block_size: int = 1,
     # dispatch options
     override: str | None = None,
     solution: str | None = None,
@@ -2979,6 +2981,12 @@ def mla_decode_with_kvcache(
         qk_rope_head_dim: RoPE q/k head dim.
         softmax_scale: Scale applied to QK logits before softmax.
         logit_cap: Optional soft cap applied to attention logits.
+        window_left: Number of historical positions visible to the first query
+            in a proposal block, or -1 for full attention. A non-causal row
+            also sees the whole proposal block. DFlash2 passes the model's
+            ``sliding_window - 1`` value here.
+        noncausal_block_size: Number of flattened proposal rows per request.
+            Use one for ordinary causal decode.
         return_lse: Whether to also return log-sum-exp values.
         out: Optional output tensor with shape [batch, q_len, num_q_heads,
             kv_lora_rank]. When ``value_weight`` is provided, this is required
@@ -2998,6 +3006,17 @@ def mla_decode_with_kvcache(
     """
     if gate is not None and value_weight is None:
         raise ValueError("gate requires value_weight")
+    if window_left < -1:
+        raise ValueError(f"window_left must be -1 or non-negative, got {window_left}")
+    if noncausal_block_size <= 0:
+        raise ValueError(
+            f"noncausal_block_size must be positive, got {noncausal_block_size}"
+        )
+    if 0 <= window_left < noncausal_block_size - 1:
+        raise ValueError(
+            "window_left must cover the complete non-causal block; got "
+            f"window_left={window_left}, block_size={noncausal_block_size}"
+        )
 
     projected_value = value_weight is not None
     if projected_value:
@@ -3030,6 +3049,43 @@ def mla_decode_with_kvcache(
         if return_lse:
             raise ValueError("projected MLA decode does not support return_lse")
 
+    # The portable Triton MLA kernel is currently the exact implementation of
+    # DFlash2's non-causal sliding mask. Keep full-attention dispatch unchanged;
+    # optimized projected-value kernels can add this trait independently.
+    if window_left >= 0:
+        if override not in (None, "triton_mla_decode_with_kvcache"):
+            raise ValueError(
+                "sliding MLA decode currently requires "
+                "triton_mla_decode_with_kvcache"
+            )
+        if solution not in (None, "triton"):
+            raise ValueError("sliding MLA decode currently requires solution='triton'")
+        override = "triton_mla_decode_with_kvcache"
+        solution = "triton"
+        if projected_value:
+            attention = mla_decode_with_kvcache(
+                q=q,
+                kv_cache=kv_cache,
+                page_table=page_table,
+                cache_seqlens=cache_seqlens,
+                max_seqlen_k=max_seqlen_k,
+                qk_nope_head_dim=qk_nope_head_dim,
+                kv_lora_rank=kv_lora_rank,
+                qk_rope_head_dim=qk_rope_head_dim,
+                softmax_scale=softmax_scale,
+                logit_cap=logit_cap,
+                window_left=window_left,
+                noncausal_block_size=noncausal_block_size,
+                override=override,
+                solution=solution,
+            )
+            return mla_project_value(
+                attention.reshape(q.shape[0], q.shape[2], kv_lora_rank),
+                value_weight,
+                gate=gate,
+                out=out,
+            )
+
     traits = {
         "batch_size": q.shape[0],
         "page_size": kv_cache.shape[1],
@@ -3041,6 +3097,7 @@ def mla_decode_with_kvcache(
         "qk_rope_head_dim": qk_rope_head_dim,
         "support_logit_cap": logit_cap != 0.0,
         "return_lse": return_lse,
+        "sliding_window": window_left >= 0,
     }
     if projected_value:
         traits.update(
@@ -3115,6 +3172,8 @@ def mla_decode_with_kvcache(
         "kv_lora_rank": kv_lora_rank,
         "qk_rope_head_dim": qk_rope_head_dim,
         "max_seqlen_k": max_seqlen_k,
+        "window_left": window_left,
+        "noncausal_block_size": noncausal_block_size,
     }
     if projected_value:
         shape_params["value_head_dim"] = value_weight.shape[2]
@@ -3149,7 +3208,7 @@ def mla_decode_with_kvcache(
                 out=out,
                 logit_cap=logit_cap,
             )
-        return kernel(
+        kernel_kwargs = dict(
             q=q,
             kv_cache=kv_cache,
             page_table=page_table,
@@ -3163,6 +3222,12 @@ def mla_decode_with_kvcache(
             return_lse=return_lse,
             out=out,
         )
+        if window_left >= 0:
+            kernel_kwargs.update(
+                window_left=window_left,
+                noncausal_block_size=noncausal_block_size,
+            )
+        return kernel(**kernel_kwargs)
 
 
 # ===-----------------------------------------------------------------------===#
diff --git a/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/triton/mla_decode.py b/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/triton/mla_decode.py
index 903a012701..915461541b 100644
--- a/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/triton/mla_decode.py
+++ b/tokenspeed-kernel/python/tokenspeed_kernel/ops/attention/triton/mla_decode.py
@@ -66,6 +66,8 @@ def _mla_decode_kernel(
     BLOCK_ROPE: tl.constexpr,
     BLOCK_N: tl.constexpr,
     HAS_LSE: tl.constexpr,
+    WINDOW_LEFT: tl.constexpr,
+    NONCAUSAL_BLOCK_SIZE: tl.constexpr,
 ):
     cur_batch = tl.program_id(0)
     cur_q = tl.program_id(1)
@@ -85,15 +87,30 @@ def _mla_decode_kernel(
     )
 
     cache_len = tl.load(cache_seqlens + cur_batch)
+    if WINDOW_LEFT >= 0:
+        # DFlash2 flattens each non-causal proposal block into one decode row
+        # per position. Every row sees the whole proposal block, while its
+        # historical context follows the reference mask
+        #     query_position - key_position <= window_left.
+        block_position = cur_batch % NONCAUSAL_BLOCK_SIZE
+        context_len = cache_len - NONCAUSAL_BLOCK_SIZE
+        window_start = tl.maximum(
+            0,
+            context_len - WINDOW_LEFT + block_position,
+        )
+        loop_start = (window_start // BLOCK_N) * BLOCK_N
+    else:
+        window_start = 0
+        loop_start = 0
     offs_n = tl.arange(0, BLOCK_N)
     acc = tl.zeros([BLOCK_R], dtype=tl.float32)
     e_sum = 0.0
     e_max = -float("inf")
 
-    for start_n in tl.range(0, cache_len, BLOCK_N, num_stages=2):
+    for start_n in tl.range(loop_start, cache_len, BLOCK_N, num_stages=2):
         start_n = tl.multiple_of(start_n, BLOCK_N)
         token_offsets = start_n + offs_n
-        mask_n = token_offsets < cache_len
+        mask_n = (token_offsets >= window_start) & (token_offsets < cache_len)
         page_indices = token_offsets // PAGE_SIZE
         page_offsets = token_offsets - page_indices * PAGE_SIZE
         physical_pages = tl.load(
@@ -176,6 +193,8 @@ def mla_decode_fwd(
     *,
     logit_cap: float = 0.0,
     lse: torch.Tensor | None = None,
+    window_left: int = -1,
+    noncausal_block_size: int = 1,
 ) -> None:
     if q.dim() != 4:
         raise ValueError(
@@ -197,6 +216,22 @@ def mla_decode_fwd(
         raise ValueError("q and out must have contiguous last dimension")
     if lse is not None and lse.shape != q.shape[:-1]:
         raise ValueError(f"lse shape must be {q.shape[:-1]}, got {tuple(lse.shape)}")
+    if window_left < -1:
+        raise ValueError(f"window_left must be -1 or non-negative, got {window_left}")
+    if noncausal_block_size <= 0:
+        raise ValueError(
+            f"noncausal_block_size must be positive, got {noncausal_block_size}"
+        )
+    if 0 <= window_left < noncausal_block_size - 1:
+        raise ValueError(
+            "window_left must cover the complete non-causal block; got "
+            f"window_left={window_left}, block_size={noncausal_block_size}"
+        )
+    if window_left >= 0 and q.shape[0] % noncausal_block_size:
+        raise ValueError(
+            "sliding MLA decode rows must contain complete non-causal blocks: "
+            f"batch={q.shape[0]}, block_size={noncausal_block_size}"
+        )
 
     kv_cache = _normalize_kv_cache(kv_cache)
     if kv_cache.shape[2] != 1:
@@ -245,6 +280,8 @@ def mla_decode_fwd(
         BLOCK_ROPE=block_rope,
         BLOCK_N=block_n,
         HAS_LSE=lse is not None,
+        WINDOW_LEFT=window_left,
+        NONCAUSAL_BLOCK_SIZE=noncausal_block_size,
         num_warps=8,
         num_stages=2,
     )
@@ -260,6 +297,7 @@ def mla_decode_fwd(
     priority=Priority.PORTABLE,
     traits={
         "q_len": frozenset({1}),
+        "sliding_window": frozenset({False, True}),
         "support_logit_cap": frozenset({False, True}),
         "return_lse": frozenset({False, True}),
     },
@@ -279,6 +317,8 @@ def triton_mla_decode_with_kvcache(
     logit_cap: float = 0.0,
     return_lse: bool = False,
     out: torch.Tensor | None = None,
+    window_left: int = -1,
+    noncausal_block_size: int = 1,
 ) -> torch.Tensor | tuple[torch.Tensor, torch.Tensor]:
     if out is None:
         out_dtype = torch.bfloat16 if q.dtype in _FP8_DTYPES else q.dtype
@@ -304,6 +344,8 @@ def triton_mla_decode_with_kvcache(
         softmax_scale,
         logit_cap=logit_cap,
         lse=lse,
+        window_left=window_left,
+        noncausal_block_size=noncausal_block_size,
     )
     if return_lse:
         return out, lse
diff --git a/tokenspeed-kernel/test/ops/test_attention_mla.py b/tokenspeed-kernel/test/ops/test_attention_mla.py
index a4b2f3fa4f..65a35dec5d 100644
--- a/tokenspeed-kernel/test/ops/test_attention_mla.py
+++ b/tokenspeed-kernel/test/ops/test_attention_mla.py
@@ -417,6 +417,86 @@ def test_mla_decode_with_kvcache(
     torch.testing.assert_close(lse, lse_ref, rtol=8e-2, atol=8e-2)
 
 
+def test_mla_decode_noncausal_block_sliding_window_matches_reference_and_captures(
+    device: str,
+) -> None:
+    torch.manual_seed(91)
+    block_size = 8
+    context_len = 177
+    cache_len = context_len + block_size
+    window_left = 129
+    page_size = 64
+    num_heads = 2
+    kv_lora_rank = 8
+    qk_rope_head_dim = 4
+    qk_nope_head_dim = 4
+    qk_head_dim = kv_lora_rank + qk_rope_head_dim
+    max_pages = math.ceil(cache_len / page_size)
+
+    q = torch.randn(
+        block_size,
+        1,
+        num_heads,
+        qk_head_dim,
+        device=device,
+        dtype=torch.bfloat16,
+    )
+    kv_cache = torch.randn(
+        max_pages,
+        page_size,
+        1,
+        qk_head_dim,
+        device=device,
+        dtype=torch.bfloat16,
+    )
+    page_table = torch.arange(max_pages, device=device, dtype=torch.int32).repeat(
+        block_size, 1
+    )
+    cache_seqlens = torch.full(
+        (block_size,), cache_len, device=device, dtype=torch.int32
+    )
+    softmax_scale = 1.0 / math.sqrt(qk_head_dim)
+
+    def run(query: torch.Tensor) -> torch.Tensor:
+        return mla_decode_with_kvcache(
+            q=query,
+            kv_cache=kv_cache,
+            page_table=page_table,
+            cache_seqlens=cache_seqlens,
+            max_seqlen_k=cache_len,
+            qk_nope_head_dim=qk_nope_head_dim,
+            kv_lora_rank=kv_lora_rank,
+            qk_rope_head_dim=qk_rope_head_dim,
+            softmax_scale=softmax_scale,
+            window_left=window_left,
+            noncausal_block_size=block_size,
+        )
+
+    output = run(q)
+    dense_kv = kv_cache.reshape(-1, qk_head_dim)[:cache_len].float()
+    expected = []
+    for block_position in range(block_size):
+        start = max(
+            0,
+            context_len - window_left + block_position,
+        )
+        visible = dense_kv[start:cache_len]
+        scores = torch.einsum("hd,kd->hk", q[block_position, 0].float(), visible)
+        probs = torch.softmax(scores * softmax_scale, dim=-1)
+        expected.append(torch.matmul(probs, visible[:, :kv_lora_rank]))
+    expected_output = torch.stack(expected).unsqueeze(1)
+    torch.testing.assert_close(output.float(), expected_output, rtol=8e-2, atol=8e-2)
+
+    graph_output = torch.empty_like(output)
+    graph = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(graph):
+        graph_output.copy_(run(q))
+    eager_output = output.clone()
+    graph.replay()
+    torch.cuda.synchronize()
+    torch.testing.assert_close(graph_output, eager_output, atol=0, rtol=0)
+
+
 @pytest.mark.parametrize(
     "dtype",
     [