diff --git a/cpp/tensorrt_llm/kernels/IndexerKCacheGather.h b/cpp/tensorrt_llm/kernels/IndexerKCacheGather.h
index 60422dea5910..bb3cb928485c 100644
--- a/cpp/tensorrt_llm/kernels/IndexerKCacheGather.h
+++ b/cpp/tensorrt_llm/kernels/IndexerKCacheGather.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2025, NVIDIA CORPORATION.  All rights reserved.
+ * Copyright (c) 2022-2026, NVIDIA CORPORATION.  All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,12 +24,26 @@ TRTLLM_NAMESPACE_BEGIN
 namespace kernels
 {
 
-void invokeIndexerKCacheGather(uint8_t const* k_cache, int64_t const* slot_mapping_fp8,
-    int64_t const* slot_mapping_scale, uint8_t* out_fp8, uint8_t* out_scale, int32_t k_token_start, int32_t num_tokens,
-    int32_t head_dim, int32_t scale_size, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2,
-    int32_t cache_dim_3, int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2, int64_t cache_stride_3,
+/// Gather FP8 K values and scales from a non-contiguous paged indexer K cache
+/// into contiguous output buffers. This is the inverse of invokeIndexerKCacheScatter.
+///
+/// @param k_fp8_out          Output FP8 data [num_tokens, head_dim], contiguous.
+/// @param k_scale_out        Output scale data [num_tokens, scale_size], contiguous (uint8 view).
+/// @param k_cache            Indexer K cache pool [num_blocks, block_size, 1, per_token_size] (may be non-contiguous).
+/// @param slot_mapping_fp8   Flat element index for FP8 data start position [num_tokens].
+/// @param slot_mapping_scale Flat element index for scale data start position [num_tokens].
+/// @param num_tokens         Number of tokens to gather.
+/// @param head_dim           Head dimension (must be 128).
+/// @param scale_size         Scale size in bytes (must be 4).
+/// @param cache_dim_0..3     Sizes of k_cache dimensions.
+/// @param cache_stride_0..3  Strides of k_cache dimensions (in bytes).
+/// @param stream             CUDA stream.
+void invokeIndexerKCacheGather(uint8_t* k_fp8_out, uint8_t* k_scale_out, uint8_t const* k_cache,
+    int64_t const* slot_mapping_fp8, int64_t const* slot_mapping_scale, int32_t num_tokens, int32_t head_dim,
+    int32_t scale_size, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2, int32_t cache_dim_3,
+    int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2, int64_t cache_stride_3,
     cudaStream_t stream = 0);
 
-}
+} // namespace kernels
 
 TRTLLM_NAMESPACE_END
diff --git a/cpp/tensorrt_llm/kernels/fusedCatFp8.cu b/cpp/tensorrt_llm/kernels/fusedCatFp8.cu
index 98dacf65f0ba..a2e01bac5547 100644
--- a/cpp/tensorrt_llm/kernels/fusedCatFp8.cu
+++ b/cpp/tensorrt_llm/kernels/fusedCatFp8.cu
@@ -80,7 +80,8 @@ union FP8x4
 template <bool UseUe8m0>
 __global__ __launch_bounds__(WARP_SIZE* ROWS_PER_BLOCK) void fusedCatFp8Kernel(__nv_fp8_e4m3* __restrict__ fp8_out,
     float* __restrict__ scale_out, __nv_bfloat16 const* __restrict__ pe, __nv_bfloat16 const* __restrict__ nope,
-    int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t pe_row_stride, int32_t nope_row_stride)
+    int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t pe_row_stride, int32_t nope_row_stride,
+    float output_scale_factor)
 {
     int warp_in_block = threadIdx.x / WARP_SIZE;
     int lane = threadIdx.x % WARP_SIZE;
@@ -169,7 +170,7 @@ __global__ __launch_bounds__(WARP_SIZE* ROWS_PER_BLOCK) void fusedCatFp8Kernel(_
 
     if (lane == 0)
     {
-        scale_out[row] = scale;
+        scale_out[row] = scale * output_scale_factor;
     }
 }
 
@@ -177,7 +178,7 @@ __global__ __launch_bounds__(WARP_SIZE* ROWS_PER_BLOCK) void fusedCatFp8Kernel(_
 
 void invokeFusedCatFp8(__nv_fp8_e4m3* fp8_out, float* scale_out, __nv_bfloat16 const* pe, __nv_bfloat16 const* nope,
     int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t head_dim, int32_t pe_row_stride, int32_t nope_row_stride,
-    bool use_ue8m0, cudaStream_t stream)
+    bool use_ue8m0, float output_scale_factor, cudaStream_t stream)
 {
     if (M == 0)
     {
@@ -208,12 +209,12 @@ void invokeFusedCatFp8(__nv_fp8_e4m3* fp8_out, float* scale_out, __nv_bfloat16 c
     if (use_ue8m0)
     {
         fusedCatFp8Kernel<true><<<grid, block, 0, stream>>>(
-            fp8_out, scale_out, pe, nope, M, pe_dim, nope_dim, pe_row_stride, nope_row_stride);
+            fp8_out, scale_out, pe, nope, M, pe_dim, nope_dim, pe_row_stride, nope_row_stride, output_scale_factor);
     }
     else
     {
         fusedCatFp8Kernel<false><<<grid, block, 0, stream>>>(
-            fp8_out, scale_out, pe, nope, M, pe_dim, nope_dim, pe_row_stride, nope_row_stride);
+            fp8_out, scale_out, pe, nope, M, pe_dim, nope_dim, pe_row_stride, nope_row_stride, output_scale_factor);
     }
 
     TLLM_CUDA_CHECK(cudaGetLastError());
diff --git a/cpp/tensorrt_llm/kernels/fusedCatFp8.h b/cpp/tensorrt_llm/kernels/fusedCatFp8.h
index 5118fc494de2..40b41d913846 100644
--- a/cpp/tensorrt_llm/kernels/fusedCatFp8.h
+++ b/cpp/tensorrt_llm/kernels/fusedCatFp8.h
@@ -52,11 +52,14 @@ namespace kernels
 ///                        For contiguous layout this equals pe_dim; for non-contiguous
 ///                        views (e.g. from torch.split) it may be larger.
 /// @param nope_row_stride Stride (in elements) between consecutive rows of nope.
-/// @param use_ue8m0       If true, use UE8M0 (power-of-two) scale format.
-/// @param stream          CUDA stream.
+/// @param use_ue8m0             If true, use UE8M0 (power-of-two) scale format.
+/// @param output_scale_factor   Multiplier applied to the output scale (default 1.0).
+///                              Useful for folding downstream scale multiplications
+///                              (e.g. weight_scale_factor) into the quantization kernel.
+/// @param stream                CUDA stream.
 void invokeFusedCatFp8(__nv_fp8_e4m3* fp8_out, float* scale_out, __nv_bfloat16 const* pe, __nv_bfloat16 const* nope,
     int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t head_dim, int32_t pe_row_stride, int32_t nope_row_stride,
-    bool use_ue8m0, cudaStream_t stream = 0);
+    bool use_ue8m0, float output_scale_factor = 1.0f, cudaStream_t stream = 0);
 
 } // namespace kernels
 
diff --git a/cpp/tensorrt_llm/kernels/fusedCatFp8Scatter.cu b/cpp/tensorrt_llm/kernels/fusedCatFp8Scatter.cu
new file mode 100644
index 000000000000..83ee115322fa
--- /dev/null
+++ b/cpp/tensorrt_llm/kernels/fusedCatFp8Scatter.cu
@@ -0,0 +1,282 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "fusedCatFp8Scatter.h"
+#include "tensorrt_llm/common/assert.h"
+#include "tensorrt_llm/common/config.h"
+#include "tensorrt_llm/common/cudaUtils.h"
+
+#include <cuda_bf16.h>
+#include <cuda_fp8.h>
+
+#include <cfloat>
+#include <cmath>
+#include <cstdint>
+
+TRTLLM_NAMESPACE_BEGIN
+
+namespace kernels
+{
+
+namespace
+{
+
+constexpr int HEAD_DIM = 128;
+constexpr int WARP_SIZE = 32;
+constexpr int ELEMS_PER_THREAD = 4; // 128 / 32
+constexpr int ROWS_PER_BLOCK = 8;
+constexpr float INV_FP8_E4M3_MAX = 1.0f / 448.0f;
+constexpr float MIN_AMAX = 1.0e-12f;
+constexpr int32_t SCALE_SIZE = 4;
+constexpr int32_t PER_TOKEN_SIZE = HEAD_DIM + SCALE_SIZE; // 132
+
+/// Warp-wide max reduction
+__device__ __forceinline__ float warpReduceMax(float val)
+{
+    for (int offset = WARP_SIZE / 2; offset > 0; offset >>= 1)
+    {
+        val = fmaxf(val, __shfl_xor_sync(0xFFFFFFFF, val, offset));
+    }
+    return val;
+}
+
+/// Helper union for vectorized BF16 loads (4 BF16 values = 8 bytes).
+union BF16x4
+{
+    int2 vec;
+    __nv_bfloat162 bf16x2[2];
+};
+
+/// Helper union for vectorized FP8 stores (4 FP8 values = 4 bytes).
+union FP8x4
+{
+    uint32_t u32;
+    __nv_fp8_e4m3 fp8[4];
+};
+
+/**
+ * Unravel a flat element index into a byte offset for the indexer K cache.
+ * Specialized for the DeepSeek-V3.2 layout: d2=1, d3=132 (compile-time constants).
+ * This enables the compiler to use multiplication-based integer division (4 cycles)
+ * instead of general-purpose int64 division (20+ cycles) for the d3 unravel.
+ */
+__device__ __forceinline__ int64_t flatIndexToMemoryOffset(
+    int64_t flat_idx, int32_t d1_mask, int32_t d1_shift, int64_t s0, int64_t s1, int64_t s3)
+{
+    // d3 = PER_TOKEN_SIZE = 132 (compile-time constant → fast multiply-based reduction)
+    int32_t i3 = flat_idx % PER_TOKEN_SIZE;
+    flat_idx /= PER_TOKEN_SIZE;
+    // d2 = 1: skip (always 0)
+    // d1 is power of 2 → bitwise AND/shift instead of integer division
+    int32_t i1 = static_cast<int32_t>(flat_idx) & d1_mask;
+    int32_t i0 = static_cast<int32_t>(flat_idx) >> d1_shift;
+    return i0 * s0 + i1 * s1 + i3 * s3;
+}
+
+/// Fused kernel: cat + FP8 quantization + scatter to paged K cache.
+///
+/// Grid: (ceil(M / ROWS_PER_BLOCK),)
+/// Block: (WARP_SIZE * ROWS_PER_BLOCK,)   i.e., (256,)
+///
+/// Each warp handles one row:
+///   - Loads from pe/nope, quantizes to FP8
+///   - Writes FP8 data to contiguous output AND to paged K cache
+///   - Lane 0 writes scale to contiguous output AND to paged K cache
+///
+/// Optimizations:
+///   1. Compile-time d3=132, d2=1: fast integer division via multiplication reduction.
+///   2. __ldg for read-only inputs (pe, nope, slot_mappings).
+///   3. All threads compute own cache offset in SIMT lockstep (no shuffle overhead).
+template <bool UseUe8m0>
+__global__ __launch_bounds__(WARP_SIZE* ROWS_PER_BLOCK) void fusedCatFp8ScatterKernel(
+    __nv_fp8_e4m3* __restrict__ fp8_out, float* __restrict__ scale_out, uint8_t* __restrict__ k_cache,
+    __nv_bfloat16 const* __restrict__ pe, __nv_bfloat16 const* __restrict__ nope,
+    int64_t const* __restrict__ slot_mapping_fp8, int64_t const* __restrict__ slot_mapping_scale,
+    int32_t const* __restrict__ num_tokens_ptr, int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t pe_row_stride,
+    int32_t nope_row_stride, int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_3, int32_t d1_mask,
+    int32_t d1_shift)
+{
+    int warp_in_block = threadIdx.x / WARP_SIZE;
+    int lane = threadIdx.x % WARP_SIZE;
+    int row = blockIdx.x * ROWS_PER_BLOCK + warp_in_block;
+
+    if (row >= M)
+    {
+        return;
+    }
+
+    // ---- Stage 1: Load + Concat (vectorized 8-byte loads) ----
+    float v0, v1, v2, v3;
+    {
+        int base = lane * ELEMS_PER_THREAD;
+        __nv_bfloat16 const* pe_row = pe + static_cast<int64_t>(row) * pe_row_stride;
+        __nv_bfloat16 const* nope_row = nope + static_cast<int64_t>(row) * nope_row_stride;
+
+        bool from_pe = (base < pe_dim);
+        __nv_bfloat16 const* src = from_pe ? pe_row : nope_row;
+        int col = from_pe ? base : (base - pe_dim);
+
+        BF16x4 loaded;
+        loaded.vec = __ldg(reinterpret_cast<int2 const*>(src + col));
+
+        float2 f0 = __bfloat1622float2(loaded.bf16x2[0]);
+        float2 f1 = __bfloat1622float2(loaded.bf16x2[1]);
+        v0 = f0.x;
+        v1 = f0.y;
+        v2 = f1.x;
+        v3 = f1.y;
+    }
+
+    // ---- Stage 2: FP8 Quantization ----
+    float local_max = fmaxf(fmaxf(fabsf(v0), fabsf(v1)), fmaxf(fabsf(v2), fabsf(v3)));
+    float amax = warpReduceMax(local_max);
+    amax = fmaxf(amax, MIN_AMAX);
+
+    float scale;
+    if constexpr (UseUe8m0)
+    {
+        float ratio = amax * INV_FP8_E4M3_MAX;
+        uint32_t bits = __float_as_uint(ratio);
+        uint32_t mantissa = bits & 0x007FFFFFu;
+        uint32_t exp_bits = bits & 0x7F800000u;
+        if (mantissa != 0u)
+        {
+            exp_bits += 0x00800000u;
+        }
+        scale = __uint_as_float(exp_bits);
+    }
+    else
+    {
+        scale = amax * INV_FP8_E4M3_MAX;
+    }
+
+    float inv_scale;
+    asm("rcp.approx.ftz.f32 %0, %1;" : "=f"(inv_scale) : "f"(scale));
+
+    auto quantize = [&](float val) -> __nv_fp8_e4m3
+    {
+        float scaled = val * inv_scale;
+        return __nv_fp8_e4m3(scaled);
+    };
+
+    // ---- Stage 3: Store to contiguous output ----
+    FP8x4 packed;
+    packed.fp8[0] = quantize(v0);
+    packed.fp8[1] = quantize(v1);
+    packed.fp8[2] = quantize(v2);
+    packed.fp8[3] = quantize(v3);
+
+    int base_out = row * HEAD_DIM + lane * ELEMS_PER_THREAD;
+    *reinterpret_cast<uint32_t*>(fp8_out + base_out) = packed.u32;
+
+    if (lane == 0)
+    {
+        scale_out[row] = scale;
+    }
+
+    // ---- Stage 4: Scatter to paged K cache ----
+    // Under CUDA graph capture, M is the padded batch size but only the first
+    // num_tokens rows carry valid slot_mapping entries. Skip scatter for rows
+    // beyond num_tokens to avoid corrupting stale cache locations. The negative
+    // sentinel branch below additionally guards against invalid slots in the
+    // non-capture path (e.g., explicit skip markers from callers).
+    int32_t const num_valid = (num_tokens_ptr != nullptr) ? __ldg(num_tokens_ptr) : M;
+    int64_t flat_idx_fp8_base = (row < num_valid) ? __ldg(&slot_mapping_fp8[row]) : int64_t{-1};
+    if (flat_idx_fp8_base >= 0)
+    {
+        int32_t head_dim_idx = lane * ELEMS_PER_THREAD;
+        int64_t flat_idx = flat_idx_fp8_base + head_dim_idx;
+
+        int64_t dst_offset
+            = flatIndexToMemoryOffset(flat_idx, d1_mask, d1_shift, cache_stride_0, cache_stride_1, cache_stride_3);
+
+        *reinterpret_cast<uint32_t*>(&k_cache[dst_offset]) = packed.u32;
+
+        if (lane == 0)
+        {
+            int64_t flat_idx_scale_base = __ldg(&slot_mapping_scale[row]);
+            if (flat_idx_scale_base >= 0)
+            {
+                int64_t dst_offset_scale = flatIndexToMemoryOffset(
+                    flat_idx_scale_base, d1_mask, d1_shift, cache_stride_0, cache_stride_1, cache_stride_3);
+
+                *reinterpret_cast<float*>(&k_cache[dst_offset_scale]) = scale;
+            }
+        }
+    }
+}
+
+} // anonymous namespace
+
+void invokeFusedCatFp8Scatter(__nv_fp8_e4m3* fp8_out, float* scale_out, uint8_t* k_cache, __nv_bfloat16 const* pe,
+    __nv_bfloat16 const* nope, int64_t const* slot_mapping_fp8, int64_t const* slot_mapping_scale,
+    int32_t const* num_tokens_ptr, int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t head_dim, int32_t pe_row_stride,
+    int32_t nope_row_stride, bool use_ue8m0, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2,
+    int32_t cache_dim_3, int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2, int64_t cache_stride_3,
+    cudaStream_t stream)
+{
+    if (M == 0)
+    {
+        return;
+    }
+
+    TLLM_CHECK_WITH_INFO(head_dim == HEAD_DIM, "fusedCatFp8Scatter: head_dim must be 128, got %d", head_dim);
+    TLLM_CHECK_WITH_INFO(pe_dim + nope_dim == head_dim,
+        "fusedCatFp8Scatter: pe_dim (%d) + nope_dim (%d) != head_dim (%d)", pe_dim, nope_dim, head_dim);
+    TLLM_CHECK_WITH_INFO((head_dim & (head_dim - 1)) == 0, "fusedCatFp8Scatter: head_dim must be power of 2");
+    TLLM_CHECK_WITH_INFO(pe_dim % ELEMS_PER_THREAD == 0, "fusedCatFp8Scatter: pe_dim (%d) must be a multiple of %d",
+        pe_dim, ELEMS_PER_THREAD);
+    TLLM_CHECK_WITH_INFO(pe_row_stride >= pe_dim, "fusedCatFp8Scatter: pe_row_stride (%d) must be >= pe_dim (%d)",
+        pe_row_stride, pe_dim);
+    TLLM_CHECK_WITH_INFO(nope_row_stride >= nope_dim,
+        "fusedCatFp8Scatter: nope_row_stride (%d) must be >= nope_dim (%d)", nope_row_stride, nope_dim);
+    TLLM_CHECK_WITH_INFO(pe_row_stride % ELEMS_PER_THREAD == 0,
+        "fusedCatFp8Scatter: pe_row_stride (%d) must be a multiple of %d", pe_row_stride, ELEMS_PER_THREAD);
+    TLLM_CHECK_WITH_INFO(nope_row_stride % ELEMS_PER_THREAD == 0,
+        "fusedCatFp8Scatter: nope_row_stride (%d) must be a multiple of %d", nope_row_stride, ELEMS_PER_THREAD);
+    TLLM_CHECK_WITH_INFO(cache_dim_2 == 1, "fusedCatFp8Scatter: cache_dim_2 must be 1 (got %d)", cache_dim_2);
+    TLLM_CHECK_WITH_INFO(cache_dim_3 == PER_TOKEN_SIZE,
+        "fusedCatFp8Scatter: cache_dim_3 must be %d (head_dim + scale_size, got %d)", PER_TOKEN_SIZE, cache_dim_3);
+    TLLM_CHECK_WITH_INFO((cache_dim_1 & (cache_dim_1 - 1)) == 0,
+        "fusedCatFp8Scatter: cache_dim_1 (block_size=%d) must be a power of 2", cache_dim_1);
+
+    // Pre-compute mask and shift for power-of-2 block_size division
+    int32_t d1_mask = cache_dim_1 - 1;
+    int32_t d1_shift = __builtin_ctz(cache_dim_1);
+
+    int num_blocks = (M + ROWS_PER_BLOCK - 1) / ROWS_PER_BLOCK;
+    dim3 grid(num_blocks);
+    dim3 block(WARP_SIZE * ROWS_PER_BLOCK); // 256 threads per block
+
+    if (use_ue8m0)
+    {
+        fusedCatFp8ScatterKernel<true><<<grid, block, 0, stream>>>(fp8_out, scale_out, k_cache, pe, nope,
+            slot_mapping_fp8, slot_mapping_scale, num_tokens_ptr, M, pe_dim, nope_dim, pe_row_stride, nope_row_stride,
+            cache_stride_0, cache_stride_1, cache_stride_3, d1_mask, d1_shift);
+    }
+    else
+    {
+        fusedCatFp8ScatterKernel<false><<<grid, block, 0, stream>>>(fp8_out, scale_out, k_cache, pe, nope,
+            slot_mapping_fp8, slot_mapping_scale, num_tokens_ptr, M, pe_dim, nope_dim, pe_row_stride, nope_row_stride,
+            cache_stride_0, cache_stride_1, cache_stride_3, d1_mask, d1_shift);
+    }
+
+    TLLM_CUDA_CHECK(cudaGetLastError());
+}
+
+} // namespace kernels
+
+TRTLLM_NAMESPACE_END
diff --git a/cpp/tensorrt_llm/kernels/fusedCatFp8Scatter.h b/cpp/tensorrt_llm/kernels/fusedCatFp8Scatter.h
new file mode 100644
index 000000000000..0add27a5917e
--- /dev/null
+++ b/cpp/tensorrt_llm/kernels/fusedCatFp8Scatter.h
@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "tensorrt_llm/common/config.h"
+#include "tensorrt_llm/common/cudaUtils.h"
+
+#include <cuda_bf16.h>
+#include <cuda_fp8.h>
+#include <cuda_runtime.h>
+
+TRTLLM_NAMESPACE_BEGIN
+
+namespace kernels
+{
+
+/// Fused concat + FP8 quantization + scatter into paged K cache.
+///
+/// Combines the functionality of fusedCatFp8 and indexerKCacheScatter into a single kernel.
+/// The FP8-quantized values are written to both:
+///   1. A contiguous output buffer (for use by the prefill logits path)
+///   2. The non-contiguous paged K cache (for later decode-phase gather)
+///
+/// This eliminates one global memory round trip compared to running the two kernels sequentially.
+///
+/// @param fp8_out            Output FP8 data [M, head_dim], row-major, contiguous.
+/// @param scale_out          Output scales [M, 1], float32, contiguous.
+/// @param k_cache            Paged K cache [num_blocks, block_size, 1, per_token_size] (may be non-contiguous).
+/// @param pe                 Input PE part, BF16.
+/// @param nope               Input non-PE part, BF16.
+/// @param slot_mapping_fp8   Flat element index for FP8 data start in k_cache [M].
+/// @param slot_mapping_scale Flat element index for scale start in k_cache [M].
+/// @param M                  Number of rows (may be the padded size under CUDA graph capture).
+/// @param num_tokens_ptr     Optional device pointer to an int32 scalar holding the actual number
+///                           of valid rows. When non-null, rows whose index is >= *num_tokens_ptr
+///                           skip the paged-K-cache scatter (the contiguous fp8_out/scale_out are
+///                           still written for them). When null, all M rows scatter. This is
+///                           required when the kernel is captured in a CUDA graph but M is padded.
+/// @param pe_dim             PE dimension.
+/// @param nope_dim           Non-PE dimension.
+/// @param head_dim           Total head dimension (must be 128).
+/// @param pe_row_stride      Row stride for pe input.
+/// @param nope_row_stride    Row stride for nope input.
+/// @param use_ue8m0          If true, use UE8M0 scale format.
+/// @param cache_dim_0..3     Sizes of k_cache dimensions.
+/// @param cache_stride_0..3  Strides of k_cache dimensions (in bytes).
+/// @param stream             CUDA stream.
+void invokeFusedCatFp8Scatter(__nv_fp8_e4m3* fp8_out, float* scale_out, uint8_t* k_cache, __nv_bfloat16 const* pe,
+    __nv_bfloat16 const* nope, int64_t const* slot_mapping_fp8, int64_t const* slot_mapping_scale,
+    int32_t const* num_tokens_ptr, int32_t M, int32_t pe_dim, int32_t nope_dim, int32_t head_dim, int32_t pe_row_stride,
+    int32_t nope_row_stride, bool use_ue8m0, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2,
+    int32_t cache_dim_3, int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2, int64_t cache_stride_3,
+    cudaStream_t stream = 0);
+
+} // namespace kernels
+
+TRTLLM_NAMESPACE_END
diff --git a/cpp/tensorrt_llm/kernels/indexerKCacheGather.cu b/cpp/tensorrt_llm/kernels/indexerKCacheGather.cu
index dcb5a212b319..d7552a58d551 100644
--- a/cpp/tensorrt_llm/kernels/indexerKCacheGather.cu
+++ b/cpp/tensorrt_llm/kernels/indexerKCacheGather.cu
@@ -26,135 +26,121 @@ namespace kernels
 
 namespace
 {
+
+constexpr int WARP_SIZE = 32;
+constexpr int VEC_SIZE = 4;
+constexpr int TOKENS_PER_BLOCK = 8;
+constexpr int32_t HEAD_DIM = 128;
+constexpr int32_t SCALE_SIZE = 4;
+constexpr int32_t PER_TOKEN_SIZE = HEAD_DIM + SCALE_SIZE; // 132
+
 /**
- * Given a flat element index and tensor shape [d0, d1, d2, d3] with strides [s0, s1, s2, s3],
- * find the actual memory offset within the given k cache pool using the strides.
+ * Unravel a flat element index into a byte offset for the indexer K cache.
+ * Specialized for the DeepSeek-V3.2 layout: d2=1, d3=132 (compile-time constant).
+ * Block size (d1) must be power of 2 → bitwise AND/shift instead of integer division.
  */
 __device__ __forceinline__ int64_t flatIndexToMemoryOffset(
-    int64_t flat_idx, int32_t d0, int32_t d1, int32_t d2, int32_t d3, int64_t s0, int64_t s1, int64_t s2, int64_t s3)
+    int64_t flat_idx, int32_t d1_mask, int32_t d1_shift, int64_t s0, int64_t s1, int64_t s3)
 {
-    // Unravel from innermost to outermost dimension
-    int32_t i3 = flat_idx % d3;
-    flat_idx /= d3;
-
-    int32_t i2 = flat_idx % d2;
-    flat_idx /= d2;
-
-    int32_t i1 = flat_idx % d1;
-    flat_idx /= d1;
-
-    int32_t i0 = flat_idx;
-
-    // Compute memory offset using strides
-    return i0 * s0 + i1 * s1 + i2 * s2 + i3 * s3;
+    // d3 = PER_TOKEN_SIZE = 132 (compile-time constant → fast multiply-based reduction)
+    int32_t i3 = flat_idx % PER_TOKEN_SIZE;
+    flat_idx /= PER_TOKEN_SIZE;
+    // d2 = 1: skip (always 0)
+    // d1 is power of 2 → bitwise AND/shift instead of integer division
+    int32_t i1 = static_cast<int32_t>(flat_idx) & d1_mask;
+    int32_t i0 = static_cast<int32_t>(flat_idx) >> d1_shift;
+    return i0 * s0 + i1 * s1 + i3 * s3;
 }
 
 } // anonymous namespace
 
 /**
- * CUDA kernel to gather both FP8 K values and scales from the indexer k cache pool.
- * This is the inverse of indexerKCacheScatterUnifiedKernel.
+ * Optimized gather kernel for FP8 K values and scales from paged indexer K cache.
  *
- * @param k_cache             Indexer k cache pool with shape [num_blocks, block_size, 1, per_token_size]
- *                            (can be non-contiguous)
- * @param slot_mapping_fp8    Flat element index for FP8 data start position [total_kv_len]
- * @param slot_mapping_scale  Flat element index for scale data start position [total_kv_len]
- * @param out_fp8             Output FP8 data [num_tokens, head_dim] contiguous
- * @param out_scale           Output scale data [num_tokens, scale_size] contiguous
- * @param k_token_start       Start offset into slot_mapping arrays
- * @param num_tokens          Number of tokens to gather
- * @param head_dim            Head dimension (must be 128)
- * @param scale_size          Scale size in bytes (must be 4)
- * @param cache_stride_0      Stride for k_cache dimension 0 (in bytes)
- * @param cache_stride_1      Stride for k_cache dimension 1 (in bytes)
- * @param cache_stride_2      Stride for k_cache dimension 2 (in bytes)
- * @param cache_stride_3      Stride for k_cache dimension 3 (in bytes)
- * @param cache_dim_0         Size of k_cache dimension 0
- * @param cache_dim_1         Size of k_cache dimension 1
- * @param cache_dim_2         Size of k_cache dimension 2
- * @param cache_dim_3         Size of k_cache dimension 3
+ * Optimizations:
+ *   1. Multi-token blocks: 8 tokens/block (256 threads) for warp-level latency hiding.
+ *   2. Compile-time d3=132: enables fast multiplication-based division (~4 cycles)
+ *      instead of general-purpose int64 division (20+ cycles).
+ *   3. Power-of-2 d1 (block_size): d1 division replaced by bitwise AND/shift.
+ *   4. __ldg for read-only cache and slot mapping loads.
+ *   5. All threads compute own offset in SIMT lockstep (no shuffle serialization).
  */
-__global__ void indexerKCacheGatherUnifiedKernel(uint8_t const* __restrict__ k_cache,
-    int64_t const* __restrict__ slot_mapping_fp8, int64_t const* __restrict__ slot_mapping_scale,
-    uint8_t* __restrict__ out_fp8, uint8_t* __restrict__ out_scale, int32_t k_token_start, int32_t num_tokens,
-    int32_t head_dim, int32_t scale_size, int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2,
-    int64_t cache_stride_3, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2, int32_t cache_dim_3)
+__global__ __launch_bounds__(WARP_SIZE* TOKENS_PER_BLOCK) void indexerKCacheGatherUnifiedKernel(
+    uint8_t* __restrict__ k_fp8_out, uint8_t* __restrict__ k_scale_out, uint8_t const* __restrict__ k_cache,
+    int64_t const* __restrict__ slot_mapping_fp8, int64_t const* __restrict__ slot_mapping_scale, int32_t num_tokens,
+    int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_3, int32_t d1_mask, int32_t d1_shift)
 {
-    // For head_dim=128, each thread handles 4 bytes/elements per read/write instruction
-    constexpr int VEC_SIZE = 4;
-
-    // Token index from block.x
-    int32_t token_idx = blockIdx.x;
+    int warp_in_block = threadIdx.x / WARP_SIZE;
+    int lane = threadIdx.x % WARP_SIZE;
+    int32_t token_idx = blockIdx.x * TOKENS_PER_BLOCK + warp_in_block;
 
     if (token_idx >= num_tokens)
     {
         return;
     }
 
-    // Index into slot_mapping with k_token_start offset
-    int32_t slot_idx = k_token_start + token_idx;
-
-    int64_t flat_idx_fp8_base = slot_mapping_fp8[slot_idx];
-    int64_t flat_idx_scale_base = slot_mapping_scale[slot_idx];
+    int64_t flat_idx_fp8_base = __ldg(&slot_mapping_fp8[token_idx]);
+    int64_t flat_idx_scale_base = __ldg(&slot_mapping_scale[token_idx]);
 
     if (flat_idx_fp8_base < 0 || flat_idx_scale_base < 0)
     {
         return;
     }
 
-    int32_t head_dim_idx = threadIdx.x * VEC_SIZE;
+    // Each thread computes its own memory offset (SIMT lockstep, no extra cost)
+    int32_t head_dim_idx = lane * VEC_SIZE;
     int64_t flat_idx = flat_idx_fp8_base + head_dim_idx;
+    int64_t src_offset
+        = flatIndexToMemoryOffset(flat_idx, d1_mask, d1_shift, cache_stride_0, cache_stride_1, cache_stride_3);
+    int64_t dst_offset = static_cast<int64_t>(token_idx) * HEAD_DIM + head_dim_idx;
 
-    // Convert flat index to memory offset using strides (k cache pool from cpp kv cache manager is non-contiguous)
-    int64_t src_offset = flatIndexToMemoryOffset(flat_idx, cache_dim_0, cache_dim_1, cache_dim_2, cache_dim_3,
-        cache_stride_0, cache_stride_1, cache_stride_2, cache_stride_3);
-    int64_t dst_offset = token_idx * head_dim + head_dim_idx;
-
-    // 4 bytes read from non-contiguous cache, write to contiguous output
-    *reinterpret_cast<uint32_t*>(&out_fp8[dst_offset]) = *reinterpret_cast<uint32_t const*>(&k_cache[src_offset]);
+    // Gather FP8 data: 4 bytes per thread, 128 bytes per warp
+    *reinterpret_cast<uint32_t*>(&k_fp8_out[dst_offset])
+        = __ldg(reinterpret_cast<uint32_t const*>(&k_cache[src_offset]));
 
-    // Only thread 0 reads the single 4 bytes scale value
-    if (threadIdx.x == 0)
+    // Lane 0: gather scale (4 bytes)
+    if (lane == 0)
     {
-        int64_t src_offset_scale = flatIndexToMemoryOffset(flat_idx_scale_base, cache_dim_0, cache_dim_1, cache_dim_2,
-            cache_dim_3, cache_stride_0, cache_stride_1, cache_stride_2, cache_stride_3);
-        int64_t dst_offset_scale = token_idx * scale_size; // scale_size = 4
-
-        // 4 bytes read for scale
-        *reinterpret_cast<uint32_t*>(&out_scale[dst_offset_scale])
-            = *reinterpret_cast<uint32_t const*>(&k_cache[src_offset_scale]);
+        int64_t src_offset_scale = flatIndexToMemoryOffset(
+            flat_idx_scale_base, d1_mask, d1_shift, cache_stride_0, cache_stride_1, cache_stride_3);
+        int64_t dst_offset_scale = static_cast<int64_t>(token_idx) * SCALE_SIZE;
+        *reinterpret_cast<uint32_t*>(&k_scale_out[dst_offset_scale])
+            = __ldg(reinterpret_cast<uint32_t const*>(&k_cache[src_offset_scale]));
     }
 }
 
-void invokeIndexerKCacheGather(uint8_t const* k_cache, int64_t const* slot_mapping_fp8,
-    int64_t const* slot_mapping_scale, uint8_t* out_fp8, uint8_t* out_scale, int32_t k_token_start, int32_t num_tokens,
-    int32_t head_dim, int32_t scale_size, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2,
-    int32_t cache_dim_3, int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2, int64_t cache_stride_3,
-    cudaStream_t stream)
+void invokeIndexerKCacheGather(uint8_t* k_fp8_out, uint8_t* k_scale_out, uint8_t const* k_cache,
+    int64_t const* slot_mapping_fp8, int64_t const* slot_mapping_scale, int32_t num_tokens, int32_t head_dim,
+    int32_t scale_size, int32_t cache_dim_0, int32_t cache_dim_1, int32_t cache_dim_2, int32_t cache_dim_3,
+    int64_t cache_stride_0, int64_t cache_stride_1, int64_t cache_stride_2, int64_t cache_stride_3, cudaStream_t stream)
 {
     if (num_tokens == 0)
     {
         return;
     }
 
-    // Assertions for DeepSeek-V3.2 configuration
-    constexpr int32_t QUANT_BLOCK_SIZE = 128;
     TLLM_CHECK_WITH_INFO(
-        head_dim == QUANT_BLOCK_SIZE, "head_dim must equal 128 for DeepSeek-V3 indexer cache (got %d)", head_dim);
+        head_dim == HEAD_DIM, "head_dim must equal 128 for DeepSeek-V3 indexer cache (got %d)", head_dim);
     TLLM_CHECK_WITH_INFO(
-        scale_size == 4, "scale_size must equal 4 bytes (1 float32 scale per token, got %d)", scale_size);
+        scale_size == SCALE_SIZE, "scale_size must equal 4 bytes (1 float32 scale per token, got %d)", scale_size);
+    TLLM_CHECK_WITH_INFO(cache_dim_2 == 1, "Optimized gather requires cache_dim_2 == 1 (got %d)", cache_dim_2);
+    TLLM_CHECK_WITH_INFO(cache_dim_3 == PER_TOKEN_SIZE,
+        "Optimized gather requires cache_dim_3 == %d (head_dim + scale_size, got %d)", PER_TOKEN_SIZE, cache_dim_3);
+    TLLM_CHECK_WITH_INFO((cache_dim_1 & (cache_dim_1 - 1)) == 0,
+        "Optimized gather requires cache_dim_1 (block_size=%d) to be a power of 2", cache_dim_1);
+
+    // Pre-compute mask and shift for power-of-2 block_size division
+    int32_t d1_mask = cache_dim_1 - 1;
+    int32_t d1_shift = __builtin_ctz(cache_dim_1);
 
-    // For head_dim=128, we use 32 threads to handle 128 bytes per token and extra 4 bytes for scale
-    constexpr int32_t THREADS_PER_BLOCK = 32;
+    constexpr int32_t THREADS_PER_BLOCK = WARP_SIZE * TOKENS_PER_BLOCK; // 256
 
     dim3 block(THREADS_PER_BLOCK);
-    dim3 grid(num_tokens);
+    dim3 grid((num_tokens + TOKENS_PER_BLOCK - 1) / TOKENS_PER_BLOCK);
 
-    indexerKCacheGatherUnifiedKernel<<<grid, block, 0, stream>>>(k_cache, slot_mapping_fp8, slot_mapping_scale, out_fp8,
-        out_scale, k_token_start, num_tokens, head_dim, scale_size, cache_stride_0, cache_stride_1, cache_stride_2,
-        cache_stride_3, cache_dim_0, cache_dim_1, cache_dim_2, cache_dim_3);
+    indexerKCacheGatherUnifiedKernel<<<grid, block, 0, stream>>>(k_fp8_out, k_scale_out, k_cache, slot_mapping_fp8,
+        slot_mapping_scale, num_tokens, cache_stride_0, cache_stride_1, cache_stride_3, d1_mask, d1_shift);
 
-    // Check for kernel launch errors
     TLLM_CUDA_CHECK(cudaGetLastError());
 }
 
diff --git a/cpp/tensorrt_llm/thop/CMakeLists.txt b/cpp/tensorrt_llm/thop/CMakeLists.txt
index 5d715a87fece..3dd130f2e811 100644
--- a/cpp/tensorrt_llm/thop/CMakeLists.txt
+++ b/cpp/tensorrt_llm/thop/CMakeLists.txt
@@ -90,6 +90,7 @@ add_library(
   fp4BlockScaleMoe.cpp
   noAuxTcOp.cpp
   fusedCatFp8Op.cpp
+  FusedCatFp8ScatterOp.cpp
   IndexerKCacheGatherOp.cpp
   IndexerKCacheScatterOp.cpp
   IndexerTopKOp.cpp
diff --git a/cpp/tensorrt_llm/thop/FusedCatFp8ScatterOp.cpp b/cpp/tensorrt_llm/thop/FusedCatFp8ScatterOp.cpp
new file mode 100644
index 000000000000..0cf09c41297d
--- /dev/null
+++ b/cpp/tensorrt_llm/thop/FusedCatFp8ScatterOp.cpp
@@ -0,0 +1,119 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "tensorrt_llm/kernels/fusedCatFp8Scatter.h"
+#include "tensorrt_llm/thop/thUtils.h"
+
+#include <ATen/cuda/EmptyTensor.h>
+
+TRTLLM_NAMESPACE_BEGIN
+
+namespace torch_ext
+{
+
+std::tuple<at::Tensor, at::Tensor> fused_cat_fp8_scatter(at::Tensor const& pe, at::Tensor const& nope, bool use_ue8m0,
+    at::Tensor& k_cache, at::Tensor const& slot_mapping_fp8, at::Tensor const& slot_mapping_scale,
+    std::optional<at::Tensor> const& num_tokens)
+{
+    CHECK_TH_CUDA(pe);
+    CHECK_TH_CUDA(nope);
+    CHECK_TH_CUDA(k_cache);
+    CHECK_TH_CUDA(slot_mapping_fp8);
+    CHECK_TH_CUDA(slot_mapping_scale);
+
+    TORCH_CHECK(pe.scalar_type() == at::ScalarType::BFloat16, "pe must be BF16, got ", pe.scalar_type());
+    TORCH_CHECK(nope.scalar_type() == at::ScalarType::BFloat16, "nope must be BF16, got ", nope.scalar_type());
+    TORCH_CHECK(pe.dim() >= 2, "pe must be >= 2D, got ", pe.dim(), "D");
+    TORCH_CHECK(nope.dim() >= 2, "nope must be >= 2D, got ", nope.dim(), "D");
+
+    TORCH_CHECK(pe.stride(-1) == 1, "pe must have contiguous innermost dim");
+    TORCH_CHECK(nope.stride(-1) == 1, "nope must have contiguous innermost dim");
+
+    TORCH_CHECK(k_cache.dim() == 4, "k_cache must be 4D [num_blocks, block_size, 1, per_token_size]");
+    TORCH_CHECK(slot_mapping_fp8.dim() == 1, "slot_mapping_fp8 must be 1D");
+    TORCH_CHECK(slot_mapping_scale.dim() == 1, "slot_mapping_scale must be 1D");
+    TORCH_CHECK(slot_mapping_fp8.scalar_type() == torch::kInt64, "slot_mapping_fp8 must be int64");
+    TORCH_CHECK(slot_mapping_scale.scalar_type() == torch::kInt64, "slot_mapping_scale must be int64");
+
+    int32_t const* num_tokens_ptr = nullptr;
+    if (num_tokens.has_value())
+    {
+        CHECK_TH_CUDA(num_tokens.value());
+        TORCH_CHECK(num_tokens.value().scalar_type() == torch::kInt32, "num_tokens must be int32");
+        TORCH_CHECK(num_tokens.value().numel() == 1, "num_tokens must be a 1-element tensor");
+        num_tokens_ptr = num_tokens.value().data_ptr<int32_t>();
+    }
+
+    auto const pe_dim = static_cast<int32_t>(pe.size(-1));
+    auto const nope_dim = static_cast<int32_t>(nope.size(-1));
+    auto const head_dim = pe_dim + nope_dim;
+
+    TORCH_CHECK(head_dim == 128, "head_dim (pe_dim + nope_dim) must be 128, got ", head_dim);
+
+    auto const pe_M = pe.numel() / pe_dim;
+    auto const nope_M = nope.numel() / nope_dim;
+    TORCH_CHECK(pe_M == nope_M, "pe and nope must have same number of rows");
+    auto const M = static_cast<int32_t>(pe_M);
+
+    TORCH_CHECK(slot_mapping_fp8.size(0) == M, "slot_mapping_fp8 length must equal num rows");
+    TORCH_CHECK(slot_mapping_scale.size(0) == M, "slot_mapping_scale length must equal num rows");
+
+    auto const pe_row_stride = static_cast<int32_t>(pe.stride(-2));
+    auto const nope_row_stride = static_cast<int32_t>(nope.stride(-2));
+
+    // Allocate contiguous output tensors
+    at::Tensor fp8_out
+        = at::detail::empty_cuda({M, head_dim}, at::ScalarType::Float8_e4m3fn, pe.device(), std::nullopt);
+    at::Tensor scale_out = at::detail::empty_cuda({M, 1}, at::ScalarType::Float, pe.device(), std::nullopt);
+
+    int32_t cache_dim_0 = static_cast<int32_t>(k_cache.size(0));
+    int32_t cache_dim_1 = static_cast<int32_t>(k_cache.size(1));
+    int32_t cache_dim_2 = static_cast<int32_t>(k_cache.size(2));
+    int32_t cache_dim_3 = static_cast<int32_t>(k_cache.size(3));
+
+    int64_t cache_stride_0 = static_cast<int64_t>(k_cache.stride(0));
+    int64_t cache_stride_1 = static_cast<int64_t>(k_cache.stride(1));
+    int64_t cache_stride_2 = static_cast<int64_t>(k_cache.stride(2));
+    int64_t cache_stride_3 = static_cast<int64_t>(k_cache.stride(3));
+
+    auto stream = at::cuda::getCurrentCUDAStream(pe.get_device());
+
+    tensorrt_llm::kernels::invokeFusedCatFp8Scatter(reinterpret_cast<__nv_fp8_e4m3*>(fp8_out.data_ptr()),
+        reinterpret_cast<float*>(scale_out.data_ptr()), k_cache.data_ptr<uint8_t>(),
+        reinterpret_cast<__nv_bfloat16 const*>(pe.data_ptr()), reinterpret_cast<__nv_bfloat16 const*>(nope.data_ptr()),
+        slot_mapping_fp8.data_ptr<int64_t>(), slot_mapping_scale.data_ptr<int64_t>(), num_tokens_ptr, M, pe_dim,
+        nope_dim, head_dim, pe_row_stride, nope_row_stride, use_ue8m0, cache_dim_0, cache_dim_1, cache_dim_2,
+        cache_dim_3, cache_stride_0, cache_stride_1, cache_stride_2, cache_stride_3, stream);
+
+    return {fp8_out, scale_out};
+}
+
+} // namespace torch_ext
+
+TRTLLM_NAMESPACE_END
+
+TORCH_LIBRARY_FRAGMENT(trtllm, m)
+{
+    m.def(
+        "fused_cat_fp8_scatter(Tensor pe, Tensor nope, bool use_ue8m0, "
+        "Tensor(a!) k_cache, Tensor slot_mapping_fp8, Tensor slot_mapping_scale, "
+        "Tensor? num_tokens=None) -> (Tensor, Tensor)");
+}
+
+TORCH_LIBRARY_IMPL(trtllm, CUDA, m)
+{
+    m.impl("fused_cat_fp8_scatter", &tensorrt_llm::torch_ext::fused_cat_fp8_scatter);
+}
diff --git a/cpp/tensorrt_llm/thop/IndexerKCacheGatherOp.cpp b/cpp/tensorrt_llm/thop/IndexerKCacheGatherOp.cpp
index 27fbc33bd1be..86e0a178f660 100644
--- a/cpp/tensorrt_llm/thop/IndexerKCacheGatherOp.cpp
+++ b/cpp/tensorrt_llm/thop/IndexerKCacheGatherOp.cpp
@@ -28,29 +28,16 @@ TRTLLM_NAMESPACE_BEGIN
 namespace torch_ext
 {
 
-std::tuple<th::Tensor, th::Tensor> indexer_k_cache_gather_op(th::Tensor const& k_cache,
-    th::Tensor const& slot_mapping_fp8, th::Tensor const& slot_mapping_scale, int64_t k_token_start, int64_t num_tokens)
+std::tuple<th::Tensor, th::Tensor> indexer_k_cache_gather_op(
+    th::Tensor const& k_cache, th::Tensor const& slot_mapping_fp8, th::Tensor const& slot_mapping_scale)
 {
-    constexpr int32_t HEAD_DIM = 128;
-    constexpr int32_t SCALE_SIZE = 4;
-
-    auto device = k_cache.device();
-
-    // Early return for empty gather
-    if (num_tokens == 0)
-    {
-        auto k_fp8 = th::empty({0, HEAD_DIM}, th::TensorOptions().dtype(torch::kFloat8_e4m3fn).device(device));
-        auto k_scale = th::empty({0, 1}, th::TensorOptions().dtype(torch::kFloat32).device(device));
-        return std::make_tuple(std::move(k_fp8), std::move(k_scale));
-    }
-
     // Validate all tensors are CUDA tensors
     TORCH_CHECK(k_cache.is_cuda() && slot_mapping_fp8.is_cuda() && slot_mapping_scale.is_cuda(),
         "All tensors must be CUDA tensors");
 
     // Validate tensor dimensions
-    TORCH_CHECK(slot_mapping_fp8.dim() == 1, "slot_mapping_fp8 must be a 1D Tensor");
-    TORCH_CHECK(slot_mapping_scale.dim() == 1, "slot_mapping_scale must be a 1D Tensor");
+    TORCH_CHECK(slot_mapping_fp8.dim() == 1, "slot_mapping_fp8 must be a 1D Tensor [num_tokens]");
+    TORCH_CHECK(slot_mapping_scale.dim() == 1, "slot_mapping_scale must be a 1D Tensor [num_tokens]");
 
     // Enforce k_cache is 4D tensor
     TORCH_CHECK(k_cache.dim() == 4,
@@ -61,47 +48,44 @@ std::tuple<th::Tensor, th::Tensor> indexer_k_cache_gather_op(th::Tensor const& k
     TORCH_CHECK(slot_mapping_fp8.scalar_type() == torch::kInt64, "slot_mapping_fp8 must be int64");
     TORCH_CHECK(slot_mapping_scale.scalar_type() == torch::kInt64, "slot_mapping_scale must be int64");
 
+    // Validate shapes are consistent
+    auto num_tokens = static_cast<int32_t>(slot_mapping_fp8.size(0));
+    TORCH_CHECK(
+        slot_mapping_scale.size(0) == num_tokens, "slot_mapping_scale length must equal slot_mapping_fp8 length");
+
     // Validate tensors are contiguous (except k_cache which may be non-contiguous)
-    // k_cache can be non-contiguous - we handle this via strides
     TORCH_CHECK(slot_mapping_fp8.is_contiguous(), "slot_mapping_fp8 must be contiguous");
     TORCH_CHECK(slot_mapping_scale.is_contiguous(), "slot_mapping_scale must be contiguous");
 
-    // Validate slot_mapping has enough elements
-    TORCH_CHECK(slot_mapping_fp8.size(0) >= k_token_start + num_tokens,
-        "slot_mapping_fp8 too short for k_token_start + num_tokens");
-    TORCH_CHECK(slot_mapping_scale.size(0) >= k_token_start + num_tokens,
-        "slot_mapping_scale too short for k_token_start + num_tokens");
+    // DeepSeek-V3.2 constants
+    constexpr int32_t head_dim = 128;
+    constexpr int32_t scale_size = 4; // float32 = 4 bytes
 
-    int32_t cache_dim_0 = static_cast<int32_t>(k_cache.size(0)); // num_blocks
-    int32_t cache_dim_1 = static_cast<int32_t>(k_cache.size(1)); // block_size
-    int32_t cache_dim_2 = static_cast<int32_t>(k_cache.size(2)); // num_kv_heads
-    int32_t cache_dim_3 = static_cast<int32_t>(k_cache.size(3)); // per_token_size
-
-    // Validation for indexer k cache pool for DeepSeek-V3.2 constraints
+    int32_t cache_dim_2 = static_cast<int32_t>(k_cache.size(2));
     TORCH_CHECK(cache_dim_2 == 1, "k_cache dimension 2 must be 1 for DeepSeek-V3.2, got %d", cache_dim_2);
 
+    int32_t cache_dim_0 = static_cast<int32_t>(k_cache.size(0));
+    int32_t cache_dim_1 = static_cast<int32_t>(k_cache.size(1));
+    int32_t cache_dim_3 = static_cast<int32_t>(k_cache.size(3));
+
     int64_t cache_stride_0 = static_cast<int64_t>(k_cache.stride(0));
     int64_t cache_stride_1 = static_cast<int64_t>(k_cache.stride(1));
     int64_t cache_stride_2 = static_cast<int64_t>(k_cache.stride(2));
     int64_t cache_stride_3 = static_cast<int64_t>(k_cache.stride(3));
 
-    // Allocate output buffers
-    auto num_tokens_i32 = static_cast<int32_t>(num_tokens);
-    auto out_fp8 = th::empty({num_tokens, HEAD_DIM}, th::TensorOptions().dtype(torch::kUInt8).device(device));
-    auto out_scale = th::empty({num_tokens, SCALE_SIZE}, th::TensorOptions().dtype(torch::kUInt8).device(device));
+    // Allocate output tensors
+    auto options_u8 = th::TensorOptions().dtype(torch::kUInt8).device(k_cache.device());
+    th::Tensor k_fp8_out = th::empty({num_tokens, head_dim}, options_u8);
+    th::Tensor k_scale_out = th::empty({num_tokens, scale_size}, options_u8);
 
     auto stream = at::cuda::getCurrentCUDAStream(k_cache.get_device());
 
-    tk::invokeIndexerKCacheGather(k_cache.data_ptr<uint8_t>(), slot_mapping_fp8.data_ptr<int64_t>(),
-        slot_mapping_scale.data_ptr<int64_t>(), out_fp8.data_ptr<uint8_t>(), out_scale.data_ptr<uint8_t>(),
-        static_cast<int32_t>(k_token_start), num_tokens_i32, HEAD_DIM, SCALE_SIZE, cache_dim_0, cache_dim_1,
-        cache_dim_2, cache_dim_3, cache_stride_0, cache_stride_1, cache_stride_2, cache_stride_3, stream);
-
-    // View-cast to final dtypes (no copy)
-    auto k_fp8 = out_fp8.view(torch::kFloat8_e4m3fn);
-    auto k_scale = out_scale.view(torch::kFloat32).view({num_tokens, 1});
+    tk::invokeIndexerKCacheGather(k_fp8_out.data_ptr<uint8_t>(), k_scale_out.data_ptr<uint8_t>(),
+        k_cache.data_ptr<uint8_t>(), slot_mapping_fp8.data_ptr<int64_t>(), slot_mapping_scale.data_ptr<int64_t>(),
+        num_tokens, head_dim, scale_size, cache_dim_0, cache_dim_1, cache_dim_2, cache_dim_3, cache_stride_0,
+        cache_stride_1, cache_stride_2, cache_stride_3, stream);
 
-    return std::make_tuple(std::move(k_fp8), std::move(k_scale));
+    return {k_fp8_out, k_scale_out};
 }
 
 } // namespace torch_ext
@@ -111,8 +95,8 @@ TRTLLM_NAMESPACE_END
 TORCH_LIBRARY_FRAGMENT(trtllm, m)
 {
     m.def(
-        "indexer_k_cache_gather_op(Tensor k_cache, Tensor slot_mapping_fp8, "
-        "Tensor slot_mapping_scale, int k_token_start, int num_tokens) -> (Tensor, Tensor)");
+        "indexer_k_cache_gather_op(Tensor k_cache, "
+        "Tensor slot_mapping_fp8, Tensor slot_mapping_scale) -> (Tensor, Tensor)");
 }
 
 TORCH_LIBRARY_IMPL(trtllm, CUDA, m)
diff --git a/cpp/tensorrt_llm/thop/fusedCatFp8Op.cpp b/cpp/tensorrt_llm/thop/fusedCatFp8Op.cpp
index e4e0de19c5a6..a0ef4867f9be 100644
--- a/cpp/tensorrt_llm/thop/fusedCatFp8Op.cpp
+++ b/cpp/tensorrt_llm/thop/fusedCatFp8Op.cpp
@@ -24,7 +24,8 @@ TRTLLM_NAMESPACE_BEGIN
 namespace torch_ext
 {
 
-std::tuple<at::Tensor, at::Tensor> fused_cat_fp8(at::Tensor const& pe, at::Tensor const& nope, bool use_ue8m0)
+std::tuple<at::Tensor, at::Tensor> fused_cat_fp8(
+    at::Tensor const& pe, at::Tensor const& nope, bool use_ue8m0, double output_scale_factor)
 {
     CHECK_TH_CUDA(pe);
     CHECK_TH_CUDA(nope);
@@ -67,7 +68,7 @@ std::tuple<at::Tensor, at::Tensor> fused_cat_fp8(at::Tensor const& pe, at::Tenso
     tensorrt_llm::kernels::invokeFusedCatFp8(reinterpret_cast<__nv_fp8_e4m3*>(fp8_out.data_ptr()),
         reinterpret_cast<float*>(scale_out.data_ptr()), reinterpret_cast<__nv_bfloat16 const*>(pe.data_ptr()),
         reinterpret_cast<__nv_bfloat16 const*>(nope.data_ptr()), M, pe_dim, nope_dim, head_dim, pe_row_stride,
-        nope_row_stride, use_ue8m0, stream);
+        nope_row_stride, use_ue8m0, static_cast<float>(output_scale_factor), stream);
 
     return {fp8_out, scale_out};
 }
@@ -78,7 +79,9 @@ TRTLLM_NAMESPACE_END
 
 TORCH_LIBRARY_FRAGMENT(trtllm, m)
 {
-    m.def("fused_cat_fp8(Tensor pe, Tensor nope, bool use_ue8m0=False) -> (Tensor, Tensor)");
+    m.def(
+        "fused_cat_fp8(Tensor pe, Tensor nope, bool use_ue8m0=False, float output_scale_factor=1.0) -> (Tensor, "
+        "Tensor)");
 }
 
 TORCH_LIBRARY_IMPL(trtllm, CUDA, m)
diff --git a/tensorrt_llm/_torch/attention_backend/sparse/dsa.py b/tensorrt_llm/_torch/attention_backend/sparse/dsa.py
index d9a410fe62f2..c8d3d4f19c77 100644
--- a/tensorrt_llm/_torch/attention_backend/sparse/dsa.py
+++ b/tensorrt_llm/_torch/attention_backend/sparse/dsa.py
@@ -448,6 +448,23 @@ def __post_init__(self):
             device='cpu',
             pin_memory=prefer_pinned(),
         )
+        # Scalar device tensor holding the actual num_tokens for the current
+        # forward pass. The fused cat+scatter kernel reads this to bound the
+        # scatter writes when running under CUDA graph capture (where the K
+        # tensor is padded to max_num_tokens but only the first num_tokens
+        # slot_mapping entries are refreshed).
+        self.indexer_num_tokens_cuda = self.get_empty(
+            self.cuda_graph_buffers,
+            (1, ),
+            cache_name="indexer_num_tokens",
+            dtype=torch.int32,
+            capture_graph=capture_graph,
+        )
+        self.host_indexer_num_tokens = torch.zeros_like(
+            self.indexer_num_tokens_cuda,
+            device='cpu',
+            pin_memory=prefer_pinned(),
+        )
         # Per-token request index buffer for topk_indices conversion
         self.req_idx_per_token = self.get_empty(
             self.cuda_graph_buffers,
@@ -978,6 +995,15 @@ def on_update_kv_lens(self):
             self.slot_mapping_fp8[:self.num_tokens] = fp8_indices
             self.slot_mapping_scale[:self.num_tokens] = scale_indices
 
+        # Refresh the scalar num_tokens device tensor for the fused
+        # cat+scatter kernel. Under CUDA graph capture, the K projection
+        # tensor is padded to max_num_tokens but only the first num_tokens
+        # slot_mapping entries above are valid; the kernel uses this
+        # scalar to bound its scatter writes.
+        self.host_indexer_num_tokens[0] = self.num_tokens
+        self.indexer_num_tokens_cuda.copy_(self.host_indexer_num_tokens,
+                                           non_blocking=True)
+
         if self.num_generations > 0:
             torch.cumsum(
                 self.kv_lens_cuda[self.num_contexts:self.
@@ -1447,31 +1473,54 @@ def prepare(metadata: DSAtrtllmAttentionMetadata):
             metadata.slot_mapping_fp8_fullkv = metadata.slot_mapping_fp8
             metadata.slot_mapping_scale_fullkv = metadata.slot_mapping_scale
 
-    def _update_k_cache(self, k_fp8: torch.Tensor, k_scale: torch.Tensor,
-                        metadata: DSAtrtllmAttentionMetadata) -> None:
+    def _gather_k_cache_for_chunk(
+        self,
+        metadata: DSAtrtllmAttentionMetadata,
+        chunk: IndexerPrefillChunkMetadata,
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
         """
-        Insert/append k values and scales into the indexer k cache using pre-computed slot mappings.
-        Uses flat byte indices with vectorized scatter.
+        Gather K values from indexer cache for a specific chunk.
+
+        Uses pre-computed extended slot mappings that cover cached + current batch context tokens.
+        chunk.k_token_start/k_token_end directly index into the extended slot mapping.
 
         Args:
-            k_fp8: FP8 quantized k tensor, shape [total_tokens, head_dim]
-            k_scale: Scaling factors, shape [total_tokens, head_dim // quant_block_size]
+            metadata: Attention metadata
+            chunk: Chunk metadata with k_token_start/end as indices into extended slot mapping
+
+        Returns:
+            k_fp8: FP8 quantized k tensor, shape [num_k_tokens, head_dim]
+            k_scale: Scaling factors, shape [num_k_tokens, 1]
         """
-        if metadata.kv_cache_manager is None or metadata.slot_mapping_fp8 is None:
-            return
+        assert metadata.slot_mapping_fp8_fullkv is not None, \
+            "_gather_k_cache_for_chunk requires extended slot mappings (only available with cached tokens)"
 
         k_cache = metadata.kv_cache_manager.get_indexer_k_cache_buffers(
             self.layer_idx)
 
-        num_tokens = k_fp8.shape[0]
+        head_dim = self.head_dim
+        scale_size = 4  # float32 = 4 bytes
 
-        # The C++ op reinterprets k_fp8 (FP8) and k_scale (float32) as raw
-        # bytes internally and only reads the first num_tokens entries from
-        # the slot mapping buffers, avoiding Python-side view/slice overhead.
-        torch.ops.trtllm.indexer_k_cache_scatter_op(k_fp8, k_scale, k_cache,
-                                                    metadata.slot_mapping_fp8,
-                                                    metadata.slot_mapping_scale,
-                                                    num_tokens)
+        # Extract slot mappings using chunk's k_token_start/end
+        # These indices point directly into the extended slot mapping array
+        k_token_start = chunk.k_token_start
+        k_token_end = chunk.k_token_end
+        num_k_tokens = k_token_end - k_token_start
+
+        slot_mapping_fp8_chunk = metadata.slot_mapping_fp8_fullkv[
+            k_token_start:k_token_end]
+        slot_mapping_scale_chunk = metadata.slot_mapping_scale_fullkv[
+            k_token_start:k_token_end]
+
+        # Fused CUDA gather: single kernel replaces ~12 Python tensor ops.
+        k_fp8_bytes, k_scale_bytes = torch.ops.trtllm.indexer_k_cache_gather_op(
+            k_cache, slot_mapping_fp8_chunk, slot_mapping_scale_chunk)
+
+        k_fp8 = k_fp8_bytes.view(torch.float8_e4m3fn).view(
+            num_k_tokens, head_dim)
+        k_scale = k_scale_bytes.view(torch.float32).view(num_k_tokens, 1)
+
+        return k_fp8, k_scale
 
     def sparse_attn_indexer(
         self,
@@ -1483,12 +1532,15 @@ def sparse_attn_indexer(
         weights: torch.Tensor,
         use_custom_topk: bool = True,
     ) -> torch.Tensor:
-        """Run the indexer TopK kernel for both prefill and decode phases."""
+        """Run the indexer TopK kernel for both prefill and decode phases.
+
+        The K cache is populated upstream by the fused cat+quant+scatter
+        kernel invoked from pre_indexer_proj, so this method does not need
+        to run a separate scatter before the prefill chunks gather from it.
+        """
         assert metadata.kv_cache_manager is None or \
             metadata.kv_cache_manager.quant_block_size == 128, \
             "Only support quant_block_size = 128 for now"
-        # Update the indexer k cache before prefill chunks gather from it.
-        self._update_k_cache(k_fp8, k_scale, metadata)
 
         num_contexts = metadata.num_contexts
         num_generations = metadata.num_generations
@@ -1518,15 +1570,9 @@ def sparse_attn_indexer(
                     tp_rank = metadata.mapping.tp_rank
                     tp_size = metadata.mapping.tp_size
 
-                k_cache_4d = metadata.kv_cache_manager.get_indexer_k_cache_buffers(
-                    self.layer_idx)
-
                 for chunk in metadata.indexer_prefill_chunks:
-                    num_k_tokens = chunk.k_token_end - chunk.k_token_start
-                    chunk_k_fp8, chunk_k_scale = torch.ops.trtllm.indexer_k_cache_gather_op(
-                        k_cache_4d, metadata.slot_mapping_fp8_fullkv,
-                        metadata.slot_mapping_scale_fullkv, chunk.k_token_start,
-                        num_k_tokens)
+                    chunk_k_fp8, chunk_k_scale = self._gather_k_cache_for_chunk(
+                        metadata, chunk)
 
                     chunk_num_token = chunk.token_end - chunk.token_start
                     apply_q_split = q_split_eligible and chunk_num_token >= q_split_threshold
@@ -1759,12 +1805,6 @@ def sparse_attn_indexer(
                 metadata.topk_indices_buffer[num_ctx_tokens:num_tokens, :]
         return topk_indices_buffer
 
-    def _weight_scale(self, weights: torch.Tensor,
-                      q_scale: torch.Tensor) -> torch.Tensor:
-        """Apply quantization scale to indexer attention weights."""
-        weights = _scale(weights, q_scale, self.weight_scale_factor)
-        return weights
-
     def _qk_projection_and_rope(self, qr: torch.Tensor, indexer_k: torch.Tensor,
                                 position_ids: torch.Tensor):
         """Project Q/K and apply RoPE"""
@@ -1779,21 +1819,60 @@ def _qk_projection_and_rope(self, qr: torch.Tensor, indexer_k: torch.Tensor,
         k_pe = k_pe[:, 0, :]
         return q_pe, q_nope, k_pe, k_nope
 
-    def _prep_q_or_k(self, qk_pe: torch.Tensor, qk_nope: torch.Tensor):
-        """Concatenate and FP8 quantize for Q or K via fused kernel."""
+    def _prep_q(self, q_pe: torch.Tensor, q_nope: torch.Tensor):
+        """Concatenate, FP8 quantize Q, and fold weight_scale_factor into scale output."""
+        fp8_out, scale = torch.ops.trtllm.fused_cat_fp8(
+            q_pe, q_nope, self.scale_fmt == "ue8m0", self.weight_scale_factor)
+        return fp8_out, scale
+
+    def _prep_k(self, k_pe: torch.Tensor, k_nope: torch.Tensor):
+        """Concatenate and FP8 quantize K (no scale factor folding)."""
         fp8_out, scale = torch.ops.trtllm.fused_cat_fp8(
-            qk_pe, qk_nope, self.scale_fmt == "ue8m0")
+            k_pe, k_nope, self.scale_fmt == "ue8m0")
         return fp8_out, scale
 
+    def _prep_k_and_scatter(self, k_pe: torch.Tensor, k_nope: torch.Tensor,
+                            metadata: DSAtrtllmAttentionMetadata):
+        """Fused: concatenate K PE/noPE, FP8 quantize, and scatter to K cache in one kernel.
+
+        The scatter target slots are read from metadata.slot_mapping_fp8 /
+        slot_mapping_scale. Under CUDA graph capture, k_pe is padded to the
+        captured batch size, so we slice the slot mapping buffers to match
+        k_pe.shape[0] and rely on metadata.indexer_num_tokens_cuda to bound
+        the in-kernel scatter writes to only the valid rows.
+
+        When no kv_cache_manager is attached (e.g. unit-test harnesses that
+        only exercise the indexer TopK path) we fall back to the
+        non-scattering cat+quant so the rest of the pipeline still works.
+        """
+        if metadata.kv_cache_manager is None or metadata.slot_mapping_fp8 is None:
+            return self._prep_k(k_pe, k_nope)
+
+        k_cache = metadata.kv_cache_manager.get_indexer_k_cache_buffers(
+            self.layer_idx)
+        num_rows = k_pe.shape[0]
+        slot_mapping_fp8 = metadata.slot_mapping_fp8[:num_rows]
+        slot_mapping_scale = metadata.slot_mapping_scale[:num_rows]
+
+        k_fp8, k_scale = torch.ops.trtllm.fused_cat_fp8_scatter(
+            k_pe, k_nope, self.scale_fmt == "ue8m0", k_cache, slot_mapping_fp8,
+            slot_mapping_scale, metadata.indexer_num_tokens_cuda)
+        return k_fp8, k_scale
+
     def pre_indexer_proj(
-        self, qr: torch.Tensor, hidden_states: torch.Tensor,
-        position_ids: torch.Tensor
+        self,
+        qr: torch.Tensor,
+        hidden_states: torch.Tensor,
+        position_ids: torch.Tensor,
+        metadata: DSAtrtllmAttentionMetadata,
     ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
-        """Pure token-wise projections (CUDA-graph-capturable).
+        """Token-wise projections plus fused K-cache scatter (CUDA-graph-capturable).
 
-        Runs cublas_mm, qk_projection_and_rope, FP8 quantize, and weight
-        scaling.  Does NOT touch the k cache or any batch-specific metadata,
-        so this can safely run inside a captured CUDA graph partition.
+        Runs cublas_mm, qk_projection_and_rope, FP8 quantize, weight scaling,
+        and (on the K path) a fused cat+quant+scatter that writes directly
+        into the paged indexer K cache. The scatter is bounded at runtime by
+        metadata.indexer_num_tokens_cuda so this remains safe under graph
+        capture where tensors are padded to the captured batch size.
 
         Returns (q_fp8, k_fp8, k_scale, weights).
         """
@@ -1811,9 +1890,11 @@ def pre_indexer_proj(
 
         q_pe, q_nope, k_pe, k_nope = self._qk_projection_and_rope(
             qr, indexer_k, position_ids)
+        # Q path: cat + FP8 quantize + fold weight_scale_factor into scale
+        # K path: cat + FP8 quantize + scatter to paged K cache (fused)
         q, k = maybe_execute_in_parallel(
-            lambda: self._prep_q_or_k(q_pe, q_nope),
-            lambda: self._prep_q_or_k(k_pe, k_nope),
+            lambda: self._prep_q(q_pe, q_nope),
+            lambda: self._prep_k_and_scatter(k_pe, k_nope, metadata),
             self.ln_events[0],
             self.ln_events[1],
             self.aux_stream,
@@ -1823,21 +1904,11 @@ def pre_indexer_proj(
         q_fp8 = q_fp8.view(-1, self.n_heads, self.head_dim)
         q_scale = q_scale.view(-1, self.n_heads, 1)
 
-        weights = self._weight_scale(weights, q_scale)
+        # weight_scale_factor is already folded into q_scale by _prep_q
+        weights = weights * q_scale.squeeze(-1)
 
         return q_fp8, k_fp8, k_scale, weights
 
-    @torch.inference_mode()
-    def forward(self, qr: torch.Tensor, hidden_states: torch.Tensor,
-                metadata: DSAtrtllmAttentionMetadata,
-                position_ids: torch.Tensor):
-        q_fp8, k_fp8, k_scale, weights = self.pre_indexer_proj(
-            qr, hidden_states, position_ids)
-
-        # Return topk indices buffer for sparse attention [num_tokens, index_topk]
-        return self.sparse_attn_indexer(metadata, hidden_states, q_fp8, k_fp8,
-                                        k_scale, weights)
-
 
 class DSATrtllmAttention(TrtllmAttention):
     """TRT-LLM attention layer with DSA sparse indexer for MLA models."""
diff --git a/tensorrt_llm/_torch/custom_ops/cpp_custom_ops.py b/tensorrt_llm/_torch/custom_ops/cpp_custom_ops.py
index f75ee36cd088..99599cdadbd5 100644
--- a/tensorrt_llm/_torch/custom_ops/cpp_custom_ops.py
+++ b/tensorrt_llm/_torch/custom_ops/cpp_custom_ops.py
@@ -574,7 +574,37 @@ def _(input: torch.Tensor, use_ue8m0: bool = False):
                                     sz, dtype=torch.float)
 
     @torch.library.register_fake("trtllm::fused_cat_fp8")
-    def _(pe: torch.Tensor, nope: torch.Tensor, use_ue8m0: bool = False):
+    def _(pe: torch.Tensor,
+          nope: torch.Tensor,
+          use_ue8m0: bool = False,
+          output_scale_factor: float = 1.0):
+        pe_dim = pe.shape[-1]
+        nope_dim = nope.shape[-1]
+        head_dim = pe_dim + nope_dim
+        M = pe.numel() // pe_dim
+        fp8_out = pe.new_empty((M, head_dim), dtype=torch.float8_e4m3fn)
+        scale_out = pe.new_empty((M, 1), dtype=torch.float32)
+        return fp8_out, scale_out
+
+    @torch.library.register_fake("trtllm::indexer_k_cache_gather_op")
+    def _(k_cache: torch.Tensor, slot_mapping_fp8: torch.Tensor,
+          slot_mapping_scale: torch.Tensor):
+        num_tokens = slot_mapping_fp8.shape[0]
+        head_dim = 128
+        scale_size = 4
+        k_fp8_out = k_cache.new_empty((num_tokens, head_dim), dtype=torch.uint8)
+        k_scale_out = k_cache.new_empty((num_tokens, scale_size),
+                                        dtype=torch.uint8)
+        return k_fp8_out, k_scale_out
+
+    @torch.library.register_fake("trtllm::fused_cat_fp8_scatter")
+    def _(pe: torch.Tensor,
+          nope: torch.Tensor,
+          use_ue8m0: bool,
+          k_cache: torch.Tensor,
+          slot_mapping_fp8: torch.Tensor,
+          slot_mapping_scale: torch.Tensor,
+          num_tokens: Optional[torch.Tensor] = None):
         pe_dim = pe.shape[-1]
         nope_dim = nope.shape[-1]
         head_dim = pe_dim + nope_dim
@@ -1117,11 +1147,3 @@ def _(req_id: torch.Tensor, block_table: torch.Tensor,
           token_indices: torch.Tensor, block_size: int, num_topk_tokens: int,
           stride_factor: int, layer_id: int) -> torch.Tensor:
         return torch.empty_like(token_indices)
-
-    @torch.library.register_fake("trtllm::indexer_k_cache_gather_op")
-    def _(k_cache: torch.Tensor, slot_mapping_fp8: torch.Tensor,
-          slot_mapping_scale: torch.Tensor, k_token_start: int,
-          num_tokens: int) -> Tuple[torch.Tensor, torch.Tensor]:
-        k_fp8 = k_cache.new_empty([num_tokens, 128], dtype=torch.float8_e4m3fn)
-        k_scale = k_cache.new_empty([num_tokens, 1], dtype=torch.float32)
-        return k_fp8, k_scale
diff --git a/tensorrt_llm/_torch/modules/attention.py b/tensorrt_llm/_torch/modules/attention.py
index fd0714e25a2a..e3179f3068d7 100644
--- a/tensorrt_llm/_torch/modules/attention.py
+++ b/tensorrt_llm/_torch/modules/attention.py
@@ -1748,11 +1748,13 @@ def forward_dsa_proj(
         if use_short_mha_for_ctx and attn_metadata.num_generations == 0:
             return [q, compressed_kv, k_pe, latent_cache]
 
-        # pre_indexer_proj is the CUDA-graph-safe portion: pure token-wise
-        # compute (cublas_mm, rope, FP8 quantize, weight scaling) with no
-        # access to batch-specific metadata or the k cache.
+        # pre_indexer_proj is the CUDA-graph-safe portion: token-wise compute
+        # (cublas_mm, rope, FP8 quantize, weight scaling) plus a fused
+        # cat+quant+scatter into the paged indexer K cache. The scatter is
+        # bounded by attn_metadata.indexer_num_tokens_cuda at kernel time, so
+        # padded rows under graph capture do not corrupt the cache.
         q_fp8, k_fp8, k_scale, weights = self.mqa.indexer.pre_indexer_proj(
-            qr, hidden_states, position_ids)
+            qr, hidden_states, position_ids, attn_metadata)
 
         return [
             q, compressed_kv, k_pe, latent_cache, q_fp8, k_fp8, k_scale, weights
diff --git a/tensorrt_llm/_torch/pyexecutor/sampler.py b/tensorrt_llm/_torch/pyexecutor/sampler.py
index dc8fa941ab8b..7a7dfe43964e 100644
--- a/tensorrt_llm/_torch/pyexecutor/sampler.py
+++ b/tensorrt_llm/_torch/pyexecutor/sampler.py
@@ -290,7 +290,7 @@ class SampleStateWithMMResult(SampleState[SampleStateTensors, SampleStateTensors
 
 
 @dataclass(kw_only=True, frozen=True, slots=True)
-class RequestGroupKey(Generic[GenericStrategyKeyType]):  # type: ignore[misc]
+class RequestGroupKey(Generic[GenericStrategyKeyType]):
     strategy_key: GenericStrategyKeyType
     needs_probs: bool
 
diff --git a/tests/unittest/_torch/attention/sparse/test_cpp_custom_ops.py b/tests/unittest/_torch/attention/sparse/test_cpp_custom_ops.py
index fd1a00b34c62..ce520bb4019d 100644
--- a/tests/unittest/_torch/attention/sparse/test_cpp_custom_ops.py
+++ b/tests/unittest/_torch/attention/sparse/test_cpp_custom_ops.py
@@ -150,10 +150,16 @@ def test_indexer_k_cache_gather_contiguous(
         total_kv_len, num_blocks, block_size, per_token_size, device
     )
 
-    # C++ op
-    cpp_fp8, cpp_scale = torch.ops.trtllm.indexer_k_cache_gather_op(
-        k_cache, slot_fp8, slot_scale, k_token_start, num_tokens
+    # Pre-slice slot mappings for the 3-arg C++ op API
+    sliced_fp8 = slot_fp8[k_token_start : k_token_start + num_tokens]
+    sliced_scale = slot_scale[k_token_start : k_token_start + num_tokens]
+
+    # C++ op returns raw bytes; reinterpret to typed tensors (same as dsa.py)
+    cpp_fp8_bytes, cpp_scale_bytes = torch.ops.trtllm.indexer_k_cache_gather_op(
+        k_cache, sliced_fp8, sliced_scale
     )
+    cpp_fp8 = cpp_fp8_bytes.view(torch.float8_e4m3fn)
+    cpp_scale = cpp_scale_bytes.view(torch.float32).view(num_tokens, 1)
 
     # Reference
     ref_fp8, ref_scale = _reference_indexer_k_cache_gather(
@@ -217,10 +223,16 @@ def test_indexer_k_cache_gather_noncontiguous(
     slot_fp8 = flat_starts
     slot_scale = flat_starts + HEAD_DIM
 
-    # C++ op (handles non-contiguous strides internally)
-    cpp_fp8, cpp_scale = torch.ops.trtllm.indexer_k_cache_gather_op(
-        k_cache_nc, slot_fp8, slot_scale, k_token_start, num_tokens
+    # Pre-slice slot mappings for the 3-arg C++ op API
+    sliced_fp8 = slot_fp8[k_token_start : k_token_start + num_tokens]
+    sliced_scale = slot_scale[k_token_start : k_token_start + num_tokens]
+
+    # C++ op returns raw bytes; reinterpret to typed tensors (same as dsa.py)
+    cpp_fp8_bytes, cpp_scale_bytes = torch.ops.trtllm.indexer_k_cache_gather_op(
+        k_cache_nc, sliced_fp8, sliced_scale
     )
+    cpp_fp8 = cpp_fp8_bytes.view(torch.float8_e4m3fn)
+    cpp_scale = cpp_scale_bytes.view(torch.float32).view(num_tokens, 1)
 
     # Reference uses contiguous copy
     ref_fp8, ref_scale = _reference_indexer_k_cache_gather(
@@ -241,12 +253,16 @@ def test_indexer_k_cache_gather_empty():
     """Zero-length gather should return correctly shaped empty tensors."""
     device = torch.device("cuda")
     k_cache = torch.randint(0, 256, (4, 8, 1, BYTES_PER_TOKEN), dtype=torch.uint8, device=device)
-    slot_fp8 = torch.zeros(10, dtype=torch.int64, device=device)
-    slot_scale = torch.zeros(10, dtype=torch.int64, device=device)
+    # Pass empty (0-length) slot mappings for the 3-arg API
+    slot_fp8 = torch.zeros(0, dtype=torch.int64, device=device)
+    slot_scale = torch.zeros(0, dtype=torch.int64, device=device)
 
-    k_fp8, k_scale = torch.ops.trtllm.indexer_k_cache_gather_op(
-        k_cache, slot_fp8, slot_scale, k_token_start=5, num_tokens=0
+    # C++ op returns raw bytes; reinterpret to typed tensors (same as dsa.py)
+    k_fp8_bytes, k_scale_bytes = torch.ops.trtllm.indexer_k_cache_gather_op(
+        k_cache, slot_fp8, slot_scale
     )
+    k_fp8 = k_fp8_bytes.view(torch.float8_e4m3fn)
+    k_scale = k_scale_bytes.view(torch.float32).reshape(0, 1)
 
     assert k_fp8.shape == (0, HEAD_DIM)
     assert k_scale.shape == (0, 1)
diff --git a/tests/unittest/_torch/attention/sparse/test_dsa_indexer.py b/tests/unittest/_torch/attention/sparse/test_dsa_indexer.py
index 79fa32dc74e0..847cc1f18822 100644
--- a/tests/unittest/_torch/attention/sparse/test_dsa_indexer.py
+++ b/tests/unittest/_torch/attention/sparse/test_dsa_indexer.py
@@ -157,6 +157,25 @@ def __init__(self, head_dim):
     return indexer
 
 
+def _update_k_cache_for_test(indexer, k_fp8: torch.Tensor,
+                             k_scale: torch.Tensor, metadata) -> None:
+    """Test-only helper that scatters quantized K tensors into the paged
+    indexer K cache. Replaces the former ``Indexer._update_k_cache`` method,
+    which was removed in favor of the fused cat+quant+scatter kernel
+    invoked from ``pre_indexer_proj``. Tests that start from pre-quantized
+    ``k_fp8`` / ``k_scale`` still need a plain scatter path.
+    """
+    if metadata.kv_cache_manager is None or metadata.slot_mapping_fp8 is None:
+        return
+    k_cache = metadata.kv_cache_manager.get_indexer_k_cache_buffers(
+        indexer.layer_idx)
+    num_tokens = k_fp8.shape[0]
+    torch.ops.trtllm.indexer_k_cache_scatter_op(k_fp8, k_scale, k_cache,
+                                                metadata.slot_mapping_fp8,
+                                                metadata.slot_mapping_scale,
+                                                num_tokens)
+
+
 def _calc_diff(x: torch.Tensor, y: torch.Tensor):
     """Return a global difference metric for unit tests.
 
@@ -879,7 +898,7 @@ def test_fp8_k_cache_roundtrip():
     k_fp8, k_scale = fp8_utils.fp8_quantize_1x128_sf_transpose(k_original)
 
     # Write to cache
-    indexer._update_k_cache(k_fp8, k_scale, metadata)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata)
 
     # Verify scales for both requests
     cache_flat = cache_manager.indexer_k_cache_pool_per_layer[
@@ -1010,7 +1029,8 @@ def test_indexer_decode_with_paged_kv_cache(batch_size, next_n):
     k_context_fp8, k_context_scale = fp8_utils.fp8_quantize_1x128_sf_transpose(
         k_context_bf16)
 
-    indexer._update_k_cache(k_context_fp8, k_context_scale, metadata_context)
+    _update_k_cache_for_test(indexer, k_context_fp8, k_context_scale,
+                             metadata_context)
     print(f"✓ Wrote {total_context_tokens} FP8 context tokens to cache")
 
     # Phase 2: Write generation tokens (next_n per sequence) as FP8
@@ -1035,7 +1055,7 @@ def test_indexer_decode_with_paged_kv_cache(batch_size, next_n):
 
     k_gen_fp8, k_gen_scale = fp8_utils.fp8_quantize_1x128_sf_transpose(
         k_gen_bf16)
-    indexer._update_k_cache(k_gen_fp8, k_gen_scale, metadata_gen)
+    _update_k_cache_for_test(indexer, k_gen_fp8, k_gen_scale, metadata_gen)
     print(
         f"✓ Wrote {batch_size * num_gen_tokens} FP8 generation tokens to cache")
 
@@ -1579,7 +1599,7 @@ def test_indexer_chunked_prefill(chunk_size, seq_lens_list, chunking_type):
             f"  Chunk {i}: Q[{chunk.token_start}:{chunk.token_end}] ({num_q} tokens), "
             f"K[{chunk.k_token_start}:{chunk.k_token_end}] ({num_k} tokens)")
 
-    indexer._update_k_cache(k_fp8, k_scale, metadata_chunked)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_chunked)
     topk_indices_chunked = indexer.sparse_attn_indexer(metadata_chunked,
                                                        hidden_states, q_fp8,
                                                        k_fp8, k_scale, weights)
@@ -1611,7 +1631,7 @@ def test_indexer_chunked_prefill(chunk_size, seq_lens_list, chunking_type):
             f"✓ Created {num_baseline_chunks} chunk(s) (effectively non-chunked)"
         )
 
-    indexer._update_k_cache(k_fp8, k_scale, metadata_baseline)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_baseline)
     topk_indices_baseline = indexer.sparse_attn_indexer(metadata_baseline,
                                                         hidden_states, q_fp8,
                                                         k_fp8, k_scale, weights)
@@ -1833,7 +1853,8 @@ def test_indexer_decode_custom_vs_fallback(batch_size, next_n, index_topk,
         max_draft_tokens=next_n - 1,
     )
     Indexer.prepare(metadata_context)
-    indexer._update_k_cache(k_context_fp8, k_context_scale, metadata_context)
+    _update_k_cache_for_test(indexer, k_context_fp8, k_context_scale,
+                             metadata_context)
 
     # Generate decode phase test data
     q = torch.randn((num_gen_tokens, heads, head_dim),
@@ -1866,7 +1887,7 @@ def test_indexer_decode_custom_vs_fallback(batch_size, next_n, index_topk,
         max_draft_tokens=next_n - 1,
     )
     Indexer.prepare(metadata_gen_write)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_gen_write)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_gen_write)
 
     # Test with custom CUDA kernel
     metadata_custom = _create_mock_metadata(request_ids,
@@ -1883,7 +1904,7 @@ def test_indexer_decode_custom_vs_fallback(batch_size, next_n, index_topk,
                                             max_draft_tokens=next_n - 1)
 
     Indexer.prepare(metadata_custom)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_custom)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_custom)
 
     try:
         topk_indices_custom = indexer.sparse_attn_indexer(metadata_custom,
@@ -1911,7 +1932,7 @@ def test_indexer_decode_custom_vs_fallback(batch_size, next_n, index_topk,
                                               max_draft_tokens=next_n - 1)
 
     Indexer.prepare(metadata_fallback)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_fallback)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_fallback)
     topk_indices_fallback = indexer.sparse_attn_indexer(metadata_fallback,
                                                         hidden_states,
                                                         q_fp8,
@@ -1937,7 +1958,7 @@ def test_indexer_decode_custom_vs_fallback(batch_size, next_n, index_topk,
                                               enable_indexer_skip=True)
 
         Indexer.prepare(metadata_skip)
-        indexer._update_k_cache(k_fp8, k_scale, metadata_skip)
+        _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_skip)
 
         try:
             topk_indices_skip = indexer.sparse_attn_indexer(
@@ -2050,7 +2071,7 @@ def test_indexer_prefill_chunked_custom_vs_fallback(batch_size, index_topk,
                                             total_tokens, chunk_size)
 
     Indexer.prepare(metadata_custom)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_custom)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_custom)
 
     assert metadata_custom.indexer_prefill_chunks is not None
 
@@ -2074,7 +2095,7 @@ def test_indexer_prefill_chunked_custom_vs_fallback(batch_size, index_topk,
                                               chunk_size)
 
     Indexer.prepare(metadata_fallback)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_fallback)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_fallback)
     topk_indices_fallback = indexer.sparse_attn_indexer(metadata_fallback,
                                                         hidden_states,
                                                         q_fp8,
@@ -2158,7 +2179,7 @@ def test_indexer_prefill_single_pass_custom_vs_fallback(batch_size, index_topk,
                                             total_tokens, max_model_len)
 
     Indexer.prepare(metadata_custom)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_custom)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_custom)
     # Force single-pass path by setting indexer_prefill_chunks to None
     metadata_custom.indexer_prefill_chunks = None
 
