diff --git a/examples/auto_deploy/model_registry/configs/super_v3_mtp.yaml b/examples/auto_deploy/model_registry/configs/super_v3_mtp.yaml
index e4ea9ab6ee0a..c3b5911ff956 100644
--- a/examples/auto_deploy/model_registry/configs/super_v3_mtp.yaml
+++ b/examples/auto_deploy/model_registry/configs/super_v3_mtp.yaml
@@ -2,17 +2,17 @@
 #
 runtime: trtllm
 compile_backend: torch-cudagraph
-max_batch_size: 128
-max_seq_len: 65536 # tunable
+max_batch_size: 64
+max_seq_len: 8192 # tunable
+max_num_tokens: 2048
 enable_chunked_prefill: false # TODO: Support chunked prefill w/ spec dec
 attn_backend: trtllm
 model_factory: AutoModelForCausalLM
 skip_loading_weights: false
 cuda_graph_config:
-  batch_sizes: [1, 2, 4, 8, 16, 24, 32, 64, 128]
+  max_batch_size: 64
+  enable_padding: true
 kv_cache_config:
-  # tunable mamba cache dtype
-  # --> use float32 for accuracy and default (auto) for speed
   mamba_ssm_cache_dtype: auto
 speculative_config:
   decoding_type: MTP
@@ -54,6 +54,7 @@ transforms:
   # for speculative Mamba state caching in this configuration.
   insert_cached_ssm_attention:
     backend: flashinfer_ssm
+    ssm_replay: true
   insert_cached_causal_conv:
     backend: triton_causal_conv
   fuse_nvfp4_moe:
diff --git a/tensorrt_llm/_torch/auto_deploy/custom_ops/attention_interface.py b/tensorrt_llm/_torch/auto_deploy/custom_ops/attention_interface.py
index 838a1731791a..cf907ee08e39 100644
--- a/tensorrt_llm/_torch/auto_deploy/custom_ops/attention_interface.py
+++ b/tensorrt_llm/_torch/auto_deploy/custom_ops/attention_interface.py
@@ -402,7 +402,7 @@ class BatchInfo:
     Args:
         batch_info_host: The batch info tensor on the host.
 
-    The information is stored in a 12-element batch_info_host tensor as follows:
+    The information is stored in a 14-element batch_info_host tensor as follows:
 
     Slots 0-5 (batch composition):
     - [0] num_prefill: number of prefill requests
@@ -425,10 +425,13 @@ class BatchInfo:
     Slot 12 (spec-decoding info, set once at runtime init):
     - [12] max_draft_len: maximum draft length per request (0 when not spec dec)
 
+    Slot 13 (replay mode flag, set once at runtime init):
+    - [13] use_replay: 1 if SSM replay state-update path is active, 0 otherwise
+
     All fields can be accessed and updated with the convenience functions below.
     """
 
-    _NUM_ELEMENTS = 13
+    _NUM_ELEMENTS = 14
 
     def __init__(self, batch_info_host: Optional[torch.Tensor] = None):
         if batch_info_host is None:
@@ -551,6 +554,16 @@ def update_max_draft_len(self, max_draft_len: int) -> None:
     def get_max_draft_len(self) -> int:
         return int(self._batch_info[12])
 
+    # --- replay mode flag (slot 13) writer ---
+
+    def update_use_replay(self, use_replay: bool) -> None:
+        self._batch_info[13] = int(use_replay)
+
+    # --- replay mode flag (slot 13) reader ---
+
+    def is_use_replay(self) -> bool:
+        return bool(self._batch_info[13])
+
 
 class SequenceInfo:
     """An interface to hold information about how the sequence is laid out and stored in cache.
@@ -2098,6 +2111,149 @@ def from_base(cls, base: Optional["SSMResourceHandler"]) -> Optional["SpecSSMRes
         )
 
 
+class ReplayOldXHandler(StateResourceHandler):
+    """Per-layer old_x cache for the replay SSM kernel (single-buffered, bf16).
+
+    Shape: (max_batch, T, num_heads, head_dim) — T is determined by the manager's
+    spec_config (max_draft_len + 1), not by this handler.  Acts as a type marker.
+    Routes to MambaHybridCacheManager via get_replay_old_x(layer_idx).
+    """
+
+    def __init__(self, num_heads: int, head_dim: int, dtype: torch.dtype) -> None:
+        self.num_heads = num_heads
+        self.head_dim = head_dim
+        super().__init__(dtype=dtype)
+
+    @property
+    def state_shape(self) -> Tuple[int, int]:
+        return (self.num_heads, self.head_dim)
+
+    def allocate(self, sequence_info) -> torch.Tensor:
+        raise RuntimeError("ReplayOldXHandler must be backed by MambaHybridCacheManager")
+
+    def __eq__(self, other) -> bool:
+        return (
+            isinstance(other, ReplayOldXHandler)
+            and self.num_heads == other.num_heads
+            and self.head_dim == other.head_dim
+            and self.dtype == other.dtype
+        )
+
+
+class ReplayOldBHandler(StateResourceHandler):
+    """Per-layer old_B cache for the replay SSM kernel (double-buffered, bf16).
+
+    Shape: (max_batch, 2, T, n_groups, d_state) — T from manager.
+    Routes to MambaHybridCacheManager via get_replay_old_B(layer_idx).
+    """
+
+    def __init__(self, n_groups: int, d_state: int, dtype: torch.dtype) -> None:
+        self.n_groups = n_groups
+        self.d_state = d_state
+        super().__init__(dtype=dtype)
+
+    @property
+    def state_shape(self) -> Tuple[int, int]:
+        return (self.n_groups, self.d_state)
+
+    def allocate(self, sequence_info) -> torch.Tensor:
+        raise RuntimeError("ReplayOldBHandler must be backed by MambaHybridCacheManager")
+
+    def __eq__(self, other) -> bool:
+        return (
+            isinstance(other, ReplayOldBHandler)
+            and self.n_groups == other.n_groups
+            and self.d_state == other.d_state
+            and self.dtype == other.dtype
+        )
+
+
+class ReplayOldDtHandler(StateResourceHandler):
+    """Per-layer old_dt cache for the replay SSM kernel (double-buffered, fp32).
+
+    Shape: (max_batch, 2, num_heads, T) — T from manager.
+    Routes to MambaHybridCacheManager via get_replay_old_dt(layer_idx).
+    """
+
+    def __init__(self, num_heads: int) -> None:
+        self.num_heads = num_heads
+        super().__init__(dtype=torch.float32)
+
+    @property
+    def state_shape(self) -> Tuple[int]:
+        return (self.num_heads,)
+
+    def allocate(self, sequence_info) -> torch.Tensor:
+        raise RuntimeError("ReplayOldDtHandler must be backed by MambaHybridCacheManager")
+
+    def __eq__(self, other) -> bool:
+        return isinstance(other, ReplayOldDtHandler) and self.num_heads == other.num_heads
+
+
+class ReplayOldDAcumsumHandler(StateResourceHandler):
+    """Per-layer old_dA_cumsum cache for the replay SSM kernel (double-buffered, fp32).
+
+    Shape: (max_batch, 2, num_heads, T) — T from manager.
+    Routes to MambaHybridCacheManager via get_replay_old_dA_cumsum(layer_idx).
+    """
+
+    def __init__(self, num_heads: int) -> None:
+        self.num_heads = num_heads
+        super().__init__(dtype=torch.float32)
+
+    @property
+    def state_shape(self) -> Tuple[int]:
+        return (self.num_heads,)
+
+    def allocate(self, sequence_info) -> torch.Tensor:
+        raise RuntimeError("ReplayOldDAcumsumHandler must be backed by MambaHybridCacheManager")
+
+    def __eq__(self, other) -> bool:
+        return isinstance(other, ReplayOldDAcumsumHandler) and self.num_heads == other.num_heads
+
+
+class ReplayCacheBufIdxHandler(StateResourceHandler):
+    """Global cache_buf_idx tensor for the replay SSM kernel (shared across all layers, int32).
+
+    Shape: (max_batch,).  Routes to MambaHybridCacheManager.get_replay_cache_buf_idx().
+    The same tensor is returned for every SSM layer — the FX graph has one constant-address
+    input per layer, all pointing to the same buffer.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(dtype=torch.int32)
+
+    @property
+    def state_shape(self) -> Tuple[()]:
+        return ()
+
+    def allocate(self, sequence_info) -> torch.Tensor:
+        raise RuntimeError("ReplayCacheBufIdxHandler must be backed by MambaHybridCacheManager")
+
+    def __eq__(self, other) -> bool:
+        return isinstance(other, ReplayCacheBufIdxHandler)
+
+
+class ReplayPrevNumAcceptedHandler(StateResourceHandler):
+    """Global prev_num_accepted_tokens tensor for the replay SSM kernel (int32, shared).
+
+    Shape: (max_batch,).  Routes to MambaHybridCacheManager.get_replay_prev_num_accepted_tokens().
+    """
+
+    def __init__(self) -> None:
+        super().__init__(dtype=torch.int32)
+
+    @property
+    def state_shape(self) -> Tuple[()]:
+        return ()
+
+    def allocate(self, sequence_info) -> torch.Tensor:
+        raise RuntimeError("ReplayPrevNumAcceptedHandler must be backed by MambaHybridCacheManager")
+
+    def __eq__(self, other) -> bool:
+        return isinstance(other, ReplayPrevNumAcceptedHandler)
+
+
 class SpecCausalConvResourceHandler(StateResourceHandler):
     """Intermediate conv state cache descriptor for speculative decoding.
 
