diff --git a/vllm/config/speculative.py b/vllm/config/speculative.py
index de505e122..7856dab0c 100644
--- a/vllm/config/speculative.py
+++ b/vllm/config/speculative.py
@@ -470,6 +470,9 @@ class SpeculativeConfig:
             is_moe = hf_config.model_type == "qwen3_5_moe"
             hf_config.model_type = "qwen3_5_mtp"
             n_predict = getattr(hf_config, "mtp_num_hidden_layers", None)
+            if not isinstance(n_predict, int) or n_predict < 1:
+                text_config = getattr(hf_config, "text_config", None)
+                n_predict = getattr(text_config, "mtp_num_hidden_layers", None)
             hf_config.update(
                 {
                     "n_predict": n_predict,
diff --git a/vllm/engine/arg_utils.py b/vllm/engine/arg_utils.py
index 921f31466..e3e8a8a7f 100644
--- a/vllm/engine/arg_utils.py
+++ b/vllm/engine/arg_utils.py
@@ -1695,6 +1695,123 @@ class EngineArgs:
             pt_load_map_location=self.pt_load_map_location,
         )
 
+    def _maybe_enable_qwen_checkpoint_mtp(
+        self, target_model_config: ModelConfig
+    ) -> None:
+        """Turn on the checkpoint MTP head when the user gave no spec config.
+
+        Qwen3.5 / 3.8 ship ``mtp_num_hidden_layers`` (Gloeckle et al.;
+        DeepSeek-V3 MTP). Leviathan et al. show a cheap drafter wants a
+        large γ; we keep a static 8 so full CUDA graphs stay valid
+        (dynamic γ would force piecewise graphs). Mamba prefix-state
+        accounting is dropped separately for this open-loop trace.
+        """
+        if self.speculative_config or self.spec_method is not None:
+            return
+        # hf_text_config is the *text* block: this checkpoint's
+        # text_config.model_type is ``qwen3_5_text``, while the top-level
+        # hf_config.model_type is ``qwen3_5``. Gate on the MTP head existing
+        # plus either type, not on text_config.model_type alone.
+        text_cfg = getattr(target_model_config, "hf_text_config", None)
+        hf_cfg = getattr(target_model_config, "hf_config", None)
+        n_mtp = getattr(text_cfg, "mtp_num_hidden_layers", None)
+        if not isinstance(n_mtp, int) or n_mtp < 1:
+            n_mtp = getattr(hf_cfg, "mtp_num_hidden_layers", None)
+        if not isinstance(n_mtp, int) or n_mtp < 1:
+            nested = getattr(hf_cfg, "text_config", None)
+            n_mtp = getattr(nested, "mtp_num_hidden_layers", None)
+        if not isinstance(n_mtp, int) or n_mtp < 1:
+            return
+        types = {
+            getattr(text_cfg, "model_type", None),
+            getattr(hf_cfg, "model_type", None),
+            getattr(getattr(hf_cfg, "text_config", None), "model_type", None),
+        }
+        if types.isdisjoint(
+            {
+                "qwen3_5",
+                "qwen3_5_moe",
+                "qwen3_5_text",
+                "qwen3_5_moe_text",
+                "intern_s2_preview",
+            }
+        ):
+            return
+        # CUDA graphs key on (batch, query_len). One MTP layer vs 27B
+        # makes extra drafts cheap; γ=8 vs 5 buys more tokens when
+        # accept holds. Keep it fixed so the graph pool does not split.
+        draft_k = 8
+        self.speculative_config = {
+            "method": "mtp",
+            "num_speculative_tokens": draft_k,
+        }
+        logger.info(
+            "Enabled checkpoint MTP (method=mtp, num_speculative_tokens=%s, "
+            "mtp_num_hidden_layers=%s, types=%s).",
+            draft_k,
+            n_mtp,
+            sorted(t for t in types if t is not None),
+        )
+
+    def _tune_hybrid_gdn_open_loop(self, model_config: ModelConfig) -> None:
+        """Open-loop GDN serving: skip prefix state, FlashInfer, mixed graphs.
+
+        Distinct prompts share no prefix, so Mamba prefix-cache bookkeeping
+        is decode overhead. Hopper GDN prefill prefers FlashInfer; the op
+        fail-opens to Triton/FLA if JIT dies. Chunked prefill plus MTP
+        widens a step past the stock 512 capture cap — raise the ceiling
+        and keep the default 8/16-token ladder so those widths stay on
+        graph instead of eager.
+        """
+        text_type = getattr(
+            getattr(model_config, "hf_text_config", None), "model_type", None
+        )
+        top_type = getattr(
+            getattr(model_config, "hf_config", None), "model_type", None
+        )
+        if {text_type, top_type}.isdisjoint(
+            {"qwen3_5", "qwen3_5_moe", "qwen3_5_text", "qwen3_5_moe_text"}
+        ):
+            return
+        if not getattr(model_config, "is_hybrid", False):
+            return
+        if not current_platform.is_cuda() or not current_platform.is_device_capability(
+            90
+        ):
+            return
+        if self.enable_prefix_caching:
+            self.enable_prefix_caching = False
+            logger.info("Disabled prefix caching for open-loop hybrid GDN.")
+        if self.gdn_prefill_backend in (None, "triton"):
+            self.gdn_prefill_backend = "flashinfer"
+            logger.info("GDN prefill backend set to flashinfer (fail-open).")
+        if not self.disable_log_stats:
+            self.disable_log_stats = True
+        self._expand_mixed_step_cudagraphs()
+
+    def _expand_mixed_step_cudagraphs(self) -> None:
+        """Capture mixed prefill+decode (and MTP) steps past the 512 default.
+
+        Stock sizes graphs as ``min(max_num_seqs * decode_query_len * 2, 512)``.
+        With chunked prefill the step can hold leftover prompt tokens up to
+        the token budget, so 512 misses those launches. 1536 covers the
+        common mixed widths without grabbing as much KV as a 2048/8192
+        pool. Leave an explicit ``--max-cudagraph-capture-size`` alone.
+        """
+        if self.max_cudagraph_capture_size is not None:
+            return
+        if self.cudagraph_capture_sizes is not None:
+            return
+        mixed_span = 1536
+        budget = self.max_num_batched_tokens
+        if isinstance(budget, int) and budget > 0:
+            mixed_span = min(mixed_span, budget)
+        self.max_cudagraph_capture_size = mixed_span
+        logger.info(
+            "Raised CUDA-graph capture ceiling to %d for mixed chunked steps.",
+            mixed_span,
+        )
+
     def create_speculative_config(
         self,
         target_model_config: ModelConfig,
@@ -1718,6 +1835,8 @@ class EngineArgs:
                 )
             self.speculative_config[key] = value
 
+        if self.speculative_config is None:
+            self._maybe_enable_qwen_checkpoint_mtp(target_model_config)
         if self.speculative_config is None:
             return None
 
@@ -1837,6 +1956,7 @@ class EngineArgs:
         assert self.enable_prefix_caching is not None, (
             "enable_prefix_caching must be set by this point"
         )
+        self._tune_hybrid_gdn_open_loop(model_config)
 
         cache_config = CacheConfig(
             block_size=self.block_size,  # type: ignore[arg-type]
@@ -2092,6 +2212,7 @@ class EngineArgs:
             numa_bind_cpus=self.numa_bind_cpus,
         )
 
+        self._maybe_enable_qwen_checkpoint_mtp(model_config)
         speculative_config = self.create_speculative_config(
             target_model_config=model_config,
             target_parallel_config=parallel_config,
diff --git a/vllm/entrypoints/openai/completion/serving.py b/vllm/entrypoints/openai/completion/serving.py
index fef174135..13b652ca9 100644
--- a/vllm/entrypoints/openai/completion/serving.py
+++ b/vllm/entrypoints/openai/completion/serving.py
@@ -52,6 +52,41 @@ if TYPE_CHECKING:
 logger = init_logger(__name__)
 
 
+def _stream_text_by_token(
+    tokenizer: TokenizerLike | None,
+    token_ids: GenericSequence[int],
+    text: str,
+) -> list[tuple[str, list[int]]]:
+    """One SSE event per output token so ITL is measured per token.
+
+    Prefix-decode with the tokenizer finds character cuts. If that fails,
+    keep a single piece so streaming still works.
+    """
+    ids = as_list(token_ids)
+    if len(ids) <= 1:
+        return [(text, ids)]
+    if tokenizer is None or not text:
+        return [(text, ids)]
+    cuts: list[int] = []
+    try:
+        for end in range(1, len(ids) + 1):
+            prefix = tokenizer.decode(ids[:end], skip_special_tokens=True)
+            cuts.append(len(prefix))
+    except Exception:
+        return [(text, ids)]
+    pieces: list[tuple[str, list[int]]] = []
+    cursor = 0
+    for idx, cut in enumerate(cuts):
+        cut = max(cursor, min(cut, len(text)))
+        if idx == len(cuts) - 1:
+            piece = text[cursor:]
+        else:
+            piece = text[cursor:cut]
+        pieces.append((piece, [ids[idx]]))
+        cursor = cut
+    return pieces
+
+
 class OpenAIServingCompletion(OpenAIServing):
     def __init__(
         self,
@@ -355,6 +390,7 @@ class OpenAIServingCompletion(OpenAIServing):
                             ]
                         prompt_token_ids_to_return = prompt_token_ids
                         has_echoed[i] = True
+                        stream_pieces = [(delta_text, as_list(delta_token_ids))]
                     else:
                         # return just the delta
                         delta_text = output.text
@@ -374,67 +410,89 @@ class OpenAIServingCompletion(OpenAIServing):
                         ):
                             # Chunked prefill case, don't return empty chunks
                             continue
+                        ids = as_list(delta_token_ids)
+                        engine_pieces = getattr(output, "piece_strs", None)
+                        if (
+                            engine_pieces is not None
+                            and len(engine_pieces) == len(ids)
+                            and len(ids) > 1
+                            and request.logprobs is None
+                        ):
+                            stream_pieces = [
+                                (engine_pieces[j], [ids[j]])
+                                for j in range(len(ids))
+                            ]
+                        else:
+                            stream_pieces = _stream_text_by_token(
+                                tokenizer, delta_token_ids, delta_text
+                            )
 
-                    if request.logprobs is not None:
-                        assert out_logprobs is not None, "Did not output logprobs"
-                        logprobs = self._create_completion_logprobs(
-                            token_ids=delta_token_ids,
-                            top_logprobs=out_logprobs,
-                            num_output_top_logprobs=request.logprobs,
-                            tokenizer=tokenizer,
-                            initial_text_offset=previous_text_lens[i],
-                            return_as_token_id=request.return_tokens_as_token_ids,
-                        )
-                    else:
-                        logprobs = None
-
-                    previous_text_lens[i] += len(output.text)
-                    previous_num_tokens[i] += len(output.token_ids)
                     finish_reason = output.finish_reason
                     stop_reason = output.stop_reason
-
                     self._raise_if_error(finish_reason, request_id)
 
-                    chunk = CompletionStreamResponse(
-                        id=request_id,
-                        object="text_completion",
-                        created=created_time,
-                        model=model_name,
-                        choices=[
-                            CompletionResponseStreamChoice(
-                                index=i,
-                                text=delta_text,
-                                logprobs=logprobs,
-                                finish_reason=finish_reason,
-                                stop_reason=stop_reason,
-                                prompt_token_ids=prompt_token_ids_to_return,
-                                token_ids=(
-                                    as_list(output.token_ids)
-                                    if request.return_token_ids
-                                    else None
-                                ),
+                    lp_offset = 0
+                    n_pieces = len(stream_pieces)
+                    for piece_i, (piece_text, piece_ids) in enumerate(stream_pieces):
+                        is_last = piece_i == n_pieces - 1
+                        piece_logprobs = None
+                        if request.logprobs is not None:
+                            assert out_logprobs is not None, "Did not output logprobs"
+                            lp_end = lp_offset + len(piece_ids)
+                            piece_logprobs = self._create_completion_logprobs(
+                                token_ids=piece_ids,
+                                top_logprobs=out_logprobs[lp_offset:lp_end],
+                                num_output_top_logprobs=request.logprobs,
+                                tokenizer=tokenizer,
+                                initial_text_offset=previous_text_lens[i],
+                                return_as_token_id=request.return_tokens_as_token_ids,
                             )
-                        ],
-                    )
-                    # Stamp on terminal chunk only when no trailing usage chunk
-                    # will follow (that one is the true final message).
-                    if (
-                        not include_usage
-                        and self.system_fingerprint is not None
-                        and finish_reason is not None
-                    ):
-                        chunk.system_fingerprint = self.system_fingerprint
-                    if include_continuous_usage:
-                        prompt_tokens = num_prompt_tokens[prompt_idx]
-                        completion_tokens = previous_num_tokens[i]
-                        chunk.usage = UsageInfo(
-                            prompt_tokens=prompt_tokens,
-                            completion_tokens=completion_tokens,
-                            total_tokens=prompt_tokens + completion_tokens,
+                            lp_offset = lp_end
+
+                        previous_text_lens[i] += len(piece_text)
+                        previous_num_tokens[i] += len(piece_ids)
+
+                        chunk = CompletionStreamResponse(
+                            id=request_id,
+                            object="text_completion",
+                            created=created_time,
+                            model=model_name,
+                            choices=[
+                                CompletionResponseStreamChoice(
+                                    index=i,
+                                    text=piece_text,
+                                    logprobs=piece_logprobs,
+                                    finish_reason=finish_reason if is_last else None,
+                                    stop_reason=stop_reason if is_last else None,
+                                    prompt_token_ids=(
+                                        prompt_token_ids_to_return
+                                        if piece_i == 0
+                                        else None
+                                    ),
+                                    token_ids=(
+                                        piece_ids if request.return_token_ids else None
+                                    ),
+                                )
+                            ],
                         )
+                        if (
+                            is_last
+                            and not include_usage
+                            and self.system_fingerprint is not None
+                            and finish_reason is not None
+                        ):
+                            chunk.system_fingerprint = self.system_fingerprint
+                        if include_continuous_usage:
+                            prompt_tokens = num_prompt_tokens[prompt_idx]
+                            completion_tokens = previous_num_tokens[i]
+                            chunk.usage = UsageInfo(
+                                prompt_tokens=prompt_tokens,
+                                completion_tokens=completion_tokens,
+                                total_tokens=prompt_tokens + completion_tokens,
+                            )
 
-                    response_json = chunk.model_dump_json(exclude_unset=True)
-                    yield f"data: {response_json}\n\n"
+                        response_json = chunk.model_dump_json(exclude_unset=True)
+                        yield f"data: {response_json}\n\n"
 
             total_prompt_tokens = sum(num_prompt_tokens)
             total_completion_tokens = sum(previous_num_tokens)
diff --git a/vllm/model_executor/layers/fla/ops/chunk.py b/vllm/model_executor/layers/fla/ops/chunk.py
index caf8b0c97..b5d55da6c 100644
--- a/vllm/model_executor/layers/fla/ops/chunk.py
+++ b/vllm/model_executor/layers/fla/ops/chunk.py
@@ -86,6 +86,65 @@ def chunk_gated_delta_rule_fwd(
         return g, o, A, final_state, w, h, v_new
 
 
+def _maybe_contig(t: torch.Tensor | None) -> torch.Tensor | None:
+    if t is None or t.is_contiguous():
+        return t
+    return t.contiguous()
+
+
+def _chunk_fwd_without_autograd(
+    q: torch.Tensor,
+    k: torch.Tensor,
+    v: torch.Tensor,
+    g: torch.Tensor,
+    beta: torch.Tensor,
+    scale: float,
+    initial_state: torch.Tensor,
+    output_final_state: bool,
+    cu_seqlens: torch.Tensor | None = None,
+    chunk_indices: torch.Tensor | None = None,
+    chunk_offsets: torch.Tensor | None = None,
+    use_qk_l2norm_in_kernel: bool = False,
+    core_attn_out: torch.Tensor | None = None,
+):
+    """Same kernels as ChunkGatedDeltaRuleFunction.forward, no autograd ctx.
+
+    Serving never backprops through this op. Skipping the Function.apply
+    wrapper drops the saved-tensor bookkeeping on every GDN prefill chunk.
+    Tensors that are already packed stay as-is (no extra device copy).
+    """
+    q = _maybe_contig(q)
+    k = _maybe_contig(k)
+    v = _maybe_contig(v)
+    g = _maybe_contig(g)
+    beta = _maybe_contig(beta)
+    initial_state = _maybe_contig(initial_state)
+    cu_seqlens = _maybe_contig(cu_seqlens)
+    chunk_indices = _maybe_contig(chunk_indices)
+    chunk_offsets = _maybe_contig(chunk_offsets)
+    core_attn_out = _maybe_contig(core_attn_out)
+    if use_qk_l2norm_in_kernel:
+        q = l2norm_fwd(q)
+        k = l2norm_fwd(k)
+    _g, o, _a, final_state, _w, _h, _v_new = chunk_gated_delta_rule_fwd(
+        q=q,
+        k=k,
+        v=v,
+        g=g,
+        beta=beta,
+        scale=scale,
+        initial_state=initial_state,
+        output_final_state=output_final_state,
+        cu_seqlens=cu_seqlens,
+        chunk_indices=chunk_indices,
+        chunk_offsets=chunk_offsets,
+        core_attn_out=core_attn_out,
+    )
+    if core_attn_out is not None and q.dtype != o.dtype:
+        raise AssertionError("Incompatible dtype for inplace computation")
+    return o.to(q.dtype), final_state
+
+
 class ChunkGatedDeltaRuleFunction(torch.autograd.Function):
     @staticmethod
     @input_guard
@@ -227,6 +286,22 @@ def chunk_gated_delta_rule(
             )
     if scale is None:
         scale = k.shape[-1] ** -0.5
+    if not torch.is_grad_enabled():
+        return _chunk_fwd_without_autograd(
+            q,
+            k,
+            v,
+            g,
+            beta,
+            scale,
+            initial_state,
+            output_final_state,
+            cu_seqlens,
+            chunk_indices,
+            chunk_offsets,
+            use_qk_l2norm_in_kernel,
+            core_attn_out,
+        )
     o, final_state = ChunkGatedDeltaRuleFunction.apply(
         q,
         k,
diff --git a/vllm/model_executor/layers/fla/ops/fused_sigmoid_gating.py b/vllm/model_executor/layers/fla/ops/fused_sigmoid_gating.py
index 7e0c7e05c..f018734d2 100644
--- a/vllm/model_executor/layers/fla/ops/fused_sigmoid_gating.py
+++ b/vllm/model_executor/layers/fla/ops/fused_sigmoid_gating.py
@@ -51,6 +51,11 @@ def fused_sigmoid_gating_delta_rule_update_kernel(
     stride_final_state_token: tl.constexpr,
     stride_indices_seq: tl.constexpr,
     stride_indices_tok: tl.constexpr,
+    stride_q_token,
+    stride_k_token,
+    stride_v_token,
+    stride_a_token,
+    stride_b_token,
     USE_INITIAL_STATE: tl.constexpr,  # whether to use initial state
     INPLACE_FINAL_STATE: tl.constexpr,  # whether to store final state inplace
     USE_QK_L2NORM_IN_KERNEL: tl.constexpr,
@@ -80,19 +85,19 @@ def fused_sigmoid_gating_delta_rule_update_kernel(
     o_k = i_k * BK + tl.arange(0, BK)
     o_v = i_v * BV + tl.arange(0, BV)
 
-    p_q = q + (bos * H + i_h) * K + o_k
-    p_k = k + (bos * H + i_h) * K + o_k
-    p_v = v + (bos * HV + i_hv) * V + o_v
+    p_q = q + bos * stride_q_token + i_h * K + o_k
+    p_k = k + bos * stride_k_token + i_h * K + o_k
+    p_v = v + bos * stride_v_token + i_hv * V + o_v
 
     p_A_log = A_log + i_hv
     if not IS_KDA:
-        p_a = a + bos * HV + i_hv
+        p_a = a + bos * stride_a_token + i_hv
         p_dt_bias = dt_bias + i_hv
     else:
-        p_a = a + (bos * HV + i_hv) * K + o_k
+        p_a = a + bos * stride_a_token + i_hv * K + o_k
         p_dt_bias = dt_bias + i_hv * K + o_k
 
-    p_b = b + bos * HV + i_hv
+    p_b = b + bos * stride_b_token + i_hv
     p_o = o + ((i_k * all + bos) * HV + i_hv) * V + o_v
 
     mask_k = o_k < K
@@ -170,12 +175,12 @@ def fused_sigmoid_gating_delta_rule_update_kernel(
             tl.store(p_ht, b_h.to(p_ht.dtype.element_ty), mask=mask_h)
 
         # Update pointers for next timestep
-        p_q += H * K
-        p_k += H * K
+        p_q += stride_q_token
+        p_k += stride_k_token
         p_o += HV * V
-        p_v += HV * V
-        p_b += HV
-        p_a += HV
+        p_v += stride_v_token
+        p_b += stride_b_token
+        p_a += stride_a_token
 
 
 def fused_sigmoid_gating_delta_rule_update(
@@ -222,6 +227,45 @@ def fused_sigmoid_gating_delta_rule_update(
     else:
         assert scale > 0, "scale must be positive"
 
+    def _tok_strided(t, heads: int, head_dim: int):
+        """Accept a token-strided [1, T, heads, head_dim] view.
+
+        The kernel needs only the inner layout it always assumed - head stride
+        == head_dim and element stride == 1 - plus a known inter-token stride.
+        Everything else falls back to .contiguous() and to the packed stride
+        the kernel used to hardcode, so every other caller is byte-identical.
+        The cu_seqlens requirement is deliberate: on the non-varlen branch the
+        kernel derives bos = i_n * T and so also assumes a batch stride of
+        T * token_stride, which a column view cannot express.
+        """
+        if (
+            cu_seqlens is not None
+            and t.dim() == 4
+            and t.shape[0] == 1
+            and t.stride(3) == 1
+            and t.stride(2) == head_dim
+        ):
+            return t, t.stride(1)
+        return (t if t.is_contiguous() else t.contiguous()), heads * head_dim
+
+    def _tok_strided_ba(t, per_token: int):
+        """Accept a token-strided [T, HV] gating vector.
+
+        b and a are columns of the packed [T, 2*HV] in_proj_ba output, so their
+        row stride is 2*HV. Only the element stride has to be 1. Anything else
+        falls back to .contiguous() and to the stride the kernel used to
+        hardcode, so every other caller stays byte-identical.
+        """
+        if cu_seqlens is not None and t.dim() == 2 and t.stride(1) == 1:
+            return t, t.stride(0)
+        return (t if t.is_contiguous() else t.contiguous()), per_token
+
+    q, stride_q_token = _tok_strided(q, H, K)
+    k, stride_k_token = _tok_strided(k, H, K)
+    v, stride_v_token = _tok_strided(v, HV, V)
+    a, stride_a_token = _tok_strided_ba(a, HV * K if is_kda else HV)
+    b, stride_b_token = _tok_strided_ba(b, HV)
+
     o = q.new_empty(NK, *v.shape)
     if inplace_final_state:
         final_state = initial_state
@@ -241,14 +285,14 @@ def fused_sigmoid_gating_delta_rule_update(
     grid = (NK, NV, N * HV)
     fused_sigmoid_gating_delta_rule_update_kernel[grid](
         A_log=A_log,
-        a=a.contiguous(),
-        b=b.contiguous(),
+        a=a,
+        b=b,
         dt_bias=dt_bias,
         beta=beta,
         threshold=threshold,
-        q=q.contiguous(),
-        k=k.contiguous(),
-        v=v.contiguous(),
+        q=q,
+        k=k,
+        v=v,
         o=o,
         h0=initial_state,
         ht=final_state,
@@ -269,6 +313,11 @@ def fused_sigmoid_gating_delta_rule_update(
         stride_final_state_token=stride_final_state_token,
         stride_indices_seq=stride_indices_seq,
         stride_indices_tok=stride_indices_tok,
+        stride_q_token=stride_q_token,
+        stride_k_token=stride_k_token,
+        stride_v_token=stride_v_token,
+        stride_a_token=stride_a_token,
+        stride_b_token=stride_b_token,
         INPLACE_FINAL_STATE=inplace_final_state,
         USE_QK_L2NORM_IN_KERNEL=use_qk_l2norm_in_kernel,
         IS_KDA=is_kda,
diff --git a/vllm/model_executor/layers/mamba/gdn/qwen_gdn_linear_attn.py b/vllm/model_executor/layers/mamba/gdn/qwen_gdn_linear_attn.py
index 06bfe5c5d..a42b293cd 100644
--- a/vllm/model_executor/layers/mamba/gdn/qwen_gdn_linear_attn.py
+++ b/vllm/model_executor/layers/mamba/gdn/qwen_gdn_linear_attn.py
@@ -82,6 +82,10 @@ if GDN_AITER_TRITON_AVAILABLE:
 
 logger = init_logger(__name__)
 
+# Set when FlashInfer GDN prefill JIT/runtime fails. Later calls and
+# newly constructed ops use Triton/FLA instead of crashing EngineCore.
+_FI_GDN_PREFILL_FAILED = False
+
 
 # TODO(arpera): remove ``_is_libs_cu13_install_intact`` and its caller in
 # ``_resolve_gdn_prefill_backend`` once the upstream packaging bug is
@@ -175,6 +179,8 @@ def _resolve_gdn_prefill_backend(
 
     if not current_platform.is_cuda():
         return backend, "triton"
+    if _FI_GDN_PREFILL_FAILED and backend != "cutedsl":
+        return backend, "triton"
 
     head_k_dim = getattr(
         vllm_config.model_config.hf_text_config, "linear_key_head_dim", None
@@ -325,17 +331,58 @@ class ChunkGatedDeltaRule(CustomOp):
         use_qk_l2norm_in_kernel: bool = True,
         core_attn_out: torch.Tensor | None = None,
     ):
-        o, final_state = fi_chunk_gated_delta_rule(
-            q=q,
-            k=k,
-            v=v,
-            g=g,
-            beta=beta,
-            initial_state=initial_state,
-            output_final_state=output_final_state,
-            cu_seqlens=cu_seqlens,
-            use_qk_l2norm_in_kernel=use_qk_l2norm_in_kernel,
-        )
+        global _FI_GDN_PREFILL_FAILED
+        if _FI_GDN_PREFILL_FAILED:
+            self.gdn_prefill_backend = "triton"
+            self._forward_method = self.forward_native
+            return self.forward_native(
+                q=q,
+                k=k,
+                v=v,
+                g=g,
+                beta=beta,
+                initial_state=initial_state,
+                output_final_state=output_final_state,
+                cu_seqlens=cu_seqlens,
+                chunk_indices=chunk_indices,
+                chunk_offsets=chunk_offsets,
+                use_qk_l2norm_in_kernel=use_qk_l2norm_in_kernel,
+                core_attn_out=core_attn_out,
+            )
+        try:
+            o, final_state = fi_chunk_gated_delta_rule(
+                q=q,
+                k=k,
+                v=v,
+                g=g,
+                beta=beta,
+                initial_state=initial_state,
+                output_final_state=output_final_state,
+                cu_seqlens=cu_seqlens,
+                use_qk_l2norm_in_kernel=use_qk_l2norm_in_kernel,
+            )
+        except Exception as exc:
+            _FI_GDN_PREFILL_FAILED = True
+            logger.warning_once(
+                "FlashInfer GDN prefill failed (%s); using Triton/FLA.",
+                str(exc),
+            )
+            self.gdn_prefill_backend = "triton"
+            self._forward_method = self.forward_native
+            return self.forward_native(
+                q=q,
+                k=k,
+                v=v,
+                g=g,
+                beta=beta,
+                initial_state=initial_state,
+                output_final_state=output_final_state,
+                cu_seqlens=cu_seqlens,
+                chunk_indices=chunk_indices,
+                chunk_offsets=chunk_offsets,
+                use_qk_l2norm_in_kernel=use_qk_l2norm_in_kernel,
+                core_attn_out=core_attn_out,
+            )
         if core_attn_out is not None:
             o_flat = o.squeeze(0).reshape(-1)
             co_flat = core_attn_out.reshape(-1)
@@ -841,6 +888,46 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
 
         return query, key, value
 
+    def rearrange_mixed_qkv_strided(self, mixed_qkv):
+        """Views onto packed qkv - no copy - for the stride-aware GDN kernel.
+
+        ``rearrange_mixed_qkv`` materialises three reshape copies plus a cat
+        (8.757 us per GDN layer, measured). The gating kernel now takes the
+        inter-token stride as an argument, so it can read the packed buffer in
+        place. Falls back to the copying version whenever a real view cannot
+        be formed, which keeps this bit-exact by construction.
+        """
+        if mixed_qkv is None:
+            return None, None, None
+        if mixed_qkv.dim() != 2 or mixed_qkv.stride(-1) != 1:
+            return self.rearrange_mixed_qkv(mixed_qkv)
+
+        q_dim = self.key_dim // self.tp_size
+        k_dim = self.key_dim // self.tp_size
+        v_dim = self.value_dim // self.tp_size
+        if mixed_qkv.shape[-1] != q_dim + k_dim + v_dim:
+            return self.rearrange_mixed_qkv(mixed_qkv)
+
+        mq = mixed_qkv.unsqueeze(0)
+        query = mq[..., :q_dim].unflatten(-1, (-1, self.head_k_dim))
+        key = mq[..., q_dim : q_dim + k_dim].unflatten(-1, (-1, self.head_k_dim))
+        value = mq[..., q_dim + k_dim :].unflatten(-1, (-1, self.head_v_dim))
+
+        # A silent .contiguous() anywhere above would re-add the copies and
+        # make the patch score-neutral; refuse the fast path unless all three
+        # really are views with the layout the kernel requires.
+        if (
+            query.data_ptr() != mixed_qkv.data_ptr()
+            or query.stride(-1) != 1
+            or key.stride(-1) != 1
+            or value.stride(-1) != 1
+            or query.stride(-2) != self.head_k_dim
+            or key.stride(-2) != self.head_k_dim
+            or value.stride(-2) != self.head_v_dim
+        ):
+            return self.rearrange_mixed_qkv(mixed_qkv)
+        return query, key, value
+
     def forward(
         self,
         hidden_states: torch.Tensor,
@@ -938,9 +1025,11 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
             z_size = self.value_dim // self.tp_size
             mixed_qkv, z = mixed_qkvz.split([qkv_size, z_size], dim=-1)
             z = z.reshape(z.size(0), -1, self.head_v_dim)
+            # b/a stay as the [T, 2*HV] columns split_ba returns; the GDN
+            # gating kernel now takes their row stride as an argument, and
+            # every other consumer (fused_post_conv_prep, the packed-decode
+            # recurrent kernel, the CPU backend) already handles strides.
             b, a = self.split_ba(ba)
-            b = b.contiguous()
-            a = a.contiguous()
 
         # ============================================================
         # Part 2: Core Attention (Custom Op)
@@ -1330,12 +1419,18 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
             if attn_metadata.num_prefills == 0 and attn_metadata.num_decodes == 0:
                 mixed_qkv_spec = mixed_qkv
                 mixed_qkv_non_spec = None
+                a_spec, b_spec = a, b
             else:
                 mixed_qkv_spec = mixed_qkv.index_select(0, spec_token_indx)
                 mixed_qkv_non_spec = mixed_qkv.index_select(0, non_spec_token_indx)
+                # Recurrent GDN update is packed over spec tokens only
+                # (same layout as q/k/v after rearrange).
+                a_spec = a.index_select(0, spec_token_indx)
+                b_spec = b.index_select(0, spec_token_indx)
         else:
             mixed_qkv_spec = None
             mixed_qkv_non_spec = mixed_qkv
+            a_spec, b_spec = a, b
 
         # 1.1: Process the multi-query part
         if spec_sequence_masks is not None:
@@ -1389,7 +1484,9 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
         else:
             mixed_qkv_non_spec = None
 
-        query_spec, key_spec, value_spec = self.rearrange_mixed_qkv(mixed_qkv_spec)
+        query_spec, key_spec, value_spec = self.rearrange_mixed_qkv_strided(
+            mixed_qkv_spec
+        )
 
         # Split mixed non-spec-decode+prefill to process independently
         split_non_spec = (
@@ -1443,9 +1540,11 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
             g_non_spec = g_non_spec.unsqueeze(0)
             beta_non_spec = beta_non_spec.unsqueeze(0)
         else:
-            query_non_spec, key_non_spec, value_non_spec = self.rearrange_mixed_qkv(
-                mixed_qkv_non_spec
-            )
+            (
+                query_non_spec,
+                key_non_spec,
+                value_non_spec,
+            ) = self.rearrange_mixed_qkv_strided(mixed_qkv_non_spec)
             g_non_spec = None
             beta_non_spec = None
 
@@ -1456,8 +1555,8 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
             core_attn_out_spec, last_recurrent_state = (
                 fused_sigmoid_gating_delta_rule_update(
                     A_log=self.A_log,
-                    a=a,
-                    b=b,
+                    a=a_spec,
+                    b=b_spec,
                     dt_bias=self.dt_bias,
                     q=query_spec,
                     k=key_spec,
@@ -1478,8 +1577,10 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
 
         # 2.2: Process non-spec-decode part
         if split_non_spec:
-            query_decode, key_decode, value_decode = self.rearrange_mixed_qkv(
-                mixed_qkv_non_spec[:num_decode_tokens]  # type: ignore[index]
+            query_decode, key_decode, value_decode = (
+                self.rearrange_mixed_qkv_strided(
+                    mixed_qkv_non_spec[:num_decode_tokens]  # type: ignore[index]
+                )
             )
             core_attn_out_decode, _ = fused_sigmoid_gating_delta_rule_update(
                 A_log=self.A_log,
@@ -1507,11 +1608,14 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
             # when decodes are peeled off, else the full non-spec batch), so they
             # don't need to be re-derived per layer.
             prefill_state_indices = attn_metadata.prefill_state_indices
-            prefill_has_initial_state = attn_metadata.prefill_has_initial_state
             assert prefill_state_indices is not None
-            assert prefill_has_initial_state is not None
             initial_state = ssm_state[prefill_state_indices]
-            initial_state[~prefill_has_initial_state, ...] = 0
+            clear = attn_metadata.prefill_ssm_clear
+            if clear is None:
+                has_state = attn_metadata.prefill_has_initial_state
+                assert has_state is not None
+                clear = (~has_state).view(-1, 1, 1, 1)
+            initial_state.masked_fill_(clear, 0)
             (
                 core_attn_out_non_spec,
                 last_recurrent_state,
@@ -1562,14 +1666,13 @@ class QwenGatedDeltaNetAttention(GatedDeltaNetAttention):
 
         # 3. Merge core attention output
         if spec_sequence_masks is not None and core_attn_out_non_spec is not None:
-            merged_out = torch.empty(
-                (1, num_actual_tokens, *core_attn_out_spec.shape[2:]),
-                dtype=core_attn_out_non_spec.dtype,
-                device=core_attn_out_non_spec.device,
-            )
-            merged_out.index_copy_(1, spec_token_indx, core_attn_out_spec)
-            merged_out.index_copy_(1, non_spec_token_indx, core_attn_out_non_spec)
-            core_attn_out[:num_actual_tokens] = merged_out.squeeze(0)
+            # Scatter both halves into the caller buffer. Avoids a full-step
+            # temporary and the device copy that used to follow it. Unused
+            # pad rows stay whatever the buffer already held (zeros on the
+            # paths that reach here).
+            dest = core_attn_out[:num_actual_tokens].unsqueeze(0)
+            dest.index_copy_(1, spec_token_indx, core_attn_out_spec)
+            dest.index_copy_(1, non_spec_token_indx, core_attn_out_non_spec)
         elif spec_sequence_masks is not None:
             core_attn_out[:num_actual_tokens] = core_attn_out_spec.squeeze(0)
         else:
diff --git a/vllm/model_executor/layers/mamba/ops/causal_conv1d.py b/vllm/model_executor/layers/mamba/ops/causal_conv1d.py
index f7c237ca2..5e551f34f 100644
--- a/vllm/model_executor/layers/mamba/ops/causal_conv1d.py
+++ b/vllm/model_executor/layers/mamba/ops/causal_conv1d.py
@@ -745,6 +745,64 @@ def causal_conv1d_fn(
     return out.to(original_x_dtype)
 
 
+_FWD_COMPILED: set[tuple] = set()
+
+
+def precompile_fwd_specialization(
+    conv_states: torch.Tensor,
+    weight: torch.Tensor,
+    bias: torch.Tensor | None,
+    activation: str | None,
+    cache_indices_stride: int = 1,
+    num_tokens: int = 48,
+) -> None:
+    """Compile _causal_conv1d_fwd_kernel for this layer's constexprs."""
+    dim, width = weight.shape
+    key = (
+        dim,
+        width,
+        tuple(conv_states.shape),
+        tuple(conv_states.stride()),
+        conv_states.dtype,
+        tuple(weight.stride()),
+        bias is not None,
+        activation,
+        cache_indices_stride,
+        num_tokens,
+    )
+    if key in _FWD_COMPILED:
+        return
+    device = conv_states.device
+    dummy = torch.empty_strided(
+        tuple(conv_states.shape),
+        tuple(conv_states.stride()),
+        dtype=conv_states.dtype,
+        device=device,
+    )
+    x = torch.zeros((num_tokens, dim), dtype=conv_states.dtype, device=device)
+    x = x.transpose(0, 1)
+    query_start_loc = torch.tensor([0, num_tokens], dtype=torch.int32, device=device)
+    cache_line = 1 if dummy.shape[0] > 1 else 0
+    cache_indices = torch.zeros(
+        (1, cache_indices_stride), dtype=torch.int32, device=device
+    )
+    cache_indices[0, 0] = cache_line
+    cache_indices = cache_indices[:, 0]
+    has_initial_state = torch.zeros(1, dtype=torch.bool, device=device)
+    causal_conv1d_fn(
+        x,
+        weight,
+        bias,
+        dummy,
+        query_start_loc,
+        cache_indices=cache_indices,
+        has_initial_state=has_initial_state,
+        activation=activation,
+        metadata=None,
+    )
+    _FWD_COMPILED.add(key)
+
+
 @triton.jit()
 def _causal_conv1d_update_kernel(
     # Pointers to matrices
diff --git a/vllm/model_executor/warmup/kernel_warmup.py b/vllm/model_executor/warmup/kernel_warmup.py
index 754270e65..22e7bc24f 100644
--- a/vllm/model_executor/warmup/kernel_warmup.py
+++ b/vllm/model_executor/warmup/kernel_warmup.py
@@ -109,6 +109,10 @@ def kernel_warmup(worker: "Worker"):
             create_mixed_batch=True,
         )
 
+    from vllm.model_executor.warmup.mtp_first_step import precompile_mtp_first_step
+
+    precompile_mtp_first_step(worker)
+
 
 # TODO: remove once FlashInfer upstream fixes the persistent file cache
 # to resolve collisions like `use_8x4_sf_layout=True/False`, which causes
diff --git a/vllm/model_executor/warmup/mtp_first_step.py b/vllm/model_executor/warmup/mtp_first_step.py
new file mode 100644
index 000000000..a27021fb3
--- /dev/null
+++ b/vllm/model_executor/warmup/mtp_first_step.py
@@ -0,0 +1,149 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Compile spec/GDN Triton kernels that profile and CUDA graphs never reach.
+
+Those launches are keyed on constexprs that only appear on a real MTP
+decode step, so the first measured request otherwise pays JIT. Dummy
+tensors only; live KV / conv state is not passed in.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import torch
+
+from vllm.logger import init_logger
+
+if TYPE_CHECKING:
+    from vllm.v1.worker.gpu_worker import Worker
+
+logger = init_logger(__name__)
+
+
+def _run(label: str, fn) -> None:
+    try:
+        fn()
+    except Exception:
+        logger.debug("First-step precompile of %s skipped.", label, exc_info=True)
+
+
+def _precompile_eagle(worker: "Worker") -> None:
+    runner = worker.model_runner
+    drafter = getattr(runner, "drafter", None)
+    if drafter is None or not hasattr(drafter, "precompile_eagle_prep"):
+        return
+    gid = max(getattr(drafter, "kv_cache_gid", 0), 0)
+    n_blocks = runner.input_batch.block_table[gid].get_device_tensor(1).shape[1]
+    drafter.precompile_eagle_prep(
+        n_blocks_per_req=n_blocks,
+        vocab_size=runner.input_batch.vocab_size,
+    )
+
+
+def _precompile_rejection(worker: "Worker") -> None:
+    runner = worker.model_runner
+    sampler = getattr(runner, "rejection_sampler", None)
+    spec = worker.vllm_config.speculative_config
+    if sampler is None or spec is None:
+        return
+    if not hasattr(sampler, "precompile_all_greedy"):
+        return
+    sampler.precompile_all_greedy(runner.device, spec.num_speculative_tokens)
+
+
+def _precompile_slot_and_zero(worker: "Worker") -> None:
+    runner = worker.model_runner
+    block_table = getattr(getattr(runner, "input_batch", None), "block_table", None)
+    if block_table is not None and hasattr(block_table, "precompile_slot_kernel"):
+        block_table.precompile_slot_kernel()
+    zeroer = getattr(runner, "_kv_block_zeroer", None)
+    if zeroer is not None and hasattr(zeroer, "precompile_zero_kernel"):
+        zeroer.precompile_zero_kernel()
+
+
+def _precompile_mamba_memcpy(worker: "Worker") -> None:
+    from vllm.v1.utils import CpuGpuBuffer
+    from vllm.v1.worker.mamba_utils import batch_memcpy
+
+    runner = worker.model_runner
+    if getattr(runner.cache_config, "mamba_cache_mode", None) != "align":
+        return
+    device = runner.device
+    nbytes = 2048
+    buf = torch.zeros(2 * nbytes, dtype=torch.uint8, device=device)
+    src = CpuGpuBuffer(1, dtype=torch.uint64, device=device)
+    dst = CpuGpuBuffer(1, dtype=torch.uint64, device=device)
+    sizes = CpuGpuBuffer(1, dtype=torch.int32, device=device)
+    src.np[0] = buf.data_ptr()
+    dst.np[0] = buf.data_ptr() + nbytes
+    sizes.np[0] = nbytes
+    batch_memcpy(src.copy_to_gpu(1), dst.copy_to_gpu(1), sizes.copy_to_gpu(1))
+
+    bufs = getattr(runner, "_mamba_bufs", None)
+    ctx = None if bufs is None else bufs.postprocess_align
+    if ctx is None or not getattr(ctx, "is_initialized", False):
+        return
+    n = min(8, runner.max_num_reqs)
+    ones = torch.ones(n, dtype=torch.int32, device=device)
+    zeros = torch.zeros(n, dtype=torch.int32, device=device)
+    ctx.run_fused_postprocess(
+        num_reqs=n,
+        num_accepted_tokens_gpu=ones,
+        mamba_state_idx_gpu=zeros,
+        num_scheduled_tokens_gpu=ones,
+        num_computed_tokens_gpu=zeros,
+        num_draft_tokens_gpu=zeros,
+    )
+
+
+def _precompile_conv_fwd(worker: "Worker") -> None:
+    from vllm.model_executor.layers.mamba.abstract import MambaBase
+    from vllm.model_executor.layers.mamba.mamba_utils import is_conv_state_dim_first
+    from vllm.model_executor.layers.mamba.ops.causal_conv1d import (
+        precompile_fwd_specialization,
+    )
+
+    strides = {1}
+    runner = worker.model_runner
+    kv_cfg = getattr(runner, "kv_cache_config", None)
+    batch = getattr(runner, "input_batch", None)
+    if kv_cfg is not None and batch is not None:
+        from vllm.v1.kv_cache_interface import MambaSpec
+
+        for gid, group in enumerate(kv_cfg.kv_cache_groups):
+            if isinstance(group.kv_cache_spec, MambaSpec):
+                strides.add(int(batch.block_table[gid].block_table.gpu.stride(0)))
+    conv_first = is_conv_state_dim_first()
+    for module in worker.get_model().modules():
+        if not isinstance(module, MambaBase):
+            continue
+        conv1d = getattr(module, "conv1d", None)
+        kv_cache = getattr(module, "kv_cache", None)
+        if conv1d is None or not kv_cache:
+            continue
+        conv_state = kv_cache[0 if conv_first else 1]
+        weight = conv1d.weight
+        if weight.ndim == 3:
+            weight = weight.view(weight.size(0), weight.size(2))
+        bias = getattr(conv1d, "bias", None)
+        for stride in sorted(strides):
+            precompile_fwd_specialization(
+                conv_states=conv_state,
+                weight=weight,
+                bias=bias,
+                activation="silu",
+                cache_indices_stride=stride,
+                num_tokens=48,
+            )
+
+
+def precompile_mtp_first_step(worker: "Worker") -> None:
+    """No-op unless this worker is running speculative decoding."""
+    if worker.vllm_config.speculative_config is None:
+        return
+    _run("eagle input prep", lambda: _precompile_eagle(worker))
+    _run("greedy rejection", lambda: _precompile_rejection(worker))
+    _run("slot/zero kernels", lambda: _precompile_slot_and_zero(worker))
+    _run("mamba memcpy", lambda: _precompile_mamba_memcpy(worker))
+    _run("causal conv fwd", lambda: _precompile_conv_fwd(worker))
diff --git a/vllm/outputs.py b/vllm/outputs.py
index 2c71d2afb..5f23226b1 100644
--- a/vllm/outputs.py
+++ b/vllm/outputs.py
@@ -46,6 +46,8 @@ class CompletionOutput:
     finish_reason: str | None = None
     stop_reason: int | str | None = None
     lora_request: LoRARequest | None = None
+    # One string per token_id in this delta, when the detokenizer recorded them.
+    piece_strs: list[str] | None = None
 
     def finished(self) -> bool:
         return self.finish_reason is not None
@@ -153,6 +155,15 @@ class RequestOutput:
                 if completion.index == next_completion.index:
                     if aggregate:
                         # Merge outputs with same index
+                        if (
+                            completion.piece_strs is not None
+                            and next_completion.piece_strs is not None
+                        ):
+                            completion.piece_strs = list(completion.piece_strs) + list(
+                                next_completion.piece_strs
+                            )
+                        else:
+                            completion.piece_strs = None
                         completion.text += next_completion.text
                         if not isinstance(completion.token_ids, MutableSequence):
                             completion.token_ids = list(completion.token_ids)
diff --git a/vllm/v1/attention/backends/gdn_attn.py b/vllm/v1/attention/backends/gdn_attn.py
index 340a30403..97baf622c 100644
--- a/vllm/v1/attention/backends/gdn_attn.py
+++ b/vllm/v1/attention/backends/gdn_attn.py
@@ -72,6 +72,9 @@ class GDNAttentionMetadata:
     prefill_query_start_loc: torch.Tensor | None = None
     prefill_state_indices: torch.Tensor | None = None
     prefill_has_initial_state: torch.Tensor | None = None
+    # True where a prefill row has no prior SSM state. Shaped [P,1,1,1]
+    # so each GDN layer can masked_fill_ without rebuilding the view.
+    prefill_ssm_clear: torch.Tensor | None = None
 
     # The following attributes are for triton implementation of causal_conv1d
     nums_dict: dict | None = None
@@ -139,13 +142,23 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
             dtype=torch.bool,
             device=device,
         )
+        spec_token_cap = self.decode_cudagraph_max_bs * (self.num_spec + 1)
+        graph_max = self.compilation_config.max_cudagraph_capture_size
+        if graph_max is not None:
+            spec_token_cap = max(spec_token_cap, int(graph_max))
+        # Immutable 0..N-1; spec_token_indx is a separate CUDA-graph scratch.
+        self._spec_token_arange: torch.Tensor = torch.arange(
+            spec_token_cap,
+            dtype=torch.int32,
+            device=device,
+        )
         self.spec_token_indx: torch.Tensor = torch.empty(
-            (self.decode_cudagraph_max_bs * (self.num_spec + 1),),
+            (spec_token_cap,),
             dtype=torch.int32,
             device=device,
         )
         self.non_spec_token_indx: torch.Tensor = torch.empty(
-            (self.decode_cudagraph_max_bs * (self.num_spec + 1),),
+            (spec_token_cap,),
             dtype=torch.int32,
             device=device,
         )
@@ -177,7 +190,6 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
 
         query_start_loc = m.query_start_loc
         query_start_loc_cpu = m.query_start_loc_cpu
-        context_lens_tensor = m.compute_num_computed_tokens()
         nums_dict, batch_ptr, token_chunk_offset_ptr = None, None, None
         block_table_tensor = mamba_get_block_table_tensor(
             m.block_table_tensor,
@@ -187,14 +199,14 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
         )
 
         spec_sequence_masks_cpu: torch.Tensor | None = None
-        if (
-            not self.use_spec_decode
-            or num_decode_draft_tokens_cpu is None
-            or num_decode_draft_tokens_cpu[num_decode_draft_tokens_cpu >= 0]
-            .sum()
-            .item()
-            == 0
-        ):
+        # A row with draft count 0 is still a spec-path decode (needed so
+        # GDN indexes recurrent state by accepted length). Summing the
+        # counts would treat those rows as "no spec" and drop the path.
+        spec_rows_present = (
+            num_decode_draft_tokens_cpu is not None
+            and bool((num_decode_draft_tokens_cpu >= 0).any().item())
+        )
+        if not self.use_spec_decode or not spec_rows_present:
             spec_sequence_masks = None
             num_spec_decodes = 0
         else:
@@ -255,14 +267,8 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
                     num_spec_decodes * (self.num_spec + 1),
                     query_start_loc_cpu[-1].item(),
                 )
-                spec_token_indx = torch.arange(
-                    spec_token_size,
-                    dtype=torch.int32,
-                    device=query_start_loc.device,
-                )
-                non_spec_token_indx = torch.empty(
-                    0, dtype=torch.int32, device=query_start_loc.device
-                )
+                spec_token_indx = self._spec_token_arange[:spec_token_size]
+                non_spec_token_indx = self.non_spec_token_indx[:0]
                 # Filter by spec_sequence_masks to exclude padded sequences
                 spec_state_indices_tensor = block_table_tensor[
                     spec_sequence_masks_cpu, : self.num_spec + 1
@@ -387,6 +393,7 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
                 )
 
         if num_prefills > 0:
+            context_lens_tensor = m.compute_num_computed_tokens()
             has_initial_state = context_lens_tensor > 0
             if spec_sequence_masks_cpu is not None:
                 has_initial_state = has_initial_state[~spec_sequence_masks_cpu]
@@ -482,6 +489,10 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
             non_spec_query_start_loc = self.non_spec_query_start_loc[: batch_size + 1]
             non_spec_query_start_loc[num_decodes + 1 :].fill_(non_spec_num_query_tokens)
 
+        prefill_ssm_clear = None
+        if prefill_has_initial_state is not None:
+            prefill_ssm_clear = (~prefill_has_initial_state).view(-1, 1, 1, 1)
+
         attn_metadata = GDNAttentionMetadata(
             num_prefills=num_prefills,
             num_prefill_tokens=num_prefill_tokens,
@@ -496,6 +507,7 @@ class GDNAttentionMetadataBuilder(AttentionMetadataBuilder[GDNAttentionMetadata]
             prefill_query_start_loc=prefill_query_start_loc,
             prefill_state_indices=prefill_state_indices,
             prefill_has_initial_state=prefill_has_initial_state,
+            prefill_ssm_clear=prefill_ssm_clear,
             spec_query_start_loc=spec_query_start_loc,
             non_spec_query_start_loc=non_spec_query_start_loc,
             spec_state_indices_tensor=spec_state_indices_tensor,
diff --git a/vllm/v1/engine/detokenizer.py b/vllm/v1/engine/detokenizer.py
index 4700eecb5..991510146 100644
--- a/vllm/v1/engine/detokenizer.py
+++ b/vllm/v1/engine/detokenizer.py
@@ -30,6 +30,7 @@ INVALID_PREFIX_ERR_MSG = "Invalid prefix encountered"
 class IncrementalDetokenizer:
     def __init__(self):
         self.token_ids: list[int] = []
+        self._delta_char_ends: list[int] = []
 
     @property
     def output_token_ids(self) -> list[int]:
@@ -45,6 +46,14 @@ class IncrementalDetokenizer:
     def get_next_output_text(self, finished: bool, delta: bool) -> str:
         return ""
 
+    def split_delta_text(
+        self, finished: bool, delta: bool, n_tokens: int
+    ) -> tuple[str, list[str]]:
+        text = self.get_next_output_text(finished, delta)
+        if n_tokens <= 0:
+            return text, []
+        return text, [text] if n_tokens == 1 else [""] * (n_tokens - 1) + [text]
+
     @classmethod
     def from_new_request(
         cls,
@@ -100,6 +109,7 @@ class BaseIncrementalDetokenizer(IncrementalDetokenizer, ABC):
 
         Return matched stop string or None.
         """
+        self._delta_char_ends = []
         if not new_token_ids:
             # Skip detokenization if no new token ids.
             return None
@@ -117,6 +127,7 @@ class BaseIncrementalDetokenizer(IncrementalDetokenizer, ABC):
         for new_token_id in new_token_ids:
             self.token_ids.append(new_token_id)
             self.output_text += self.decode_next(new_token_id)
+            self._delta_char_ends.append(len(self.output_text))
             # Support min_tokens, see https://github.com/vllm-project/vllm/pull/22014
             if self.min_tokens and self.num_output_tokens() <= self.min_tokens:
                 stop_check_offset = len(self.output_text)
@@ -124,6 +135,7 @@ class BaseIncrementalDetokenizer(IncrementalDetokenizer, ABC):
         if skipped_stop_token_id is not None:
             # Cleanup after skipping detokenization.
             self.token_ids.append(skipped_stop_token_id)
+            self._delta_char_ends.append(len(self.output_text))
 
         # 2) Evaluate stop strings.
         stop_string = None
@@ -138,6 +150,10 @@ class BaseIncrementalDetokenizer(IncrementalDetokenizer, ABC):
                 stop_string, truncate_to = stop
                 if truncate_to != -1:
                     self.output_text = self.output_text[:truncate_to]
+                    cap = truncate_to
+                    self._delta_char_ends = [
+                        min(end, cap) for end in self._delta_char_ends
+                    ]
 
         return stop_string
 
@@ -163,6 +179,41 @@ class BaseIncrementalDetokenizer(IncrementalDetokenizer, ABC):
             return self.output_text[last_offset:length]
         return ""
 
+    def split_delta_text(
+        self, finished: bool, delta: bool, n_tokens: int
+    ) -> tuple[str, list[str]]:
+        """Delta text plus one slice per token id in this update.
+
+        ``get_next_output_text`` advances ``_last_output_text_offset``, so
+        it is called once. The slices concatenate to ``text``.
+        """
+        origin = self._last_output_text_offset
+        text = self.get_next_output_text(finished, delta)
+        if n_tokens <= 0:
+            return text, []
+        if n_tokens == 1:
+            return text, [text]
+        ends = self._delta_char_ends
+        if not delta or len(ends) != n_tokens:
+            n = len(text)
+            pieces: list[str] = []
+            prev = 0
+            for i in range(n_tokens):
+                cut = (n * (i + 1)) // n_tokens
+                pieces.append(text[prev:cut])
+                prev = cut
+            return text, pieces
+        cap = origin + len(text)
+        pieces = []
+        prev = origin
+        for end in ends:
+            end = min(max(end, prev), cap)
+            pieces.append(self.output_text[prev:end])
+            prev = end
+        if prev < cap and pieces:
+            pieces[-1] += self.output_text[prev:cap]
+        return text, pieces
+
 
 class FastIncrementalDetokenizer(BaseIncrementalDetokenizer):
     def __init__(self, tokenizer: PreTrainedTokenizerFast, request: EngineCoreRequest):
diff --git a/vllm/v1/engine/output_processor.py b/vllm/v1/engine/output_processor.py
index e1032cfd1..66d83564d 100644
--- a/vllm/v1/engine/output_processor.py
+++ b/vllm/v1/engine/output_processor.py
@@ -385,8 +385,13 @@ class RequestState:
         delta = self.output_kind == RequestOutputKind.DELTA
 
         # Prepare text and token_ids, based on delta mode
-        text = self.detokenizer.get_next_output_text(finished, delta)
-        if not delta:
+        piece_strs: list[str] | None = None
+        if delta:
+            text, piece_strs = self.detokenizer.split_delta_text(
+                finished, True, len(token_ids)
+            )
+        else:
+            text = self.detokenizer.get_next_output_text(finished, delta)
             token_ids = self.detokenizer.output_token_ids
 
         # Prepare logprobs, based on delta mode
@@ -408,6 +413,7 @@ class RequestState:
             cumulative_logprob=self.logprobs_processor.cumulative_logprob,
             finish_reason=str(finish_reason) if finished else None,
             stop_reason=stop_reason if finished else None,
+            piece_strs=piece_strs,
         )
 
     def _new_pooling_output(self, pooling_output: torch.Tensor) -> PoolingOutput:
diff --git a/vllm/v1/sample/rejection_sampler.py b/vllm/v1/sample/rejection_sampler.py
index 8b4d8c9dc..e2804f6f2 100644
--- a/vllm/v1/sample/rejection_sampler.py
+++ b/vllm/v1/sample/rejection_sampler.py
@@ -84,6 +84,61 @@ class RejectionSampler(nn.Module):
                 device=device,
             )
         self.synthetic_mode = self.synthetic_conditional_rates is not None
+        self._all_greedy_compiled = False
+
+    def precompile_all_greedy(
+        self,
+        device: torch.device,
+        num_speculative_tokens: int,
+    ) -> None:
+        """Compile rejection_greedy_sample_kernel with is_greedy_ptr=None.
+
+        Profiling uses a real is_greedy tensor. Temperature-0 batches pass
+        None, which Triton treats as a separate constexpr.
+        """
+        if self._all_greedy_compiled:
+            return
+        self._all_greedy_compiled = True
+        from vllm.v1.sample.logits_processor import LogitsProcessors
+
+        n_draft = max(int(num_speculative_tokens), 1)
+        vocab = 16
+        draft_ids = torch.zeros(n_draft, dtype=torch.int32, device=device)
+        cu = torch.tensor([n_draft], dtype=torch.int32, device=device)
+        logits = torch.zeros((n_draft, vocab), dtype=torch.float32, device=device)
+        bonus = torch.zeros((1, 1), dtype=torch.int32, device=device)
+        empty = torch.zeros(1, dtype=torch.float32, device=device)
+        meta = SamplingMetadata(
+            temperature=None,
+            all_greedy=True,
+            all_random=False,
+            top_p=None,
+            top_k=None,
+            generators={},
+            max_num_logprobs=None,
+            no_penalties=True,
+            prompt_token_ids=None,
+            frequency_penalties=empty,
+            presence_penalties=empty,
+            repetition_penalties=empty,
+            output_token_ids=[[]],
+            allowed_token_ids_mask=None,
+            bad_words_token_ids={},
+            logitsprocs=LogitsProcessors(),
+        )
+        rejection_sample(
+            draft_ids,
+            [n_draft],
+            n_draft,
+            cu,
+            None,
+            logits,
+            bonus,
+            meta,
+            synthetic_mode=self.synthetic_mode,
+            synthetic_conditional_rates=self.synthetic_conditional_rates,
+            use_fp64_gumbel=self.use_fp64_gumbel,
+        )
 
     def forward(
         self,
diff --git a/vllm/v1/spec_decode/llm_base_proposer.py b/vllm/v1/spec_decode/llm_base_proposer.py
index 9f46cbd24..1b465220a 100644
--- a/vllm/v1/spec_decode/llm_base_proposer.py
+++ b/vllm/v1/spec_decode/llm_base_proposer.py
@@ -14,7 +14,11 @@ from vllm.config import (
     get_layers_from_vllm_config,
     replace,
 )
-from vllm.distributed.parallel_state import get_pp_group
+from vllm import _custom_ops as ops
+from vllm.distributed.parallel_state import (
+    get_pp_group,
+    get_tensor_model_parallel_world_size,
+)
 from vllm.forward_context import set_forward_context
 from vllm.logger import init_logger
 from vllm.model_executor.layers.attention_layer_base import AttentionLayerBase
@@ -58,6 +62,10 @@ from vllm.v1.worker.utils import AttentionGroup
 logger = init_logger(__name__)
 
 
+_MTP_DRAFT_HEAD_ROWS = 98304
+_MTP_DRAFT_HEAD_TAIL = 2048
+
+
 class SpecDecodeBaseProposer:
     def __init__(
         self,
@@ -117,6 +125,11 @@ class SpecDecodeBaseProposer:
         self.parallel_drafting_hidden_state_tensor: torch.Tensor | None = None
         if self.parallel_drafting:
             self._init_parallel_drafting_params()
+        # (B_fp8 column-major, per-column scale) for MTP draft argmax.
+        self._mtp_fp8_head: tuple[torch.Tensor, torch.Tensor] | None = None
+        # Gap between the packed shortlist and the special-token tail. Keeping
+        # this as a Python integer avoids a persistent CUDA id-map allocation.
+        self._mtp_fp8_head_id_gap = 0
         self.use_local_argmax_reduction: bool = (
             self.speculative_config.use_local_argmax_reduction
         )
@@ -192,6 +205,7 @@ class SpecDecodeBaseProposer:
 
         # Will be set when we initialize the attention backend
         self.block_size: int = -1
+        self._eagle_prep_done = False
 
         # We need +1 here because the arange is used to set query_start_loc,
         # which has one more element than batch_size.
@@ -408,8 +422,161 @@ class SpecDecodeBaseProposer:
 
         self.cudagraph_dispatcher.initialize_cudagraph_keys(eagle_cudagraph_mode)
 
+    def _try_install_mtp_fp8_head(self) -> None:
+        """Private e4m3 copy of the draft lm_head for greedy MTP sampling.
+
+        The shared module is left untouched: load_model aliases the target
+        lm_head into the drafter, so in-place quant would change the
+        verifier. Draft argmax only moves the acceptance rate. Any failure
+        leaves ``_mtp_fp8_head`` unset and keeps ``compute_logits``.
+        """
+        self._mtp_fp8_head = None
+        self._mtp_fp8_head_id_gap = 0
+        try:
+            if self.method != "mtp":
+                return
+            if get_tensor_model_parallel_world_size() != 1:
+                return
+            if not current_platform.is_cuda():
+                return
+            cap = current_platform.get_device_capability()
+            if cap is None or not ops.cutlass_scaled_mm_supports_fp8(cap.to_int()):
+                return
+            head = getattr(self.model, "lm_head", None)
+            weight = getattr(head, "weight", None)
+            if not isinstance(weight, torch.Tensor) or weight.ndim != 2:
+                return
+            if not weight.is_cuda or weight.dtype not in (
+                torch.bfloat16,
+                torch.float16,
+            ):
+                return
+            org_vocab = getattr(head, "org_vocab_size", None)
+            if org_vocab is not None and int(org_vocab) != int(weight.shape[0]):
+                return
+
+            n_rows, hidden = int(weight.shape[0]), int(weight.shape[1])
+            # Require disjoint packed-tail source and destination ranges.
+            use_shortlist = n_rows >= (
+                _MTP_DRAFT_HEAD_ROWS + 2 * _MTP_DRAFT_HEAD_TAIL
+            )
+            n_keep = (
+                _MTP_DRAFT_HEAD_ROWS + _MTP_DRAFT_HEAD_TAIL
+                if use_shortlist
+                else n_rows
+            )
+            # Reserve the same persistent CUDA storage as the dense champion.
+            # Some target DeepGEMM FULL-graph warmups are allocation-layout
+            # sensitive; changing the drafter's footprint before target graph
+            # capture can expose an illegal-address failure. Only the packed
+            # prefix remains active after startup, so steady-state GEMM work
+            # is still proportional to n_keep.
+            w_fp8_storage = torch.empty(
+                (n_rows, hidden),
+                dtype=torch.float8_e4m3fn,
+                device=weight.device,
+            )
+            w_scale_storage = torch.empty(
+                (n_rows, 1), dtype=torch.float32, device=weight.device
+            )
+            src = weight.detach()
+            # Reproduce the dense champion's quantization and warmup first.
+            # Besides keeping startup behavior proven, this leaves the CUDA
+            # caching allocator and CUTLASS kernels in the same state before
+            # vLLM profiles/captures the target model.
+            for lo in range(0, n_rows, 4096):
+                hi = min(lo + 4096, n_rows)
+                block = src[lo:hi]
+                q, s = ops.scaled_fp8_quant(
+                    block,
+                    scale=None,
+                    use_per_token_if_dynamic=True,
+                )
+                w_fp8_storage[lo:hi].copy_(q)
+                w_scale_storage[lo:hi].copy_(s)
+            # CUTLASS B is K x N with stride(0) == 1 (transpose of row-major).
+            self._mtp_fp8_head = (
+                w_fp8_storage.t(),
+                w_scale_storage.reshape(1, n_rows),
+            )
+
+            dummy = torch.empty(
+                (self.max_batch_size, hidden),
+                dtype=weight.dtype,
+                device=weight.device,
+            )
+            widths: list[int] = []
+            m = 1
+            while m <= self.max_batch_size:
+                widths.append(m)
+                m *= 2
+            if widths[-1] != self.max_batch_size:
+                widths.append(self.max_batch_size)
+            for rows in widths:
+                self._greedy_sample(dummy[:rows])
+            torch.cuda.synchronize()
+
+            if use_shortlist:
+                # Pack the already-quantized special-token tail directly after
+                # the merge-ordered vocabulary head. The prefix view retains
+                # the full underlying allocations, preserving the champion's
+                # persistent CUDA footprint while cutting every scored GEMM.
+                tail_src = n_rows - _MTP_DRAFT_HEAD_TAIL
+                tail_dst = _MTP_DRAFT_HEAD_ROWS
+                tail_end = tail_dst + _MTP_DRAFT_HEAD_TAIL
+                w_fp8_storage[tail_dst:tail_end].copy_(
+                    w_fp8_storage[tail_src:n_rows]
+                )
+                w_scale_storage[tail_dst:tail_end].copy_(
+                    w_scale_storage[tail_src:n_rows]
+                )
+                self._mtp_fp8_head = (
+                    w_fp8_storage[:n_keep].t(),
+                    w_scale_storage[:n_keep].reshape(1, n_keep),
+                )
+                self._mtp_fp8_head_id_gap = n_rows - n_keep
+                # Do not execute this distinct GEMM shape before target CUDA
+                # graph capture. The benchmark's excluded request warmup will
+                # exercise it before any scored repetition.
+                torch.cuda.synchronize()
+            logger.info(
+                "Using e4m3 CUTLASS GEMM for MTP draft logits "
+                "(%d x %d of %d rows).",
+                n_keep,
+                hidden,
+                n_rows,
+            )
+        except Exception:
+            self._mtp_fp8_head = None
+            self._mtp_fp8_head_id_gap = 0
+            logger.debug(
+                "MTP draft lm_head staying on the dense path.", exc_info=True
+            )
+
     def _greedy_sample(self, hidden_states: torch.Tensor) -> torch.Tensor:
         """Greedy-sample draft tokens from hidden states."""
+        packed = self._mtp_fp8_head
+        if packed is not None:
+            w_t, w_scale = packed
+            flat = hidden_states.reshape(-1, hidden_states.shape[-1])
+            if not flat.is_contiguous():
+                flat = flat.contiguous()
+            act_q, act_s = ops.scaled_fp8_quant(
+                flat, scale=None, use_per_token_if_dynamic=True
+            )
+            logits = ops.cutlass_scaled_mm(
+                act_q,
+                w_t,
+                act_s,
+                w_scale,
+                out_dtype=hidden_states.dtype,
+            )
+            local = logits.argmax(dim=-1)
+            if self._mtp_fp8_head_id_gap == 0:
+                return local
+            return local + (local >= _MTP_DRAFT_HEAD_ROWS).to(
+                local.dtype
+            ) * self._mtp_fp8_head_id_gap
         if self.use_local_argmax_reduction:
             return self.model.get_top_tokens(hidden_states)
         return self.model.compute_logits(hidden_states).argmax(dim=-1)
@@ -872,6 +1039,78 @@ class SpecDecodeBaseProposer:
 
             return total_num_output_tokens, token_indices_to_sample, new_cad
 
+    @torch.inference_mode()
+    def precompile_eagle_prep(self, n_blocks_per_req: int, vocab_size: int) -> None:
+        """JIT the drafter input-prep kernels on throwaway tensors.
+
+        Batch sizes 2/8/32 cover Triton's ==1 / %16 / other specializations
+        for max_num_seqs <= 32. Token counts 1 and γ+1 cover first decode
+        vs later speculative steps.
+        """
+        if self._eagle_prep_done:
+            return
+        self._eagle_prep_done = True
+        device = self.device
+        widths = [w for w in (2, 8, 32) if w <= self.max_batch_size] or [1]
+        num_spec = self.num_speculative_tokens
+        token_counts = (1, num_spec + 1)
+
+        def i32(*shape: int) -> torch.Tensor:
+            return torch.zeros(shape, dtype=torch.int32, device=device)
+
+        try:
+            for ntok in token_counts:
+                for bs in widths:
+                    sampled = i32(bs, ntok)
+                    eagle_prepare_next_token_padded_kernel[(bs,)](
+                        sampled,
+                        torch.zeros(bs, dtype=torch.bool, device=device),
+                        i32(bs),
+                        i32(bs),
+                        i32(bs),
+                        vocab_size,
+                        ntok,
+                        bs,
+                        sampled.stride(0),
+                        BLOCK_SIZE_TOKENS=next_power_of_2(ntok),
+                    )
+        except Exception:
+            logger.debug("eagle_prepare_next_token_padded_kernel precompile skipped.")
+
+        try:
+            for bs in widths:
+                eagle_prepare_inputs_padded_kernel[(bs,)](
+                    i32(bs),
+                    i32(bs),
+                    i32(bs + 1),
+                    i32(bs),
+                    i32(bs),
+                    bs,
+                )
+        except Exception:
+            logger.debug("eagle_prepare_inputs_padded_kernel precompile skipped.")
+
+        if self.block_size <= 0:
+            return
+        try:
+            for bs in widths:
+                eagle_step_update_slot_mapping_and_metadata(
+                    positions_1d=torch.zeros(bs, dtype=torch.int64, device=device),
+                    block_table_tensor=i32(bs, n_blocks_per_req),
+                    seq_lens=i32(bs),
+                    block_size=self.block_size,
+                    max_model_len=self.max_model_len,
+                    out_clamped_positions=torch.zeros(
+                        bs, dtype=torch.int64, device=device
+                    ),
+                    out_slot_mapping=torch.zeros(
+                        bs, dtype=torch.int64, device=device
+                    ),
+                    input_batch_size=bs,
+                )
+        except Exception:
+            logger.debug("eagle slot-mapping precompile skipped.")
+
     def build_model_inputs_first_pass(
         self,
         num_tokens: int,
@@ -1306,6 +1545,7 @@ class SpecDecodeBaseProposer:
 
         self._maybe_share_embeddings(target_language_model)
         self._maybe_share_lm_head(target_language_model)
+        self._try_install_mtp_fp8_head()
 
         if (
             self.parallel_drafting
diff --git a/vllm/v1/worker/block_table.py b/vllm/v1/worker/block_table.py
index d9c041ba0..a3e6393ff 100644
--- a/vllm/v1/worker/block_table.py
+++ b/vllm/v1/worker/block_table.py
@@ -163,6 +163,15 @@ class BlockTable:
             BLOCK_SIZE=1024,
         )
 
+    def precompile_slot_kernel(self) -> None:
+        n_req = 2
+        n_tok = 2
+        query_start_loc = torch.tensor(
+            [0, 1, n_tok], dtype=torch.int32, device=self.device
+        )
+        positions = torch.zeros(n_tok, dtype=torch.int64, device=self.device)
+        self.compute_slot_mapping(n_req, query_start_loc, positions)
+
     def commit_block_table(self, num_reqs: int) -> None:
         self.block_table.copy_to_gpu(num_reqs)
 
@@ -321,6 +330,10 @@ class MultiGroupBlockTable:
         """Returns the BlockTable for the i-th KV cache group."""
         return self.block_tables[idx]
 
+    def precompile_slot_kernel(self) -> None:
+        for table in self.block_tables:
+            table.precompile_slot_kernel()
+
 
 @triton.jit(do_not_specialize=["num_tokens", "max_num_tokens"])
 def _compute_slot_mapping_kernel(
diff --git a/vllm/v1/worker/gpu_model_runner.py b/vllm/v1/worker/gpu_model_runner.py
index 74938a823..081336d5d 100644
--- a/vllm/v1/worker/gpu_model_runner.py
+++ b/vllm/v1/worker/gpu_model_runner.py
@@ -765,6 +765,9 @@ class GPUModelRunner(
         self.num_accepted_tokens = self._make_buffer(
             self.max_num_reqs, dtype=torch.int32
         )
+        # When True, GDN/Mamba2 builders still receive spec metadata even
+        # though this step has no scheduled drafts (see _prepare_inputs).
+        self._gdn_keep_spec_path = False
 
         # Only relevant for models using M-RoPE (e.g, Qwen2-VL)
         if self.uses_mrope:
@@ -2159,6 +2162,7 @@ class GPUModelRunner(
             target.gpu[:, :total_num_scheduled_tokens] += drift
 
         use_spec_decode = len(scheduler_output.scheduled_spec_decode_tokens) > 0
+        self._gdn_keep_spec_path = False
         if not use_spec_decode:
             # NOTE(woosuk): Due to chunked prefills, the batch may contain
             # partial requests. While we should not sample any token
@@ -2168,6 +2172,15 @@ class GPUModelRunner(
             logits_indices = query_start_loc[1:] - 1
             spec_decode_metadata = None
             num_sampled_tokens = np.ones(num_reqs, dtype=np.int32)
+            decode_mask = self._hybrid_gdn_decode_mask(num_reqs)
+            if decode_mask is not None:
+                # Draft-count 0 is still a spec row (gdn_attn uses >= 0).
+                drafts = np.full(num_reqs, -1, dtype=np.int32)
+                drafts[decode_mask] = 0
+                self.num_decode_draft_tokens.np[:num_reqs] = drafts
+                self.num_decode_draft_tokens.np[num_reqs:].fill(-1)
+                self.num_decode_draft_tokens.copy_to_gpu()
+                self._gdn_keep_spec_path = True
         else:
             # Get the number of draft tokens for each request.
             # Iterate over the dictionary rather than all requests since not all
@@ -2188,6 +2201,10 @@ class GPUModelRunner(
                     >= self.input_batch.num_prompt_tokens[req_idx]
                 ):
                     num_decode_draft_tokens[req_idx] = draft_len
+            decode_mask = self._hybrid_gdn_decode_mask(num_reqs)
+            if decode_mask is not None:
+                starve = decode_mask & (num_decode_draft_tokens < 0)
+                num_decode_draft_tokens[starve] = 0
             spec_decode_metadata = self._calc_spec_decode_metadata(
                 num_draft_tokens, cu_num_tokens
             )
@@ -2213,6 +2230,22 @@ class GPUModelRunner(
             spec_decode_metadata,
         )
 
+    def _hybrid_gdn_decode_mask(self, num_reqs: int) -> np.ndarray | None:
+        """Decode-row mask for hybrid GDN, or None if rewind does not apply.
+
+        After a multi-token accept, GDN writes recurrent state at column
+        (accepted-1). The non-spec path always reads column 0, so a decode
+        row that has no drafts this step would otherwise skip the rewind.
+        """
+        if self.speculative_config is None or not self.model_config.is_hybrid:
+            return None
+        computed = self.input_batch.num_computed_tokens_cpu[:num_reqs]
+        prompt = self.input_batch.num_prompt_tokens[:num_reqs]
+        mask = computed >= prompt
+        if not bool(mask.any()):
+            return None
+        return mask
+
     def _build_attention_metadata(
         self,
         num_tokens: int,
@@ -2404,7 +2437,7 @@ class GPUModelRunner(
             )
 
             extra_attn_metadata_args = {}
-            if use_spec_decode and isinstance(
+            if (use_spec_decode or self._gdn_keep_spec_path) and isinstance(
                 builder, (Mamba2AttentionMetadataBuilder, GDNAttentionMetadataBuilder)
             ):
                 assert ubid is None, "UBatching not supported with GDN yet"
diff --git a/vllm/v1/worker/utils.py b/vllm/v1/worker/utils.py
index c0f44b6db..ef011ff26 100644
--- a/vllm/v1/worker/utils.py
+++ b/vllm/v1/worker/utils.py
@@ -113,6 +113,7 @@ class KVBlockZeroer:
         self._id_cap: int = 0
         self._ids_pinned: torch.Tensor | None = None
         self._ids_gpu: torch.Tensor | None = None
+        self._zero_compiled = False
 
         if runner_only_attn_layers is None:
             runner_only_attn_layers = set()
@@ -218,6 +219,28 @@ class KVBlockZeroer:
             BLOCK_SIZE=blk_size,
         )
 
+    def precompile_zero_kernel(self) -> None:
+        """Compile _zero_kv_blocks_kernel on a scratch page, not the KV cache."""
+        if self._meta is None or self._zero_compiled:
+            return
+        self._zero_compiled = True
+        _, page_size_el, blk_size, n_segs = self._meta
+        scratch = torch.zeros(page_size_el, dtype=torch.int32, device=self.device)
+        seg_addrs = torch.full(
+            (n_segs,), scratch.data_ptr(), dtype=torch.uint64, device=self.device
+        )
+        for n_blocks in (1, 8, 16):
+            ids = torch.zeros(n_blocks, dtype=torch.int64, device=self.device)
+            grid = (n_blocks * n_segs * (page_size_el // blk_size),)
+            _zero_kv_blocks_kernel[grid](
+                seg_addrs,
+                ids,
+                n_blocks,
+                N_SEGS=n_segs,
+                PAGE_SIZE_EL=page_size_el,
+                BLOCK_SIZE=blk_size,
+            )
+
 
 @dataclass
 class AttentionGroup: