diff --git a/tensorrt_llm/_torch/attention_backend/flashinfer.py b/tensorrt_llm/_torch/attention_backend/flashinfer.py
index c16d2ffaee36..291ed21cf19f 100644
--- a/tensorrt_llm/_torch/attention_backend/flashinfer.py
+++ b/tensorrt_llm/_torch/attention_backend/flashinfer.py
@@ -587,7 +587,10 @@ def _sanitize_swa_page_indices(self, page_indices: torch.Tensor,
         """Replace evicted SWA pages with a safe in-range page index."""
         window_vec = getattr(self.kv_cache_manager, 'max_attention_window_vec',
                              None)
-        if not window_vec or window_vec[layer_idx % len(window_vec)] is None:
+        if not window_vec:
+            return
+        local_layer_idx = self.kv_cache_manager.layer_offsets[layer_idx]
+        if window_vec[local_layer_idx] is None:
             return
 
         # KVCacheManagerV2 marks evicted out-of-window pages with -1.
diff --git a/tensorrt_llm/_torch/attention_backend/trtllm.py b/tensorrt_llm/_torch/attention_backend/trtllm.py
index 101474ca00f1..eee3993026cb 100644
--- a/tensorrt_llm/_torch/attention_backend/trtllm.py
+++ b/tensorrt_llm/_torch/attention_backend/trtllm.py
@@ -2541,12 +2541,10 @@ def forward(
         # the model does not specify one, so paged-context attention does not
         # read stale page-table entries (BAD_PAGE_INDEX).
         if forward_args.attention_window_size is None and metadata.kv_cache_manager is not None:
-            window_vec = getattr(metadata.kv_cache_manager,
-                                 'max_attention_window_vec', None)
-            if window_vec:
-                window = window_vec[self.local_layer_idx % len(window_vec)]
-                if window is not None:
-                    forward_args.attention_window_size = window
+            window = metadata.kv_cache_manager.max_attention_window_vec[
+                self.local_layer_idx]
+            if window is not None:
+                forward_args.attention_window_size = window
         if forward_args.attention_window_size is None:
             forward_args.attention_window_size = metadata.max_seq_len
 
diff --git a/tensorrt_llm/_torch/pyexecutor/_util.py b/tensorrt_llm/_torch/pyexecutor/_util.py
index 50aaf77d7185..b7aa1105ab9e 100644
--- a/tensorrt_llm/_torch/pyexecutor/_util.py
+++ b/tensorrt_llm/_torch/pyexecutor/_util.py
@@ -15,7 +15,7 @@
 import copy
 import dataclasses
 import os
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Sequence, Union
 
 import torch
 
@@ -524,6 +524,25 @@ def draft_config_defines_attention_layout(
         or bool(getattr(draft_pretrained_config, "layer_types", None)))
 
 
+def _expand_attention_window_pattern_to_global_layers(
+    max_attention_window: Optional[Sequence[int]],
+    layer_mask: Sequence[bool],
+) -> Optional[List[int]]:
+    """Expand an enabled-layer pattern into physical global-layer order."""
+    if max_attention_window is None:
+        return None
+
+    pattern = list(max_attention_window)
+    global_windows = [pattern[0]] * len(layer_mask)
+    enabled_layer_offset = 0
+    for layer_idx, enabled in enumerate(layer_mask):
+        if enabled:
+            global_windows[layer_idx] = pattern[enabled_layer_offset %
+                                                len(pattern)]
+            enabled_layer_offset += 1
+    return global_windows
+
+
 def _derive_draft_max_attention_window(
     kv_cache_config: KvCacheConfig,
     draft_pretrained_config: object,
@@ -1560,6 +1579,11 @@ def _create_one_model_draft_kv_cache_manager(
             kv_cache_config,
             max_seq_len,
             estimating_kv_cache=estimating_kv_cache)
+        draft_kv_config.max_attention_window = (
+            _expand_attention_window_pattern_to_global_layers(
+                draft_kv_config.max_attention_window,
+                spec_dec_layer_mask,
+            ))
         if (not uses_vswa_kv_cache_layout(draft_kv_config.max_attention_window)
                 and draft_kv_config.pool_ratio is not None
                 and len(draft_kv_config.pool_ratio) != 1):
diff --git a/tensorrt_llm/_torch/pyexecutor/kv_cache_manager_v2.py b/tensorrt_llm/_torch/pyexecutor/kv_cache_manager_v2.py
index 73d9c515ecfe..f18a56249568 100644
--- a/tensorrt_llm/_torch/pyexecutor/kv_cache_manager_v2.py
+++ b/tensorrt_llm/_torch/pyexecutor/kv_cache_manager_v2.py
@@ -97,10 +97,11 @@
     BaseResourceManager,
     CacheTypeCpp,
     DataType,
-    KVCacheManager,
     ModelConfigCpp,
     ModelConfigPython,
+    _clamp_max_attention_window_vec,
     _populate_dummy_mrope_config,
+    _project_max_attention_window_vec,
     get_pp_layers,
     request_context,
 )
@@ -386,6 +387,17 @@ def _get_single_swa_pool_slot_bytes(
     return sum(layer_sizes) * int(tokens_per_block)
 
 
+def _resolve_v2_max_attention_window_vec(
+    max_attention_window_vec: Optional[Sequence[int]],
+    max_seq_len: int,
+    pp_layers: Sequence[int],
+) -> List[Optional[int]]:
+    """Resolve a V2 window pattern into exact cache-local layer order."""
+    clamped_window_pattern = _clamp_max_attention_window_vec(max_attention_window_vec, max_seq_len)
+    local_windows = _project_max_attention_window_vec(clamped_window_pattern, pp_layers)
+    return [None if window == max_seq_len else window for window in local_windows]
+
+
 def _get_static_cache_size_layer_components(
     model_config: ModelConfigPython,
     mapping: Mapping,
@@ -424,24 +436,38 @@ def _get_static_cache_size_layer_components(
         )
         layer_size = cache_size_per_token * 2
 
-    num_attention_layers = KVCacheManager._resolve_num_attention_layers(
-        model_config, mapping, num_layers
-    )
+    if num_layers is None:
+        total_attention_layers = model_config.get_num_attention_layers()
+        local_layer_ids, _ = get_pp_layers(total_attention_layers, mapping)
+    else:
+        local_layer_ids = list(range(max(num_layers, 1)))
+    num_attention_layers = len(local_layer_ids)
     layer_sizes = [layer_size] * num_attention_layers
     window_pattern = kv_cache_config.max_attention_window if kv_cache_config is not None else None
 
-    def get_window_size(layer_idx: int) -> Optional[int]:
-        if window_pattern is None or not isinstance(window_pattern, (list, tuple)):
-            return None
-        window_size = window_pattern[layer_idx % len(window_pattern)]
+    # Static estimation accepts an unknown max_seq_len and treats recurrent-state
+    # sentinels as full-attention cost. Runtime resolution must preserve those
+    # sentinels for layer construction, so it cannot be reused here directly.
+    def normalize_window_size(window_size: Optional[int]) -> Optional[int]:
         if window_size is None or window_size <= 0:
             return None
         window_size = int(window_size)
+        if max_seq_len is not None:
+            window_size = min(window_size, int(max_seq_len))
         if max_seq_len is not None and window_size == int(max_seq_len):
             return None
         return window_size
 
-    attention_windows = [get_window_size(layer_idx) for layer_idx in range(num_attention_layers)]
+    if window_pattern is None or not isinstance(window_pattern, (list, tuple)):
+        attention_windows = [None] * num_attention_layers
+    else:
+        local_window_pattern = _project_max_attention_window_vec(
+            window_pattern,
+            local_layer_ids,
+        )
+        attention_windows = [
+            normalize_window_size(window_size) for window_size in local_window_pattern
+        ]
     return layer_sizes, attention_windows
 
 
@@ -740,31 +766,43 @@ def _update_kv_cache_draft_token_location(
     else:
         use_paged_kv_cache = False
     assert use_paged_kv_cache, "Only paged kv cache is supported"
-    assert len(cache_manager.max_attention_window_vec) == 1, (
+    assert len(set(cache_manager.max_attention_window_vec)) == 1, (
         "Currently, only one max attention window size is supported."
     )
+    max_attention_window = cache_manager.max_attention_window_vec[0]
+    if max_attention_window is None:
+        max_attention_window = cache_manager.max_seq_len
 
     if use_paged_kv_cache:
         assert len(set(cache_manager.num_kv_heads_per_layer)) == 1, (
             "update_kv_cache_draft_token_location requires uniform num_kv_heads across all layers, "
             f"but got {cache_manager.num_kv_heads_per_layer}"
         )
+        local_pool_ids = {
+            int(cache_manager.kv_cache_pool_mapping[layer_idx][0])
+            for layer_idx in range(cache_manager.num_local_layers)
+        }
+        assert len(local_pool_ids) == 1, (
+            "update_kv_cache_draft_token_location requires all local layers "
+            f"in one KV pool, got pools {sorted(local_pool_ids)}"
+        )
+        pool_idx = local_pool_ids.pop()
         torch.ops.tensorrt_llm.update_kv_cache_draft_token_location(
             accepted_draft_token_offsets,
             packed_accepted_draft_tokens_indices,
             past_key_value_lengths,
             True,
-            cache_manager.num_layers,
+            cache_manager.num_local_layers,
             # Use TP-sharded num_kv_heads (per-rank) instead of the unsharded
             # total so the C++ kernel computes correct strides and grid dims.
             cache_manager.num_kv_heads_per_layer[0],
             int(cache_manager.head_dim * kv_cache_dtype_byte_size),
             cache_manager.max_total_draft_tokens,
-            cache_manager.max_attention_window_vec[0],
+            max_attention_window,
             rewind_draft_token_separate_adjustments,
             None,
-            cache_manager.kv_cache_pool_pointers,
-            attn_metadata.kv_cache_block_offsets,
+            cache_manager.kv_cache_pool_pointers[pool_idx],
+            attn_metadata.kv_cache_block_offsets[pool_idx],
             cache_manager.max_blocks_per_seq,
             cache_manager.tokens_per_block,
             None,
@@ -966,24 +1004,13 @@ def __init__(
         )
         logger.info(f"[KVCacheManager] execution_stream: {self._stream}")
 
-        # Determine max_attention_window_vec
-        if kv_cache_config.max_attention_window is not None:
-            self.max_attention_window_vec = (
-                kv_cache_config.max_attention_window.copy()
-            )  # Make a copy to avoid modifying original
-            # Clamp all window sizes to max_seq_len before calculating the
-            # number of KV cache blocks. This prevents the KV cache pool from
-            # being skewed by the largest window values.
-            self.max_attention_window_vec = [
-                min(max_seq_len, w) for w in self.max_attention_window_vec
-            ]
-
-            self.max_attention_window_vec = [
-                None if w == max_seq_len else w for w in self.max_attention_window_vec
-            ]
-
-        else:
-            self.max_attention_window_vec = [None]
+        # Materialize an exact per-local-layer vector for cache and attention consumers.
+        self.max_attention_window_vec = _resolve_v2_max_attention_window_vec(
+            kv_cache_config.max_attention_window,
+            max_seq_len,
+            self.pp_layers,
+        )
+        assert len(self.max_attention_window_vec) == self.num_local_layers
 
         event_window_size = max(
             self.max_seq_len if window_size is None else int(window_size)
@@ -1604,14 +1631,11 @@ def _kv_pool_mapping_offset(
     def _get_runtime_cache_size_layer_components(self) -> tuple[List[int], List[Optional[int]]]:
         layer_sizes = []
         attention_windows = []
-        pattern_len = len(self.max_attention_window_vec)
         for local_layer_idx in range(self.num_local_layers):
             layer_sizes.append(
                 self.get_layer_bytes_per_token(local_layer_idx=local_layer_idx, data_role=Role.ALL)
             )
-            attention_windows.append(
-                self.max_attention_window_vec[self.pp_layers[local_layer_idx] % pattern_len]
-            )
+            attention_windows.append(self.max_attention_window_vec[local_layer_idx])
         return layer_sizes, attention_windows
 
     def _get_max_tokens_from_quota(self, quota: int) -> float:
@@ -2083,9 +2107,7 @@ def _build_base_config(
                 AttentionLayerConfig(
                     layer_id=layer_id,
                     buffers=buffers,
-                    sliding_window_size=self.max_attention_window_vec[
-                        self.pp_layers[layer_id] % len(self.max_attention_window_vec)
-                    ],
+                    sliding_window_size=self.max_attention_window_vec[layer_id],
                     num_sink_tokens=None,
                 )
             )
diff --git a/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py b/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py
index 3f0aaed94e52..44eb4b14a912 100644
--- a/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py
+++ b/tensorrt_llm/_torch/pyexecutor/mamba_cache_manager.py
@@ -2332,12 +2332,13 @@ def __init__(
             )
             kv_cache_config.enable_partial_reuse = False
 
-        kv_cache_config.max_attention_window = []
-        for i in range(len(layer_mask)):
-            if layer_mask[i]:
-                kv_cache_config.max_attention_window.append(
-                    LinearCacheType.RECURRENT_STATES.
-                    value if mamba_layer_mask[i] else max_seq_len)
+        # Keep the vector in physical global-layer order. Disabled entries are
+        # placeholders that are never projected into this manager.
+        kv_cache_config.max_attention_window = [
+            LinearCacheType.RECURRENT_STATES.value
+            if mamba_layer_mask[layer_idx] else max_seq_len
+            for layer_idx in range(len(layer_mask))
+        ]
 
         recurrent_states_window = LinearCacheType.RECURRENT_STATES.value
         local_windows = {
diff --git a/tensorrt_llm/_torch/pyexecutor/resource_manager.py b/tensorrt_llm/_torch/pyexecutor/resource_manager.py
index bd8016d25efd..6e4fa18e7e82 100644
--- a/tensorrt_llm/_torch/pyexecutor/resource_manager.py
+++ b/tensorrt_llm/_torch/pyexecutor/resource_manager.py
@@ -24,15 +24,13 @@
                     Set, Tuple, Union)
 
 import torch
-from mpi4py import MPI
 
 import tensorrt_llm
 import tensorrt_llm.bindings
 from tensorrt_llm._torch.distributed.communicator import Distributed, ReduceOp
 from tensorrt_llm._torch.peft.lora.config import LoraConfig
 from tensorrt_llm._torch.peft.lora.manager import LoraManager, LoraModelConfig
-from tensorrt_llm._utils import (get_size_in_bytes, mpi_comm, mpi_disabled,
-                                 prefer_pinned, torch_comm,
+from tensorrt_llm._utils import (get_size_in_bytes, prefer_pinned,
                                  torch_dtype_to_binding)
 from tensorrt_llm.bindings.internal.batch_manager import (
     LinearAttentionMetadata, LinearCacheType)
@@ -80,6 +78,63 @@
     int]]  # window_size -> (blocks_in_primary_pool, blocks_in_secondary_pool)
 
 
+def _clamp_max_attention_window_vec(
+    max_attention_window_vec: Optional[Sequence[int]],
+    max_seq_len: int,
+) -> List[int]:
+    """Copy and clamp a window pattern without mutating it."""
+    windows = ([max_seq_len] if max_attention_window_vec is None else
+               list(max_attention_window_vec))
+    return [min(max_seq_len, window) for window in windows]
+
+
+def _project_max_attention_window_vec(
+    max_attention_window_vec: Sequence[int],
+    pp_layers: Sequence[int],
+) -> List[int]:
+    """Project a global window vector into cache-local layer order.
+
+    Window vectors are repeating patterns anchored at global model layer zero.
+    The projected result has one entry for every PP-local cache layer, in local
+    layer order.
+    """
+    pattern_len = len(max_attention_window_vec)
+    return [
+        max_attention_window_vec[layer_idx % pattern_len]
+        for layer_idx in pp_layers
+    ]
+
+
+def _get_minimum_blocks_per_window(
+    local_blocks_per_window: BlocksPerWindow,
+    rank_blocks_per_window: Sequence[BlocksPerWindow],
+    rank_attention_window_sets: Sequence[Set[int]],
+) -> BlocksPerWindow:
+    """Conservatively align local pool capacities across distributed ranks."""
+    if any(window_set != rank_attention_window_sets[0]
+           for window_set in rank_attention_window_sets[1:]):
+        raise RuntimeError(
+            "Asymmetrical pipeline parallelism is not supported by KV cache "
+            "manager V1: ranks host different attention window sets: "
+            f"{rank_attention_window_sets}")
+
+    reduced_blocks_per_window = {}
+    for window_size in local_blocks_per_window:
+        # Attention window sets were validated above. Recurrent-state pools
+        # use different units and may be absent on some PP stages, so both
+        # kinds of pool are reduced only against their matching key.
+        candidate_blocks = [
+            rank_blocks[window_size] for rank_blocks in rank_blocks_per_window
+            if window_size in rank_blocks
+        ]
+        assert candidate_blocks
+        reduced_blocks_per_window[window_size] = (
+            min(blocks[0] for blocks in candidate_blocks),
+            min(blocks[1] for blocks in candidate_blocks),
+        )
+    return reduced_blocks_per_window
+
+
 @dataclass
 class PoolConfiguration:
     """Configuration of a single KV pool.
@@ -477,8 +532,6 @@ def append_to_kv_heads_per_layer(num_kv_heads_per_layer: List[int],
         self.max_attention_window_vec = self._resolve_max_attention_window_vec(
             kv_cache_config=kv_cache_config,
             max_seq_len=max_seq_len,
-            num_layers=num_layers,
-            layer_mask=layer_mask,
             pool_configurations=self.pool_configurations,
         )
 
@@ -553,38 +606,6 @@ def append_to_kv_heads_per_layer(num_kv_heads_per_layer: List[int],
                     kv_cache_config=kv_cache_config,
                     extra_cost_memory=0,
                 )
-                if mapping.world_size > 1:
-                    # make sure all ranks use the same number of primary/secondary blocks
-                    if mpi_disabled():
-                        for window_size, (
-                                primary_blocks,
-                                secondary_blocks) in blocks_per_window.items():
-                            reduced_primary_blocks = torch_comm().allreduce(
-                                primary_blocks,
-                                op=torch.distributed.ReduceOp.MIN)
-                            reduced_secondary_blocks = torch_comm().allreduce(
-                                secondary_blocks,
-                                op=torch.distributed.ReduceOp.MIN)
-                            blocks_per_window[window_size] = (
-                                reduced_primary_blocks,
-                                reduced_secondary_blocks)
-                    else:
-                        for window_size, (
-                                primary_blocks,
-                                secondary_blocks) in blocks_per_window.items():
-                            reduced_primary_blocks = mpi_comm().allreduce(
-                                primary_blocks, op=MPI.MIN)
-                            reduced_secondary_blocks = mpi_comm().allreduce(
-                                secondary_blocks, op=MPI.MIN)
-                            blocks_per_window[window_size] = (
-                                reduced_primary_blocks,
-                                reduced_secondary_blocks)
-                    logger.info(
-                        f"[MPI rank={mapping.rank}] Original blocks_per_window: {blocks_per_window}"
-                    )
-                    logger.info(
-                        f"[MPI rank={mapping.rank}] Reduced blocks_per_window: {blocks_per_window}"
-                    )
             else:
                 # Standard case: use original Python implementation
                 self.blocks_in_primary_pool, self.blocks_in_secondary_pool = self.calculate_max_num_blocks(
@@ -594,11 +615,50 @@ def append_to_kv_heads_per_layer(num_kv_heads_per_layer: List[int],
                     mapping=mapping,
                     dtype=dtype,
                     kv_factor=self.kv_factor,
+                    synchronize_across_ranks=False,
                 )
                 blocks_per_window = {
                     self.max_attention_window_vec[0]:
                     (self.blocks_in_primary_pool, self.blocks_in_secondary_pool)
                 }
+            if mapping.world_size > 1:
+                # PP ranks can own different window sets or cache types.
+                # Gather once after either sizing path so every rank executes
+                # the same collective, then align local pool capacities.
+                local_blocks_per_window = blocks_per_window.copy()
+                local_attention_windows = {
+                    window_size
+                    for window_size in self.max_attention_window_vec
+                    if window_size != LinearCacheType.RECURRENT_STATES.value
+                }
+                rank_sizing_states = Distributed.get(mapping).allgather(
+                    (blocks_per_window, local_attention_windows))
+                rank_blocks_per_window = [
+                    state[0] for state in rank_sizing_states
+                ]
+                rank_attention_window_sets = [
+                    state[1] for state in rank_sizing_states
+                ]
+                blocks_per_window = _get_minimum_blocks_per_window(
+                    local_blocks_per_window,
+                    rank_blocks_per_window,
+                    rank_attention_window_sets,
+                )
+                logger.info(
+                    f"[MPI rank={mapping.rank}] Original blocks_per_window: {local_blocks_per_window}"
+                )
+                logger.info(
+                    f"[MPI rank={mapping.rank}] Reduced blocks_per_window: {blocks_per_window}"
+                )
+
+        if len(blocks_per_window) == 1:
+            window_size, pool_blocks = next(iter(blocks_per_window.items()))
+            if window_size != LinearCacheType.RECURRENT_STATES.value:
+                # Single-pool consumers use these legacy scalar attributes.
+                # Refresh them after distributed reduction so they match the
+                # capacity passed to the C++ block manager.
+                (self.blocks_in_primary_pool,
+                 self.blocks_in_secondary_pool) = pool_blocks
 
         # Validate and adjust attention windows against their upper bounds if needed
         blocks_per_window, self.max_seq_len, self.max_attention_window_vec, window_adjustments = self._validate_and_adjust_attention_windows(
@@ -730,11 +790,9 @@ def append_to_kv_heads_per_layer(num_kv_heads_per_layer: List[int],
                 for config in self.impl.pool_configurations
             }
             # Match the local layer order used by C++ to build the pointer
-            # mapping. The Python window helpers additionally account for
-            # global PP layer IDs and therefore do not describe these rows.
+            # mapping.
             layer_pool_dtypes = [
-                dtype_by_window[self.max_attention_window_vec[
-                    layer_offset % len(self.max_attention_window_vec)]]
+                dtype_by_window[self.max_attention_window_vec[layer_offset]]
                 for layer_offset in range(self.num_local_layers)
             ]
             self.kv_cache_pool_pointers = _merge_kv_cache_pool_pointers(
@@ -1452,23 +1510,16 @@ def _resolve_max_attention_window_vec(
         self,
         kv_cache_config: KvCacheConfig,
         max_seq_len: int,
-        num_layers: int,
-        layer_mask: Optional[List[bool]],
         pool_configurations: Optional[List["PoolConfiguration"]] = None,
     ) -> List[int]:
         """Compute the per-local-layer attention window vector.
 
-        Three input shapes are supported:
+        Two input shapes are supported and projected into local layer order:
 
-        * ``max_attention_window is None``: use ``max_seq_len`` as the only
-          entry (single-window default).
-        * ``len(max_attention_window) == num_layers``: the user supplied a
-          global per-layer pattern. Shard it down to this PP rank using
-          ``layer_mask`` + ``self.pp_layers`` / ``self.layer_offsets``,
-          clamping each entry to ``max_seq_len``.
-        * Otherwise: use the user-supplied vector verbatim, clamped
-          element-wise to ``max_seq_len`` so the largest window can't skew
-          the KV cache pool sizing.
+        * ``max_attention_window is None``: use ``max_seq_len`` for every
+          local layer (single-window default).
+        * Otherwise: treat the user-supplied vector as a repeating pattern
+          anchored at global model layer zero.
 
         ``pool_configurations`` (if given) are clamped in place to the same
         ``max_seq_len`` bound so their window keys stay consistent with the
@@ -1476,29 +1527,13 @@ def _resolve_max_attention_window_vec(
         """
         for pc in pool_configurations or []:
             pc.window_size = min(pc.window_size, max_seq_len)
-        if kv_cache_config.max_attention_window is None:
-            return [max_seq_len]
-        if len(kv_cache_config.max_attention_window) == num_layers:
-            if layer_mask is not None:
-                global_enabled_layers = [
-                    layer_idx for layer_idx in range(len(layer_mask))
-                    if layer_mask[layer_idx]
-                ]
-            else:
-                global_enabled_layers = list(range(num_layers))
-            pp_rank_offset = global_enabled_layers.index(self.pp_layers[0])
-            sharded = []
-            for layer_idx in self.pp_layers:
-                if layer_mask is not None and not layer_mask[layer_idx]:
-                    continue
-                window_size = kv_cache_config.max_attention_window[
-                    pp_rank_offset + self.layer_offsets[layer_idx]]
-                sharded.append(min(window_size, max_seq_len))
-            return sharded
-        # General case: clamp each user-supplied entry to max_seq_len.
-        return [
-            min(max_seq_len, w) for w in kv_cache_config.max_attention_window
-        ]
+        clamped_window_pattern = _clamp_max_attention_window_vec(
+            kv_cache_config.max_attention_window, max_seq_len)
+        local_windows = _project_max_attention_window_vec(
+            clamped_window_pattern,
+            self.pp_layers,
+        )
+        return local_windows
 
     @staticmethod
     def _resolve_num_attention_layers(
@@ -1594,13 +1629,15 @@ def get_cache_bytes_per_token(self):
                 scaling_factor_dtype=DataType.FP8)
         return cache_size_bytes_per_token
 
-    def calculate_max_num_blocks(self,
-                                 kv_cache_config: KvCacheConfig,
-                                 head_dim: int,
-                                 tokens_per_block: int,
-                                 mapping: Mapping,
-                                 dtype: DataType,
-                                 kv_factor: int = 2):
+    def calculate_max_num_blocks(
+            self,
+            kv_cache_config: KvCacheConfig,
+            head_dim: int,
+            tokens_per_block: int,
+            mapping: Mapping,
+            dtype: DataType,
+            kv_factor: int = 2,
+            synchronize_across_ranks: bool = True) -> Tuple[int, int]:
         free_mem_fraction = (kv_cache_config.free_gpu_memory_fraction
                              if kv_cache_config.free_gpu_memory_fraction
                              is not None else 0.9)
@@ -1626,7 +1663,7 @@ def calculate_max_num_blocks(self,
                     f"max_tokens is set by kv_cache_config.max_tokens: {max_tokens}"
                 )
 
-        if mapping.world_size > 1:
+        if synchronize_across_ranks and mapping.world_size > 1:
             # make sure all ranks use same value for maxTokens
             dist = Distributed.get(mapping)
             max_tokens = dist.allreduce(
@@ -2073,12 +2110,7 @@ def get_pool_for_layer(self,
     def _get_layer_offset_to_window_size(self) -> Dict[int, int]:
         """Inverse of _get_window_size_to_layers: layer_offset -> window_size.
 
-        Asserts every local layer is mapped exactly once.  This is the
-        explicit, length-mismatch-safe replacement for
-        ``max_attention_window_vec[layer_offset % len(max_attention_window_vec)]``
-        — that modulo silently masks length mismatches between the window
-        pattern and num_local_layers; this helper catches them via the
-        assert below.
+        Asserts every local layer is mapped exactly once.
         """
         window_size_to_layers = self._get_window_size_to_layers()
         layer_offset_to_window_size: Dict[int, int] = {}
@@ -2100,7 +2132,7 @@ def _get_window_size_to_layers(self) -> dict[int, list[int]]:
         Get the window size to layers mapping.
         The returned map has window sizes as keys and lists of layer indices as values.
 
-        max_attention_window_vec is treated as a repeating pattern.
+        max_attention_window_vec contains one entry per local cache layer.
         """
         window_size_to_layers_map = defaultdict(list)
 
@@ -2114,20 +2146,13 @@ def _get_window_size_to_layers(self) -> dict[int, list[int]]:
             return {
             }  # Return an empty dict if no local layers or if somehow vec is empty and no layers.
 
-        # Treat max_attention_window_vec as a repeating pattern.
-        pattern_len = len(
-            self.max_attention_window_vec
-        )  # `sliding_window_pattern`, in HF config terms, e.g. https://huggingface.co/google/gemma-3-1b-it/blob/main/config.json#L32
-        # early return if max_attention_window_vec is a single value(SWA)
-        if pattern_len == 1:
-            return {
-                self.max_attention_window_vec[0]:
-                list(range(self.num_local_layers))
-            }
-        for local_layer_idx in range(self.num_local_layers):
-            global_layer_idx = self.pp_layers[local_layer_idx]
-            window_size = self.max_attention_window_vec[global_layer_idx %
-                                                        pattern_len]
+        if len(self.max_attention_window_vec) != self.num_local_layers:
+            raise ValueError(
+                "max_attention_window_vec must contain one entry per local "
+                f"layer, got {len(self.max_attention_window_vec)} entries "
+                f"for {self.num_local_layers} layers")
+        for local_layer_idx, window_size in enumerate(
+                self.max_attention_window_vec):
             window_size_to_layers_map[window_size].append(local_layer_idx)
         return window_size_to_layers_map
 
diff --git a/tensorrt_llm/_torch/speculative/eagle3_dynamic_tree.py b/tensorrt_llm/_torch/speculative/eagle3_dynamic_tree.py
index 5711ecfa5060..f9aa0177816f 100644
--- a/tensorrt_llm/_torch/speculative/eagle3_dynamic_tree.py
+++ b/tensorrt_llm/_torch/speculative/eagle3_dynamic_tree.py
@@ -426,20 +426,37 @@ def _relocate_kv_eagerly(self, attn_metadata, batch_size):
             "update_kv_cache_draft_token_location_2d requires uniform num_kv_heads across all layers, "
             f"but got {cache_mgr.num_kv_heads_per_layer}"
         )
+        assert len(set(cache_mgr.max_attention_window_vec)) == 1, (
+            "update_kv_cache_draft_token_location_2d requires uniform "
+            "attention windows across all local layers, but got "
+            f"{cache_mgr.max_attention_window_vec}"
+        )
+        max_attention_window = cache_mgr.max_attention_window_vec[0]
+        if max_attention_window is None:
+            max_attention_window = cache_mgr.max_seq_len
+        local_pool_ids = {
+            int(cache_mgr.kv_cache_pool_mapping[layer_idx][0])
+            for layer_idx in range(cache_mgr.num_local_layers)
+        }
+        assert len(local_pool_ids) == 1, (
+            "update_kv_cache_draft_token_location_2d requires all local "
+            f"layers in one KV pool, got pools {sorted(local_pool_ids)}"
+        )
+        pool_idx = local_pool_ids.pop()
         torch.ops.tensorrt_llm.update_kv_cache_draft_token_location_2d(
             self._accepted_draft_indices_tensor[:batch_size],
             self._num_accepted_tokens_buf[:batch_size],
             attn_metadata.kv_lens_cuda[:batch_size],
             True,
-            cache_mgr.num_layers,
+            cache_mgr.num_local_layers,
             # Use TP-sharded num_kv_heads (per-rank) instead of the unsharded
             # total so the C++ kernel computes correct strides and grid dims.
             cache_mgr.num_kv_heads_per_layer[0],
             self._kv_head_dim_bytes,
             cache_mgr.max_total_draft_tokens,
-            cache_mgr.max_attention_window_vec[0],
-            cache_mgr.kv_cache_pool_pointers,
-            attn_metadata.kv_cache_block_offsets,
+            max_attention_window,
+            cache_mgr.kv_cache_pool_pointers[pool_idx],
+            attn_metadata.kv_cache_block_offsets[pool_idx],
             cache_mgr.max_blocks_per_seq,
             cache_mgr.tokens_per_block,
             None,
diff --git a/tests/unittest/_torch/attention/test_flashinfer_attention.py b/tests/unittest/_torch/attention/test_flashinfer_attention.py
index 0b933623acd5..ebd696196703 100644
--- a/tests/unittest/_torch/attention/test_flashinfer_attention.py
+++ b/tests/unittest/_torch/attention/test_flashinfer_attention.py
@@ -27,6 +27,7 @@
 from tensorrt_llm.bindings.executor import KvCacheConfig
 from tensorrt_llm.functional import AttentionMaskType
 from tensorrt_llm.mapping import Mapping
+from tensorrt_llm.runtime.kv_cache_manager_v2._common import BAD_PAGE_INDEX
 
 
 class TestingFlashInferAttentionMetadata(FlashInferAttentionMetadata):
@@ -71,6 +72,30 @@ class CUDAGraphTestScenario:
 
 class TestFlashInferAttention(unittest.TestCase):
 
+    def test_swa_page_sanitization_uses_local_window_order(self) -> None:
+        metadata = object.__new__(FlashInferAttentionMetadata)
+        metadata.kv_cache_manager = SimpleNamespace(
+            layer_offsets={
+                4: 0,
+                5: 1,
+                6: 2
+            },
+            max_attention_window_vec=[128, None, 128],
+        )
+
+        sliding_page_indices = torch.tensor([BAD_PAGE_INDEX, 3],
+                                            dtype=torch.int32)
+        metadata._sanitize_swa_page_indices(sliding_page_indices, layer_idx=4)
+        torch.testing.assert_close(sliding_page_indices,
+                                   torch.tensor([0, 3], dtype=torch.int32))
+
+        full_page_indices = torch.tensor([BAD_PAGE_INDEX, 3], dtype=torch.int32)
+        metadata._sanitize_swa_page_indices(full_page_indices, layer_idx=5)
+        torch.testing.assert_close(
+            full_page_indices,
+            torch.tensor([BAD_PAGE_INDEX, 3], dtype=torch.int32),
+        )
+
     def test_generation_page_table_uses_reserved_block_count(self):
         manager = SimpleNamespace(get_batch_cache_indices=mock.Mock(
             return_value=[list(range(325))]))
diff --git a/tests/unittest/_torch/executor/test_kv_cache_estimation.py b/tests/unittest/_torch/executor/test_kv_cache_estimation.py
index c984016658bf..76fba30b31ce 100644
--- a/tests/unittest/_torch/executor/test_kv_cache_estimation.py
+++ b/tests/unittest/_torch/executor/test_kv_cache_estimation.py
@@ -701,6 +701,37 @@ def get_num_attention_layers(self):
     assert cost == CacheCost(slope=0, intercept=expected_configured_slots * slot_bytes)
 
 
+def test_v2_static_cache_size_preserves_window_pattern_phase_across_pp() -> None:
+    class FakeModelConfig:
+        quant_config = None
+        pretrained_config = SimpleNamespace(
+            hidden_size=32,
+            num_attention_heads=4,
+            num_key_value_heads=2,
+        )
+
+        def get_num_attention_layers(self) -> int:
+            return 5
+
+    mapping = Mock(enable_attention_dp=False, tp_size=1)
+    mapping.pp_layers.return_value = [3]
+
+    cache_cost = CacheCost.from_raw(
+        KVCacheManagerV2.get_cache_size_per_token(
+            FakeModelConfig(),
+            mapping,
+            tokens_per_block=64,
+            max_seq_len=256,
+            max_batch_size=1,
+            kv_cache_config=KvCacheConfig(max_attention_window=[128, 256]),
+        )
+    )
+
+    # Global layer 3 selects the full-attention entry, so its 64-byte layer
+    # cost is entirely per-token with no fixed SWA allocation.
+    assert cache_cost == CacheCost(slope=64, intercept=0)
+
+
 def test_creator_uses_v2_affine_cache_cost():
     class FakeV2Manager(KVCacheManagerV2):
         @staticmethod
@@ -724,7 +755,7 @@ def test_v2_quota_from_max_tokens_models_context_swa_scratch():
     manager = object.__new__(KVCacheManagerV2)
     manager._has_cp_helix = False
     manager.num_local_layers = 3
-    manager.pp_layers = [0, 1, 2]
+    manager.pp_layers = [4, 5, 6]
     manager.max_attention_window_vec = [128, 128, None]
     manager.tokens_per_block = 64
     manager.max_batch_size = 4
diff --git a/tests/unittest/_torch/executor/test_kv_cache_manager_v2.py b/tests/unittest/_torch/executor/test_kv_cache_manager_v2.py
index d2636a492161..54abfb345976 100644
--- a/tests/unittest/_torch/executor/test_kv_cache_manager_v2.py
+++ b/tests/unittest/_torch/executor/test_kv_cache_manager_v2.py
@@ -22,12 +22,15 @@
 import torch
 
 from tensorrt_llm._torch.distributed.communicator import Distributed, ReduceOp
+from tensorrt_llm._torch.pyexecutor import kv_cache_manager_v2 as kv_cache_v2_module
 from tensorrt_llm._torch.pyexecutor.kv_cache_manager_v2 import (
     BlockReusePolicy,
     KVCacheManagerV2,
     _KVCacheManagerInitStatus,
     _sync_kv_cache_manager_init_status,
+    _update_kv_cache_draft_token_location,
 )
+from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequestState
 from tensorrt_llm._torch.pyexecutor.scheduler import ScheduledRequests
 from tensorrt_llm.bindings import DataType
 from tensorrt_llm.bindings.BuildInfo import ENABLE_MULTI_DEVICE
@@ -83,19 +86,27 @@ def _make_cache_config_for_test(
     max_num_tokens: int | None = None,
     max_draft_len: int = 0,
     num_extra_kv_tokens: int = 0,
+    max_attention_window_vec: list[int | None] | None = None,
+    pp_layers: list[int] | None = None,
 ) -> KVCacheManagerConfig:
+    if max_attention_window_vec is None:
+        max_attention_window_vec = [None]
+    if pp_layers is None:
+        pp_layers = list(range(len(max_attention_window_vec)))
+    assert len(max_attention_window_vec) == len(pp_layers)
+
     cache_manager = object.__new__(KVCacheManagerV2)
     cache_manager.kv_cache_type = CacheType.SELFKONLY
     cache_manager.dtype = DataType.HALF
-    cache_manager.head_dim_per_layer = [128]
+    cache_manager.head_dim_per_layer = [128] * len(pp_layers)
     cache_manager.enable_swa_scratch_reuse = False
     cache_manager.num_extra_kv_tokens = num_extra_kv_tokens
     cache_manager.enable_stats = False
     cache_manager.block_reuse_policy = BlockReusePolicy(kv_cache_config.block_reuse_config.policy)
     cache_manager.is_draft = is_draft
-    cache_manager.num_local_layers = 1
-    cache_manager.pp_layers = [0]
-    cache_manager.max_attention_window_vec = [None]
+    cache_manager.num_local_layers = len(pp_layers)
+    cache_manager.pp_layers = pp_layers
+    cache_manager.max_attention_window_vec = max_attention_window_vec
     cache_manager.max_seq_len = max_seq_len
     cache_manager.max_batch_size = max_batch_size
     cache_manager.max_num_tokens = max_num_tokens
@@ -227,6 +238,117 @@ def _multi_rank_host_fallback_consensus_worker() -> tuple[int, int, int, bool]:
     )
 
 
+def test_base_config_uses_local_attention_window_order() -> None:
+    config = _make_cache_config_for_test(
+        KvCacheConfig(),
+        max_attention_window_vec=[128, None],
+        pp_layers=[3, 4],
+    )
+
+    assert [layer.sliding_window_size for layer in config.layers] == [
+        128,
+        None,
+    ]
+
+
+def test_draft_token_relocation_uses_local_cache_layout(monkeypatch: pytest.MonkeyPatch) -> None:
+    request = SimpleNamespace(
+        state=LlmRequestState.GENERATION_IN_PROGRESS,
+        py_num_accepted_draft_tokens=1,
+        py_num_accepted_draft_tokens_indices=[0],
+    )
+    batch = ScheduledRequests()
+    batch.generation_requests = [request]
+
+    accepted_offsets = object()
+    accepted_indices = object()
+    rewind_adjustments = object()
+
+    def locate_accepted_draft_tokens(
+        requests: list[object],
+    ) -> tuple[object, object, object]:
+        del requests
+        return accepted_offsets, accepted_indices, rewind_adjustments
+
+    monkeypatch.setattr(
+        kv_cache_v2_module,
+        "_locate_accepted_draft_tokens",
+        locate_accepted_draft_tokens,
+    )
+
+    local_pool_pointers = object()
+    local_block_offsets = object()
+    cache_manager = SimpleNamespace(
+        num_layers=8,
+        num_local_layers=2,
+        num_kv_heads_per_layer=[8, 8],
+        head_dim=128,
+        max_attention_window_vec=[None, None],
+        max_seq_len=8192,
+        max_total_draft_tokens=31,
+        max_blocks_per_seq=256,
+        tokens_per_block=32,
+        kv_cache_pool_mapping=[[0, 0], [0, 1]],
+        kv_cache_pool_pointers=[local_pool_pointers],
+    )
+    attention_metadata = SimpleNamespace(
+        kv_lens_cuda=torch.tensor([128], dtype=torch.int32),
+        kv_cache_block_offsets=[local_block_offsets],
+        host_kv_cache_pool_pointers=object(),
+        host_kv_cache_pool_mapping=object(),
+    )
+    update_op = Mock()
+    monkeypatch.setattr(
+        torch.ops.tensorrt_llm,
+        "update_kv_cache_draft_token_location",
+        update_op,
+        raising=False,
+    )
+
+    _update_kv_cache_draft_token_location(
+        cache_manager,
+        batch,
+        attention_metadata,
+        kv_cache_dtype_byte_size=2,
+    )
+
+    update_op.assert_called_once()
+    (
+        actual_accepted_offsets,
+        actual_accepted_indices,
+        past_key_value_lengths,
+        use_paged_kv_cache,
+        layer_count,
+        num_kv_heads,
+        head_size_in_bytes,
+        rewind_draft_token_count,
+        max_kv_cache_len,
+        actual_rewind_adjustments,
+        past_key_value_list,
+        pool_pointers,
+        block_offsets,
+        max_blocks_per_seq,
+        tokens_per_block,
+        stream,
+    ) = update_op.call_args.args
+    assert actual_accepted_offsets is accepted_offsets
+    assert actual_accepted_indices is accepted_indices
+    assert torch.equal(past_key_value_lengths, attention_metadata.kv_lens_cuda)
+    assert use_paged_kv_cache is True
+    assert layer_count == cache_manager.num_local_layers
+    assert num_kv_heads == 8
+    assert head_size_in_bytes == 256
+    assert rewind_draft_token_count == cache_manager.max_total_draft_tokens
+    assert max_kv_cache_len == cache_manager.max_seq_len
+    assert actual_rewind_adjustments is rewind_adjustments
+    assert past_key_value_list is None
+    assert pool_pointers is local_pool_pointers
+    assert block_offsets is local_block_offsets
+    assert max_blocks_per_seq == cache_manager.max_blocks_per_seq
+    assert tokens_per_block == cache_manager.tokens_per_block
+    assert stream is None
+
+
 @pytest.mark.parametrize(
     ("enable_block_reuse", "block_reuse_policy", "is_draft", "commit_min_snapshot"),
     [
diff --git a/tests/unittest/_torch/executor/test_mamba_cache_manager.py b/tests/unittest/_torch/executor/test_mamba_cache_manager.py
index 2f0a0efd5b34..06378da704c2 100644
--- a/tests/unittest/_torch/executor/test_mamba_cache_manager.py
+++ b/tests/unittest/_torch/executor/test_mamba_cache_manager.py
@@ -3640,6 +3640,12 @@ def test_cpp_hybrid_merges_compact_scale_rows_with_unmanaged_layers():
     )
 
     assert mgr.pp_layers == [0, 2, 3]
+    recurrent_window = LinearCacheType.RECURRENT_STATES.value
+    assert mgr.max_attention_window_vec == [
+        recurrent_window,
+        recurrent_window,
+        mgr.max_seq_len,
+    ]
     assert mgr.kv_cache_pool_mapping[:, 0].tolist() == [0, 0, 1]
     compact_scale_pointers = mgr.impl.get_block_scale_pool_pointers()
     assert compact_scale_pointers.shape == (1, 2)
diff --git a/tests/unittest/_torch/executor/test_resource_manager.py b/tests/unittest/_torch/executor/test_resource_manager.py
index 34db8b0bf8e3..0601620bd0f7 100644
--- a/tests/unittest/_torch/executor/test_resource_manager.py
+++ b/tests/unittest/_torch/executor/test_resource_manager.py
@@ -22,14 +22,14 @@
 from tensorrt_llm._torch.pyexecutor.py_executor import PyExecutor
 from tensorrt_llm._torch.pyexecutor.resource_manager import (
     KVCacheManager, PeftCacheManager, ResourceManager, ResourceManagerType,
-    _merge_kv_cache_pool_pointers,
+    _get_minimum_blocks_per_window, _merge_kv_cache_pool_pointers,
     _warn_if_unsupported_v1_kv_cache_event_hash_algo)
 from tensorrt_llm._torch.pyexecutor.scheduler import ScheduledRequests
 from tensorrt_llm.bindings import LayerType
 from tensorrt_llm.bindings import ModelConfig as ModelConfigCpp
 from tensorrt_llm.bindings import executor as tllm
-from tensorrt_llm.bindings.internal.batch_manager import \
-    PeftTaskNotCachedException
+from tensorrt_llm.bindings.internal.batch_manager import (
+    LinearCacheType, PeftTaskNotCachedException)
 from tensorrt_llm.bindings.internal.testing import \
     simulate_prefill_completion_only_use_for_testing
 from tensorrt_llm.llmapi.llm_args import KvCacheConfig, PeftCacheConfig
@@ -75,6 +75,80 @@ def test_v1_kv_cache_event_hash_algo_no_warning_for_auto():
     warning.assert_not_called()
 
 
+def test_minimum_blocks_per_window_aligns_attention_pool_capacity() -> None:
+    recurrent_window = LinearCacheType.RECURRENT_STATES.value
+    local_blocks = {
+        64: (8, 4),
+        128: (7, 3),
+        recurrent_window: (20, 0),
+    }
+    rank_blocks = [
+        local_blocks,
+        {
+            64: (5, 2),
+            128: (6, 1),
+        },
+        {
+            64: (6, 3),
+            128: (9, 2),
+            recurrent_window: (18, 0),
+        },
+    ]
+
+    assert _get_minimum_blocks_per_window(
+        local_blocks,
+        rank_blocks,
+        [{64, 128}, {64, 128}, {64, 128}],
+    ) == {
+        64: (5, 2),
+        128: (6, 1),
+        recurrent_window: (18, 0),
+    }
+
+
+def test_minimum_blocks_per_window_rejects_asymmetric_hosted_windows() -> None:
+    recurrent_window = LinearCacheType.RECURRENT_STATES.value
+    # Linear-attention sizing emits a positive placeholder key even when the
+    # rank only hosts recurrent-state layers. The hosted-window set, not the
+    # sizing keys, must expose the PP asymmetry.
+    with pytest.raises(RuntimeError, match="Asymmetrical pipeline parallelism"):
+        _get_minimum_blocks_per_window(
+            {
+                128: (8, 4),
+                recurrent_window: (20, 0),
+            },
+            [
+                {
+                    128: (8, 4),
+                    recurrent_window: (20, 0),
+                },
+                {
+                    128: (5, 2)
+                },
+            ],
+            [set(), {128}],
+        )
+
+
+def test_v1_window_resolution_and_grouping_use_local_layer_order() -> None:
+    manager = object.__new__(KVCacheManager)
+    manager.pp_layers = [3, 5]
+    manager.num_local_layers = 2
+    config = KvCacheConfig(max_attention_window=[64, 128, 512])
+
+    manager.max_attention_window_vec = manager._resolve_max_attention_window_vec(
+        config,
+        max_seq_len=256,
+    )
+
+    assert manager.max_attention_window_vec == [64, 256]
+    assert config.max_attention_window == [64, 128, 512]
+    assert manager._get_window_size_to_layers() == {
+        64: [0],
+        256: [1],
+    }
+
+
 class TestMergeKVCachePoolPointers(unittest.TestCase):
 
     def test_mixed_half_nvfp4_pools_align_scale_rows(self):
diff --git a/tests/unittest/_torch/speculative/test_eagle3.py b/tests/unittest/_torch/speculative/test_eagle3.py
index b372a8f77a17..e3e743fe3999 100644
--- a/tests/unittest/_torch/speculative/test_eagle3.py
+++ b/tests/unittest/_torch/speculative/test_eagle3.py
@@ -36,12 +36,15 @@
 from tensorrt_llm._torch.attention_backend.trtllm import TrtllmAttentionMetadata
 from tensorrt_llm._torch.metadata import KVCacheParams
 from tensorrt_llm._torch.peft.lora.config import LoraConfig
-from tensorrt_llm._torch.pyexecutor._util import \
-    _derive_draft_max_attention_window
+from tensorrt_llm._torch.pyexecutor._util import (
+    _derive_draft_max_attention_window,
+    _expand_attention_window_pattern_to_global_layers)
 from tensorrt_llm._torch.pyexecutor.py_executor_creator import \
     _extend_full_attention_windows_for_spec_decode
 from tensorrt_llm._torch.speculative.eagle3 import (Eagle3OneModelSpecMetadata,
                                                     MTPEagleWorker)
+from tensorrt_llm._torch.speculative.eagle3_dynamic_tree import \
+    Eagle3OneModelDynamicTreeWorker
 from tensorrt_llm._torch.speculative.interface import \
     INVALID_PROMPT_LOOKAHEAD_TOKEN
 from tensorrt_llm._torch.speculative.mtp_dynamic_tree import \
@@ -188,6 +191,77 @@ def test_mtp_dynamic_tree_relocation_uses_full_attention_window(
     assert args[10] is attention_block_offsets
 
 
+def test_eagle3_dynamic_tree_relocation_uses_local_cache_shape(
+        monkeypatch: pytest.MonkeyPatch) -> None:
+    worker = object.__new__(Eagle3OneModelDynamicTreeWorker)
+    worker._kv_head_dim_bytes = 256
+    worker._accepted_draft_indices_tensor = torch.tensor([[0, 1], [2, -1]],
+                                                         dtype=torch.int32)
+    worker._num_accepted_tokens_buf = torch.tensor([2, 1], dtype=torch.int32)
+
+    local_pool_pointers = object()
+    local_block_offsets = object()
+    cache_manager = SimpleNamespace(
+        num_layers=8,
+        num_local_layers=2,
+        num_kv_heads_per_layer=[8, 8],
+        max_attention_window_vec=[None, None],
+        max_seq_len=8192,
+        max_total_draft_tokens=31,
+        max_blocks_per_seq=256,
+        tokens_per_block=32,
+        kv_cache_pool_mapping=[[1, 0], [1, 1]],
+        kv_cache_pool_pointers=[object(), local_pool_pointers],
+    )
+    attention_metadata = SimpleNamespace(
+        kv_cache_manager=cache_manager,
+        kv_lens_cuda=torch.tensor([128, 256], dtype=torch.int32),
+        kv_cache_block_offsets=[object(), local_block_offsets],
+    )
+    update_op = MagicMock()
+    monkeypatch.setattr(
+        torch.ops.tensorrt_llm,
+        "update_kv_cache_draft_token_location_2d",
+        update_op,
+        raising=False,
+    )
+
+    worker._relocate_kv_eagerly(attention_metadata, batch_size=2)
+
+    update_op.assert_called_once()
+    (
+        accepted_draft_indices,
+        num_accepted_tokens,
+        past_key_value_lengths,
+        use_paged_kv_cache,
+        layer_count,
+        num_kv_heads,
+        head_size_in_bytes,
+        rewind_draft_token_count,
+        max_kv_cache_len,
+        pool_pointers,
+        block_offsets,
+        max_blocks_per_seq,
+        tokens_per_block,
+        stream,
+    ) = update_op.call_args.args
+    assert torch.equal(accepted_draft_indices,
+                       worker._accepted_draft_indices_tensor)
+    assert torch.equal(num_accepted_tokens, worker._num_accepted_tokens_buf)
+    assert torch.equal(past_key_value_lengths, attention_metadata.kv_lens_cuda)
+    assert use_paged_kv_cache is True
+    assert layer_count == cache_manager.num_local_layers
+    assert num_kv_heads == 8
+    assert head_size_in_bytes == worker._kv_head_dim_bytes
+    assert rewind_draft_token_count == cache_manager.max_total_draft_tokens
+    assert max_kv_cache_len == cache_manager.max_seq_len
+    assert pool_pointers is local_pool_pointers
+    assert block_offsets is local_block_offsets
+    assert max_blocks_per_seq == cache_manager.max_blocks_per_seq
+    assert tokens_per_block == cache_manager.tokens_per_block
+    assert stream is None
+
+
 def test_eagle3_draft_kv_cache_uses_full_window_when_draft_has_no_swa() -> None:
     kv_cache_config = KvCacheConfig(max_attention_window=[128, 131072])
     draft_pretrained_config = SimpleNamespace(num_hidden_layers=3)
@@ -202,7 +276,7 @@ def test_eagle3_draft_kv_cache_uses_full_window_when_draft_has_no_swa() -> None:
     assert max_attention_window is None
 
 
-def test_eagle3_draft_kv_cache_uses_draft_layer_types_for_swa() -> None:
+def test_eagle3_draft_kv_cache_expands_swa_in_global_layer_order() -> None:
     kv_cache_config = KvCacheConfig(max_attention_window=[128, 131072])
     draft_pretrained_config = SimpleNamespace(
         sliding_window=512,
@@ -216,7 +290,14 @@ def test_eagle3_draft_kv_cache_uses_draft_layer_types_for_swa() -> None:
         num_draft_layers=3,
     )
 
-    assert max_attention_window == [512, 4096, 512]
+    global_windows = _expand_attention_window_pattern_to_global_layers(
+        max_attention_window=max_attention_window,
+        layer_mask=[False, False, False, False, False, True, True, True],
+    )
+
+    # Slots 0-4 are inactive fillers; physical draft layers 5-7 preserve the
+    # derived sliding/full/sliding pattern.
+    assert global_windows == [512, 512, 512, 512, 512, 512, 4096, 512]
 
 
 def test_eagle3_draft_kv_cache_rejects_multiple_sliding_window_sizes() -> None: