'''
Description  :  
Author       : Boxin Zhang
Version      : 0.1.0
'''
# Adapted from
# https://github.com/huggingface/transformers/blob/v4.41.2/src/transformers/cache_utils.py
# Copyright 2018- The Hugging Face team. All rights reserved.
# Copyright (c) 2024 by KVCache.AI, All Rights Reserved.
import torch
import torch.nn as nn
import transformers
from transformers import Cache, PretrainedConfig
from typing import List, Optional, Dict, Any, Tuple

try:
    import torch_npu
    from ktransformers.util import utils
    from ktransformers.server.balance_serve.inference.forward_batch import ForwardMiniBatchCombine, ForwardMiniBatchSplit
    
    use_torch_npu = torch_npu.npu.is_available()
except:
    use_torch_npu = False
from transformers.models.llama.modeling_llama import LlamaDecoderLayer
from ktransformers.server.balance_serve.settings import sched_ext

class StaticCache(transformers.StaticCache):
    """
    Static Cache class to be used with `torch.compile(model)`.

    Parameters:
        config (`PretrainedConfig):
            The configuration file defining the shape-related attributes required to initialize the static cache.
        max_batch_size (`int`):
            The maximum batch size with which the model will be used.
        max_cache_len (`int`):
            The maximum sequence length with which the model will be used.
        device (`torch.device` or `dict`):
            The device on which the cache should be initialized. Should be the same as the layer.
            If a `dict`, it should contain the `device` key with the device name as the value.
        dtype (*optional*, defaults to `torch.float32`):
            The default `dtype` to use when initializing the layer.
    """

    def __init__(self, config: PretrainedConfig, max_batch_size: int, max_cache_len: int, device: torch.device| dict, dtype=None) -> None:
        Cache.__init__(self, layer_class_to_replicate=LlamaDecoderLayer)
        self._max_batch_size = max_batch_size

        if use_torch_npu:
            self.position = [0]

        self._max_cache_len = config.max_position_embeddings if max_cache_len is None else max_cache_len
        # Some model define a custom `head_dim` != config.hidden_size // config.num_attention_heads
        if config.architectures[0] == "DeepseekV3ForCausalLM":
            self.head_dim = config.qk_rope_head_dim
        else:
            self.head_dim = (
                config.head_dim if hasattr(config, "head_dim") else config.hidden_size // config.num_attention_heads
            )

        self.dtype = dtype if dtype is not None else torch.float32
        self.num_key_value_heads = (
            config.num_attention_heads if config.num_key_value_heads is None else config.num_key_value_heads
        )

        self.key_cache: List[torch.Tensor] = []
        self.value_cache: List[torch.Tensor] = []
        cache_shape = (max_batch_size, self.num_key_value_heads, self._max_cache_len, self.head_dim)
        if config.architectures[0] == "DeepseekV2ForCausalLM" or config.architectures[0] == "DeepseekV3ForCausalLM":
            # TODO: for deepseek, cache_shape is different whether using Absorbed MLA, check it automatically

            if use_torch_npu:
                self.page_size = 128
                self.page_size_tensor = torch.tensor(
                self.page_size,
                dtype=torch.int32,
                    ).npu()
                self.max_pages_per_batch = (self._max_cache_len + self.page_size - 1) // self.page_size
                self.max_pages = (self._max_cache_len + self.page_size - 1) // self.page_size * self._max_batch_size
            else:
                self.page_size = 64
                self.max_pages = (self._max_cache_len + self.page_size - 1) // self.page_size
            latent_shape = (self.max_pages, self.page_size, 1, config.kv_lora_rank + config.qk_rope_head_dim)
            self.kv_lora_rank = config.kv_lora_rank
            self.qk_rope_head_dim = config.qk_rope_head_dim
            # TODO: support real page table
            self.page_table_map = dict()
            self.page_table_list = []
            for idx in range(config.num_hidden_layers):
                if isinstance(device, dict):
                    target_device = device[f"blk.{idx}.self_attn"]["generate_device"]
                else:
                    target_device = device
                
                if target_device not in self.page_table_map:
                    if use_torch_npu:
                        page_table = torch.zeros((max_batch_size, self.max_pages_per_batch), dtype=torch.int32, device=target_device)
                        for seq_id in range(max_batch_size):
                            page_table[seq_id, :] = torch.arange(seq_id * self.max_pages_per_batch, seq_id * self.max_pages_per_batch + self.max_pages_per_batch, dtype=torch.int32, device=target_device)
                    else:
                        page_table = torch.zeros((max_batch_size, self.max_pages), dtype=torch.int32, device=target_device)
                        for seq_id in range(max_batch_size):
                            page_table[seq_id, :] = torch.arange(seq_id * self.max_pages, seq_id * self.max_pages + self.max_pages, dtype=torch.int32, device=target_device)
                    self.page_table_map[target_device] = page_table
                    
                self.page_table_list.append(self.page_table_map[target_device])
                    
            self.is_MLA = True
            self.is_page = True
        else:
            key_shape = cache_shape
            value_shape = cache_shape
            self.is_MLA = False

        self.past_tokens = []
        self.num_hidden_layers = config.num_hidden_layers
        for idx in range(self.num_hidden_layers):
            # Note: `mark_static_address` is used to tag the cache as an fixed data pointer, preventing cuda graph
            # breaks when updating the cache.
            if isinstance(device, dict):
                target_device = device[f"blk.{idx}.self_attn"]["generate_device"]
            else:
                target_device = device
            
            if self.is_MLA:
                new_layer_key_cache = torch.zeros(latent_shape, dtype=self.dtype, device=target_device)
                new_layer_value_cache = None
                torch._dynamo.mark_static_address(new_layer_key_cache)
            else:
                new_layer_key_cache = torch.zeros(key_shape, dtype=self.dtype, device=target_device)
                new_layer_value_cache = torch.zeros(value_shape, dtype=self.dtype, device=target_device)
                torch._dynamo.mark_static_address(new_layer_key_cache)
                torch._dynamo.mark_static_address(new_layer_value_cache)
                
            self.key_cache.append(new_layer_key_cache)
            self.value_cache.append(new_layer_value_cache)
            self.past_tokens.append(0)

    @property
    def max_batch_size(self):
        return self._max_batch_size

    @property
    def max_cache_len(self):
        return self._max_cache_len

    def update(
        self,
        key_states: torch.Tensor,
        value_states: torch.Tensor,
        layer_idx: int,
        cache_kwargs: Optional[Dict[str, Any]] = None,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Updates the cache with the new `key_states` and `value_states` for the layer `layer_idx`.
        It is VERY important to index using a tensor, otherwise you introduce a copy to the device.

        Parameters:
            key_states (`torch.Tensor`):
                The new key states to cache.
            value_states (`torch.Tensor`):
                The new value states to cache.
            layer_idx (`int`):
                The index of the layer to cache the states for.
            cache_kwargs (`Dict[str, Any]`, `optional`):
                Additional arguments for the cache subclass. The `StaticCache` needs the `cache_position` input
                to know how where to write in the cache.

        Return:
            A tuple containing the updated key and value states.
        """
        cache_position = cache_kwargs.get("cache_position")
        k_out = self.key_cache[layer_idx]
        v_out = self.value_cache[layer_idx]
        self.past_tokens[layer_idx] += cache_position.size(0)
        #print(cache_position)
        if self.is_MLA:
            if use_torch_npu:
                page_idx = cache_position // self.page_size_tensor
                page_offset = cache_position % self.page_size_tensor

                page_idx = page_idx.unsqueeze(0).expand(self.max_batch_size, -1)
                page_offset = page_offset.unsqueeze(0).expand(self.max_batch_size, -1)

                page_idx_offset = torch.arange(self.max_batch_size, device=page_idx.device) * self.max_pages_per_batch
                page_idx = page_idx + page_idx_offset.unsqueeze(1)

                combined = torch.cat([key_states, value_states], dim=-1)
                combined = combined.contiguous()
                # key shape (self.max_pages, self.page_size, 1, config.kv_lora_rank + config.qk_rope_head_dim)
                k_out[page_idx, page_offset] = combined
            else:
                page_idx = cache_position // self.page_size
                page_offset = cache_position % self.page_size
                # key shape (self.max_pages, self.page_size, 1, config.kv_lora_rank + config.qk_rope_head_dim)
                k_out[page_idx, page_offset, :, :self.kv_lora_rank] = key_states
                k_out[page_idx, page_offset, :, self.kv_lora_rank:] = value_states
            return k_out, self.page_table_list[layer_idx]
        else:
            k_out[:, :, cache_position] = key_states
            v_out[:, :, cache_position] = value_states
            return k_out, v_out

    def get_seq_length(self, layer_idx: Optional[int] = 0) -> int:
        """Returns the sequence length of the cached states that were seen by the model."""
        # Occupied cache == any slot in the 3rd dim (sequence length) holds a non-zero value. To save on compute, let's
        # limit the check to the first batch member and head dimension.
        # TODO: deprecate this function in favor of `cache_position`
        return self.past_tokens[layer_idx]
    
    def change_seq_length(self, bias: Optional[int] = 0) -> int:
        """Returns the sequence length of the cached states that were seen by the model."""
        # Occupied cache == any slot in the 3rd dim (sequence length) holds a non-zero value. To save on compute, let's
        # limit the check to the first batch member and head dimension.
        # TODO: deprecate this function in favor of `cache_position`
        for layer_idx in range(self.num_hidden_layers):
            self.past_tokens[layer_idx] += bias

    def get_max_length(self) -> Optional[int]:
        """Returns the maximum sequence length of the cached states."""
        return self.max_cache_len
    
    def get_usable_length(self, kv_seq_len, layer_idx: Optional[int] = 0) -> int:
        return 0

    def reset(self):
        """Resets the cache values while preserving the objects"""
        for layer_idx in range(len(self.key_cache)):
            # In-place ops prevent breaking the static address
            self.key_cache[layer_idx].zero_()
            if self.value_cache[layer_idx] is not None:
                self.value_cache[layer_idx].zero_()
            self.past_tokens[layer_idx] = 0
        
        if use_torch_npu:
            self.position = [0]

    def remove_suffix(self, start_pos):
        for layer_idx in range(len(self.key_cache)):
            # In-place ops prevent breaking the static address
            if self.is_MLA:
                k_cache = self.key_cache[layer_idx]
                k_cache.view(-1, k_cache.shape[-1])[start_pos:].zero_()
            else:
                self.key_cache[layer_idx][..., start_pos:, :].zero_()
                self.value_cache[layer_idx][..., start_pos:, :].zero_()
            self.past_tokens[layer_idx] = start_pos
    
    def get_max_cache_shape(self) -> Tuple[int, int, int, int]:
        """Returns the maximum shape of the cache."""
        return self.max_cache_len

class KVC2StaticCache:
    """
    Static Cache class connect with KVC2
    remind: page_idx & page_offset info need to refs to forward batching, only contains KV Block Tensor here
    """
    def __init__(self, config: PretrainedConfig, max_batch_size, page_size: int = 256, dtype=torch.bfloat16, device=None) -> None:
        super().__init__()
        self.config = config
        self.dtype = dtype
        self.device = torch.device("npu:0")
        self.kv_lora_rank = config.kv_lora_rank
        self.max_batch_size = max_batch_size
        self.page_size = page_size
        self.k_caches = []
        self.v_caches = []

        self.num_hidden_layers = config.num_hidden_layers

        self.is_MLA = True if config.architectures[0] in ["DeepseekV2ForCausalLM", "DeepseekV3ForCausalLM"] else False
        # kv cache stored in kvc2
        # self.past_tokens = []

    def load(self, inference_context):
        # assert self.is_MLA and len(inference_context.k_cache) == 1, "currently only support MLA and Cache Pool TP=1"
        from ktransformers.util.utils import get_current_device
        for i in range(self.config.num_hidden_layers):
            new_layer_key_cache = inference_context.k_cache[int(torch.distributed.get_rank())][i].to(get_current_device())
            torch._dynamo.mark_static_address(new_layer_key_cache)

            self.k_caches.append(
                new_layer_key_cache  # [TP_idx, layer_idx, page_idx, page_size, kv_head_num, kv_head_dim]
            )

            self.v_caches.append(None)
        self.max_cache_len = self.k_caches[0].shape[0] * self.k_caches[0].shape[1]  # page_len * page_size

    def update(
        self,
        combined: torch.Tensor,
        layer_idx: int,
        cache_kwargs: Optional[Dict[str, Any]] = None,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Updates the cache with the new `key_states` and `value_states` for the layer `layer_idx`.

        Parameters:
            key_states (`torch.Tensor`):
                The new key states to cache.
            value_states (`torch.Tensor`):
                The new value states to cache.
            layer_idx (`int`):
                The index of the layer to cache the states for.
            cache_kwargs (`Dict[str, Any]`, `optional`):
                must have page_idx (`torch.Tensor`): & page_offset (`torch.Tensor`) & cache_position (`torch.Tensor`)

        Return:
            A tuple containing the updated key and value states.
        """
        page_idx, page_offset = cache_kwargs.get("page_idx"), cache_kwargs.get("page_offset")
        if page_idx is None or page_offset is None:
            raise ValueError('[ERROR] block info:page_idx & page_offset missing!')

        k_out = self.k_caches[layer_idx]
        assert self.is_MLA, "currently only support DeepSeekV3 on NPU balance server"

        if page_idx.dim() == 1:
            page_idx_tmp = page_idx.unsqueeze(0)
            page_offset_tmp = page_offset.unsqueeze(0)
        else:
             page_idx_tmp = page_idx
             page_offset_tmp = page_offset

        k_out[page_idx_tmp, page_offset_tmp] = combined
        return k_out, page_idx

    def get_seq_length(self, layer_idx: Optional[int] = 0) -> int:
        """Returns the sequence length of the cached states that were seen by the model."""
        raise ValueError('kvc2 cache pool no longer hold seq_length info, refer to forward batching')

    def get_usable_length(self, kv_seq_len, layer_idx: Optional[int] = 0) -> int:
        return 0

    def change_seq_length(self, bias: Optional[int] = 0) -> int:
        """Returns the sequence length of the cached states that were seen by the model."""
        raise ValueError('kvc2 cache pool no longer hold seq_length info, refer to forward batching')

    def get_max_length(self) -> Optional[int]:
        """Returns the maximum sequence length of the cached states."""
        return self.max_cache_len

    def reset(self, inference_context):
        assert self.is_MLA and len(inference_context.k_cache) == 1, "currently only support MLA and Cache Pool TP=1"
        self.k_caches = []
        self.v_caches = []
        for i in range(self.config.num_hidden_layers):
            self.k_caches.append(
                inference_context.k_cache[0][i]
            )
            self.v_caches.append(None)
        self.max_cache_len = self.k_caches[0].shape[0] * self.k_caches[0].shape[1]  # page_len * page_size

    def get_page_table(self, mini_batch, bsz_tensors: torch.tensor = None, is_prefill=True):
        if is_prefill:
            # TODO add padding support
            q_lens = [mini_batch.p_q_len[idx] for idx in range(mini_batch.prefill_batch)]
            page_local_idx = -1 * torch.ones(mini_batch.prefill_batch, max(q_lens),
                                             dtype=mini_batch.p_position_ids.dtype, device=mini_batch.p_position_ids.device)
            page_offset = -1 * torch.ones_like(page_local_idx)
            # convert merged into batched
            start_ids = 0
            for i in range(mini_batch.prefill_batch):
                page_offset[i, 0:q_lens[i]] = mini_batch.p_position_ids[start_ids:start_ids+q_lens[i]] % self.page_size
                page_local_idx[i, 0:q_lens[i]] = mini_batch.p_position_ids[start_ids:start_ids+q_lens[i]] // self.page_size
                for j in range(q_lens[i]):
                    # get global page idx index by local page idx from block table, as followed decode
                    page_local_idx[i, j] = mini_batch.p_block_tables[i, page_local_idx[i, j]]
                start_ids += q_lens[i]
            page_idx = page_local_idx
            # only padding will cause page_local_idx/page_offset still have -1 value
            # you can use following code as check
            # indices = torch.where(page_offset == -1)
            # assert not indices[0].numel() > 0, 'there still have un-calculated page_idx value'
        else:
            page_local_idx = mini_batch.d_position_ids // self.page_size

            page_offset = mini_batch.d_position_ids % self.page_size
            
            for i in range(mini_batch.decode_batch):
                page_local_idx[i] = mini_batch.d_block_tables[i, page_local_idx[i]]
            
            page_idx = page_local_idx
            
        return page_idx, page_offset

class KDeepSeekV3Cache(nn.Module):
    def __init__(
        self,
        config: PretrainedConfig,
        page_size: int = 256,
        dtype=torch.bfloat16,
        device=torch.device("cuda:0"),
        
    ):
        super().__init__()
        self.config = config
        self.dtype = dtype
        self.device = device
        self.kv_lora_rank = config.kv_lora_rank
        self.page_size = page_size
        self.k_caches = []
        self.v_caches = []
        

    def load(self, inference_context: "sched_ext.InferenceContext"):
        
        for i in range(self.config.num_hidden_layers):
            self.k_caches.append(
                inference_context.k_cache[0][i] 
            )
        self.max_cache_len = self.k_caches[0].shape[0]*self.k_caches[0].shape[1]

    def update(
        self,
        key_states: torch.Tensor,
        value_states: torch.Tensor,
        layer_idx: int,

        page_idx: torch.Tensor,
        page_offset: torch.Tensor,

        cache_kwargs: Optional[Dict[str, Any]] = None,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Updates the cache with the new `key_states` and `value_states` for the layer `layer_idx`.
        It is VERY important to index using a tensor, otherwise you introduce a copy to the device.

        Parameters:
            key_states (`torch.Tensor`):
                The new key states to cache.
            value_states (`torch.Tensor`):
                The new value states to cache.
            layer_idx (`int`):
                The index of the layer to cache the states for.
            cache_kwargs (`Dict[str, Any]`, `optional`):
                Additional arguments for the cache subclass. The `StaticCache` needs the `cache_position` input
                to know how where to write in the cache.

        Return:
            A tuple containing the updated key and value states.
        """
        k_out = self.k_caches[layer_idx]

        k_out[page_idx, page_offset, :, :self.kv_lora_rank] = key_states.reshape(-1, *key_states.shape[2:])
        k_out[page_idx, page_offset, :, self.kv_lora_rank:] = value_states.reshape(-1, *value_states.shape[2:])
        return k_out

        
    def get_page_table(self, cache_position: torch.Tensor, q_indptr: torch.Tensor, kv_indptr: torch.Tensor, kv_indices: torch.Tensor, bsz_tensors: torch.tensor):
        page_offset = cache_position % self.page_size  
        page_idx_local = cache_position // self.page_size  
        query_ids = torch.zeros_like(cache_position)
        for i in range(len(q_indptr) - 1):
            start_idx = q_indptr[i]
            end_idx = q_indptr[i + 1]
            query_ids[start_idx:end_idx] = i
        page_idx = torch.zeros_like(page_idx_local)
        for i in range(bsz_tensors[0]):
            query_id = query_ids[i]
            local_block = page_idx_local[i]
            start_block = kv_indptr[query_id]
            if local_block < kv_indptr[query_id + 1] - kv_indptr[query_id]:
                page_idx[i] = kv_indices[start_block + local_block]
        
        return page_idx, page_offset
    
class KGQACache(nn.Module):
    def __init__(
        self,
        config: PretrainedConfig,
        page_size: int = 256,
        dtype=torch.bfloat16,
        device=torch.device("cuda:0"),
        
    ):
        super().__init__()
        self.config = config
        self.dtype = dtype
        self.device = device
        self.page_size = page_size
        self.k_caches = []
        self.v_caches = []
        

    def load(self, inference_context: "sched_ext.InferenceContext"):
        print(self.config.num_hidden_layers)
        for i in range(self.config.num_hidden_layers):
            self.k_caches.append(
                inference_context.k_cache[0][i] 
            )
            self.v_caches.append(
                inference_context.v_cache[0][i]
            )


        self.max_cache_len = self.k_caches[0].shape[0]*self.k_caches[0].shape[1]


        
    def get_page_table(self, cache_position: torch.Tensor, q_indptr: torch.Tensor, kv_indptr: torch.Tensor, kv_indices: torch.Tensor, bsz_tensors: torch.tensor):
        page_offset = cache_position % self.page_size  
        page_idx_local = cache_position // self.page_size  
        query_ids = torch.zeros_like(cache_position)
        for i in range(len(q_indptr) - 1):
            start_idx = q_indptr[i]
            end_idx = q_indptr[i + 1]
            query_ids[start_idx:end_idx] = i
        page_idx = torch.zeros_like(page_idx_local)
        for i in range(bsz_tensors[0]):
            query_id = query_ids[i]
            local_block = page_idx_local[i]
            start_block = kv_indptr[query_id]
            if local_block < kv_indptr[query_id + 1] - kv_indptr[query_id]:
                page_idx[i] = kv_indices[start_block + local_block]
        
        return page_idx, page_offset

    def get_k_cache(self, layer_idx):
        return self.k_caches[layer_idx]

    def get_v_cache(self, layer_idx):
        return self.v_caches[layer_idx]


class KVC2Qwen3Cache(nn.Module):

    def __init__(self, config, max_batch_size, page_size=256,
                 dtype=torch.bfloat16, device=None):
        super().__init__()
        self.config = config
        self.max_batch_size = max_batch_size
        self.page_size = page_size
        self.dtype = dtype
        self.device = device if device else torch.device("npu:0")

        self.num_layers = config.num_hidden_layers
        self.num_kv_heads = config.num_key_value_heads
        self.head_dim = config.head_dim

        self.k_caches = []
        self.v_caches = []


    # ------------------------- 绑定到底层 kvc2 pool -------------------------

    def load(self, inference_context):
        from ktransformers.util.utils import get_current_device
        dev = get_current_device()

        self.k_caches = []
        self.v_caches = []

        rank = (
            torch.distributed.get_rank()
            if (torch.distributed.is_available() and torch.distributed.is_initialized())
            else 0
        )

        for i in range(self.num_layers):
            k_buf = inference_context.k_cache[rank][i].to(dev).to(self.dtype)
            v_buf = inference_context.v_cache[rank][i].to(dev).to(self.dtype)

            torch._dynamo.mark_static_address(k_buf)
            torch._dynamo.mark_static_address(v_buf)

            self.k_caches.append(k_buf)
            self.v_caches.append(v_buf)

        # num_pages * page_size
        self.max_cache_len = self.k_caches[0].shape[0] * self.k_caches[0].shape[1]

    # ------------------------- 写 KV -------------------------
    @torch.no_grad()
    def update(
        self,
        key_states: torch.Tensor,
        value_states: torch.Tensor,
        layer_idx: int,
        cache_kwargs: Optional[Dict[str, Any]] = None,
    ):
        if cache_kwargs is None:
            raise ValueError("[KVC2Qwen3Cache] cache_kwargs must contain page_idx & page_offset")

        page_idx: Optional[torch.Tensor] = cache_kwargs.get("page_idx", None)
        page_offset: Optional[torch.Tensor] = cache_kwargs.get("page_offset", None)

        if page_idx is None or page_offset is None:
            raise ValueError("[KVC2Qwen3Cache] page_idx & page_offset are required in cache_kwargs")

        k_out = self.k_caches[layer_idx]
        v_out = self.v_caches[layer_idx]

        # -------- 1) 修正维度顺序：[B, KvH, Q, D] -> [B, Q, KvH, D] --------
        if key_states.dim() == 4 and key_states.shape[1] == self.num_kv_heads:
            key_states = key_states.transpose(1, 2).contiguous()
            value_states = value_states.transpose(1, 2).contiguous()

        if key_states.shape != value_states.shape:
            raise ValueError(
                f"[KVC2Qwen3Cache] key_states.shape {key_states.shape} "
                f"!= value_states.shape {value_states.shape}"
            )

        if key_states.dim() != 4:
            raise ValueError(
                f"[KVC2Qwen3Cache] expect key_states dim=4, got {key_states.dim()} "
                f"(shape={key_states.shape})"
            )

        bsz, q_len, kv_heads, head_dim = key_states.shape

        if kv_heads != self.num_kv_heads or head_dim != self.head_dim:
            raise ValueError(
                f"[KVC2Qwen3Cache] KV shape mismatch: "
                f"got num_kv_heads={kv_heads}, head_dim={head_dim}, "
                f"expected num_kv_heads={self.num_kv_heads}, head_dim={self.head_dim}"
            )

        # -------- 2) flatten page_idx / page_offset 为一维 --------
        page_idx = page_idx.reshape(-1)
        page_offset = page_offset.reshape(-1)

        # -------- 3) flatten KV，并强制 dtype 与 cache 对齐 --------
        val_dtype = k_out.dtype
        flat_k = key_states.to(val_dtype).reshape(-1, kv_heads, head_dim)
        flat_v = value_states.to(val_dtype).reshape(-1, kv_heads, head_dim)

        # -------- 4) 真正写入 K / V --------
        # k_out / v_out: [num_pages, page_size, num_kv_heads, head_dim]
        k_out[page_idx, page_offset] = flat_k
        v_out[page_idx, page_offset] = flat_v

    # ------------------------- get K/V -------------------------
    def get_k_cache(self, layer_idx):
        return self.k_caches[layer_idx]

    def get_v_cache(self, layer_idx):
        return self.v_caches[layer_idx]

    # ------------------------- page table 计算 -------------------------
    def get_page_table(
        self,
        mini_batch,
        bsz_tensors: torch.Tensor = None,
        is_prefill: bool = True,
    ):
        if is_prefill:
            # prefill: merged positions => batched (B, T_chunk)
            q_lens = [int(mini_batch.p_q_len[idx]) for idx in range(mini_batch.prefill_batch)]
            if len(q_lens) == 0:
                return None, None

            max_q_len = max(q_lens)

            page_local_idx = -1 * torch.ones(
                mini_batch.prefill_batch,
                max_q_len,
                dtype=mini_batch.p_position_ids.dtype,
                device=mini_batch.p_position_ids.device,
            )
            page_offset = -1 * torch.ones_like(page_local_idx)

            start_ids = 0
            for i in range(mini_batch.prefill_batch):
                cur_len = q_lens[i]
                pos = mini_batch.p_position_ids[start_ids:start_ids + cur_len]  # global pos of this chunk

                # local block + offset by page_size
                page_offset[i, 0:cur_len] = pos % self.page_size
                page_local_idx[i, 0:cur_len] = pos // self.page_size

                # local block -> global page id via block_tables
                for j in range(cur_len):
                    blk = page_local_idx[i, j]
                    page_local_idx[i, j] = mini_batch.p_block_tables[i, blk]

                start_ids += cur_len

            page_idx = page_local_idx
        else:
            # decode: decode_batch = 当前 step 的 batch_size, 每条样本通常 1 个 token
            page_local_idx = mini_batch.d_position_ids // self.page_size
            page_offset = mini_batch.d_position_ids % self.page_size

            for i in range(mini_batch.decode_batch):
                blk = page_local_idx[i]
                page_local_idx[i] = mini_batch.d_block_tables[i, blk]

            page_idx = page_local_idx

        return page_idx, page_offset
