diff --git a/cpp/tensorrt_llm/kernels/unfusedAttentionKernels.h b/cpp/tensorrt_llm/kernels/unfusedAttentionKernels.h
index 04bcdfbbe2ac..3a757c3263d9 100644
--- a/cpp/tensorrt_llm/kernels/unfusedAttentionKernels.h
+++ b/cpp/tensorrt_llm/kernels/unfusedAttentionKernels.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2023, NVIDIA CORPORATION.  All rights reserved.
+ * Copyright (c) 2020-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.
@@ -418,6 +418,16 @@ template <typename T, typename KVCacheBuffer>
 void invokeDebugSparseKvCacheParams(
     QKVPreprocessingParams<T, KVCacheBuffer> params, int* debug_output, cudaStream_t stream);
 
+//! Compact a uniform group of KVCacheManagerV2 layer pools in one batched launch
+//! (per request and head, moves are ascending and never overtake their sources:
+//! the copy runs in place).
+template <typename T>
+void invokeSparseKvCacheCompactLayers(int64_t const* poolPointers, int32_t const* pageTable, int32_t numLayers,
+    int64_t pageTableRequestStride, int32_t const* sparseKvIndices, int32_t const* sourceLayerIndices,
+    int64_t sourceLayerStride, int64_t sourceHeadStride, int32_t const* sparseKvOffsets,
+    int32_t const* destinationBases, int32_t batchSize, int32_t numKvHeads, int32_t tokensPerBlock, int32_t headDim,
+    cudaStream_t stream);
+
 template <typename T, typename KVCacheBuffer>
 void invokeKvCachePostprocessing(QKVPreprocessingParams<T, KVCacheBuffer> params, cudaStream_t stream)
 {
diff --git a/cpp/tensorrt_llm/kernels/unfusedAttentionKernels/unfusedAttentionKernels_2_template.h b/cpp/tensorrt_llm/kernels/unfusedAttentionKernels/unfusedAttentionKernels_2_template.h
index 560a153ffa70..22e17b438d58 100644
--- a/cpp/tensorrt_llm/kernels/unfusedAttentionKernels/unfusedAttentionKernels_2_template.h
+++ b/cpp/tensorrt_llm/kernels/unfusedAttentionKernels/unfusedAttentionKernels_2_template.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2024, NVIDIA CORPORATION.  All rights reserved.
+ * Copyright (c) 2019-2026, NVIDIA CORPORATION.  All rights reserved.
  * Copyright (c) 2021, NAVER Corp.  Authored by CLOVA.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -29,6 +29,8 @@
 #include "tensorrt_llm/kernels/quantization.cuh"
 #include "tensorrt_llm/kernels/unfusedAttentionKernels.h"
 
+#include <type_traits>
+
 using namespace tensorrt_llm::common;
 
 TRTLLM_NAMESPACE_BEGIN
@@ -1754,6 +1756,213 @@ void invokeUpdateCyclicKvCacheAfterFmha(QKVPreprocessingParams<T, KVCacheBuffer>
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
+#ifdef ENABLE_BF16
+
+// Pipelined bf16 compaction kernels: double-buffered cp.async copies that
+// compact paged KV pools in place through the V2 block-offset table.
+
+namespace compact_detail
+{
+// Vendored cp.async wrappers (xqa's ldgsts.cuh cannot be included here).
+template <uint32_t size>
+__device__ __forceinline__ void copyAsync(void* dst, void const* src, uint32_t srcSize = size)
+{
+    static_assert(size == 16, "only the 16B cp.async variant is vendored");
+    // srcSize == 0 zero-fills shared memory instead of reading global.
+    if (srcSize == 0)
+    {
+        src = nullptr;
+    }
+    asm volatile(
+        "cp.async.cg.shared.global [%0], [%1], 16, %2;\n" ::"l"(__cvta_generic_to_shared(dst)), "l"(src), "r"(srcSize));
+}
+
+__device__ __forceinline__ void commitGroup()
+{
+    asm volatile("cp.async.commit_group;\n");
+}
+
+// Wait until at most InFlightGroups cp.async groups remain in flight.
+template <uint32_t InFlightGroups>
+__device__ __forceinline__ void waitGroup()
+{
+    asm volatile("cp.async.wait_group %0;\n" ::"n"(InFlightGroups));
+}
+} // namespace compact_detail
+
+// One pipeline stage moves a 32-token tile regardless of the page geometry.
+constexpr int32_t kSparseKvCompactTokensPerTile = 32;
+
+//! Launch parameters for the pipelined bf16 fast-path compaction kernels.
+struct SparseKvCacheCompactBf16Params
+{
+    int64_t const* poolPointers;
+    int32_t const* pageTable;
+    int32_t const* sourceIndices;
+    int32_t const* sourceOffsets;
+    int32_t const* sourceLayerIndices;
+    int32_t const* destinationBases;
+    int64_t sourceLayerStride;
+    int64_t sourceHeadStride;
+    int64_t pageTableRequestStride;
+    int32_t numLayers;
+    int32_t batchSize;
+    int32_t numKvHeads;
+    size_t bytesPerKvHalf;
+    size_t bytesPerPage;
+};
+
+//! Double-buffered cp.async pipeline: the next tile streams in while the
+//! current tile drains. One CTA per (layer, KV head, request).
+template <typename T, int32_t HeadDim, int32_t TokensPerBlock>
+__global__ __launch_bounds__(HeadDim * sizeof(T) / sizeof(uint4)
+    * kSparseKvCompactTokensPerTile) void sparseKvCacheCompactV2Bf16PipelineKernel(SparseKvCacheCompactBf16Params
+        params)
+{
+    static_assert(std::is_same_v<T, __nv_bfloat16>);
+    static_assert(HeadDim == 64 || HeadDim == 128);
+    // 128-token pages are the geometry the kernel was written for; 32-token
+    // pages cover the supported production configuration (one tile == one page).
+    static_assert(TokensPerBlock == 32 || TokensPerBlock == 128);
+    constexpr int32_t kVectorsPerHead = HeadDim * sizeof(T) / sizeof(uint4);
+    constexpr int32_t kTokensPerTile = kSparseKvCompactTokensPerTile;
+    constexpr int32_t kVectorsPerTile = kTokensPerTile * kVectorsPerHead;
+    // A buffer holds one K tile plus one V tile; two buffers ping-pong.
+    constexpr int32_t kVectorsPerBuffer = 2 * kVectorsPerTile;
+
+    int32_t const layerIdx = static_cast<int32_t>(blockIdx.x);
+    int32_t const kvHeadIdx = static_cast<int32_t>(blockIdx.y);
+    int32_t const batchIdx = static_cast<int32_t>(blockIdx.z);
+    int32_t const moveBegin = params.sourceOffsets[batchIdx];
+    int32_t const moveEnd = params.sourceOffsets[batchIdx + 1];
+    int32_t const moveCount = moveEnd - moveBegin;
+    if (moveCount <= 0)
+    {
+        return;
+    }
+
+    // Without an explicit layer map, launch layer i reads source plane i.
+    int32_t const sourceLayer = params.sourceLayerIndices == nullptr ? layerIdx : params.sourceLayerIndices[layerIdx];
+    int64_t const sourceMoveBase = static_cast<int64_t>(sourceLayer) * params.sourceLayerStride
+        + static_cast<int64_t>(kvHeadIdx) * params.sourceHeadStride + moveBegin;
+    int32_t const destinationBase = params.destinationBases[batchIdx];
+    auto* const pool = reinterpret_cast<uint8_t*>(static_cast<uintptr_t>(params.poolPointers[layerIdx]));
+    // Block-offset entries decode to a page with >> 1.
+    int32_t const* const pageTable = params.pageTable + static_cast<int64_t>(batchIdx) * params.pageTableRequestStride;
+
+    extern __shared__ uint4 sharedVectors[];
+    int32_t const sharedVector
+        = static_cast<int32_t>(threadIdx.y) * kVectorsPerHead + static_cast<int32_t>(threadIdx.x);
+    int32_t currentRequestMove = static_cast<int32_t>(threadIdx.y);
+    bool currentValid = currentRequestMove < moveCount;
+    int32_t currentSourceToken = currentValid ? params.sourceIndices[sourceMoveBase + currentRequestMove] : -1;
+    uint4* currentSharedK = sharedVectors;
+    uint4* currentSharedV = currentSharedK + kVectorsPerTile;
+
+    // Prologue: explicitly wait for tile 0 and synchronize the CTA before any thread stores it.
+    uint4 const* currentSourceKVector = nullptr;
+    uint4 const* currentSourceVVector = nullptr;
+    if (currentValid)
+    {
+        int32_t const sourcePage = pageTable[currentSourceToken / TokensPerBlock] >> 1;
+        auto* const sourcePageBase = pool + static_cast<size_t>(sourcePage) * params.bytesPerPage;
+        auto const* const sourceK = reinterpret_cast<uint4 const*>(sourcePageBase);
+        auto const* const sourceV = reinterpret_cast<uint4 const*>(sourcePageBase + params.bytesPerKvHalf);
+        int32_t const localVector = (kvHeadIdx * TokensPerBlock + currentSourceToken % TokensPerBlock) * kVectorsPerHead
+            + static_cast<int32_t>(threadIdx.x);
+        currentSourceKVector = &sourceK[localVector];
+        currentSourceVVector = &sourceV[localVector];
+    }
+    uint32_t const currentSourceBytes = currentValid ? sizeof(uint4) : 0U;
+    compact_detail::copyAsync<sizeof(uint4)>(&currentSharedK[sharedVector], currentSourceKVector, currentSourceBytes);
+    compact_detail::copyAsync<sizeof(uint4)>(&currentSharedV[sharedVector], currentSourceVVector, currentSourceBytes);
+    compact_detail::commitGroup();
+    compact_detail::waitGroup<0>();
+    __syncthreads();
+
+    for (int32_t nextTileBegin = kTokensPerTile; nextTileBegin < moveCount; nextTileBegin += kTokensPerTile)
+    {
+        int32_t const nextRequestMove = nextTileBegin + static_cast<int32_t>(threadIdx.y);
+        bool const nextValid = nextRequestMove < moveCount;
+        int32_t const nextSourceToken = nextValid ? params.sourceIndices[sourceMoveBase + nextRequestMove] : -1;
+        int32_t const nextBuffer = (nextTileBegin / kTokensPerTile) & 1;
+        uint4* const nextSharedK = sharedVectors + nextBuffer * kVectorsPerBuffer;
+        uint4* const nextSharedV = nextSharedK + kVectorsPerTile;
+
+        uint4 const* nextSourceKVector = nullptr;
+        uint4 const* nextSourceVVector = nullptr;
+        if (nextValid)
+        {
+            int32_t const sourcePage = pageTable[nextSourceToken / TokensPerBlock] >> 1;
+            auto* const sourcePageBase = pool + static_cast<size_t>(sourcePage) * params.bytesPerPage;
+            auto const* const sourceK = reinterpret_cast<uint4 const*>(sourcePageBase);
+            auto const* const sourceV = reinterpret_cast<uint4 const*>(sourcePageBase + params.bytesPerKvHalf);
+            int32_t const localVector
+                = (kvHeadIdx * TokensPerBlock + nextSourceToken % TokensPerBlock) * kVectorsPerHead
+                + static_cast<int32_t>(threadIdx.x);
+            nextSourceKVector = &sourceK[localVector];
+            nextSourceVVector = &sourceV[localVector];
+        }
+        uint32_t const nextSourceBytes = nextValid ? sizeof(uint4) : 0U;
+        compact_detail::copyAsync<sizeof(uint4)>(&nextSharedK[sharedVector], nextSourceKVector, nextSourceBytes);
+        compact_detail::copyAsync<sizeof(uint4)>(&nextSharedV[sharedVector], nextSourceVVector, nextSourceBytes);
+        compact_detail::commitGroup();
+
+        // In-place safety: sources are strictly increasing with dst(i) <= src(i),
+        // so current stores never alias future prefetch sources.
+        int32_t const destinationToken = destinationBase + currentRequestMove;
+        if (currentValid && currentSourceToken != destinationToken)
+        {
+            int32_t const destinationPage = pageTable[destinationToken / TokensPerBlock] >> 1;
+            auto* const destinationPageBase = pool + static_cast<size_t>(destinationPage) * params.bytesPerPage;
+            auto* const destinationK = reinterpret_cast<uint4*>(destinationPageBase);
+            auto* const destinationV = reinterpret_cast<uint4*>(destinationPageBase + params.bytesPerKvHalf);
+            int32_t const localVector
+                = (kvHeadIdx * TokensPerBlock + destinationToken % TokensPerBlock) * kVectorsPerHead
+                + static_cast<int32_t>(threadIdx.x);
+            destinationK[localVector] = currentSharedK[sharedVector];
+            destinationV[localVector] = currentSharedV[sharedVector];
+        }
+
+        // waitGroup<0> completes the next tile's copies before the buffer swap.
+        compact_detail::waitGroup<0>();
+        __syncthreads();
+        currentRequestMove = nextRequestMove;
+        currentValid = nextValid;
+        currentSourceToken = nextSourceToken;
+        currentSharedK = nextSharedK;
+        currentSharedV = nextSharedV;
+    }
+
+    // Epilogue: the final tile already completed its async wait and CTA barrier.
+    int32_t const destinationToken = destinationBase + currentRequestMove;
+    if (currentValid && currentSourceToken != destinationToken)
+    {
+        int32_t const destinationPage = pageTable[destinationToken / TokensPerBlock] >> 1;
+        auto* const destinationPageBase = pool + static_cast<size_t>(destinationPage) * params.bytesPerPage;
+        auto* const destinationK = reinterpret_cast<uint4*>(destinationPageBase);
+        auto* const destinationV = reinterpret_cast<uint4*>(destinationPageBase + params.bytesPerKvHalf);
+        int32_t const localVector = (kvHeadIdx * TokensPerBlock + destinationToken % TokensPerBlock) * kVectorsPerHead
+            + static_cast<int32_t>(threadIdx.x);
+        destinationK[localVector] = currentSharedK[sharedVector];
+        destinationV[localVector] = currentSharedV[sharedVector];
+    }
+}
+
+template <typename T, int32_t HeadDim, int32_t TokensPerBlock>
+void launchSparseKvCacheCompactV2Bf16Pipeline(SparseKvCacheCompactBf16Params const& params, cudaStream_t stream)
+{
+    constexpr int32_t kVectorsPerHead = HeadDim * sizeof(T) / sizeof(uint4);
+    dim3 const block(kVectorsPerHead, kSparseKvCompactTokensPerTile);
+    dim3 const grid(params.numLayers, params.numKvHeads, params.batchSize);
+    // Two ping-pong buffers of (K + V) tiles; both geometries fit the 48 KiB
+    // per-CTA dynamic shared memory default.
+    size_t const sharedBytes = 4 * kSparseKvCompactTokensPerTile * kVectorsPerHead * sizeof(uint4);
+    sparseKvCacheCompactV2Bf16PipelineKernel<T, HeadDim, TokensPerBlock><<<grid, block, sharedBytes, stream>>>(params);
+}
+
+#endif // ENABLE_BF16
+
 template <typename T, typename TCache, int BLOCK_SIZE, int Dh, typename KVCacheBuffer>
 __global__ __launch_bounds__(BLOCK_SIZE) void updateSparseKvCacheAfterFmha(
     QKVPreprocessingParams<T, KVCacheBuffer> params)
@@ -1876,6 +2085,60 @@ void invokeUpdateSparseKvCacheAfterFmha(QKVPreprocessingParams<T, KVCacheBuffer>
     }
 }
 
+template <typename T>
+void invokeSparseKvCacheCompactLayers(int64_t const* poolPointers, int32_t const* pageTable, int32_t numLayers,
+    int64_t pageTableRequestStride, int32_t const* sparseKvIndices, int32_t const* sourceLayerIndices,
+    int64_t sourceLayerStride, int64_t sourceHeadStride, int32_t const* sparseKvOffsets,
+    int32_t const* destinationBases, int32_t batchSize, int32_t numKvHeads, int32_t tokensPerBlock, int32_t headDim,
+    cudaStream_t stream)
+{
+#ifdef ENABLE_BF16
+    if constexpr (std::is_same_v<T, __nv_bfloat16>)
+    {
+        if ((headDim == 64 || headDim == 128) && (tokensPerBlock == 32 || tokensPerBlock == 128))
+        {
+            SparseKvCacheCompactBf16Params fastParams{};
+            fastParams.poolPointers = poolPointers;
+            fastParams.pageTable = pageTable;
+            fastParams.sourceIndices = sparseKvIndices;
+            fastParams.sourceOffsets = sparseKvOffsets;
+            fastParams.sourceLayerIndices = sourceLayerIndices;
+            fastParams.destinationBases = destinationBases;
+            fastParams.sourceLayerStride = sourceLayerStride;
+            fastParams.sourceHeadStride = sourceHeadStride;
+            fastParams.pageTableRequestStride = pageTableRequestStride;
+            fastParams.numLayers = numLayers;
+            fastParams.batchSize = batchSize;
+            fastParams.numKvHeads = numKvHeads;
+            fastParams.bytesPerKvHalf = static_cast<size_t>(numKvHeads) * tokensPerBlock * headDim * sizeof(T);
+            fastParams.bytesPerPage = 2 * fastParams.bytesPerKvHalf;
+            if (headDim == 64 && tokensPerBlock == 32)
+            {
+                launchSparseKvCacheCompactV2Bf16Pipeline<T, 64, 32>(fastParams, stream);
+            }
+            else if (headDim == 64 && tokensPerBlock == 128)
+            {
+                launchSparseKvCacheCompactV2Bf16Pipeline<T, 64, 128>(fastParams, stream);
+            }
+            else if (headDim == 128 && tokensPerBlock == 32)
+            {
+                launchSparseKvCacheCompactV2Bf16Pipeline<T, 128, 32>(fastParams, stream);
+            }
+            else
+            {
+                launchSparseKvCacheCompactV2Bf16Pipeline<T, 128, 128>(fastParams, stream);
+            }
+            return;
+        }
+    }
+#endif // ENABLE_BF16
+
+    TLLM_CHECK_WITH_INFO(false,
+        "Sparse KV compaction ships only the pipelined bf16 kernels (head size 64/128, page size 32/128 "
+        "tokens); got element size %zu, head size %d, %d tokens per page",
+        sizeof(T), headDim, tokensPerBlock);
+}
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 #define INSTANTIATE_ATTENTION_INPUT_PROCESSING(T, TCache, KVCacheBuffer)                                               \
@@ -1891,6 +2154,11 @@ void invokeUpdateSparseKvCacheAfterFmha(QKVPreprocessingParams<T, KVCacheBuffer>
         QKVPreprocessingParams<T, KVCacheBuffer> params, cudaStream_t stream);                                         \
     ////////////////////////////////////////////////////////////////////////////////////////////////////
 
+#define INSTANTIATE_SPARSE_KV_CACHE_COMPACT_LAYERS(T)                                                                  \
+    template void invokeSparseKvCacheCompactLayers<T>(int64_t const*, int32_t const*, int32_t, int64_t,                \
+        int32_t const*, int32_t const*, int64_t, int64_t, int32_t const*, int32_t const*, int32_t, int32_t, int32_t,   \
+        int32_t, cudaStream_t);
+
 } // namespace kernels
 
 TRTLLM_NAMESPACE_END
diff --git a/cpp/tensorrt_llm/thop/CMakeLists.txt b/cpp/tensorrt_llm/thop/CMakeLists.txt
index b95426f580fe..273ffa838a21 100644
--- a/cpp/tensorrt_llm/thop/CMakeLists.txt
+++ b/cpp/tensorrt_llm/thop/CMakeLists.txt
@@ -113,6 +113,7 @@ add_library(
   IndexerKCacheGatherOp.cpp
   IndexerKCacheScatterOp.cpp
   IndexerTopKOp.cpp
+  sparseKvCacheCompactOp.cpp
   mlaRopeInplaceOp.cpp
   ncclCommunicatorOp.cpp
   allocateOutput.cpp
diff --git a/cpp/tensorrt_llm/thop/sparseKvCacheCompactOp.cpp b/cpp/tensorrt_llm/thop/sparseKvCacheCompactOp.cpp
new file mode 100644
index 000000000000..b0446b0ef574
--- /dev/null
+++ b/cpp/tensorrt_llm/thop/sparseKvCacheCompactOp.cpp
@@ -0,0 +1,158 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES.
+ * All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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/common/opUtils.h"
+#include "tensorrt_llm/kernels/unfusedAttentionKernels.h"
+#include "tensorrt_llm/runtime/torchUtils.h"
+
+#include <limits>
+#include <optional>
+#include <vector>
+
+namespace th = torch;
+namespace tk = tensorrt_llm::kernels;
+
+TRTLLM_NAMESPACE_BEGIN
+
+namespace torch_ext
+{
+
+//! Compact one uniform group of KVCacheManagerV2 HND layer pools in one batched
+//! launch (per request and KV head, moves are ascending and never overtake their
+//! sources: the copy runs in place).
+void sparseKvCacheCompactLayers(std::vector<th::Tensor> const& pools, th::Tensor const& poolPointers,
+    th::Tensor const& pageTable, th::Tensor const& sourceIndices, th::Tensor const& sourceOffsets,
+    th::Tensor const& destinationBases, std::optional<th::Tensor> const& sourceLayerIndices)
+{
+    TORCH_CHECK(!pools.empty(), "sparse_kv_cache_compact_layers: pools must be non-empty");
+
+    auto const& firstPool = pools.front();
+    TORCH_CHECK(firstPool.is_cuda() && firstPool.dim() == 5 && firstPool.size(1) == 2 && firstPool.is_contiguous(),
+        "sparse_kv_cache_compact_layers: pools must be contiguous CUDA "
+        "[pages, 2, kv_heads, tokens_per_block, head_dim] tensors");
+    TORCH_CHECK(pageTable.is_cuda() && pageTable.dim() == 2 && pageTable.scalar_type() == th::kInt32
+            && pageTable.stride(1) == 1,
+        "sparse_kv_cache_compact_layers: K block offsets must be CUDA int32 "
+        "[batch, max_pages] tensors with a contiguous block dimension");
+
+    auto const device = firstPool.get_device();
+    auto const dtype = firstPool.scalar_type();
+    auto const numLayers = static_cast<int32_t>(pools.size());
+    auto const numKvHeads = static_cast<int32_t>(firstPool.size(2));
+    auto const tokensPerBlock = static_cast<int32_t>(firstPool.size(3));
+    auto const headDim = static_cast<int32_t>(firstPool.size(4));
+    auto const batchSize = static_cast<int32_t>(pageTable.size(0));
+    auto const pageTableRequestStride = pageTable.stride(0);
+
+    // Layer 0 defined the reference geometry in the firstPool checks above.
+    for (int32_t layer = 1; layer < numLayers; ++layer)
+    {
+        auto const& pool = pools[layer];
+        TORCH_CHECK(pool.is_cuda() && pool.get_device() == device && pool.scalar_type() == dtype && pool.dim() == 5
+                && pool.size(1) == 2 && pool.is_contiguous(),
+            "sparse_kv_cache_compact_layers: all pools must have one device, dtype, layout, and contiguous storage");
+        TORCH_CHECK(pool.size(2) == numKvHeads && pool.size(3) == tokensPerBlock && pool.size(4) == headDim,
+            "sparse_kv_cache_compact_layers: all pools must share KV-head, block, and head-dimension geometry");
+    }
+    TORCH_CHECK(
+        pageTable.get_device() == device, "sparse_kv_cache_compact_layers: block offsets must be on the pool device");
+
+    TORCH_CHECK(poolPointers.is_cuda() && poolPointers.get_device() == device
+            && poolPointers.scalar_type() == th::kInt64 && poolPointers.dim() == 1 && poolPointers.size(0) == numLayers
+            && poolPointers.is_contiguous(),
+        "sparse_kv_cache_compact_layers: pool_pointers must be contiguous CUDA int64 [num_layers]");
+
+    TORCH_CHECK(sourceIndices.is_cuda() && sourceIndices.get_device() == device
+            && sourceIndices.scalar_type() == th::kInt32 && sourceIndices.is_contiguous()
+            && (sourceIndices.dim() == 2 || sourceIndices.dim() == 3),
+        "sparse_kv_cache_compact_layers: source_indices must be contiguous CUDA int32 "
+        "[kv_heads, total] or [source_layers, kv_heads, total]");
+    int64_t sourceLayerStride = 0;
+    int32_t const* sourceLayerPtr = nullptr;
+    if (sourceIndices.dim() == 2)
+    {
+        TORCH_CHECK(sourceIndices.size(0) == numKvHeads,
+            "sparse_kv_cache_compact_layers: source_indices KV-head dimension mismatch");
+        TORCH_CHECK(!sourceLayerIndices.has_value(),
+            "sparse_kv_cache_compact_layers: source_layer_indices require 3-D per-layer source_indices");
+    }
+    else
+    {
+        TORCH_CHECK(sourceIndices.size(0) > 0 && sourceIndices.size(1) == numKvHeads,
+            "sparse_kv_cache_compact_layers: per-layer source_indices geometry mismatch");
+        TORCH_CHECK(sourceLayerIndices.has_value(),
+            "sparse_kv_cache_compact_layers: per-layer source_indices require source_layer_indices");
+        sourceLayerStride = sourceIndices.stride(0);
+    }
+    if (sourceLayerIndices.has_value())
+    {
+        auto const& layerIndices = *sourceLayerIndices;
+        TORCH_CHECK(layerIndices.is_cuda() && layerIndices.get_device() == device
+                && layerIndices.scalar_type() == th::kInt32 && layerIndices.is_contiguous() && layerIndices.dim() == 1
+                && layerIndices.size(0) == numLayers,
+            "sparse_kv_cache_compact_layers: source_layer_indices must be contiguous CUDA int32 [num_layers]");
+        sourceLayerPtr = layerIndices.data_ptr<int32_t>();
+    }
+
+    // source_offsets carve each request's move range; device-resident, the kernel trusts them.
+    TORCH_CHECK(sourceOffsets.is_cuda() && sourceOffsets.get_device() == device
+            && sourceOffsets.scalar_type() == th::kInt32 && sourceOffsets.is_contiguous() && sourceOffsets.dim() == 1
+            && sourceOffsets.size(0) == batchSize + 1,
+        "sparse_kv_cache_compact_layers: source_offsets must be contiguous CUDA int32 [batch + 1]");
+    // Per-request landing positions: one launch covers a cohort with mixed
+    // pinned-prompt lengths. Values live on device; the kernel trusts them.
+    TORCH_CHECK(destinationBases.is_cuda() && destinationBases.get_device() == device
+            && destinationBases.scalar_type() == th::kInt32 && destinationBases.dim() == 1
+            && destinationBases.size(0) == batchSize && destinationBases.is_contiguous(),
+        "sparse_kv_cache_compact_layers: destination_bases must be contiguous CUDA int32 [batch]");
+
+    auto const stream = at::cuda::getCurrentCUDAStream(device);
+    auto const* bases = destinationBases.data_ptr<int32_t>();
+    auto const sourceHeadStride = sourceIndices.size(-1);
+    if (dtype == th::kBFloat16)
+    {
+        tk::invokeSparseKvCacheCompactLayers<__nv_bfloat16>(poolPointers.data_ptr<int64_t>(),
+            pageTable.data_ptr<int32_t>(), numLayers, pageTableRequestStride, sourceIndices.data_ptr<int32_t>(),
+            sourceLayerPtr, sourceLayerStride, sourceHeadStride, sourceOffsets.data_ptr<int32_t>(), bases, batchSize,
+            numKvHeads, tokensPerBlock, headDim, stream);
+    }
+    else
+    {
+        TORCH_CHECK(false,
+            "sparse_kv_cache_compact_layers ships only the pipelined bf16 kernels (head size 64/128, page size "
+            "32/128 tokens); got pool dtype ",
+            dtype);
+    }
+}
+
+} // namespace torch_ext
+
+TRTLLM_NAMESPACE_END
+
+TORCH_LIBRARY_FRAGMENT(trtllm, m)
+{
+    m.def(
+        "sparse_kv_cache_compact_layers(Tensor(a!)[] pools, Tensor pool_pointers, Tensor page_table, Tensor "
+        "source_indices, Tensor source_offsets, Tensor destination_bases, "
+        "Tensor? source_layer_indices=None) -> ()");
+}
+
+TORCH_LIBRARY_IMPL(trtllm, CUDA, m)
+{
+    m.impl("sparse_kv_cache_compact_layers", &tensorrt_llm::torch_ext::sparseKvCacheCompactLayers);
+}
diff --git a/tensorrt_llm/_torch/kv_cache_compression/compaction.py b/tensorrt_llm/_torch/kv_cache_compression/compaction.py
new file mode 100644
index 000000000000..5d7ef1765262
--- /dev/null
+++ b/tensorrt_llm/_torch/kv_cache_compression/compaction.py
@@ -0,0 +1,269 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# 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.
+
+"""Batched physical KV-cache compaction: an algorithm-neutral mover.
+
+``build_compaction_params`` pre-binds one cache's launch parameters;
+each round the caller writes its keep decision into the agreed rows and
+``compact`` packs every cache's move sources and fires its native launches.
+"""
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Tuple
+
+import torch
+import triton
+import triton.language as tl
+
+
+@triton.jit
+def _pack_move_sources_kernel(
+    kept_ordinal_rows,
+    valid_seq_lens,
+    dense_move_offsets,
+    dense_move_indices,
+    swa_move_offsets,
+    swa_move_indices,
+    KEEP_COUNT: tl.constexpr,
+    DECISION_ROWS: tl.constexpr,
+    MOVE_CAPACITY: tl.constexpr,
+    NUM_KV_HEADS: tl.constexpr,
+    PER_LAYER: tl.constexpr,
+    DENSE_TOTAL: tl.constexpr,
+    SWA_TOTAL: tl.constexpr,
+    SWA_WINDOW: tl.constexpr,
+    BLOCK: tl.constexpr = 256,
+):
+    """Pack one decision row into one family's move sources: dense rows emit the
+    kept tokens then the protected tail; SWA rows emit the latest window
+    (ascending order: the native copy moves in place)."""
+    BROADCAST: tl.constexpr = DECISION_ROWS == 1
+    HAS_SWA: tl.constexpr = SWA_TOTAL > 0
+    request = tl.program_id(0)
+    decision_row = tl.program_id(1)
+    row = request * DECISION_ROWS + decision_row
+    kept_row = kept_ordinal_rows + row * KEEP_COUNT
+    dense_begin = tl.load(dense_move_offsets + request)
+    dense_end = tl.load(dense_move_offsets + request + 1)
+    dense_count = dense_end - dense_begin
+    valid_len = tl.load(valid_seq_lens + request)
+    if HAS_SWA:
+        swa_begin = tl.load(swa_move_offsets + request)
+        swa_end = tl.load(swa_move_offsets + request + 1)
+        swa_count = swa_end - swa_begin
+    for move_start in tl.static_range(0, MOVE_CAPACITY, BLOCK):
+        move = move_start + tl.arange(0, BLOCK)
+        kept = tl.load(
+            kept_row + move,
+            mask=move < KEEP_COUNT,
+            other=0,
+        )
+        dense_source = tl.where(move < KEEP_COUNT, kept, valid_len + move - KEEP_COUNT)
+        if BROADCAST:
+            # The one decision row per request feeds every KV head's packed row.
+            for head in tl.static_range(0, NUM_KV_HEADS):
+                tl.store(
+                    dense_move_indices + head * DENSE_TOTAL + dense_begin.to(tl.int64) + move,
+                    dense_source,
+                    mask=move < dense_count,
+                )
+        else:
+            dense_output = decision_row.to(tl.int64) * DENSE_TOTAL + dense_begin.to(tl.int64) + move
+            tl.store(dense_move_indices + dense_output, dense_source, mask=move < dense_count)
+        if HAS_SWA:
+            swa_source = valid_len - SWA_WINDOW + move
+            if BROADCAST:
+                for head in tl.static_range(0, NUM_KV_HEADS):
+                    tl.store(
+                        swa_move_indices + head * SWA_TOTAL + swa_begin.to(tl.int64) + move,
+                        swa_source,
+                        mask=move < swa_count,
+                    )
+            else:
+                swa_mask = move < swa_count
+                if PER_LAYER:
+                    # SWA has one shared row per head; the first layer's decision rows write it.
+                    swa_mask = swa_mask & (decision_row < NUM_KV_HEADS)
+                head = decision_row % NUM_KV_HEADS
+                swa_output = head.to(tl.int64) * SWA_TOTAL + swa_begin.to(tl.int64) + move
+                tl.store(
+                    swa_move_indices + swa_output,
+                    swa_source,
+                    mask=swa_mask,
+                )
+
+
+@dataclass
+class CompactionParams:
+    decision_rows: int = 0
+    pack_args: Tuple[Optional[torch.Tensor], ...] = ()
+    pack_constexprs: Dict[str, object] = field(default_factory=dict)
+    compact_args: List[Tuple[object, ...]] = field(default_factory=list)
+
+
+def build_compaction_params(
+    layout: Dict[str, object],
+    *,
+    block_offsets: torch.Tensor,
+    kept_ordinals: torch.Tensor,
+    source_lengths: torch.Tensor,
+    dense_destination_bases: torch.Tensor,
+    dense_move_offsets: torch.Tensor,
+    protected_tail_capacity: int,
+    swa_move_offsets: Optional[torch.Tensor] = None,
+    swa_destination_bases: Optional[torch.Tensor] = None,
+) -> CompactionParams:
+    """Pre-bind one compacted cache's launch parameters; only :func:`compact` reads them."""
+    layer_pools = layout["layer_pools"]
+    dense_layers = tuple(int(layer) for layer in layout["dense_layers"])
+    swa_layers = tuple(int(layer) for layer in layout["swa_layers"])
+    layer_pool_ids = tuple(int(pool_id) for pool_id in layout["layer_pool_ids"])
+    kv_block_offsets = block_offsets
+    kept_ordinal_rows = kept_ordinals
+    valid_seq_lens = source_lengths
+    token_starts = dense_destination_bases
+    protected_tail_capacity = int(protected_tail_capacity)
+
+    params = CompactionParams()
+    first_pool = layer_pools[dense_layers[0]]
+    device = first_pool.device
+    max_requests = int(valid_seq_lens.shape[0])
+    keep_count = int(kept_ordinal_rows.shape[1])
+    params.decision_rows = int(kept_ordinal_rows.shape[0]) // max_requests
+    # Pool shape [pages, K/V, heads, tokens, dim].
+    num_kv_heads = int(first_pool.shape[2])
+    per_layer_sources = (
+        len(dense_layers) > 1 and params.decision_rows == len(dense_layers) * num_kv_heads
+    )
+    dense_index_prefix = (len(dense_layers), num_kv_heads) if per_layer_sources else (num_kv_heads,)
+    dense_move_indices = torch.empty(
+        (*dense_index_prefix, (keep_count + protected_tail_capacity) * max_requests),
+        dtype=torch.int32,
+        device=device,
+    )
+    dense_entries = [
+        (
+            layer,
+            layer_pools[layer],
+            kv_block_offsets[layer_pool_ids[layer], :max_requests, 0],
+        )
+        for layer in dense_layers
+    ]
+    dense_slots = (
+        {layer: slot for slot, layer in enumerate(dense_layers)} if per_layer_sources else None
+    )
+
+    swa_move_indices = None
+    swa_window = 0
+    # One move group per family axis (dense / SWA): the layers + the tensors driving their moves.
+    move_groups = [
+        (dense_entries, dense_move_indices, dense_move_offsets, token_starts, dense_slots),
+    ]
+    if swa_layers:
+        swa_window = int(layout["swa_window"])
+        swa_move_indices = torch.empty(
+            (num_kv_heads, (swa_window + protected_tail_capacity) * max_requests),
+            dtype=torch.int32,
+            device=device,
+        )
+        # SWA layers stage against their own page-table slots.
+        swa_entries = [
+            (
+                layer,
+                layer_pools[layer],
+                kv_block_offsets[layer_pool_ids[layer], :max_requests, 0],
+            )
+            for layer in swa_layers
+        ]
+        move_groups.append(
+            (swa_entries, swa_move_indices, swa_move_offsets, swa_destination_bases, None)
+        )
+
+    # Widest per-request move count any staged offsets may express.
+    move_capacity = keep_count + protected_tail_capacity
+    if swa_layers:
+        move_capacity = max(move_capacity, swa_window + protected_tail_capacity)
+
+    params.pack_args = (
+        kept_ordinal_rows,
+        valid_seq_lens,
+        dense_move_offsets,
+        dense_move_indices,
+        swa_move_offsets,
+        swa_move_indices,
+    )
+    params.pack_constexprs = dict(
+        KEEP_COUNT=keep_count,
+        DECISION_ROWS=params.decision_rows,
+        MOVE_CAPACITY=move_capacity,
+        NUM_KV_HEADS=num_kv_heads,
+        PER_LAYER=per_layer_sources,
+        DENSE_TOTAL=int(dense_move_indices.shape[-1]),
+        SWA_TOTAL=int(swa_move_indices.shape[-1]) if swa_move_indices is not None else 0,
+        SWA_WINDOW=swa_window,
+    )
+    for entries, move_indices, move_offsets, destination_bases, slots in move_groups:
+        grouped = {}
+        for layer, pool, page_table in entries:
+            key = (
+                layer_pool_ids[layer],
+                str(pool.dtype),
+                str(pool.device),
+                tuple(int(value) for value in pool.shape[1:]),
+                tuple(int(value) for value in page_table.shape),
+            )
+            grouped.setdefault(key, []).append((layer, pool, page_table))
+        for group_entries in grouped.values():
+            layers = tuple(entry[0] for entry in group_entries)
+            pools = list(entry[1] for entry in group_entries)
+            source_layer_indices = None
+            if slots is not None:
+                source_layer_indices = torch.tensor(
+                    [slots[layer] for layer in layers],
+                    dtype=torch.int32,
+                    device=device,
+                )
+            params.compact_args.append(
+                (
+                    pools,
+                    torch.tensor(
+                        [pool.data_ptr() for pool in pools],
+                        dtype=torch.int64,
+                        device=device,
+                    ),
+                    group_entries[0][2],
+                    move_indices,
+                    move_offsets,
+                    destination_bases,
+                    source_layer_indices,
+                )
+            )
+
+    return params
+
+
+def compact(
+    params: Tuple[CompactionParams, ...],
+    request_count: int,
+) -> None:
+    """Pack each cache's move sources and fire its native compacts, in order
+    (pure mover: the caller owns the decision rows and the round's completion ordering)."""
+    # One launch per (cache, pool group); each launch covers every layer in the group.
+    for cache_params in params:
+        _pack_move_sources_kernel[(request_count, cache_params.decision_rows)](
+            *cache_params.pack_args, **cache_params.pack_constexprs
+        )
+        for args in cache_params.compact_args:
+            torch.ops.trtllm.sparse_kv_cache_compact_layers(*args)
diff --git a/tensorrt_llm/_torch/pyexecutor/_util.py b/tensorrt_llm/_torch/pyexecutor/_util.py
index 0149ba7ae03a..9510b5612104 100644
--- a/tensorrt_llm/_torch/pyexecutor/_util.py
+++ b/tensorrt_llm/_torch/pyexecutor/_util.py
@@ -62,7 +62,7 @@
                                   use_py_mamba_cache_manager)
 from .model_engine import PyTorchModelEngine
 from .py_executor import PyExecutor
-from .resource_manager import (BaseKVCacheCompressionManager, KVCacheManager,
+from .resource_manager import (KVCacheCompressionManager, KVCacheManager,
                                PeftCacheManager, ResourceManager,
                                ResourceManagerType)
 from .sampler import (EarlyStopSampler, EarlyStopWithMMResult, TorchSampler,
@@ -2134,7 +2134,7 @@ def create_kv_cache_compression_manager(
     config: KvCacheCompressionConfig,
     kv_cache_manager: KVCacheManagerV2,
     draft_kv_cache_manager: Optional[KVCacheManagerV2] = None,
-) -> Optional[BaseKVCacheCompressionManager]:
+) -> Optional[KVCacheCompressionManager]:
     """Build the KV-cache compression manager for ``config.algorithm``, or return
     None if no algorithm matches.
 
diff --git a/tensorrt_llm/_torch/pyexecutor/model_engine.py b/tensorrt_llm/_torch/pyexecutor/model_engine.py
index 50aa317e9e43..3795c479bf0d 100644
--- a/tensorrt_llm/_torch/pyexecutor/model_engine.py
+++ b/tensorrt_llm/_torch/pyexecutor/model_engine.py
@@ -4940,7 +4940,9 @@ def previous_seq_slots_device():
                     and not _has_any_multimodal_request
                     and not multimodal_params_list and not lora_params
                     and attn_metadata.padded_num_tokens is None
-                    and self._get_position_id_offset() == 0):
+                    and self._get_position_id_offset() == 0
+                    and not getattr(kv_cache_manager,
+                                    "kv_compression_manages_history", False)):
                 self._steady_gen_positions_pinned[:_n_gen].copy_(
                     torch.as_tensor(num_cached_tokens_snapshot,
                                     dtype=torch.int))
diff --git a/tensorrt_llm/_torch/pyexecutor/resource_manager.py b/tensorrt_llm/_torch/pyexecutor/resource_manager.py
index e42c8e249dc6..996e7b45f5d9 100644
--- a/tensorrt_llm/_torch/pyexecutor/resource_manager.py
+++ b/tensorrt_llm/_torch/pyexecutor/resource_manager.py
@@ -2415,7 +2415,7 @@ def _free_blocks(self, block_list: list):
 # --------------------------------------------------------------------- #
 
 
-class BaseKVCacheCompressionManager(BaseResourceManager):
+class KVCacheCompressionManager(BaseResourceManager):
     """Framework-level base class for all KV-cache compression managers.
 
     Inherits :class:`BaseResourceManager` so PyExecutor's main loop
diff --git a/tests/unittest/_torch/executor/test_kv_cache_compression_manager.py b/tests/unittest/_torch/executor/test_kv_cache_compression_manager.py
index 5c6bed5af1b6..7c11357c3210 100644
--- a/tests/unittest/_torch/executor/test_kv_cache_compression_manager.py
+++ b/tests/unittest/_torch/executor/test_kv_cache_compression_manager.py
@@ -2,11 +2,11 @@
 # Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 
 """Unit tests for the KV-cache compression manager framework
-(``BaseKVCacheCompressionManager`` in ``resource_manager.py``) — the
+(``KVCacheCompressionManager`` in ``resource_manager.py``) — the
 ``BaseResourceManager``-based single-manager design.
 
 Covers:
-- :class:`BaseKVCacheCompressionManager` contract: the four lifecycle hooks
+- :class:`KVCacheCompressionManager` contract: the four lifecycle hooks
   default to no-op, zero resource counts, and it inherits
   :class:`BaseResourceManager` (so PyExecutor auto-drives it once registered).
 - The resource-manager API -> lifecycle-hook translation, gated on PyExecutor's
@@ -31,8 +31,8 @@
 from tensorrt_llm._torch.pyexecutor import _util as util_mod
 from tensorrt_llm._torch.pyexecutor._util import create_kv_cache_compression_manager
 from tensorrt_llm._torch.pyexecutor.resource_manager import (
-    BaseKVCacheCompressionManager,
     BaseResourceManager,
+    KVCacheCompressionManager,
     ResourceManager,
     ResourceManagerType,
 )
@@ -55,7 +55,7 @@ def _record(self, hook_name: str):
         self._record_list.append(f"{self._name}:{hook_name}")
 
 
-class _MockCompressionManager(_RecordingMixin, BaseKVCacheCompressionManager):
+class _MockCompressionManager(_RecordingMixin, KVCacheCompressionManager):
     """Mock manager that records the four lifecycle hooks."""
 
     def on_request_init(self, request):
@@ -71,7 +71,7 @@ def on_request_finish(self, request):
         self._record("on_request_finish")
 
 
-class _LengthAdjustingCompressionManager(BaseKVCacheCompressionManager):
+class _LengthAdjustingCompressionManager(KVCacheCompressionManager):
     adjusts_generation_kv_length: ClassVar[bool] = True
 
 
@@ -108,17 +108,17 @@ def _batch(context=(), generation=(), last_chunk=()):
 
 
 # ---------------------------------------------------------------------- #
-# 1. BaseKVCacheCompressionManager contract                               #
+# 1. KVCacheCompressionManager contract                                   #
 # ---------------------------------------------------------------------- #
 
 
 class TestBaseABC:
     def test_inherits_base_resource_manager(self):
         # So PyExecutor's main loop auto-invokes prepare/update/free_resources.
-        assert issubclass(BaseKVCacheCompressionManager, BaseResourceManager)
+        assert issubclass(KVCacheCompressionManager, BaseResourceManager)
 
     def test_four_hooks_default_noop(self, fake_kv_cache_manager):
-        m = BaseKVCacheCompressionManager(fake_kv_cache_manager)
+        m = KVCacheCompressionManager(fake_kv_cache_manager)
         assert m.on_request_init(MagicMock()) is None
         assert m.on_context_step_end([MagicMock()]) is None
         assert m.on_generation_step_begin(MagicMock()) is None
@@ -128,12 +128,12 @@ def test_four_hooks_default_noop(self, fake_kv_cache_manager):
     def test_hooks_accept_extra_kwargs(self, fake_kv_cache_manager):
         # **kwargs lets the framework pass new args later without breaking
         # existing overrides.
-        m = BaseKVCacheCompressionManager(fake_kv_cache_manager)
+        m = KVCacheCompressionManager(fake_kv_cache_manager)
         assert m.on_request_init(MagicMock(), future_arg=1) is None
         assert m.on_generation_step_end(MagicMock(), future_arg=1) is None
 
     def test_resource_counts_are_zero(self, fake_kv_cache_manager):
-        m = BaseKVCacheCompressionManager(fake_kv_cache_manager)
+        m = KVCacheCompressionManager(fake_kv_cache_manager)
         # The manager owns no physical resources (the V2 cache manager does),
         # so it must not gate the scheduler.
         assert m.get_max_resource_count() == 0
@@ -155,9 +155,9 @@ def test_length_adjustment_marks_target_and_draft_v2(self):
 
     def test_rejects_non_v2_ownership(self):
         with pytest.raises(TypeError, match="requires KVCacheManagerV2"):
-            BaseKVCacheCompressionManager(MagicMock())
+            KVCacheCompressionManager(MagicMock())
         with pytest.raises(TypeError, match="requires KVCacheManagerV2"):
-            BaseKVCacheCompressionManager(_v2_manager(is_draft=False), MagicMock())
+            KVCacheCompressionManager(_v2_manager(is_draft=False), MagicMock())
 
     def test_request_field_defaults_to_zero(self):
         """LlmRequest carries the compression count (the manager's only
@@ -300,7 +300,7 @@ def test_eviction_method_predicate_defaults_false(self):
 
         config = KvCacheCompressionConfig(algorithm="offload")
         assert config.kv_cache_compression_mode.is_eviction_method() is False
-        m = BaseKVCacheCompressionManager(_v2_manager(is_draft=False))
+        m = KVCacheCompressionManager(_v2_manager(is_draft=False))
         assert not hasattr(m, "spec_config")
 
     def test_spec_gate_only_restricts_eviction_methods(self):
@@ -326,7 +326,7 @@ def test_names_importable_from_canonical_modules(self):
 
         # Base class stays in resource_manager (it IS a resource manager); the
         # factory lives in _util next to _create_kv_cache_manager.
-        assert hasattr(resource_manager, "BaseKVCacheCompressionManager")
+        assert hasattr(resource_manager, "KVCacheCompressionManager")
         assert hasattr(_util, "create_kv_cache_compression_manager")
 
     def test_names_not_in_sparse_module(self):
@@ -334,7 +334,7 @@ def test_names_not_in_sparse_module(self):
         # sparse-attention backend); the sparse package no longer exports it.
         from tensorrt_llm._torch.attention_backend import sparse
 
-        assert not hasattr(sparse, "BaseKVCacheCompressionManager")
+        assert not hasattr(sparse, "KVCacheCompressionManager")
         assert not hasattr(sparse, "create_kv_cache_compression_manager")
 
 
@@ -354,7 +354,7 @@ def _mgr(self, enable_block_reuse):
 
     def test_raises_when_reuse_on(self):
         with pytest.raises(ValueError, match="block reuse"):
-            BaseKVCacheCompressionManager(self._mgr(enable_block_reuse=True))
+            KVCacheCompressionManager(self._mgr(enable_block_reuse=True))
 
     def test_ok_when_reuse_off(self):
-        BaseKVCacheCompressionManager(self._mgr(enable_block_reuse=False))  # no raise
+        KVCacheCompressionManager(self._mgr(enable_block_reuse=False))  # no raise
diff --git a/tests/unittest/_torch/kv_cache_compression/conftest.py b/tests/unittest/_torch/kv_cache_compression/conftest.py
new file mode 100644
index 000000000000..4e475f35a674
--- /dev/null
+++ b/tests/unittest/_torch/kv_cache_compression/conftest.py
@@ -0,0 +1,208 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared harness for the KV-cache compaction tests."""
+
+import torch
+
+
+def encode_block_offsets(page_ids: torch.Tensor) -> torch.Tensor:
+    """Native V2 [pool, request, K/V, block] layout: K = 2*page, V = K+1."""
+    if page_ids.ndim == 2:
+        page_ids = page_ids.unsqueeze(0)
+    encoded = torch.empty(
+        page_ids.shape[0],
+        page_ids.shape[1],
+        2,
+        page_ids.shape[2],
+        dtype=torch.int32,
+        device=page_ids.device,
+    )
+    encoded[:, :, 0] = page_ids.to(torch.int32) * 2
+    encoded[:, :, 1] = encoded[:, :, 0] + 1
+    return encoded
+
+
+def _write_move_offsets(compaction, offsets, moves_per_request):
+    cumulative = [0]
+    for count in moves_per_request:
+        cumulative.append(cumulative[-1] + count)
+    # Rows past the cohort are padding and contribute no moves.
+    cumulative.extend(cumulative[-1:] * (compaction["request_count"] - len(moves_per_request)))
+    offsets.copy_(torch.tensor(cumulative, dtype=torch.int32), non_blocking=True)
+
+
+def set_protected_tails(compaction, tail_lengths, draft_tail_lengths=None):
+    """Load per-request protected tails into the caller-owned move offsets."""
+    if len(tail_lengths) > compaction["request_count"]:
+        raise ValueError("the cohort exceeds the compaction request capacity")
+    if any(tail < 0 or tail > compaction["protected_tail_capacity"] for tail in tail_lengths):
+        raise ValueError("a protected tail exceeds the configured capacity")
+    _write_move_offsets(
+        compaction,
+        compaction["dense_move_offsets"],
+        [compaction["decode_keep_count"] + int(tail) for tail in tail_lengths],
+    )
+    if compaction["has_swa"]:
+        _write_move_offsets(
+            compaction,
+            compaction["swa_move_offsets"],
+            [compaction["swa_window"] + int(tail) for tail in tail_lengths],
+        )
+    if compaction["draft_move_offsets"] is not None:
+        if draft_tail_lengths is None:
+            draft_tail_lengths = [0] * len(tail_lengths)
+        if len(draft_tail_lengths) != len(tail_lengths):
+            raise ValueError("draft protected tails must match the cohort")
+        if any(
+            tail < 0 or tail > compaction["draft_protected_tail_capacity"]
+            for tail in draft_tail_lengths
+        ):
+            raise ValueError("a draft protected tail exceeds the configured capacity")
+        _write_move_offsets(
+            compaction,
+            compaction["draft_move_offsets"],
+            [compaction["decode_keep_count"] + int(tail) for tail in draft_tail_lengths],
+        )
+
+
+def make_ramp_pools(
+    count,
+    *,
+    num_kv_heads=2,
+    pages=6,
+    tokens_per_block=32,
+    head_dim=64,
+    layer_stride=37,
+    base=0,
+    device=None,
+):
+    """bf16 pools with a shifted ``arange % 251`` ramp: every wrong move
+    lands on a different byte pattern (supported geometry defaults)."""
+    return [
+        (
+            (
+                torch.arange(
+                    pages * 2 * num_kv_heads * tokens_per_block * head_dim,
+                    dtype=torch.int32,
+                    device=device,
+                )
+                + base
+                + layer * layer_stride
+            )
+            % 251
+        )
+        .view(pages, 2, num_kv_heads, tokens_per_block, head_dim)
+        .to(torch.bfloat16)
+        for layer in range(count)
+    ]
+
+
+def build_compaction(**overrides):
+    """``build_compaction_params`` with the suite's 2-layer defaults:
+    allocates the caller-owned move-offset rows (capacity cumsum) and SWA
+    destination bases, and hands the test's pre-settled
+    ``kept_token_ordinals`` in as the decision rows. Returns the opaque
+    ``params`` plus a test-side mirror of the caller-owned inputs."""
+    from tensorrt_llm._torch.kv_cache_compression.compaction import build_compaction_params
+
+    args = dict(
+        eviction_mode="union",
+        dense_layers=[0, 1],
+        swa_layers=[],
+        layer_group_representative={0: 0, 1: 1},
+        layer_pool_ids=[0, 0],
+        request_count=2,
+        decode_keep_count=4,
+        swa_window=None,
+    )
+    args.update(overrides)
+    args.pop("eviction_mode")
+    kept = args.pop("kept_token_ordinals")
+    request_count = args["request_count"]
+    keep_count = args["decode_keep_count"]
+    tail = int(args.get("protected_tail_capacity", 0))
+    draft_tail = int(args.get("draft_protected_tail_capacity") or 0)
+    has_draft = bool(args.get("draft_layers"))
+    has_swa = bool(args["swa_layers"])
+    device = args["layer_pools"][args["dense_layers"][0]].device
+    swa_window = int(args["swa_window"] or 0) if has_swa else 0
+    swa_destination_bases = torch.empty_like(args["prompt_offsets"]) if has_swa else None
+
+    def capacity_offsets(count):
+        return torch.arange(0, (request_count + 1) * count, count, dtype=torch.int32, device=device)
+
+    args.setdefault("dense_move_offsets", capacity_offsets(keep_count + tail))
+    args.setdefault("swa_move_offsets", capacity_offsets(swa_window + tail) if has_swa else None)
+    if has_draft:
+        args.setdefault("draft_move_offsets", capacity_offsets(keep_count + draft_tail))
+    params_list = [
+        build_compaction_params(
+            dict(
+                layer_pools=args["layer_pools"],
+                dense_layers=args["dense_layers"],
+                swa_layers=args["swa_layers"],
+                swa_window=args["swa_window"],
+                layer_pool_ids=args["layer_pool_ids"],
+            ),
+            block_offsets=args["kv_block_offsets"],
+            kept_ordinals=kept.reshape(-1, keep_count),
+            source_lengths=args["valid_sequence_lengths"],
+            dense_destination_bases=args["prompt_offsets"],
+            dense_move_offsets=args["dense_move_offsets"],
+            protected_tail_capacity=tail,
+            swa_move_offsets=args["swa_move_offsets"],
+            swa_destination_bases=swa_destination_bases,
+        )
+    ]
+    if has_draft:
+        params_list.append(
+            build_compaction_params(
+                dict(
+                    layer_pools=args["draft_layer_pools"],
+                    dense_layers=args["draft_layers"],
+                    swa_layers=[],
+                    layer_pool_ids=args["draft_layer_pool_ids"],
+                ),
+                block_offsets=args["draft_kv_block_offsets"],
+                kept_ordinals=kept.reshape(-1, keep_count),
+                source_lengths=args["valid_sequence_lengths"],
+                dense_destination_bases=args["prompt_offsets"],
+                dense_move_offsets=args["draft_move_offsets"],
+                protected_tail_capacity=draft_tail,
+            )
+        )
+    params = tuple(params_list)
+    # Opaque plans plus a test-side mirror of the caller-owned construction
+    # inputs (production binds the same values as manager attributes); the
+    # standalone helpers here need the move-offset rows and SWA staging back.
+    return dict(
+        params=params,
+        prompt_offsets=args["prompt_offsets"],
+        request_count=request_count,
+        decode_keep_count=keep_count,
+        protected_tail_capacity=tail,
+        draft_protected_tail_capacity=draft_tail if has_draft else 0,
+        dense_move_offsets=args["dense_move_offsets"],
+        swa_move_offsets=args["swa_move_offsets"],
+        draft_move_offsets=args["draft_move_offsets"] if has_draft else None,
+        has_swa=has_swa,
+        swa_window=swa_window,
+        swa_destination_bases=swa_destination_bases,
+        swa_rebase_delta=keep_count - swa_window,
+    )
+
+
+def run_compaction(compaction):
+    """Replica of the round's move stage in production order: SWA
+    destination rebase, then ``compact`` loops the opaque params (each packs
+    its decision rows into move sources and fires its native moves)."""
+    from tensorrt_llm._torch.kv_cache_compression.compaction import compact
+
+    if compaction["swa_destination_bases"] is not None:
+        torch.add(
+            compaction["prompt_offsets"],
+            compaction["swa_rebase_delta"],
+            out=compaction["swa_destination_bases"],
+        )
+    compact(compaction["params"], compaction["request_count"])
diff --git a/tests/unittest/_torch/kv_cache_compression/test_compaction.py b/tests/unittest/_torch/kv_cache_compression/test_compaction.py
new file mode 100644
index 000000000000..1e7913016851
--- /dev/null
+++ b/tests/unittest/_torch/kv_cache_compression/test_compaction.py
@@ -0,0 +1,324 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Physical KV-cache compaction: packed moves, protected tails, SWA windows,
+and draft co-compaction, checked byte-exactly against torch oracles."""
+
+from types import SimpleNamespace
+
+import pytest
+import torch
+from conftest import build_compaction as _build_compaction
+from conftest import encode_block_offsets as _encode_block_offsets
+from conftest import make_ramp_pools as _make_ramp_pools
+from conftest import run_compaction as _run_compaction
+from conftest import set_protected_tails as _set_protected_tails
+
+requires_sm100 = pytest.mark.skipif(
+    not torch.cuda.is_available() or torch.cuda.get_device_capability() != (10, 0),
+    reason="KV-cache compaction kernels require SM100",
+)
+
+
+def _logical_view(pool: torch.Tensor, pages: torch.Tensor) -> torch.Tensor:
+    """Gather one request's pages into [K/V, head, token, dim] order."""
+    num_kv_heads = int(pool.shape[2])
+    head_dim = int(pool.shape[4])
+    return pool.index_select(0, pages).permute(1, 2, 0, 3, 4).reshape(2, num_kv_heads, -1, head_dim)
+
+
+@pytest.mark.parametrize("eviction_mode", ["union", "per_head", "per_layer_perhead"])
+def test_eager_compaction_preserves_exact_selected_bytes_and_tail(eviction_mode):
+    # Supported bf16 geometry; kept ordinals span all three pages so moves
+    # cross page boundaries.
+    device = torch.device("cuda", torch.cuda.current_device())
+    request_count = 2
+    num_layers = 2
+    num_kv_heads = 2
+    # Mixed prompt lengths prove per-request destination rebasing.
+    prompt_lens = [2, 5]
+    decode_keep_count = 4
+    seq_len = 80
+    tokens_per_block = 32
+    pages_per_request = 3
+    head_dim = 64
+    protected_tails = [2, 1]
+    page_tables = torch.tensor([[0, 1, 2], [3, 4, 5]], dtype=torch.int32, device=device)
+    initial_pools = _make_ramp_pools(num_layers, device=device)
+    pools = [pool.clone() for pool in initial_pools]
+
+    # Decode-only kept ordinals holding absolute positions.
+    union_decode = torch.tensor(
+        [[16, 32, 56, 72], [24, 40, 48, 64]], dtype=torch.int64, device=device
+    )
+    if eviction_mode == "union":
+        keep = union_decode
+        selection_rows = 1
+    else:
+        selection_rows = num_kv_heads if eviction_mode == "per_head" else num_layers * num_kv_heads
+        keep = torch.empty(
+            request_count,
+            selection_rows,
+            decode_keep_count,
+            dtype=torch.int64,
+            device=device,
+        )
+        for request in range(request_count):
+            for row in range(selection_rows):
+                keep[request, row] = torch.tensor(
+                    sorted(
+                        {
+                            prompt_lens[request] + ((request + row + offset * 2) % 8) * 8
+                            for offset in range(decode_keep_count)
+                        }
+                    ),
+                    dtype=torch.int64,
+                    device=device,
+                )
+
+    compaction = _build_compaction(
+        eviction_mode=eviction_mode,
+        layer_pools=pools,
+        kept_token_ordinals=keep.to(torch.int32),
+        valid_sequence_lengths=torch.tensor([seq_len, seq_len], dtype=torch.int32, device=device),
+        kv_block_offsets=_encode_block_offsets(page_tables.unsqueeze(0)),
+        prompt_offsets=torch.tensor(prompt_lens, dtype=torch.int32, device=device),
+        protected_tail_capacity=max(protected_tails),
+    )
+    _set_protected_tails(compaction, protected_tails)
+    # Production settles the kept ordinals into the contract's decision
+    # rows; with pre-settled ordinals the pack launch inside compact() is
+    # its exact analog.
+    _run_compaction(compaction)
+    torch.cuda.synchronize(device)
+
+    for layer, (before_pool, after_pool) in enumerate(zip(initial_pools, pools)):
+        for request in range(request_count):
+            prompt_len = prompt_lens[request]
+            pages = page_tables[request].to(torch.long)
+            before = (
+                before_pool[pages]
+                .permute(1, 2, 0, 3, 4)
+                .reshape(2, num_kv_heads, pages_per_request * tokens_per_block, head_dim)
+            )
+            after = after_pool[pages].permute(1, 2, 0, 3, 4).reshape_as(before)
+            assert torch.equal(after[:, :, :prompt_len], before[:, :, :prompt_len])
+            for head in range(num_kv_heads):
+                if eviction_mode == "union":
+                    selected = keep[request]
+                elif eviction_mode == "per_head":
+                    selected = keep[request, head]
+                else:
+                    selected = keep[request, layer * num_kv_heads + head]
+                tail = torch.arange(
+                    seq_len,
+                    seq_len + protected_tails[request],
+                    dtype=torch.int64,
+                    device=device,
+                )
+                source = torch.cat((selected, tail))
+                destination = torch.arange(
+                    prompt_len,
+                    prompt_len + source.numel(),
+                    dtype=torch.int64,
+                    device=device,
+                )
+                assert torch.equal(
+                    after[:, head].index_select(1, destination),
+                    before[:, head].index_select(1, source),
+                )
+
+
+@requires_sm100
+def test_eager_compaction_rebases_masked_swa_window_and_tail():
+    # Supported bf16 geometry; dense and SWA moves stay page-crossing.
+    device = torch.device("cuda", torch.cuda.current_device())
+    dense_tables = torch.tensor([[2, 0, 1], [5, 3, 4]], dtype=torch.int32, device=device)
+    swa_tables = torch.tensor([[1, 2, 0], [4, 5, 3]], dtype=torch.int32, device=device)
+    initial_pools = _make_ramp_pools(2, num_kv_heads=1, device=device)
+    pools = [pool.clone() for pool in initial_pools]
+    # Decode-only kept ordinals holding absolute positions past the prompt.
+    keep = torch.tensor(
+        [[16, 32, 40, 56], [16, 24, 40, 48]],
+        dtype=torch.int64,
+        device=device,
+    )
+    valid_seq_lens = torch.tensor([64, 56], dtype=torch.int32, device=device)
+    protected_tails = [2, 1]
+    compaction = _build_compaction(
+        layer_pools=pools,
+        dense_layers=[0],
+        swa_layers=[1],
+        layer_group_representative={0: 0},
+        # Dense layer 0 stages in plane 0, the SWA layer in its own plane 1.
+        layer_pool_ids=[0, 1],
+        kept_token_ordinals=keep.to(torch.int32),
+        valid_sequence_lengths=valid_seq_lens,
+        kv_block_offsets=_encode_block_offsets(torch.stack((dense_tables, swa_tables))),
+        prompt_offsets=torch.tensor([2, 2], dtype=torch.int32, device=device),
+        swa_window=2,
+        protected_tail_capacity=max(protected_tails),
+    )
+    _set_protected_tails(compaction, protected_tails)
+    _run_compaction(compaction)
+    torch.cuda.synchronize(device)
+
+    for request, (valid_seq_len, tail_length) in enumerate(
+        zip(valid_seq_lens.tolist(), protected_tails)
+    ):
+        dense_pages = dense_tables[request].to(torch.long)
+        swa_pages = swa_tables[request].to(torch.long)
+        dense_before = initial_pools[0][dense_pages].permute(1, 2, 0, 3, 4).reshape(2, 1, -1, 64)
+        dense_after = pools[0][dense_pages].permute(1, 2, 0, 3, 4).reshape_as(dense_before)
+        swa_before = initial_pools[1][swa_pages].permute(1, 2, 0, 3, 4).reshape(2, 1, -1, 64)
+        swa_after = pools[1][swa_pages].permute(1, 2, 0, 3, 4).reshape_as(swa_before)
+        tail = torch.arange(
+            valid_seq_len,
+            valid_seq_len + tail_length,
+            dtype=torch.int64,
+            device=device,
+        )
+        dense_source = torch.cat((keep[request], tail))
+        dense_destination = torch.arange(
+            2, 2 + dense_source.numel(), dtype=torch.int64, device=device
+        )
+        swa_source = torch.arange(
+            valid_seq_len - 2,
+            valid_seq_len + tail_length,
+            dtype=torch.int64,
+            device=device,
+        )
+        swa_destination = torch.arange(4, 4 + swa_source.numel(), dtype=torch.int64, device=device)
+        assert torch.equal(dense_after[:, :, :2], dense_before[:, :, :2])
+        assert torch.equal(swa_after[:, :, :2], swa_before[:, :, :2])
+        assert torch.equal(
+            dense_after.index_select(2, dense_destination),
+            dense_before.index_select(2, dense_source),
+        )
+        assert torch.equal(
+            swa_after.index_select(2, swa_destination),
+            swa_before.index_select(2, swa_source),
+        )
+
+
+def _launched_draft_compaction(draft_protected_tails):
+    """Target and draft pools with distinct head counts (supported bf16
+    geometry, mod-251 ramp payload), compacted in one round."""
+    device = torch.device("cuda", torch.cuda.current_device())
+    request_count = 2
+    prompt_len = 2
+    target_protected_tails = [2, 1]
+    valid_seq_lens = [10, 9]
+
+    target_tables = torch.tensor([[0, 1, 2], [3, 4, 5]], dtype=torch.int32, device=device)
+    draft_tables = torch.tensor([[1, 0, 2], [5, 4, 3]], dtype=torch.int32, device=device)
+    target_pools = _make_ramp_pools(2, num_kv_heads=2, device=device)
+    draft_pool = _make_ramp_pools(1, num_kv_heads=4, base=149, device=device)[0]
+    assert target_pools[0].shape[2] != draft_pool.shape[2]
+    initial_target = [pool.clone() for pool in target_pools]
+    initial_draft = draft_pool.clone()
+
+    keep = torch.tensor([[2, 4, 7, 9], [3, 5, 6, 8]], dtype=torch.int64, device=device)
+
+    compaction = _build_compaction(
+        layer_pools=target_pools,
+        layer_pool_ids=[0, 0],
+        kept_token_ordinals=keep.to(torch.int32),
+        valid_sequence_lengths=torch.tensor(valid_seq_lens, dtype=torch.int32, device=device),
+        kv_block_offsets=_encode_block_offsets(target_tables),
+        prompt_offsets=torch.full((request_count,), prompt_len, dtype=torch.int32, device=device),
+        protected_tail_capacity=max(target_protected_tails),
+        draft_layer_pools=[draft_pool],
+        draft_layers=[0],
+        draft_layer_group_representative={0: 0},
+        draft_layer_pool_ids=[0],
+        draft_protected_tail_capacity=max(draft_protected_tails),
+        draft_kv_block_offsets=_encode_block_offsets(draft_tables),
+    )
+    _set_protected_tails(compaction, target_protected_tails, draft_protected_tails)
+    _run_compaction(compaction)
+    torch.cuda.synchronize(device)
+
+    return SimpleNamespace(
+        device=device,
+        request_count=request_count,
+        prompt_len=prompt_len,
+        keep=keep,
+        valid_seq_lens=valid_seq_lens,
+        target_protected_tails=target_protected_tails,
+        draft_protected_tails=draft_protected_tails,
+        target_tables=target_tables,
+        draft_tables=draft_tables,
+        target_pools=target_pools,
+        draft_pool=draft_pool,
+        initial_target=initial_target,
+        initial_draft=initial_draft,
+        compaction=compaction,
+    )
+
+
+def test_draft_moves_and_pack_match_keep_broadcast_and_tail_oracle():
+    # Ragged draft tails [1, 2] against target tails [2, 1]: one request's
+    # draft tail below and one above its target, subsuming the uniform row.
+    built = _launched_draft_compaction(draft_protected_tails=[1, 2])
+    device = built.device
+    prompt_len = built.prompt_len
+
+    expected_offsets = [0]
+    for request in range(built.request_count):
+        valid = built.valid_seq_lens[request]
+        # Target dense layers compact the union keep set plus the target tail.
+        target_pages = built.target_tables[request].to(torch.long)
+        target_tail = torch.arange(
+            valid,
+            valid + built.target_protected_tails[request],
+            dtype=torch.int64,
+            device=device,
+        )
+        target_source = torch.cat((built.keep[request], target_tail))
+        target_destination = torch.arange(
+            prompt_len,
+            prompt_len + target_source.numel(),
+            dtype=torch.int64,
+            device=device,
+        )
+        for before_pool, after_pool in zip(built.initial_target, built.target_pools):
+            before = _logical_view(before_pool, target_pages)
+            after = _logical_view(after_pool, target_pages)
+            assert torch.equal(after[:, :, :prompt_len], before[:, :, :prompt_len])
+            assert torch.equal(
+                after.index_select(2, target_destination),
+                before.index_select(2, target_source),
+            )
+
+        # Same kept ordinals through the draft's OWN table/heads/tail.
+        draft_pages = built.draft_tables[request].to(torch.long)
+        draft_tail = torch.arange(
+            valid,
+            valid + built.draft_protected_tails[request],
+            dtype=torch.int64,
+            device=device,
+        )
+        draft_source = torch.cat((built.keep[request], draft_tail))
+        draft_destination = torch.arange(
+            prompt_len,
+            prompt_len + draft_source.numel(),
+            dtype=torch.int64,
+            device=device,
+        )
+        before = _logical_view(built.initial_draft, draft_pages)
+        after = _logical_view(built.draft_pool, draft_pages)
+        assert torch.equal(after[:, :, :prompt_len], before[:, :, :prompt_len])
+        for head in range(int(built.draft_pool.shape[2])):
+            assert torch.equal(
+                after[:, head].index_select(1, draft_destination),
+                before[:, head].index_select(1, draft_source),
+            )
+
+        expected_offsets.append(expected_offsets[-1] + int(draft_source.numel()))
+
+    # The test-owned draft move-offset row must match the broadcast-plus-tail
+    # oracle; the packed move sources themselves are covered byte-exactly by
+    # the pool assertions above (the ramp payload makes every wrong move land
+    # on different bytes) and by the pack-kernel oracle suite.
+    assert built.compaction["draft_move_offsets"].cpu().tolist() == expected_offsets
diff --git a/tests/unittest/_torch/thop/serial/test_sparse_kv_cache_compact.py b/tests/unittest/_torch/thop/serial/test_sparse_kv_cache_compact.py
new file mode 100644
index 000000000000..ca40c09667b9
--- /dev/null
+++ b/tests/unittest/_torch/thop/serial/test_sparse_kv_cache_compact.py
@@ -0,0 +1,442 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the layered V2 sparse-KV compaction op (pipelined bf16 kernels)."""
+
+from typing import NamedTuple, Optional
+
+import pytest
+import torch
+
+import tensorrt_llm  # noqa: F401  # Register torch.ops.trtllm operators.
+
+_TOKENS_PER_BLOCK = 32
+_NUM_KV_HEADS = 2
+_BATCH_SIZE = 2
+_PAGE_INDEX_DIVISOR = 2
+
+# Profiler probe names: the pipelined bf16 kernels are the only shipped
+# path; the retired register-staging kernel must never appear.
+_FAST_KERNEL_NAME = "sparseKvCacheCompactV2Bf16PipelineKernel"
+_RETIRED_KERNEL_NAME = "updateSparseKvCacheAfterFmha"
+
+
+def _encode_k_block_offsets(
+    page_table: torch.Tensor, page_index_scale: int = _PAGE_INDEX_DIVISOR
+) -> torch.Tensor:
+    encoded = torch.empty(
+        page_table.shape[0],
+        2,
+        page_table.shape[1],
+        dtype=torch.int32,
+        device=page_table.device,
+    )
+    encoded[:, 0] = page_table * page_index_scale
+    encoded[:, 1] = encoded[:, 0] + 1
+    return encoded[:, 0]
+
+
+class _DeviceArguments(NamedTuple):
+    pool_pointers: torch.Tensor
+    source_indices: torch.Tensor
+    source_offsets: torch.Tensor
+    source_layer_indices: Optional[torch.Tensor]
+
+
+def _make_pools(
+    num_layers: int,
+    dtype: torch.dtype,
+    head_dim: int,
+    page_index_scale: int = _PAGE_INDEX_DIVISOR,
+) -> tuple[list[torch.Tensor], list[torch.Tensor], list[torch.Tensor]]:
+    pages_per_seq = 3
+    num_pages = _BATCH_SIZE * pages_per_seq * page_index_scale // _PAGE_INDEX_DIVISOR
+    shape = (
+        num_pages,
+        2,
+        _NUM_KV_HEADS,
+        _TOKENS_PER_BLOCK,
+        head_dim,
+    )
+    numel = torch.Size(shape).numel()
+    pools_cpu = [
+        ((torch.arange(numel, dtype=torch.int32) + layer * 37) % 251).reshape(shape).to(dtype)
+        for layer in range(num_layers)
+    ]
+    pools = [pool.cuda() for pool in pools_cpu]
+    raw_pages = [[4, 1, 5], [2, 0, 3]]
+    assert set(raw_pages[0]).isdisjoint(raw_pages[1])
+    raw_page_table = torch.tensor(raw_pages, dtype=torch.int32, device="cuda")
+    page_table = _encode_k_block_offsets(raw_page_table, page_index_scale)
+    page_tables = [page_table] * num_layers
+    assert page_tables[0].stride(0) == 2 * page_tables[0].shape[1]
+    return pools_cpu, pools, page_tables
+
+
+def _device_arguments(
+    pools: list[torch.Tensor],
+    source_indices: torch.Tensor,
+    source_offsets: torch.Tensor,
+    source_layer_indices: Optional[torch.Tensor] = None,
+) -> _DeviceArguments:
+    device = pools[0].device
+    return _DeviceArguments(
+        pool_pointers=torch.tensor(
+            [pool.data_ptr() for pool in pools], dtype=torch.int64, device=device
+        ),
+        source_indices=source_indices.to(device),
+        source_offsets=source_offsets.to(device),
+        source_layer_indices=(
+            None if source_layer_indices is None else source_layer_indices.to(device)
+        ),
+    )
+
+
+def _reference_compact(
+    pools: list[torch.Tensor],
+    page_tables: list[torch.Tensor],
+    source_indices: torch.Tensor,
+    source_offsets: torch.Tensor,
+    destination_base: "int | list[int]",
+    source_layer_indices: Optional[torch.Tensor] = None,
+    tokens_per_block: int = _TOKENS_PER_BLOCK,
+    batch_size: int = _BATCH_SIZE,
+) -> list[torch.Tensor]:
+    original = [pool.clone() for pool in pools]
+    expected = [pool.clone() for pool in pools]
+    for group_layer, (source_pool, destination_pool, page_table) in enumerate(
+        zip(original, expected, page_tables)
+    ):
+        # The kernel decodes K offsets as offset // 2 regardless of the
+        # encoder's scale; the reference mirrors the kernel.
+        raw_page_table = page_table // _PAGE_INDEX_DIVISOR
+        if source_indices.ndim == 2:
+            layer_sources = source_indices
+        else:
+            assert source_layer_indices is not None
+            layer_sources = source_indices[int(source_layer_indices[group_layer])]
+        for request in range(batch_size):
+            begin = int(source_offsets[request])
+            end = int(source_offsets[request + 1])
+            request_base = (
+                destination_base[request]
+                if isinstance(destination_base, (list, tuple))
+                else destination_base
+            )
+            for head in range(layer_sources.shape[0]):
+                for request_move, global_move in enumerate(range(begin, end)):
+                    source_token = int(layer_sources[head, global_move])
+                    destination_token = request_base + request_move
+                    source_page = int(raw_page_table[request, source_token // tokens_per_block])
+                    destination_page = int(
+                        raw_page_table[request, destination_token // tokens_per_block]
+                    )
+                    destination_pool[
+                        destination_page,
+                        :,
+                        head,
+                        destination_token % tokens_per_block,
+                        :,
+                    ] = source_pool[
+                        source_page,
+                        :,
+                        head,
+                        source_token % tokens_per_block,
+                        :,
+                    ]
+    return expected
+
+
+def _compact(
+    pools: list[torch.Tensor],
+    page_tables: list[torch.Tensor],
+    arguments: _DeviceArguments,
+    destination_base: "int | list[int]",
+    batch_size: int = _BATCH_SIZE,
+) -> None:
+    # The op takes per-request destination bases; scalar test parameters are
+    # broadcast to the batch here. torch.full stays CUDA-graph-capturable.
+    if isinstance(destination_base, int):
+        destination_bases = torch.full(
+            (batch_size,), destination_base, dtype=torch.int32, device="cuda"
+        )
+    else:
+        destination_bases = torch.tensor(destination_base, dtype=torch.int32, device="cuda")
+    torch.ops.trtllm.sparse_kv_cache_compact_layers(
+        pools,
+        arguments.pool_pointers,
+        page_tables[0],
+        arguments.source_indices,
+        arguments.source_offsets,
+        destination_bases,
+        arguments.source_layer_indices,
+    )
+
+
+_SMALL_ROW = [2, 5, 8, 3, 7, 10]
+# The production-shaped fast-geometry matrix below is the byte-exact anchor
+# (both head dims, both page sizes, per-request destination bases, 3-D
+# per-layer routing, multi-tile ragged moves). These two rows keep the
+# op-level contracts it does not pin: the destination-base-0 (prompt 0)
+# boundary and the fixed //2 K-offset decode against a scale-4 encoder.
+_LAYER_CASES = [
+    pytest.param(dict(head_dim=64, dest=0, scale=2), id="bf16_h64_dest0_scale2"),
+    pytest.param(dict(head_dim=64, dest=2, scale=4), id="bf16_h64_dest2_scale4"),
+]
+
+
+@pytest.mark.parametrize("case", _LAYER_CASES)
+def test_sparse_kv_cache_compact_layers(case):
+    pools_cpu, pools, page_tables = _make_pools(3, torch.bfloat16, case["head_dim"], case["scale"])
+    page_tables_cpu = [page_table.cpu() for page_table in page_tables]
+    source_offsets = torch.tensor((0, 3, 6), dtype=torch.int32)
+    source_row = torch.tensor(_SMALL_ROW, dtype=torch.int32)
+    source_indices = source_row.view(1, -1).expand(_NUM_KV_HEADS, -1).contiguous()
+    destination_base = case["dest"]
+    expected = _reference_compact(
+        pools_cpu,
+        page_tables_cpu,
+        source_indices,
+        source_offsets,
+        destination_base,
+    )
+    arguments = _device_arguments(pools, source_indices, source_offsets)
+
+    _compact(pools, page_tables, arguments, destination_base)
+    torch.cuda.synchronize()
+
+    for actual, reference in zip(pools, expected):
+        assert torch.equal(actual.cpu(), reference)
+
+
+def test_sparse_kv_cache_compact_layers_cuda_graph_replay():
+    """Check operation-level capture safety inside an externally captured graph."""
+    pools_cpu, pools, page_tables = _make_pools(3, torch.bfloat16, 64)
+    page_tables_cpu = [page_table.cpu() for page_table in page_tables]
+    source_offsets = torch.tensor([0, 3, 6], dtype=torch.int32)
+    source_row = torch.tensor(_SMALL_ROW, dtype=torch.int32)
+    source_indices = source_row.view(1, -1).expand(_NUM_KV_HEADS, -1).contiguous()
+    replay_row = torch.tensor([3, 6, 9, 2, 5, 8], dtype=torch.int32)
+    replay_indices = replay_row.view(1, -1).expand(_NUM_KV_HEADS, -1).contiguous()
+    destination_base = 2
+    expected = _reference_compact(
+        pools_cpu,
+        page_tables_cpu,
+        replay_indices,
+        source_offsets,
+        destination_base,
+    )
+    arguments = _device_arguments(pools, source_indices, source_offsets)
+
+    graph = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(graph):
+        _compact(pools, page_tables, arguments, destination_base)
+
+    for pool, initial in zip(pools, pools_cpu):
+        pool.copy_(initial)
+    arguments.source_indices.copy_(replay_indices)
+    graph.replay()
+    torch.cuda.synchronize()
+
+    for actual, reference in zip(pools, expected):
+        assert torch.equal(actual.cpu(), reference)
+
+
+# --- Production-shaped geometry for the pipelined bf16 fast path ----------
+
+_FAST_BATCH_SIZE = 3
+# Ragged+full tiles, an empty request, and a prologue/epilogue-only tile.
+_FAST_MOVE_COUNTS = (71, 0, 29)
+# Mixed prompt lengths: none tile- or page-aligned.
+_FAST_DESTINATION_BASES = [3, 9, 17]
+# Allocation-wide index buffers: padding past the round's total move count
+# makes a device-derived stride read padding and fail the byte-compare.
+_FAST_SOURCE_PAD = 37
+_FAST_IDENTITY_MOVES = 5
+
+
+class _FastGeometryCase(NamedTuple):
+    pools_cpu: list[torch.Tensor]
+    pools: list[torch.Tensor]
+    page_tables_cpu: list[torch.Tensor]
+    page_tables: list[torch.Tensor]
+    source_indices: torch.Tensor
+    source_offsets: torch.Tensor
+    source_layer_indices: Optional[torch.Tensor]
+    destination_bases: list[int]
+    tokens_per_block: int
+    batch_size: int
+
+
+def _fast_sources_row(
+    base: int, count: int, limit: int, generator: torch.Generator
+) -> torch.Tensor:
+    """Distinct sorted source tokens >= base, so src(i) >= base + i (the op's
+    in-place-safety contract). The first few moves are identities (src == dst),
+    which the kernel skips storing."""
+    if count == 0:
+        return torch.empty(0, dtype=torch.int32)
+    identity = min(_FAST_IDENTITY_MOVES, count)
+    candidates = torch.arange(base + identity, limit, dtype=torch.int32)
+    picks = torch.randperm(candidates.numel(), generator=generator)[: count - identity]
+    tail = candidates[picks].sort().values
+    return torch.cat((torch.arange(base, base + identity, dtype=torch.int32), tail))
+
+
+def _make_fast_geometry_case(
+    head_dim: int,
+    tokens_per_block: int,
+    dtype: torch.dtype = torch.bfloat16,
+    num_layers: int = 2,
+    per_layer_sources: bool = False,
+) -> _FastGeometryCase:
+    batch_size = _FAST_BATCH_SIZE
+    pages_per_seq = max(2, 128 // tokens_per_block)
+    tokens_per_seq = pages_per_seq * tokens_per_block
+    num_pages = batch_size * pages_per_seq
+    shape = (num_pages, 2, _NUM_KV_HEADS, tokens_per_block, head_dim)
+    numel = torch.Size(shape).numel()
+    pools_cpu = [
+        ((torch.arange(numel, dtype=torch.int32) + layer * 37) % 251).reshape(shape).to(dtype)
+        for layer in range(num_layers)
+    ]
+    pools = [pool.cuda() for pool in pools_cpu]
+
+    # Deterministic per-geometry inputs so any failure reproduces exactly.
+    generator = torch.Generator().manual_seed(20260720 + head_dim * 1000 + tokens_per_block)
+    raw_page_table = (
+        torch.randperm(num_pages, generator=generator)
+        .to(torch.int32)
+        .reshape(batch_size, pages_per_seq)
+        .cuda()
+    )
+    page_table = _encode_k_block_offsets(raw_page_table)
+    page_tables = [page_table] * num_layers
+    assert page_tables[0].stride(0) == 2 * pages_per_seq
+    page_tables_cpu = [table.cpu() for table in page_tables]
+
+    offsets = [0]
+    for count in _FAST_MOVE_COUNTS:
+        offsets.append(offsets[-1] + count)
+    source_offsets = torch.tensor(offsets, dtype=torch.int32)
+    width = offsets[-1] + _FAST_SOURCE_PAD
+    source_layers = 3 if per_layer_sources else 1
+    # Padding is a valid token id so a stride bug corrupts output (caught by
+    # the byte-compare) instead of faulting.
+    rows = torch.full((source_layers, _NUM_KV_HEADS, width), tokens_per_seq - 1, dtype=torch.int32)
+    for layer in range(source_layers):
+        for head in range(_NUM_KV_HEADS):
+            cursor = 0
+            for request, count in enumerate(_FAST_MOVE_COUNTS):
+                rows[layer, head, cursor : cursor + count] = _fast_sources_row(
+                    _FAST_DESTINATION_BASES[request], count, tokens_per_seq, generator
+                )
+                cursor += count
+    if per_layer_sources:
+        source_indices = rows.contiguous()
+        source_layer_indices = torch.tensor([2, 0], dtype=torch.int32)
+    else:
+        source_indices = rows[0].contiguous()
+        source_layer_indices = None
+
+    return _FastGeometryCase(
+        pools_cpu=pools_cpu,
+        pools=pools,
+        page_tables_cpu=page_tables_cpu,
+        page_tables=page_tables,
+        source_indices=source_indices,
+        source_offsets=source_offsets,
+        source_layer_indices=source_layer_indices,
+        destination_bases=list(_FAST_DESTINATION_BASES),
+        tokens_per_block=tokens_per_block,
+        batch_size=batch_size,
+    )
+
+
+def _run_fast_geometry_case(case: _FastGeometryCase) -> list[torch.Tensor]:
+    expected = _reference_compact(
+        case.pools_cpu,
+        case.page_tables_cpu,
+        case.source_indices,
+        case.source_offsets,
+        case.destination_bases,
+        case.source_layer_indices,
+        tokens_per_block=case.tokens_per_block,
+        batch_size=case.batch_size,
+    )
+    arguments = _device_arguments(
+        case.pools, case.source_indices, case.source_offsets, case.source_layer_indices
+    )
+    _compact(
+        case.pools,
+        case.page_tables,
+        arguments,
+        case.destination_bases,
+        batch_size=case.batch_size,
+    )
+    torch.cuda.synchronize()
+    return expected
+
+
+# The full fast-path gate matrix; the per-layer-source row keeps the 3-D
+# routing path runnable through the fast kernel.
+_FAST_GEOMETRY_MATRIX = [(64, 32), (128, 32), (64, 128), (128, 128)]
+
+
+@pytest.mark.parametrize(
+    "head_dim,tokens_per_block,per_layer",
+    [(h, t, False) for h, t in _FAST_GEOMETRY_MATRIX] + [(64, 32, True)],
+)
+def test_sparse_kv_cache_compact_layers_fast_geometry(head_dim, tokens_per_block, per_layer):
+    # Byte-compare against the CPU reference, plus the dispatch probe: the
+    # pipelined kernel must actually run (and the retired one must not) --
+    # every byte-equality here would still pass on a silent fallback.
+    case = _make_fast_geometry_case(head_dim, tokens_per_block, per_layer_sources=per_layer)
+    with torch.profiler.profile(activities=[torch.profiler.ProfilerActivity.CUDA]) as profiler:
+        expected = _run_fast_geometry_case(case)
+    names = [event.name for event in profiler.events()]
+    assert any(_FAST_KERNEL_NAME in name for name in names)
+    assert not any(_RETIRED_KERNEL_NAME in name for name in names)
+    for actual, reference in zip(case.pools, expected):
+        assert torch.equal(actual.cpu(), reference)
+
+
+# One representative per reject family; no fallback kernel exists, so every
+# reject must fail loudly and leave the pools untouched.
+@pytest.mark.parametrize(
+    "dtype,head_dim,tokens_per_block,flat_with_layer_indices,match",
+    [
+        pytest.param(torch.float16, 64, 32, False, "bf16|BF16", id="dtype_outside_gate"),
+        pytest.param(torch.bfloat16, 256, 32, False, "bf16|BF16", id="head_dim_outside_gate"),
+        pytest.param(torch.bfloat16, 64, 16, False, "bf16|BF16", id="page_size_outside_gate"),
+        pytest.param(
+            torch.bfloat16,
+            64,
+            32,
+            True,
+            "require 3-D per-layer source_indices",
+            id="flat_source_with_layer_indices",
+        ),
+    ],
+)
+def test_sparse_kv_cache_compact_layers_rejects_invalid_launch(
+    dtype, head_dim, tokens_per_block, flat_with_layer_indices, match
+):
+    case = _make_fast_geometry_case(head_dim, tokens_per_block, dtype=dtype)
+    source_layer_indices = case.source_layer_indices
+    if flat_with_layer_indices:
+        source_layer_indices = torch.tensor([0, 0], dtype=torch.int32)
+    arguments = _device_arguments(
+        case.pools, case.source_indices, case.source_offsets, source_layer_indices
+    )
+    with pytest.raises((RuntimeError, ValueError), match=match):
+        _compact(
+            case.pools,
+            case.page_tables,
+            arguments,
+            case.destination_bases,
+            batch_size=case.batch_size,
+        )
+    torch.cuda.synchronize()
+    for actual, reference in zip(case.pools, case.pools_cpu):
+        assert torch.equal(actual.cpu(), reference)