@@ -2182,7 +2203,7 @@ def test_indexer_prefill_single_pass_custom_vs_fallback(batch_size, index_topk,
                                               max_model_len)
 
     Indexer.prepare(metadata_fallback)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_fallback)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_fallback)
     # Force single-pass path by setting indexer_prefill_chunks to None
     metadata_fallback.indexer_prefill_chunks = None
 
@@ -2207,7 +2228,7 @@ def test_indexer_prefill_single_pass_custom_vs_fallback(batch_size, index_topk,
                                           max_model_len,
                                           enable_indexer_skip=True)
     Indexer.prepare(metadata_skip)
-    indexer._update_k_cache(k_fp8, k_scale, metadata_skip)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_skip)
     metadata_skip.indexer_prefill_chunks = None
 
     try:
@@ -2319,7 +2340,7 @@ def test_indexer_topk_multi_request_with_different_cache(enable_indexer_skip):
                                      enable_context_mla_with_cached_kv=True)
 
     Indexer.prepare(metadata)
-    indexer._update_k_cache(k_fp8, k_scale, metadata)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata)
 
     # Test custom kernel
     topk_custom = indexer.sparse_attn_indexer(metadata,
@@ -2356,7 +2377,7 @@ def test_indexer_topk_multi_request_with_different_cache(enable_indexer_skip):
             enable_context_mla_with_cached_kv=True,
             enable_indexer_skip=True)
         Indexer.prepare(metadata_skip)