diff --git a/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/flashinfer_backend_mamba.py b/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/flashinfer_backend_mamba.py
index 563cfe34c9b9..076d1fb4ba4a 100644
--- a/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/flashinfer_backend_mamba.py
+++ b/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/flashinfer_backend_mamba.py
@@ -19,11 +19,22 @@
 from flashinfer.mamba import selective_state_update as _flashinfer_ssm_update
 from torch.fx import Node
 
+from tensorrt_llm._torch.modules.mamba.replay_selective_state_update import (
+    replay_selective_state_update,
+)
+from tensorrt_llm._utils import get_sm_version
+
 from ..._compat import KvCacheConfig
 from ..attention_interface import (
     AttentionRegistry,
     BatchInfo,
     MHACallable,
+    ReplayCacheBufIdxHandler,
+    ReplayOldBHandler,
+    ReplayOldDAcumsumHandler,
+    ReplayOldDtHandler,
+    ReplayOldXHandler,
+    ReplayPrevNumAcceptedHandler,
     ResourceHandlerDict,
     SpecSSMResourceHandler,
 )
@@ -49,7 +60,16 @@ def _fi_align(t: torch.Tensor) -> torch.Tensor:
 
 @torch.library.custom_op(
     "auto_deploy::flashinfer_cached_ssm",
-    mutates_args=("ssm_state_cache", "intermediate_ssm_state_cache"),
+    mutates_args=(
+        "ssm_state_cache",
+        "intermediate_ssm_state_cache",
+        # replay buffers: written in-place by the precompute and main kernels
+        # (double-buffered B/dt/dA_cumsum, single-buffered x); None in non-replay mode
+        "replay_old_x",
+        "replay_old_b",
+        "replay_old_dt",
+        "replay_old_da_cumsum",
+    ),
 )
 def _flashinfer_cached_ssm(
     # INPUTS (dense but may be flattened across sequences)
@@ -74,7 +94,15 @@ def _flashinfer_cached_ssm(
     ssm_state_cache: torch.Tensor,  # [max_batch_size, num_heads, head_dim, ssm_state_size]
     intermediate_ssm_state_cache: Optional[
         torch.Tensor
-    ],  # [spec_state_size, max_draft_len+1, num_heads, head_dim, d_state]
+    ],  # [spec_state_size, max_draft_len+1, num_heads, head_dim, d_state]; None in replay mode
+    replay_old_x: Optional[torch.Tensor],  # [max_batch, T, nheads, head_dim]; None in non-replay
+    replay_old_b: Optional[torch.Tensor],  # [max_batch, 2, T, ngroups, dstate]; None in non-replay
+    replay_old_dt: Optional[torch.Tensor],  # [max_batch, 2, nheads, T] fp32; None in non-replay
+    replay_old_da_cumsum: Optional[
+        torch.Tensor
+    ],  # [max_batch, 2, nheads, T] fp32; None in non-replay
+    replay_cache_buf_idx: Optional[torch.Tensor],  # [max_batch] int32; None in non-replay
+    replay_prev_num_accepted: Optional[torch.Tensor],  # [max_batch] int32; None in non-replay
     # CONSTANTS
     time_step_limit: List[float],
     chunk_size: int,
@@ -89,6 +117,17 @@ def _flashinfer_cached_ssm(
     num_prefill_tokens, num_extend_tokens, num_decode_tokens = batch_info.get_num_tokens()
     num_total_tokens = num_prefill_tokens + num_extend_tokens + num_decode_tokens
 
+    # Pre-hoist index type conversions so their kernels run before the extend
+    # causal_conv1d_update kernel rather than landing in the gap between it
+    # and the PDL-dependent replay precompute kernel.
+    # slot_idx is a per-request cache slot index in [0, max_batch_size], so the
+    # int32 cast can never overflow.
+    slot_idx_i32 = slot_idx.to(torch.int32)
+    # NOTE: intermediate_state_indices_extend (torch.arange) is NOT hoisted here.
+    # It is only needed for the non-replay FlashInfer extend path. Hoisting it
+    # unconditionally would create an unnecessary arange kernel in the gap for
+    # the replay path (where use_replay=True and the arange is unused).
+
     if out is not None:
         preallocated_ssm_out = out.view(bs, num_heads, head_dim)
     else:
@@ -130,7 +169,7 @@ def _flashinfer_cached_ssm(
         A,
         D,
         dt_bias,
-        slot_idx,
+        slot_idx_i32,
         seq_start=num_prefill,
         token_start=num_prefill_tokens,
         num_seq=num_extend,
@@ -142,12 +181,6 @@ def _flashinfer_cached_ssm(
 
     if extend_inputs is not None:
         tokens_per_extend = num_extend_tokens // num_extend
-        if intermediate_ssm_state_cache.size(1) < tokens_per_extend:
-            raise RuntimeError(
-                "flashinfer_cached_ssm: intermediate_ssm_state_cache is too small "
-                f"for extend branch (size1={intermediate_ssm_state_cache.size(1)}, "
-                f"tokens_per_extend={tokens_per_extend})"
-            )
         (
             slot_idx_extend,
             x_extend,
@@ -163,27 +196,64 @@ def _flashinfer_cached_ssm(
             num_prefill_tokens : num_prefill_tokens + num_extend_tokens
         ].view(num_extend, tokens_per_extend, num_heads, head_dim)
 
-        intermediate_state_indices = torch.arange(
-            num_extend, dtype=torch.int32, device=slot_idx_extend.device
-        )
-        _flashinfer_ssm_update(
-            ssm_state_cache,
-            x_extend,
-            dt_extend,
-            A_full,
-            B_extend,
-            C_extend,
-            D_full,
-            z=None,
-            dt_bias=dt_bias_hp,
-            dt_softplus=True,
-            state_batch_indices=slot_idx_extend.to(torch.int32),
-            out=preallocated_ssm_out_e,
-            disable_state_update=True,
-            intermediate_states_buffer=intermediate_ssm_state_cache,
-            cache_steps=tokens_per_extend,
-            intermediate_state_indices=intermediate_state_indices,
-        )
+        use_replay = batch_info.is_use_replay()
+        if use_replay:
+            # Replay path: fast-forward SSM state via tl.dot on cached values.
+            # State is updated in-place; no disable_state_update needed.
+            # x_extend/B_extend/C_extend are non-contiguous views from the CUDA graph's
+            # preallocated buffer (stride_T > nheads × head_dim). The Triton kernel
+            # handles arbitrary strides via pointer arithmetic; skipping .contiguous()
+            # avoids a captured BF16 copy kernel and reduces GPU time by ~5%.
+            replay_selective_state_update(
+                ssm_state_cache,
+                replay_old_x,
+                replay_old_b,
+                replay_old_dt,
+                replay_old_da_cumsum,
+                replay_cache_buf_idx,
+                replay_prev_num_accepted,
+                x_extend,
+                dt_extend,
+                A_full,
+                B_extend,
+                C_extend,
+                out=preallocated_ssm_out_e,
+                D=D_full,
+                dt_bias=dt_bias_hp,
+                dt_softplus=True,
+                state_batch_indices=slot_idx_extend,
+                launch_with_pdl=True,  # PDL chain: triton_causal_conv extend → precompute → main
+            )
+        else:
+            if intermediate_ssm_state_cache.size(1) < tokens_per_extend:
+                raise RuntimeError(
+                    "flashinfer_cached_ssm: intermediate_ssm_state_cache is too small "
+                    f"for extend branch (size1={intermediate_ssm_state_cache.size(1)}, "
+                    f"tokens_per_extend={tokens_per_extend})"
+                )
+            # Only needed for the non-replay FlashInfer path; not hoisted to avoid
+            # an unnecessary arange kernel in the stream gap for the replay path.
+            intermediate_state_indices = torch.arange(
+                num_extend, dtype=torch.int32, device=slot_idx_extend.device
+            )
+            _flashinfer_ssm_update(
+                ssm_state_cache,
+                x_extend.contiguous(),
+                dt_extend,
+                A_full,
+                B_extend.contiguous(),
+                C_extend.contiguous(),
+                D_full,
+                z=None,
+                dt_bias=dt_bias_hp,
+                dt_softplus=True,
+                state_batch_indices=slot_idx_extend,
+                out=preallocated_ssm_out_e,
+                disable_state_update=True,
+                intermediate_states_buffer=intermediate_ssm_state_cache,
+                cache_steps=tokens_per_extend,
+                intermediate_state_indices=intermediate_state_indices,
+            )
 
     # DECODE: single-token autoregressive path
     decode_inputs = _prepare_ssm_decode_inputs(
@@ -194,7 +264,7 @@ def _flashinfer_cached_ssm(
         A,
         D,
         dt_bias,
-        slot_idx,
+        slot_idx_i32,
         num_prefill + num_extend,
         num_prefill_tokens + num_extend_tokens,
         num_decode,
@@ -220,7 +290,6 @@ def _flashinfer_cached_ssm(
         B_decode = _fi_align(B_decode)
         C_decode = _fi_align(C_decode)
 
-        slot_idx_decode_i32 = slot_idx_decode.to(torch.int32)
         y_decode = _flashinfer_ssm_update(
             ssm_state_cache,
             x_decode,
@@ -232,7 +301,7 @@ def _flashinfer_cached_ssm(
             z=None,
             dt_bias=dt_bias_hp,
             dt_softplus=True,
-            state_batch_indices=slot_idx_decode_i32,
+            state_batch_indices=slot_idx_decode,
         )
         preallocated_ssm_out[num_prefill_tokens + num_extend_tokens : num_total_tokens].copy_(
             y_decode
@@ -272,7 +341,15 @@ def _flashinfer_cached_ssm_fake(
     ssm_state_cache: torch.Tensor,  # [max_batch_size, num_heads, head_dim, ssm_state_size]
     intermediate_ssm_state_cache: Optional[
         torch.Tensor
-    ],  # [spec_state_size, max_draft_len+1, num_heads, head_dim, d_state]
+    ],  # [spec_state_size, max_draft_len+1, num_heads, head_dim, d_state]; None in replay mode
+    replay_old_x: Optional[torch.Tensor],  # [max_batch, T, nheads, head_dim]; None in non-replay
+    replay_old_b: Optional[torch.Tensor],  # [max_batch, 2, T, ngroups, dstate]; None in non-replay
+    replay_old_dt: Optional[torch.Tensor],  # [max_batch, 2, nheads, T] fp32; None in non-replay
+    replay_old_da_cumsum: Optional[
+        torch.Tensor
+    ],  # [max_batch, 2, nheads, T] fp32; None in non-replay
+    replay_cache_buf_idx: Optional[torch.Tensor],  # [max_batch] int32; None in non-replay
+    replay_prev_num_accepted: Optional[torch.Tensor],  # [max_batch] int32; None in non-replay
     # CONSTANTS
     time_step_limit: List[float],
     chunk_size: int,
@@ -294,6 +371,11 @@ def _flashinfer_cached_ssm_fake(
 
 @AttentionRegistry.register("flashinfer_ssm")
 class FlashinferBackendSSM(BaseBackendSSM):
+    # ssm_replay=True allocates replay buffers at compile time (SM80+), which causes
+    # BatchInfo.is_use_replay() to return True at runtime.  Never check ssm_replay at
+    # runtime — use batch_info.is_use_replay() instead.
+    ssm_replay: bool = False
+
     @classmethod
     def get_cached_attention_op(cls) -> MHACallable:
         return torch.ops.auto_deploy.flashinfer_cached_ssm.default
@@ -304,15 +386,44 @@ def get_cache_initializers(
     ) -> ResourceHandlerDict:
         ret = super().get_cache_initializers(source_attn_node, cache_config)
 
-        # check head_dim is supported by flashinfer
-        if ret["ssm_state_cache"].head_dim not in FLASHINFER_SUPPORTED_HEAD_DIMS:
+        # Replay requires both ssm_replay=True and SM >= 80 (Ampere+).
+        # Only when use_replay=True do we skip FlashInfer at runtime; otherwise
+        # _flashinfer_ssm_update is called and head_dim must be supported.
+        use_replay = cls.ssm_replay and get_sm_version() >= 80
+
+        if not use_replay and ret["ssm_state_cache"].head_dim not in FLASHINFER_SUPPORTED_HEAD_DIMS:
             raise ValueError(
                 f"flashinfer_ssm only supports head_dim in {FLASHINFER_SUPPORTED_HEAD_DIMS}. "
                 f"Got head_dim={ret['ssm_state_cache'].head_dim}. "
                 "Consider using 'triton_ssm' backend instead."
             )
 
-        ret["intermediate_ssm_state_cache"] = SpecSSMResourceHandler.from_base(
-            ret["ssm_state_cache"]
-        )
+        ssm_h = ret["ssm_state_cache"]
+
+        # All 7 optional caches are always registered positionally (None = unused in this mode).
+        # intermediate_ssm_state_cache: real in non-replay, None in replay.
+        # replay_old_*: real in replay mode (SM80+), None otherwise.
+        if use_replay:
+            ret["intermediate_ssm_state_cache"] = None
+            x_dtype = torch.bfloat16
+            B_fake = source_attn_node.args[2].meta["val"]
+            n_groups = B_fake.shape[-2] if B_fake.ndim >= 4 else 1
+            ret["replay_old_x"] = ReplayOldXHandler(
+                num_heads=ssm_h.num_heads, head_dim=ssm_h.head_dim, dtype=x_dtype
+            )
+            ret["replay_old_b"] = ReplayOldBHandler(
+                n_groups=n_groups, d_state=ssm_h.d_state, dtype=x_dtype
+            )
+            ret["replay_old_dt"] = ReplayOldDtHandler(num_heads=ssm_h.num_heads)
+            ret["replay_old_da_cumsum"] = ReplayOldDAcumsumHandler(num_heads=ssm_h.num_heads)
+            ret["replay_cache_buf_idx"] = ReplayCacheBufIdxHandler()
+            ret["replay_prev_num_accepted"] = ReplayPrevNumAcceptedHandler()
+        else:
+            ret["intermediate_ssm_state_cache"] = SpecSSMResourceHandler.from_base(ssm_h)
+            ret["replay_old_x"] = None
+            ret["replay_old_b"] = None
+            ret["replay_old_dt"] = None
+            ret["replay_old_da_cumsum"] = None
+            ret["replay_cache_buf_idx"] = None
+            ret["replay_prev_num_accepted"] = None
         return ret
diff --git a/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/triton_backend_causal_conv.py b/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/triton_backend_causal_conv.py
index e90245e566ab..a3f4581d7e89 100644
--- a/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/triton_backend_causal_conv.py
+++ b/tensorrt_llm/_torch/auto_deploy/custom_ops/mamba/triton_backend_causal_conv.py
@@ -100,6 +100,10 @@ def _triton_cached_causal_conv1d(
     else:
         w2d = weight
 
+    # Pre-hoist index type conversions so they run as early kernels, not between
+    # the extend conv1d update and the downstream replay precompute kernel.
+    slot_idx_i32 = slot_idx.to(torch.int32)
+
     # PREFILL: concatenate all prefill tokens and run one varlen forward
     if num_prefill > 0:
         # x_varlen: (dim, cu_seq_len)
@@ -117,7 +121,7 @@ def _triton_cached_causal_conv1d(
             conv_state_cache,
             prefill_cu_seqlen,
             seq_lens_cpu,
-            cache_indices=slot_idx[:num_prefill].to(torch.int32),
+            cache_indices=slot_idx_i32[:num_prefill],
             has_initial_state=use_initial_states[:num_prefill],
             activation=activation,
             pad_slot_id=PAD_SLOT_ID,
@@ -125,7 +129,40 @@ def _triton_cached_causal_conv1d(
         # Scatter outputs back to input buffer
         inp_flat[:num_prefill_tokens] = y_varlen.transpose(0, 1)
 
+    # DECODE: process decode before extend so decode-related kernels (slot index
+    # conversions, scatter-back) don't land in the gap between the extend
+    # causal_conv1d_update kernel and the downstream PDL-dependent replay precompute
+    # kernel. Decode and extend access disjoint slot indices so reordering is safe.
+    if num_decode > 0:
+        x_decode = inp_flat[
+            num_prefill_tokens + num_extend_tokens : num_total_tokens
+        ]  # [num_decode_tokens, C_in]
+
+        # Note: Triton causal_conv1d_update returns a new tensor (not in-place like CUDA version)
+        # so we need to capture the output and write it back
+        y_decode = causal_conv1d_update(
+            x_decode,  # [batch, dim]
+            conv_state_cache,
+            w2d,
+            bias,
+            activation=activation,
+            cache_seqlens=None,
+            conv_state_indices=slot_idx_i32[num_prefill + num_extend : num_seq],
+            pad_slot_id=PAD_SLOT_ID,
+        )
+        inp_flat[num_prefill_tokens + num_extend_tokens : num_total_tokens] = y_decode
+
+    # Zero padding positions beyond valid tokens (for piecewise CUDA graph).
+    # Run before extend so the zero_() kernel doesn't land after extend conv.
+    if num_total_tokens < bs:
+        inp_flat[num_total_tokens:].zero_()
+
     # EXTEND: use the update kernel so extend tokens write the intermediate state cache.
+    # This block runs last so that:
+    #   1. All non-extend kernels (decode, zero_()) have already executed.
+    #   2. After causal_conv1d_update emits gdc_launch_dependents(), the very next
+    #      kernel in this stream is the extend scatter-back — just one copy_ — before
+    #      the downstream flashinfer_ssm precompute kernel starts via PDL.
     if num_extend > 0:
         # num_extend_tokens == num_extend * (max_draft_len + 1) for static draft lengths
         # (dynamic lengths not supported)
@@ -136,7 +173,7 @@ def _triton_cached_causal_conv1d(
                 "that is too small for the extend branch"
             )
 
-        slot_idx_extend = slot_idx[num_prefill : num_prefill + num_extend].to(torch.int32)
+        slot_idx_extend = slot_idx_i32[num_prefill : num_prefill + num_extend]
 
         # The intermediate state cache will be stored in these indices and read by the mamba_cache_manager,
         # which expects them in the indices arange(num_extend). They are not used across requests, so we
@@ -161,34 +198,15 @@ def _triton_cached_causal_conv1d(
             intermediate_conv_window=intermediate_conv_state_cache,
             intermediate_state_indices=intermediate_state_indices,
             pad_slot_id=PAD_SLOT_ID,
+            launch_dependent_kernels=True,  # PDL signal to downstream replay precompute
         )
-        inp_flat[num_prefill_tokens : num_prefill_tokens + num_extend_tokens] = y_extend.transpose(
-            1, 2
-        ).view(-1, inp_flat.shape[1])
-
-    # DECODE: batch update for single-token sequences
-    if num_decode > 0:
-        x_decode = inp_flat[
-            num_prefill_tokens + num_extend_tokens : num_total_tokens
-        ]  # [num_decode_tokens, C_in]
-
-        # Note: Triton causal_conv1d_update returns a new tensor (not in-place like CUDA version)
-        # so we need to capture the output and write it back
-        y_decode = causal_conv1d_update(
-            x_decode,  # [batch, dim]
-            conv_state_cache,
-            w2d,
-            bias,
-            activation=activation,
-            cache_seqlens=None,
-            conv_state_indices=slot_idx[num_prefill + num_extend : num_seq].to(torch.int32),
-            pad_slot_id=PAD_SLOT_ID,
-        )
-        inp_flat[num_prefill_tokens + num_extend_tokens : num_total_tokens] = y_decode
-
-    # Zero padding positions beyond valid tokens (for piecewise CUDA graph)
-    if num_total_tokens < bs:
-        inp_flat[num_total_tokens:].zero_()
+        # Single copy_ scatter-back: y_extend [N,C,T] → inp_flat slice [N*T,C].
+        # Uses one efficient non-contiguous-to-contiguous copy instead of
+        # transpose+view+assign (previously 2+ kernels). This is now the only kernel
+        # between causal_conv1d_update and the PDL-dependent precompute.
+        inp_flat[num_prefill_tokens : num_prefill_tokens + num_extend_tokens].view(
+            num_extend, tokens_per_extend, -1
+        ).copy_(y_extend.permute(0, 2, 1))
 
 
 @_triton_cached_causal_conv1d.register_fake
diff --git a/tensorrt_llm/_torch/auto_deploy/shim/interface.py b/tensorrt_llm/_torch/auto_deploy/shim/interface.py
index e6714f513184..53662854af51 100644
--- a/tensorrt_llm/_torch/auto_deploy/shim/interface.py
+++ b/tensorrt_llm/_torch/auto_deploy/shim/interface.py
@@ -51,6 +51,12 @@
 from ..custom_ops.attention_interface import (
     CausalConvResourceHandler,
     KVPagedResourceHandler,
+    ReplayCacheBufIdxHandler,
+    ReplayOldBHandler,
+    ReplayOldDAcumsumHandler,
+    ReplayOldDtHandler,
+    ReplayOldXHandler,
+    ReplayPrevNumAcceptedHandler,
     ResourceHandler,
     ResourceHandlerDict,
     SequenceInfo,
@@ -151,6 +157,8 @@ def __init__(
         self.info.batch_info.update_max_draft_len(
             spec_config.max_draft_len if spec_config is not None else 0
         )
+        # Default to non-replay; updated in initialize_cache once resources are identified.
+        self.info.batch_info.update_use_replay(False)
 
     @property
     def args(self) -> Tuple[torch.Tensor, ...]:
@@ -461,19 +469,89 @@ def _identify_managed_state_resources(
             and handler == SpecCausalConvResourceHandler.from_base(conv_ref)
         ]
 
+        # Replay SSM buffers — per-layer (old_x, old_B, old_dt, old_dA_cumsum)
+        # and global (cache_buf_idx, prev_num_accepted_tokens).
+        replay_old_x = [
+            (name, handler)
+            for name, handler in self._resource_lookup.items()
+            if isinstance(handler, ReplayOldXHandler)
+        ]
+        replay_old_B = [
+            (name, handler)
+            for name, handler in self._resource_lookup.items()
+            if isinstance(handler, ReplayOldBHandler)
+        ]
+        replay_old_dt = [
+            (name, handler)
+            for name, handler in self._resource_lookup.items()
+            if isinstance(handler, ReplayOldDtHandler)
+        ]
+        replay_old_dA_cumsum = [
+            (name, handler)
+            for name, handler in self._resource_lookup.items()
+            if isinstance(handler, ReplayOldDAcumsumHandler)
+        ]
+        replay_cache_buf_idx = [
+            (name, handler)
+            for name, handler in self._resource_lookup.items()
+            if isinstance(handler, ReplayCacheBufIdxHandler)
+        ]
+        replay_prev_num_accepted = [
+            (name, handler)
+            for name, handler in self._resource_lookup.items()
+            if isinstance(handler, ReplayPrevNumAcceptedHandler)
+        ]
+
         # When speculative decoding is enabled, the backend must supply matching spec buffers.
         # When it is not enabled, spec buffers may still be registered by the backend (e.g.
         # triton_ssm always registers intermediate_ssm_state_cache) but will not be bound.
-        if self._spec_config is not None:
-            assert len(ssm_spec) == len(ssm_managed), (
-                f"Mismatched SSM spec layer count: expected {len(ssm_managed)}, got {len(ssm_spec)}"
+        # Exception: in replay mode intermediate_ssm_state_cache is omitted entirely from
+        # get_cache_initializers and uses the function's default (None) via kwarg routing.
+        use_replay = len(replay_old_x) > 0
+        if use_replay:
+            n = len(ssm_managed)
+            for lst, name in [
+                (replay_old_x, "replay_old_x"),
+                (replay_old_B, "replay_old_B"),
+                (replay_old_dt, "replay_old_dt"),
+                (replay_old_dA_cumsum, "replay_old_dA_cumsum"),
+            ]:
+                assert len(lst) == n, (
+                    f"Replay bundle mismatch: {name} has {len(lst)} entries, "
+                    f"expected {n} (== len(ssm_managed))"
+                )
+            assert len(replay_cache_buf_idx) == n, (
+                f"Replay bundle mismatch: replay_cache_buf_idx has {len(replay_cache_buf_idx)} "
+                f"entries, expected {n} (== len(ssm_managed))"
             )
+            assert len(replay_prev_num_accepted) == n, (
+                f"Replay bundle mismatch: replay_prev_num_accepted has "
+                f"{len(replay_prev_num_accepted)} entries, expected {n} (== len(ssm_managed))"
+            )
+        if self._spec_config is not None:
+            if not use_replay:
+                assert len(ssm_spec) == len(ssm_managed), (
+                    f"Mismatched SSM spec layer count: expected {len(ssm_managed)}, got {len(ssm_spec)}"
+                )
             assert len(conv_spec) == len(conv_managed), (
                 f"Mismatched Conv spec layer count: expected {len(conv_managed)}, "
                 f"got {len(conv_spec)}"
             )
 
-        return ssm_ref, ssm_managed, ssm_spec, conv_ref, conv_managed, conv_spec
+        return (
+            ssm_ref,
+            ssm_managed,
+            ssm_spec,
+            conv_ref,
+            conv_managed,
+            conv_spec,
+            replay_old_x,
+            replay_old_B,
+            replay_old_dt,
+            replay_old_dA_cumsum,
+            replay_cache_buf_idx,
+            replay_prev_num_accepted,
+        )
 
     def _prepare_kv_cache_config(
         self,
@@ -653,6 +731,12 @@ def _create_and_assign_state_views(
         conv_ref: Optional[CausalConvResourceHandler],
         conv_managed: list,
         conv_spec: list,
+        replay_old_x: list = (),
+        replay_old_B: list = (),
+        replay_old_dt: list = (),
+        replay_old_dA_cumsum: list = (),
+        replay_cache_buf_idx: list = (),
+        replay_prev_num_accepted: list = (),
     ) -> Tuple[MambaHybridCacheManager, int]:
         """Create MambaHybridCacheManager and assign views for state resources.
 
@@ -668,20 +752,26 @@ def _create_and_assign_state_views(
             conv_ref: Reference Conv handler or None.
             conv_managed: List of base Conv resources.
             conv_spec: List of speculative Conv resources.
+            replay_old_x/B/dt/dA_cumsum: Per-layer replay cache resource lists.
+            replay_cache_buf_idx/prev_num_accepted: Global replay resource lists.
 
         Returns:
             Tuple of (manager, num_managed_mamba_layers).
         """
+        # Detect replay mode from presence of ReplayOldXHandler resources.
+        use_replay = len(replay_old_x) > 0
+
         # Mamba state params can be derived from reference handlers and number of managed (non-speculative) resources.
         mamba_params = self._get_mamba_state_params(
             ssm_ref, len(ssm_managed), conv_ref, len(conv_managed)
         )
         num_managed_mamba_layers = mamba_params["mamba_num_layers"]
 
-        # Create the hybrid cache manager
+        # Create the hybrid cache manager, enabling replay path if detected.
         manager = MixedMambaHybridCacheManager(
             **mamba_params,
             **kv_cache_kwargs,
+            use_replay_state_update=use_replay,
         )
 
         # Retrieve and assign views for Mamba-managed resources (up to num_managed_mamba_layers).
@@ -717,6 +807,36 @@ def _create_and_assign_state_views(
                 assert spec_view.is_contiguous(), f"Non-contiguous state {spec_conv_name}"
                 self._caches[spec_conv_name] = spec_view
 
+            # Replay per-layer buffers (only when replay mode is active and spec dec enabled).
+            if use_replay and self._spec_config is not None:
+                for buf_list, getter, label in [
+                    (replay_old_x, manager.get_replay_old_x, "old_x"),
+                    (replay_old_B, manager.get_replay_old_B, "old_B"),
+                    (replay_old_dt, manager.get_replay_old_dt, "old_dt"),
+                    (replay_old_dA_cumsum, manager.get_replay_old_dA_cumsum, "old_dA_cumsum"),
+                ]:
+                    if buf_list:
+                        buf_name = buf_list[layer_idx][0]
+                        buf_view = getter(layer_idx)
+                        if buf_view is None:
+                            raise RuntimeError(
+                                f"Replay buffer {label} not allocated for layer {layer_idx}. "
+                                "Is use_replay_state_update=True in the cache manager?"
+                            )
+                        self._caches[buf_name] = buf_view
+
+        # Replay global buffers — same tensor for every layer; bind once per resource entry.
+        if use_replay and self._spec_config is not None:
+            global_buf_map = [
+                (replay_cache_buf_idx, manager.get_replay_cache_buf_idx()),
+                (replay_prev_num_accepted, manager.get_replay_prev_num_accepted_tokens()),
+            ]
+            for buf_list, global_tensor in global_buf_map:
+                if global_tensor is None:
+                    continue
+                for buf_name, _ in buf_list:
+                    self._caches[buf_name] = global_tensor
+
         return manager, num_managed_mamba_layers
 
     def _assign_kv_cache_views(self, kv_managed: Dict[str, KVPagedResourceHandler]) -> int:
@@ -879,10 +999,24 @@ def _create_kv_cache_manager(self, max_tokens: Optional[int] = None) -> Dict:
         """
         # 1. Identify managed resources and per-pool configurations
         kv_managed, pool_configurations = self._identify_managed_kv_resources()
-
-        ssm_ref, ssm_managed, ssm_spec, conv_ref, conv_managed, conv_spec = (
-            self._identify_managed_state_resources()
-        )
+        (
+            ssm_ref,
+            ssm_managed,
+            ssm_spec,
+            conv_ref,
+            conv_managed,
+            conv_spec,
+            replay_old_x,
+            replay_old_B,
+            replay_old_dt,
+            replay_old_dA_cumsum,
+            replay_cache_buf_idx,
+            replay_prev_num_accepted,
+        ) = self._identify_managed_state_resources()
+
+        # Propagate replay mode into BatchInfo so SSM backends can branch on it
+        # without inspecting the presence/absence of Optional cache tensors.
+        self.info.batch_info.update_use_replay(len(replay_old_x) > 0)
         has_state_resources = ssm_managed or conv_managed
 
         # 2. Compute the total token budget; the C++ side splits it across pools.
@@ -913,6 +1047,12 @@ def _create_kv_cache_manager(self, max_tokens: Optional[int] = None) -> Dict:
                 conv_ref,
                 conv_managed,
                 conv_spec,
+                replay_old_x=replay_old_x,
+                replay_old_B=replay_old_B,
+                replay_old_dt=replay_old_dt,
+                replay_old_dA_cumsum=replay_old_dA_cumsum,
+                replay_cache_buf_idx=replay_cache_buf_idx,
+                replay_prev_num_accepted=replay_prev_num_accepted,
             )
         else:
             self._kv_cache_manager = KVCacheManager(**kv_cache_kwargs)
diff --git a/tensorrt_llm/_torch/auto_deploy/transform/library/fuse_causal_conv.py b/tensorrt_llm/_torch/auto_deploy/transform/library/fuse_causal_conv.py
index bce8476e6ec4..e1998e901dd1 100644
--- a/tensorrt_llm/_torch/auto_deploy/transform/library/fuse_causal_conv.py
+++ b/tensorrt_llm/_torch/auto_deploy/transform/library/fuse_causal_conv.py
@@ -22,6 +22,7 @@
 from torch.fx import GraphModule, Node
 
 from ...custom_ops.mamba.cuda_backend_causal_conv import cuda_cached_causal_conv1d_wrapper
+from ...custom_ops.mamba.triton_backend_causal_conv import triton_cached_causal_conv1d_wrapper
 from ...models.factory import ModelFactory
 from ...shim.interface import CachedSequenceInterface
 from ...utils.node_utils import is_op
@@ -36,20 +37,21 @@ def _match_causal_conv_activation_pattern(
     Match the causal_conv + activation pattern in the graph.
 
     The pattern corresponds to:
-        conv_out = cuda_cached_causal_conv1d(...)
+        conv_out = cuda_cached_causal_conv1d(...)  OR triton_cached_causal_conv1d(...)
         out = activation(conv_out)
 
     Args:
         graph: The graph module to search
-        target_op: The target causal conv op to match
+        target_op: The target causal conv op to match (or a tuple/list of ops)
 
     Returns:
         A list of tuples (conv_node, activation_node, activation_name) for each match
     """
+    target_ops = target_op if isinstance(target_op, (list, tuple)) else [target_op]
     matches = []
 
     for node in graph.nodes:
-        if not is_op(node, target_op):
+        if not any(is_op(node, op) for op in target_ops):
             continue
 
         # Check if this node has exactly one user and it's an activation
@@ -101,22 +103,29 @@ def _apply(
     ) -> Tuple[GraphModule, TransformInfo]:
         graph = gm.graph
 
-        target_op = cuda_cached_causal_conv1d_wrapper
+        # Match both the CUDA-backed and Triton-backed causal conv wrappers.
+        # fuse_causal_conv_activation previously only matched the CUDA wrapper,
+        # leaving the Triton wrapper's activation (e.g. silu(conv_out)) as a
+        # separate FX node that landed between the conv and SSM kernels in the
+        # CUDA graph stream, blocking the PDL chain.
+        target_ops = [cuda_cached_causal_conv1d_wrapper, triton_cached_causal_conv1d_wrapper]
 
         # Step 1: Identify causal_conv + activation pattern
         matches = _match_causal_conv_activation_pattern(
             graph,
-            target_op=target_op,
+            target_op=target_ops,
         )
 
         # Step 2: Replace matched patterns with fused version
         for conv_node, activation_node, activation_name in matches:
+            # Determine which wrapper was matched so the replacement uses the same op
+            conv_target_op = conv_node.target
             with graph.inserting_after(conv_node):
                 # Create new call with fused activation
                 # Replace the last arg (activation=None) with activation_name
                 new_args = list(conv_node.args[:-1]) + [activation_name]
                 fused_node = graph.call_function(
-                    target_op,
+                    conv_target_op,
                     args=tuple(new_args),
                 )
 
diff --git a/tensorrt_llm/_torch/auto_deploy/transform/library/kvcache.py b/tensorrt_llm/_torch/auto_deploy/transform/library/kvcache.py
index f131d2d6292e..bb12d726b0d7 100644
--- a/tensorrt_llm/_torch/auto_deploy/transform/library/kvcache.py
+++ b/tensorrt_llm/_torch/auto_deploy/transform/library/kvcache.py
@@ -316,7 +316,7 @@ def _insert_cached_attn_node(
     ):
         """Insert a cached attention node into the graph."""
         with gm.graph.inserting_before(attn_node):
-            all_args = (
+            args = (
                 *qkv_nodes,
                 *meta_nodes_std,
                 *meta_nodes_extra,
@@ -324,11 +324,8 @@ def _insert_cached_attn_node(
                 *constants,
             )
             if prepared_attn_mask is not None:
-                all_args = (*all_args, prepared_attn_mask)
-            cached_attn_node = gm.graph.call_function(
-                cached_attn_op,
-                args=all_args,
-            )
+                args = (*args, prepared_attn_mask)
+            cached_attn_node = gm.graph.call_function(cached_attn_op, args=args)
         attn_node.replace_all_uses_with(cached_attn_node)
         gm.graph.erase_node(attn_node)
 
@@ -431,17 +428,22 @@ def _apply(
                 for k, resource_handler in attn_descriptor.get_cache_initializers(
                     attn_node, cm.kv_cache_config
                 ).items():
-                    resource_name = cm.add_resource(k, resource_handler)
-                    cache_in_nodes.append(self._process_cache_node(gm, resource_name))
-                    # Determine group from handler equality
-                    if isinstance(resource_handler, KVPagedResourceHandler):
-                        for gi, ref in enumerate(handler_groups):
-                            if resource_handler == ref:
-                                group_idx = gi
-                                break
-                        else:
-                            group_idx = len(handler_groups)
-                            handler_groups.append(resource_handler)
+                    if resource_handler is None:
+                        # None sentinel: pass literal None positionally, no resource allocated.
+                        cache_in_nodes.append(None)
+                    else:
+                        resource_name = cm.add_resource(k, resource_handler)
+                        node = self._process_cache_node(gm, resource_name)
+                        cache_in_nodes.append(node)
+                        # Determine group from handler equality
+                        if isinstance(resource_handler, KVPagedResourceHandler):
+                            for gi, ref in enumerate(handler_groups):
+                                if resource_handler == ref:
+                                    group_idx = gi
+                                    break
+                            else:
+                                group_idx = len(handler_groups)
+                                handler_groups.append(resource_handler)
                 if layer_idx is not None:
                     cache_nodes_by_layer_idx[layer_idx] = cache_in_nodes
                     group_idx_by_layer_idx[layer_idx] = group_idx
diff --git a/tensorrt_llm/_torch/auto_deploy/transform/library/ssm_cache.py b/tensorrt_llm/_torch/auto_deploy/transform/library/ssm_cache.py
index 36f2b8cb7e76..7a6b2be901b6 100644
--- a/tensorrt_llm/_torch/auto_deploy/transform/library/ssm_cache.py
+++ b/tensorrt_llm/_torch/auto_deploy/transform/library/ssm_cache.py
@@ -14,8 +14,28 @@
 # limitations under the License.
 """A set of transforms to handle SSM cache transforms."""
 
-from ..interface import TransformRegistry
-from .kvcache import _InsertCachedOperator
+from typing import Type
+
+from pydantic import Field
+
+from ..interface import TransformConfig, TransformRegistry
+from .kvcache import InsertCachedAttentionConfig, _InsertCachedOperator
+
+
+class SSMCacheTransformConfig(InsertCachedAttentionConfig):
+    """Configuration for insert_cached_ssm_attention.
+
+    Extends the base attention config with SSM-specific options.
+    """
+
+    ssm_replay: bool = Field(
+        default=False,
+        description=(
+            "Enable the replay SSM kernel (tl.dot fast-forward) for the MTP extend path. "
+            "Requires SM >= 80 (Ampere+). Falls back to FlashInfer when disabled or "
+            "when incompatible features are active (block reuse, tree attention)."
+        ),
+    )
 
 
 # TODO: think about separating valid attention backends per transform better in the future
@@ -23,6 +43,21 @@
 class SSMCacheTransform(_InsertCachedOperator):
     """A transform to handle SSM cache operations."""
 
+    config: SSMCacheTransformConfig
+
+    @classmethod
+    def get_config_class(cls) -> Type[TransformConfig]:
+        return SSMCacheTransformConfig
+
+    @property
+    def attn_descriptor(self):
+        descriptor = super().attn_descriptor
+        if self.config.ssm_replay and hasattr(descriptor, "ssm_replay"):
+            # Return a thin subclass with ssm_replay enabled so get_cache_initializers
+            # allocates replay buffers instead of intermediate_ssm_state_cache.
+            return type(descriptor.__name__ + "_Replay", (descriptor,), {"ssm_replay": True})
+        return descriptor
+
 
 @TransformRegistry.register("insert_cached_causal_conv")
 class InitializeCausalConvCache(_InsertCachedOperator):
diff --git a/tensorrt_llm/_torch/modules/mamba/ssd_combined.py b/tensorrt_llm/_torch/modules/mamba/ssd_combined.py
index c39e7328102c..d2c90a25007f 100644
--- a/tensorrt_llm/_torch/modules/mamba/ssd_combined.py
+++ b/tensorrt_llm/_torch/modules/mamba/ssd_combined.py
@@ -71,9 +71,17 @@ def _use_flashinfer_ssd():
 
 
 def _flashinfer_ssd_supported(chunk_size, dstate, headdim):
+    # The kernel was written for Nemotron-H (chunk_size=128, dstate=128) and has a
+    # bug in tma_partition_for_mma_b_operand: it tiles the B SSM operand with
+    # tile_shape_mnk_intra2[1:] = (headdim, chunk_size) instead of the correct
+    # tile_shape_mnk_intra1[1:] = (chunk_size, dstate), making gmem K-mode =
+    # chunk_size/16 mismatch smem K-mode = dstate/16. Only compiles when
+    # dstate == chunk_size (masked the bug for NemotronH). See
+    # https://github.com/flashinfer-ai/flashinfer/issues/3397 for upstream fix.
     return (chunk_size in _FLASHINFER_SSD_VALID_M_MODES
             and dstate in _FLASHINFER_SSD_VALID_M_MODES
-            and headdim in _FLASHINFER_SSD_VALID_HEAD_DIMS)
+            and headdim in _FLASHINFER_SSD_VALID_HEAD_DIMS
+            and dstate == chunk_size)
 
 
 @functools.cache
diff --git a/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py b/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py
index b4eebeb1a25c..5ab3a39a7a73 100644
--- a/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py
+++ b/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py
@@ -724,6 +724,41 @@ def get_intermediate_conv_states(self,
         return self.mamba_cache.at_layer_idx(
             layer_offset).intermediate_conv_window
 
+    def get_replay_old_x(self, layer_idx: int) -> Optional[torch.Tensor]:
+        if not self._use_replay_state_update:
+            return None
+        layer_offset = self.mamba_layer_offsets[layer_idx]
+        return self.mamba_cache.old_x[layer_offset]
+
+    def get_replay_old_B(self, layer_idx: int) -> Optional[torch.Tensor]:
+        if not self._use_replay_state_update:
+            return None
+        layer_offset = self.mamba_layer_offsets[layer_idx]
+        return self.mamba_cache.old_B[layer_offset]
+
+    def get_replay_old_dt(self, layer_idx: int) -> Optional[torch.Tensor]:
+        if not self._use_replay_state_update:
+            return None
+        layer_offset = self.mamba_layer_offsets[layer_idx]
+        return self.mamba_cache.old_dt[layer_offset]
+
+    def get_replay_old_dA_cumsum(self,
+                                 layer_idx: int) -> Optional[torch.Tensor]:
+        if not self._use_replay_state_update:
+            return None
+        layer_offset = self.mamba_layer_offsets[layer_idx]
+        return self.mamba_cache.old_dA_cumsum[layer_offset]
+
+    def get_replay_cache_buf_idx(self) -> Optional[torch.Tensor]:
+        if not self._use_replay_state_update:
+            return None
+        return self.mamba_cache.cache_buf_idx
+
+    def get_replay_prev_num_accepted_tokens(self) -> Optional[torch.Tensor]:
+        if not self._use_replay_state_update:
+            return None
+        return self.mamba_cache.prev_num_accepted_tokens
+
     def is_speculative(self) -> bool:
         return isinstance(self.mamba_cache, self.SpeculativeState)
 
@@ -948,6 +983,31 @@ def get_intermediate_conv_states(self,
         assert not self._use_cpp, "get_intermediate_conv_states is not supported in CppMambaCacheManager"
         return self._impl.get_intermediate_conv_states(layer_idx)
 
+    def get_replay_old_x(self, layer_idx: int) -> Optional[torch.Tensor]:
+        assert not self._use_cpp, "get_replay_old_x is not supported in CppMambaCacheManager"
+        return self._impl.get_replay_old_x(layer_idx)
+
+    def get_replay_old_B(self, layer_idx: int) -> Optional[torch.Tensor]:
+        assert not self._use_cpp, "get_replay_old_B is not supported in CppMambaCacheManager"
+        return self._impl.get_replay_old_B(layer_idx)
+
+    def get_replay_old_dt(self, layer_idx: int) -> Optional[torch.Tensor]:
+        assert not self._use_cpp, "get_replay_old_dt is not supported in CppMambaCacheManager"
+        return self._impl.get_replay_old_dt(layer_idx)
+
+    def get_replay_old_dA_cumsum(self,
+                                 layer_idx: int) -> Optional[torch.Tensor]:
+        assert not self._use_cpp, "get_replay_old_dA_cumsum is not supported in CppMambaCacheManager"
+        return self._impl.get_replay_old_dA_cumsum(layer_idx)
+
+    def get_replay_cache_buf_idx(self) -> Optional[torch.Tensor]:
+        assert not self._use_cpp, "get_replay_cache_buf_idx is not supported in CppMambaCacheManager"
+        return self._impl.get_replay_cache_buf_idx()
+
+    def get_replay_prev_num_accepted_tokens(self) -> Optional[torch.Tensor]:
+        assert not self._use_cpp, "get_replay_prev_num_accepted_tokens is not supported in CppMambaCacheManager"
+        return self._impl.get_replay_prev_num_accepted_tokens()
+
     def is_speculative(self) -> bool:
         return self._impl.is_speculative()
 
diff --git a/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py b/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py
index d031109ef6e4..b717fc38ca1e 100644
--- a/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py
+++ b/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py
@@ -303,7 +303,7 @@ def route_requests(
 
         def get_relax_value(req_item):
             scheduling_params = getattr(req_item.request, "py_scheduling_params", None)
-            if scheduling_params is None:
+            if scheduling_params is None or scheduling_params.attention_dp_rank is None:
                 return True
             val = scheduling_params.attention_dp_relax
             return True if val is None else val
@@ -587,7 +587,7 @@ def route_requests(
 
         def get_relax_value(req_item):
             scheduling_params = getattr(req_item.request, "py_scheduling_params", None)
-            if scheduling_params is None:
+            if scheduling_params is None or scheduling_params.attention_dp_rank is None:
                 return True
             val = scheduling_params.attention_dp_relax
             return True if val is None else val
diff --git a/tests/integration/defs/accuracy/test_llm_api_autodeploy.py b/tests/integration/defs/accuracy/test_llm_api_autodeploy.py
index 5c3d26f2e79b..a76f3edf5a87 100644
--- a/tests/integration/defs/accuracy/test_llm_api_autodeploy.py
+++ b/tests/integration/defs/accuracy/test_llm_api_autodeploy.py
@@ -148,19 +148,18 @@ def low_memory_overrides(config,
                          max_batch_size=32,
                          free_gpu_memory_fraction=0.4,
                          max_seq_len=8192,
-                         max_num_tokens=8192,
-                         cuda_graph_batch_sizes=None):
+                         max_num_tokens=8192):
     """Update and return config that reduce memory footprint for unquantized (bf16) runs."""
-    if cuda_graph_batch_sizes is None:
-        cuda_graph_batch_sizes = [
-            s for s in [1, 2, 4, 8, 16, 32, 64, 128] if s <= max_batch_size
-        ]
+    cuda_graph_batch_sizes = [
+        s for s in [1, 2, 4, 8, 16, 32, 64, 128] if s <= max_batch_size
+    ]
     config.update({
         "max_batch_size": max_batch_size,
         "max_seq_len": max_seq_len,
         "max_num_tokens": max_num_tokens,
         "cuda_graph_config": {
-            "batch_sizes": cuda_graph_batch_sizes
+            "batch_sizes": cuda_graph_batch_sizes,
+            "max_batch_size": max_batch_size,
         },
     })
     kv_cache_config = config.setdefault("kv_cache_config", {})
@@ -769,7 +768,6 @@ def test_mtp(self, world_size, attn_backend, model_id):
         low_memory_overrides(
             kwargs,
             max_batch_size=8,
-            cuda_graph_batch_sizes=[1, 2, 4, 8],
         )
         kwargs["attn_backend"] = attn_backend
 
diff --git a/tests/integration/test_lists/test-db/l0_b200.yml b/tests/integration/test_lists/test-db/l0_b200.yml
index e82c4ee7b109..9afa4974a946 100644
--- a/tests/integration/test_lists/test-db/l0_b200.yml
+++ b/tests/integration/test_lists/test-db/l0_b200.yml
@@ -331,10 +331,12 @@ l0_b200:
   # custom_ops, smoke, transformations files) and pure-FP8 tests are covered
   # on Hopper (l0_h100.yml) and not duplicated here.
   - unittest/auto_deploy/singlegpu/custom_ops/attention/test_triton_attention.py::TestSDPADispatch
+  - unittest/auto_deploy/singlegpu/custom_ops/mamba/test_flashinfer_mamba_cached_op.py
   - unittest/auto_deploy/singlegpu/custom_ops/moe/test_ad_moe_op.py
   - unittest/auto_deploy/singlegpu/custom_ops/moe/test_trtllm_moe.py
   - unittest/auto_deploy/singlegpu/custom_ops/quantization/test_quant.py
   - unittest/auto_deploy/singlegpu/smoke/test_ad_build_small_single.py -k "Nemotron-3-Nano-30B-A3B-FP8 or Nemotron-Nano-3-30B-A3.5B-dev or Llama-4-Scout"
+  - unittest/auto_deploy/singlegpu/smoke/test_ad_speculative_decoding.py
   - unittest/auto_deploy/singlegpu/transformations/library/test_fuse_relu2_quant_nvfp4.py
   - unittest/auto_deploy/singlegpu/transformations/library/test_moe_fusion.py
   - unittest/auto_deploy/singlegpu/transformations/library/test_nvfp4_swiglu.py
diff --git a/tests/unittest/auto_deploy/singlegpu/custom_ops/mamba/test_flashinfer_mamba_cached_op.py b/tests/unittest/auto_deploy/singlegpu/custom_ops/mamba/test_flashinfer_mamba_cached_op.py
index 82614a5e1395..b8db6132dc0b 100644
--- a/tests/unittest/auto_deploy/singlegpu/custom_ops/mamba/test_flashinfer_mamba_cached_op.py
+++ b/tests/unittest/auto_deploy/singlegpu/custom_ops/mamba/test_flashinfer_mamba_cached_op.py
@@ -12,12 +12,17 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+from unittest.mock import patch
+
 import pytest
 import torch
 from test_triton_mamba_cached_op import _random_params
 
 import tensorrt_llm._torch.auto_deploy  # noqa: F401
 from tensorrt_llm._torch.auto_deploy.custom_ops.attention_interface import BatchInfo
+from tensorrt_llm._torch.modules.mamba.replay_selective_state_update import (
+    replay_selective_state_update as _real_replay_selective_state_update,
+)
 
 
 @pytest.fixture
@@ -104,7 +109,13 @@ def test_flashinfer_decode_matches_triton(mamba_env):
         None,  # seq_idx_prefill
         # CACHES
         ssm_state_cache_flashinfer,
-        None,  # intermediate_ssm_state_cache (not used in decode-only path)
+        None,  # intermediate_ssm_state_cache
+        None,  # replay_old_x
+        None,  # replay_old_b
+        None,  # replay_old_dt
+        None,  # replay_old_da_cumsum
+        None,  # replay_cache_buf_idx
+        None,  # replay_prev_num_accepted
         # CONSTANTS
         time_step_limit,
         chunk_size,
@@ -120,3 +131,111 @@ def test_flashinfer_decode_matches_triton(mamba_env):
     assert torch.allclose(
         after_flashinfer.to(after_triton.dtype), after_triton, atol=atol, rtol=rtol
     )
+
+
+@pytest.mark.parametrize(
+    "head_dim",
+    [
+        pytest.param(32, id="unsupported_flashinfer_headdim"),  # not in {64, 128}
+        pytest.param(64, id="supported_flashinfer_headdim"),  # in {64, 128}
+    ],
+)
+def test_flashinfer_extend_replay_calls_replay_kernel(mamba_env, head_dim):
+    """Verify replay_selective_state_update is invoked on the extend+replay path.
+
+    Parametrized over head_dim to cover both a dim that FlashInfer does not support
+    (proving replay is independent of the FlashInfer head-dim constraint) and one
+    that it does support (ensuring the path works for production-like configs too).
+    """
+    device = mamba_env["device"]
+    dtype = mamba_env["dtype"]
+
+    num_extend = 1
+    tokens_per_extend = 2  # draft tokens per MTP verification step
+    num_heads = 4
+    n_groups, ssm_state_size = 2, 16  # ssm_state_size >= 16 for replay tl.dot
+    max_batch_size = 2  # cache slots
+
+    (hidden_states, A, B, C, D, dt, dt_bias, time_step_limit, chunk_size) = _random_params(
+        device, dtype, num_extend, tokens_per_extend, num_heads, head_dim, n_groups, ssm_state_size
+    )
+
+    ssm_state_cache = torch.zeros(
+        max_batch_size, num_heads, head_dim, ssm_state_size, device=device, dtype=dtype
+    )
+    slot_idx = torch.tensor([0], device=device, dtype=torch.int32)
+
+    # Replay buffers: all zeros (first step, nothing cached yet; kernel still runs).
+    replay_old_x = torch.zeros(
+        max_batch_size, tokens_per_extend, num_heads, head_dim, device=device, dtype=torch.bfloat16
+    )
+    replay_old_b = torch.zeros(
+        max_batch_size,
+        2,
+        tokens_per_extend,
+        n_groups,
+        ssm_state_size,
+        device=device,
+        dtype=torch.bfloat16,
+    )
+    replay_old_dt = torch.zeros(
+        max_batch_size, 2, num_heads, tokens_per_extend, device=device, dtype=torch.float32
+    )
+    replay_old_da_cumsum = torch.zeros(
+        max_batch_size, 2, num_heads, tokens_per_extend, device=device, dtype=torch.float32
+    )
+    replay_cache_buf_idx = torch.zeros(max_batch_size, device=device, dtype=torch.int32)
+    replay_prev_num_accepted = torch.zeros(max_batch_size, device=device, dtype=torch.int32)
+
+    # Extend-only batch with replay mode enabled.
+    _bi = BatchInfo()
+    _bi.update([0, 0, num_extend, num_extend * tokens_per_extend, 0, 0])
+    _bi.update_use_replay(True)
+    batch_info_host = _bi.serialize()
+
+    cu_seqlen = torch.tensor([0, tokens_per_extend], device=device, dtype=torch.int32)
+    use_initial_states = torch.zeros(num_extend, device=device, dtype=torch.bool)
+    any_prefill_use_initial_states_host = torch.tensor([False], device=device, dtype=torch.bool)
+
+    _TARGET = (
+        "tensorrt_llm._torch.auto_deploy.custom_ops.mamba"
+        ".flashinfer_backend_mamba.replay_selective_state_update"
+    )
+    with patch(_TARGET, wraps=_real_replay_selective_state_update) as mock_replay:
+        out = torch.ops.auto_deploy.flashinfer_cached_ssm(
+            hidden_states,
+            A,
+            B,
+            C,
+            D,
+            dt,
+            dt_bias,
+            # STANDARD METADATA
+            batch_info_host,
+            cu_seqlen,
+            slot_idx,
+            use_initial_states,
+            any_prefill_use_initial_states_host,
+            # EXTRA METADATA
+            None,
+            None,
+            None,  # chunk_indices, chunk_offsets, seq_idx_prefill
+            # CACHES
+            ssm_state_cache,
+            None,  # intermediate_ssm_state_cache (None in replay mode)
+            replay_old_x,
+            replay_old_b,
+            replay_old_dt,
+            replay_old_da_cumsum,
+            replay_cache_buf_idx,
+            replay_prev_num_accepted,
+            # CONSTANTS
+            time_step_limit,
+            chunk_size,
+        )
+
+    assert mock_replay.call_count == num_extend, (
+        f"Expected {num_extend} replay call(s), got {mock_replay.call_count}"
+    )
+    assert out.shape == hidden_states.shape
+    assert torch.isfinite(out).all()
diff --git a/tests/unittest/auto_deploy/singlegpu/smoke/test_ad_speculative_decoding.py b/tests/unittest/auto_deploy/singlegpu/smoke/test_ad_speculative_decoding.py
index 37f838a50294..a589f8261499 100644
--- a/tests/unittest/auto_deploy/singlegpu/smoke/test_ad_speculative_decoding.py
+++ b/tests/unittest/auto_deploy/singlegpu/smoke/test_ad_speculative_decoding.py
@@ -92,6 +92,73 @@ def test_super_mtp_smoke():
     assert len(prompts_and_outputs) == 1
 
 
+def test_super_mtp_ssm_replay_smoke():
+    """Smoke test: MTP Eagle one-model with flashinfer_ssm + ssm_replay=True compiles and runs.
+
+    Verifies that the full pipeline — transforms, cache manager init with replay buffers,
+    and MTP inference — completes without error. The AD SSM custom ops are not directly
+    invoked at runtime in this configuration (Eagle3OneModelSampler drives its own forward
+    loop); the replay kernel path is covered by test_flashinfer_extend_replay_calls_replay_kernel.
+    Uses mamba_head_dim=64 and ssm_state_size=64 to satisfy FlashInfer constraints on the
+    decode path (which IS called in this config).
+    """
+    test_prompt = "What is the capital of France?"
+    model_hub_id = "nvidia/NVIDIA-Nemotron-3-Super-120B-A12B-BF16"
+    model_path = hf_id_to_local_model_dir(model_hub_id)
+
+    # Get base small-model config and update SSM dims for FlashInfer + replay.
+    # hidden_size must equal mamba_num_heads × mamba_head_dim (4 × 64 = 256).
+    experiment_config = get_small_model_config(
+        model_hub_id,
+        transforms={
+            "insert_cached_causal_conv": {"backend": "triton_causal_conv"},
+            "insert_cached_ssm_attention": {"backend": "flashinfer_ssm", "ssm_replay": True},
+        },
+    )
+    experiment_config["args"]["model_kwargs"].update(
+        {
+            "hidden_size": 256,
+            "intermediate_size": 256,
+            "mamba_num_heads": 4,
+            "mamba_head_dim": 64,
+            "ssm_state_size": 64,
+            "moe_intermediate_size": 128,
+            "moe_shared_expert_intermediate_size": 128,
+            "moe_latent_size": 64,
+        }
+    )
+    experiment_config["args"]["model"] = model_path
+    experiment_config["args"]["runtime"] = "trtllm"
+    experiment_config["args"]["world_size"] = 1
+    experiment_config["args"]["speculative_config"] = MTPDecodingConfig(
+        num_nextn_predict_layers=3,
+        mtp_eagle_one_model=True,
+        speculative_model=model_path,
+    )
+    experiment_config["args"]["speculative_model_kwargs"] = experiment_config["args"][
+        "model_kwargs"
+    ]
+    experiment_config["args"]["attn_backend"] = "flashinfer"
+    experiment_config["args"]["disable_overlap_scheduler"] = True
+    experiment_config["args"]["compile_backend"] = "torch-simple"
+    experiment_config["args"]["max_num_tokens"] = 256
+    experiment_config["prompt"]["batch_size"] = 1
+    experiment_config["prompt"]["queries"] = test_prompt
+
+    cfg = ExperimentConfig(**experiment_config)
+    cfg.prompt.sp_kwargs = {
+        "max_tokens": 20,
+        "top_k": None,
+        "temperature": 0.0,
+        "seed": 42,
+    }
+
+    results = main(cfg)
+
+    prompts_and_outputs = results["prompts_and_outputs"]
+    assert len(prompts_and_outputs) == 1
+
+
 def test_kv_cache_extra_seq_len_for_spec_dec():
     """Test that get_extra_seq_len_for_kv_cache computes correct extra capacity."""
     from tensorrt_llm._torch.auto_deploy.llm_args import LlmArgs