-        indexer._update_k_cache(k_fp8, k_scale, metadata_skip)
+        _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata_skip)
         topk_indices_skip = indexer.sparse_attn_indexer(metadata_skip,
                                                         hidden_states,
                                                         q_fp8,
@@ -2432,6 +2453,228 @@ def test_indexer_topk_multi_request_with_different_cache(enable_indexer_skip):
             f"Custom vs indexer skip differ: avg similarity {avg_similarity:.4f} < 0.95"
 
 
+@pytest.mark.skipif(not has_deep_gemm(), reason="DeepGEMM not available")
+@skip_pre_hopper
+def test_indexer_k_cache_gather_custom_op():
+    """
+    Verify the CUDA gather kernel produces identical results to the Python reference.
+
+    Tests the new indexer_k_cache_gather_op against the original Python-based
+    _gather_k_cache_for_chunk implementation (using _unravel_indices + advanced indexing).
+    """
+    torch.manual_seed(123)
+
+    def _unravel_indices(flat_indices, shape):
+        d3 = shape[3]
+        i3 = flat_indices % d3
+        flat_indices = flat_indices // d3
+        d2 = shape[2]
+        i2 = flat_indices % d2
+        flat_indices = flat_indices // d2
+        d1 = shape[1]
+        i1 = flat_indices % d1
+        flat_indices = flat_indices // d1
+        i0 = flat_indices
+        return i0, i1, i2, i3
+
+    # Test parameters
+    head_dim = 128
+    block_size = 64
+    batch_size = 3
+    num_tokens = 96
+    max_seq_len = 512
+
+    cache_manager, sparse_attn_config = create_dsa_cache_manager(
+        batch_size=batch_size,
+        head_dim=head_dim,
+        tokens_per_block=block_size,
+        max_seq_len=max_seq_len,
+        num_layers=1)
+    indexer = create_indexer(sparse_attn_config, layer_idx=0)
+
+    request_ids = list(range(batch_size))
+    tokens_per_req = [32, 32, 32]
+    cache_manager.add_dummy_requests(request_ids,
+                                     tokens_per_req,
+                                     is_gen=False,
+                                     prepare_resource=True)
+
+    metadata = _create_mock_metadata(
+        request_ids,
+        batch_size,
+        num_contexts=batch_size,
+        num_generations=0,
+        seq_lens=torch.tensor(tokens_per_req, dtype=torch.int32),
+        kv_lens=torch.tensor(tokens_per_req, dtype=torch.int32),
+        num_cached_tokens=[0] * batch_size,
+        cache_manager=cache_manager,
+        num_ctx_tokens=num_tokens,
+        num_tokens=num_tokens,
+    )
+    Indexer.prepare(metadata)
+
+    # Generate test data, scatter to cache
+    k_original = torch.randn((num_tokens, head_dim),
+                             device="cuda",
+                             dtype=torch.bfloat16)
+    k_fp8, k_scale = fp8_utils.fp8_quantize_1x128_sf_transpose(k_original)
+    _update_k_cache_for_test(indexer, k_fp8, k_scale, metadata)
+
+    # Get k_cache and slot mappings
+    k_cache = cache_manager.get_indexer_k_cache_buffers(0)
+    slot_mapping_fp8 = metadata.slot_mapping_fp8[:num_tokens]
+    slot_mapping_scale = metadata.slot_mapping_scale[:num_tokens]
+
+    # ===== CUDA gather kernel =====
+    k_fp8_bytes_cuda, k_scale_bytes_cuda = torch.ops.trtllm.indexer_k_cache_gather_op(
+        k_cache, slot_mapping_fp8, slot_mapping_scale)
+    k_fp8_cuda = k_fp8_bytes_cuda.view(torch.float8_e4m3fn).view(
+        num_tokens, head_dim)
+    k_scale_cuda = k_scale_bytes_cuda.view(torch.float32).view(num_tokens, 1)
+
+    # ===== Python reference gather =====
+    scale_size = 4
+    byte_offsets_fp8 = torch.arange(head_dim,
+                                    device=k_cache.device).unsqueeze(0)
+    gather_indices_fp8 = slot_mapping_fp8.unsqueeze(1) + byte_offsets_fp8
+    gather_indices_fp8 = _unravel_indices(gather_indices_fp8, k_cache.shape)
+    k_fp8_bytes_ref = k_cache[gather_indices_fp8]
+    k_fp8_ref = k_fp8_bytes_ref.view(torch.float8_e4m3fn).view(
+        num_tokens, head_dim)
+
+    byte_offsets_scale = torch.arange(scale_size,
+                                      device=k_cache.device).unsqueeze(0)
+    gather_indices_scale = slot_mapping_scale.unsqueeze(1) + byte_offsets_scale
+    gather_indices_scale = _unravel_indices(gather_indices_scale, k_cache.shape)
+    k_scale_bytes_ref = k_cache[gather_indices_scale]
+    k_scale_ref = k_scale_bytes_ref.view(torch.float32).view(num_tokens, 1)
+
+    # ===== Validate =====
+    # FP8 values: byte-for-byte match
+    assert torch.equal(k_fp8_cuda.view(torch.uint8), k_fp8_ref.view(
+        torch.uint8)), "FP8 gather mismatch between CUDA and Python"
+
+    # Scale values: exact float match
+    assert torch.equal(
+        k_scale_cuda,
+        k_scale_ref), "Scale gather mismatch between CUDA and Python"
+
+    # Also verify roundtrip: gathered values match what was scattered
+    assert torch.equal(
+        k_fp8_cuda.view(torch.uint8), k_fp8.view(torch.uint8)
+    ), "Gathered FP8 values don't match original scattered values"
+
+    assert torch.allclose(
+        k_scale_cuda, k_scale,
+        atol=1e-6), "Gathered scales don't match original scattered scales"
+
+    print(
+        f"PASS: Gather kernel matches Python reference for {num_tokens} tokens")
+
+
+@pytest.mark.skipif(not has_deep_gemm(), reason="DeepGEMM not available")
+@skip_pre_hopper
+def test_fused_cat_fp8_scatter():
+    """
+    Verify the fused cat+fp8+scatter kernel matches sequential execution.
+
+    Tests that fused_cat_fp8_scatter produces:
+    1. Identical FP8 output and scale to fused_cat_fp8
+    2. Identical cache state to sequential fused_cat_fp8 + indexer_k_cache_scatter_op
+    """
+    torch.manual_seed(42)
+
+    head_dim = 128
+    block_size = 64
+    batch_size = 2
+    num_tokens = 64
+    max_seq_len = 512
+    pe_dim = 64
+    nope_dim = 64
+
+    cache_manager, _sparse_attn_config = create_dsa_cache_manager(
+        batch_size=batch_size,
+        head_dim=head_dim,
+        tokens_per_block=block_size,
+        max_seq_len=max_seq_len,
+        num_layers=2)
+
+    request_ids = list(range(batch_size))
+    tokens_per_req = [32, 32]
+    cache_manager.add_dummy_requests(request_ids,
+                                     tokens_per_req,
+                                     is_gen=False,
+                                     prepare_resource=True)
+
+    metadata = _create_mock_metadata(
+        request_ids,
+        batch_size,
+        num_contexts=batch_size,
+        num_generations=0,
+        seq_lens=torch.tensor(tokens_per_req, dtype=torch.int32),
+        kv_lens=torch.tensor(tokens_per_req, dtype=torch.int32),
+        num_cached_tokens=[0] * batch_size,
+        cache_manager=cache_manager,
+        num_ctx_tokens=num_tokens,
+        num_tokens=num_tokens,
+    )
+    Indexer.prepare(metadata)
+
+    # Generate test input (PE and noPE components)
+    k_pe = torch.randn((num_tokens, pe_dim),
+                       device="cuda",
+                       dtype=torch.bfloat16)
+    k_nope = torch.randn((num_tokens, nope_dim),
+                         device="cuda",
+                         dtype=torch.bfloat16)
+
+    slot_mapping_fp8 = metadata.slot_mapping_fp8[:num_tokens]
+    slot_mapping_scale = metadata.slot_mapping_scale[:num_tokens]
+
+    # ===== Path 1: Sequential (fused_cat_fp8 + scatter) =====
+    k_cache_seq = cache_manager.get_indexer_k_cache_buffers(0)
+    k_cache_seq.zero_()
+
+    k_fp8_seq, k_scale_seq = torch.ops.trtllm.fused_cat_fp8(
+        k_pe, k_nope, True)  # use_ue8m0=True
+
+    # indexer_k_cache_scatter_op consumes the scatter inputs in their native
+    # dtypes (FP8 for k_fp8, float32 for k_scale), so pass them through
+    # directly after ensuring the expected 2-D layout.
+    k_fp8_scatter = k_fp8_seq.contiguous().view(num_tokens, head_dim)
+    k_scale_scatter = k_scale_seq.contiguous().view(num_tokens, -1)
+
+    torch.ops.trtllm.indexer_k_cache_scatter_op(k_fp8_scatter, k_scale_scatter,
+                                                k_cache_seq, slot_mapping_fp8,
+                                                slot_mapping_scale, num_tokens)
+    torch.cuda.synchronize()
+
+    # ===== Path 2: Fused (fused_cat_fp8_scatter) =====
+    k_cache_fused = cache_manager.get_indexer_k_cache_buffers(1)
+    k_cache_fused.zero_()
+
+    k_fp8_fused, k_scale_fused = torch.ops.trtllm.fused_cat_fp8_scatter(
+        k_pe, k_nope, True, k_cache_fused, slot_mapping_fp8, slot_mapping_scale)
+    torch.cuda.synchronize()
+
+    # ===== Validate contiguous outputs =====
+    assert torch.equal(k_fp8_fused.view(torch.uint8), k_fp8_seq.view(
+        torch.uint8)), "Fused FP8 output differs from sequential"
+
+    assert torch.equal(
+        k_scale_fused,
+        k_scale_seq), "Fused scale output differs from sequential"
+
+    # ===== Validate cache state =====
+    assert torch.equal(
+        k_cache_fused,
+        k_cache_seq), "Fused cache state differs from sequential scatter"
+
+    print(
+        f"PASS: Fused cat+fp8+scatter matches sequential for {num_tokens} tokens"
+    )
+
+
 class TestPrepareRestoreAttnMetadataForDraftReplay:
     """Tests for prepare_attn_metadata_for_draft_replay and
     restore_attn_metadata_after_draft_replay."""