inference/cache/kv_cache.py

"""
Frox AI Morph 1.1 — KV Cache

Improvements over Morph 1.0:
  - Paged block allocation (small-scale PagedAttention-style idea):
    cache grows in fixed-size blocks instead of one giant
    pre-allocated [1, heads, max_seq_len, head_dim] tensor per layer.
    This means a short 50-token chat doesn't reserve the same VRAM as
    a 4096-token one, and growth doesn't require a full reallocation.
  - Dynamic growth: if a sequence exceeds its current block budget,
    new blocks are appended rather than raising an error or silently
    overwriting (the 1.0 behavior silently dropped the oldest tokens,
    which is wrong for a cache that's supposed to be exact — sliding
    eviction now only happens if the caller explicitly opts in).
  - bfloat16 default is preserved, but dtype/device now match the
    model automatically via `MorphKVCache.for_model(model)`.
"""
from __future__ import annotations

from typing import Dict, List, Optional, Tuple

import torch


class MorphKVCache:
    """
    Paged KV Cache for Morph 1.1 inference.

    Instead of one pre-allocated [1, kv_heads, max_seq_len, head_dim]
    tensor per layer (Morph 1.0 behavior — wastes VRAM for short
    sequences and requires knowing max_seq_len up front), Morph 1.1
    allocates in fixed-size blocks and appends new blocks on demand.

    Memory usage (3B model, batch=1, block_size=256):
        Per block per layer: 2 × 8 kv_heads × 256 × 128 × 2 bytes = 1MB
        A 300-token conversation needs 2 blocks/layer × 28 layers = 56MB
        (vs. 224MB pre-allocated for a 2048 max_seq_len in 1.0)
    """

    def __init__(
        self,
        num_layers: int,
        num_kv_heads: int = 8,
        head_dim: int = 128,
        block_size: int = 256,
        dtype: torch.dtype = torch.bfloat16,
        device: torch.device = torch.device("cpu"),
        max_seq_len: Optional[int] = None,   # soft cap; None = unbounded growth
        evict_on_overflow: bool = False,      # opt-in sliding eviction
    ):
        self.num_layers = num_layers
        self.num_kv_heads = num_kv_heads
        self.head_dim = head_dim
        self.block_size = block_size
        self.dtype = dtype
        self.device = device
        self.max_seq_len = max_seq_len
        self.evict_on_overflow = evict_on_overflow

        # Each layer holds a list of blocks: [1, kv_heads, block_size, head_dim]
        self._k_blocks: List[List[torch.Tensor]] = [[] for _ in range(num_layers)]
        self._v_blocks: List[List[torch.Tensor]] = [[] for _ in range(num_layers)]
        self._seq_len = 0

    @classmethod
    def for_model(
        cls,
        model,
        block_size: int = 256,
        max_seq_len: Optional[int] = None,
    ) -> "MorphKVCache":
        """Build a cache matching a MorphForCausalLM's config, dtype, device."""
        cfg = model.config
        device = next(model.parameters()).device
        dtype = next(model.parameters()).dtype
        return cls(
            num_layers=cfg.num_hidden_layers,
            num_kv_heads=cfg.num_key_value_heads,
            head_dim=cfg.head_dim,
            block_size=block_size,
            dtype=dtype,
            device=device,
            max_seq_len=max_seq_len,
        )

    @property
    def seq_len(self) -> int:
        return self._seq_len

    def _new_block(self) -> torch.Tensor:
        return torch.zeros(
            1, self.num_kv_heads, self.block_size, self.head_dim,
            dtype=self.dtype, device=self.device,
        )

    def _ensure_capacity(self, layer_idx: int, needed_len: int):
        """Append blocks until the layer can hold `needed_len` tokens."""
        current_capacity = len(self._k_blocks[layer_idx]) * self.block_size
        while current_capacity < needed_len:
            self._k_blocks[layer_idx].append(self._new_block())
            self._v_blocks[layer_idx].append(self._new_block())
            current_capacity += self.block_size

    def update(
        self,
        layer_idx: int,
        k: torch.Tensor,
        v: torch.Tensor,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Append new K/V for a layer and return the full cached sequence.

        k, v: [B, num_kv_heads, new_len, head_dim]
        returns: [B, num_kv_heads, total_len, head_dim]
        """
        new_len = k.shape[2]
        start = self._seq_len
        end = start + new_len

        if self.max_seq_len is not None and end > self.max_seq_len:
            if self.evict_on_overflow:
                self._evict_oldest(layer_idx, end - self.max_seq_len)
                start = self._seq_len
                end = start + new_len
            # else: allow growth past the soft cap (caller's responsibility
            # to manage context length upstream — we never silently corrupt data)

        self._ensure_capacity(layer_idx, end)

        # Write into the flat block list via a concatenated view.
        # Blocks are contiguous in allocation order, so we can index
        # directly into a temporary concatenation for the write, then
        # scatter back — cheap because it only touches the new_len slice.
        block_idx_start = start // self.block_size
        offset_in_block = start % self.block_size

        remaining = new_len
        src_offset = 0
        b = block_idx_start
        off = offset_in_block

        while remaining > 0:
            space_in_block = self.block_size - off
            write_len = min(space_in_block, remaining)

            self._k_blocks[layer_idx][b][:, :, off:off + write_len, :] = \
                k[:, :, src_offset:src_offset + write_len, :]
            self._v_blocks[layer_idx][b][:, :, off:off + write_len, :] = \
                v[:, :, src_offset:src_offset + write_len, :]

            remaining -= write_len
            src_offset += write_len
            b += 1
            off = 0

        k_full = self._read_range(layer_idx, self._k_blocks, 0, end)
        v_full = self._read_range(layer_idx, self._v_blocks, 0, end)

        return k_full, v_full

    def _read_range(
        self, layer_idx: int, blocks_container: List[List[torch.Tensor]],
        start: int, end: int,
    ) -> torch.Tensor:
        blocks = blocks_container[layer_idx]
        parts = []
        pos = 0
        for block in blocks:
            block_start = pos
            block_end = pos + self.block_size
            if block_end > start and block_start < end:
                lo = max(0, start - block_start)
                hi = min(self.block_size, end - block_start)
                parts.append(block[:, :, lo:hi, :])
            pos += self.block_size
            if pos >= end:
                break

        if not parts:
            # Defensive: an empty [start, end) range (e.g. end == start == 0).
            # Callers that expect real data guard against this themselves
            # (see get()); this just prevents an IndexError from a bare
            # parts[0] on the rare path that reaches here anyway.
            return torch.zeros(
                1, self.num_kv_heads, 0, self.head_dim,
                dtype=self.dtype, device=self.device,
            )
        return torch.cat(parts, dim=2) if len(parts) > 1 else parts[0]

    def _evict_oldest(self, layer_idx: int, num_to_evict: int):
        """
        Shift the cache left by `num_to_evict` tokens (sliding window).
        Only called when evict_on_overflow=True — an explicit opt-in.
        """
        k_full = self._read_range(layer_idx, self._k_blocks, 0, self._seq_len)
        v_full = self._read_range(layer_idx, self._v_blocks, 0, self._seq_len)

        k_kept = k_full[:, :, num_to_evict:, :].contiguous()
        v_kept = v_full[:, :, num_to_evict:, :].contiguous()

        # Rebuild blocks from the kept slice
        self._k_blocks[layer_idx] = []
        self._v_blocks[layer_idx] = []
        self._seq_len = 0
        if k_kept.shape[2] > 0:
            self._ensure_capacity(layer_idx, k_kept.shape[2])
            self._k_blocks[layer_idx][0][:, :, :k_kept.shape[2], :] = k_kept
            self._v_blocks[layer_idx][0][:, :, :v_kept.shape[2], :] = v_kept
        self._seq_len = max(0, self._seq_len - num_to_evict)

    def step(self, new_tokens: int = 1):
        """Advance sequence length after generating new_tokens."""
        self._seq_len += new_tokens

    def reset(self):
        """Clear cache for a new conversation."""
        self._k_blocks = [[] for _ in range(self.num_layers)]
        self._v_blocks = [[] for _ in range(self.num_layers)]
        self._seq_len = 0

    def get(self, layer_idx: int) -> Tuple[torch.Tensor, torch.Tensor]:
        """Get cached KV for a layer as [B, kv_heads, seq_len, head_dim]."""
        if self._seq_len == 0:
            empty = torch.zeros(
                1, self.num_kv_heads, 0, self.head_dim,
                dtype=self.dtype, device=self.device,
            )
            return empty, empty
        k = self._read_range(layer_idx, self._k_blocks, 0, self._seq_len)
        v = self._read_range(layer_idx, self._v_blocks, 0, self._seq_len)
        return k, v

    def memory_mb(self) -> float:
        """Current allocated memory (not just used) across all layers."""
        total_bytes = 0
        for layer_blocks in self._k_blocks:
            for block in layer_blocks:
                total_bytes += block.numel() * block.element_size()
        for layer_blocks in self._v_blocks:
            for block in layer_blocks:
                total_bytes += block.numel() * block.element_size()
        return total_bytes / (1024 ** 2)

    def utilization(self) -> float:
        """Fraction of allocated blocks actually holding real tokens."""
        allocated = len(self._k_blocks[0]) * self.block_size if self._k_blocks[0] else 0
        if allocated == 0:
            return 0.0
        return round(self._seq_len / allocated, 3)

    def to(self, device: torch.device) -> "MorphKVCache":
        self._k_blocks = [[b.to(device) for b in layer] for layer in self._k_blocks]
        self._v_blocks = [[b.to(device) for b in layer] for layer in self._v_blocks]
        self.device = device
        return self


class MorphSessionCache:
    """
    Per-session KV cache manager.
    Maintains separate paged caches for concurrent chat sessions,
    evicting the least-recently-used session when at capacity.
    """

    def __init__(
        self,
        num_layers: int = 28,
        max_sessions: int = 16,
        block_size: int = 256,
        num_kv_heads: int = 8,
        head_dim: int = 128,
        dtype: torch.dtype = torch.bfloat16,
        device: torch.device = torch.device("cpu"),
        max_seq_len_per_session: Optional[int] = None,
    ):
        self.num_layers = num_layers
        self.max_sessions = max_sessions
        self.block_size = block_size
        self.num_kv_heads = num_kv_heads
        self.head_dim = head_dim
        self.dtype = dtype
        self.device = device
        self.max_seq_len_per_session = max_seq_len_per_session

        self._caches: Dict[str, MorphKVCache] = {}
        self._lru_order: List[str] = []   # NEW 1.1: real LRU tracking

    def get_or_create(self, session_id: str) -> MorphKVCache:
        if session_id in self._caches:
            self._touch(session_id)
            return self._caches[session_id]

        if len(self._caches) >= self.max_sessions:
            lru_id = self._lru_order.pop(0)
            del self._caches[lru_id]

        self._caches[session_id] = MorphKVCache(
            num_layers=self.num_layers,
            block_size=self.block_size,
            num_kv_heads=self.num_kv_heads,
            head_dim=self.head_dim,
            dtype=self.dtype,
            device=self.device,
            max_seq_len=self.max_seq_len_per_session,
        )
        self._lru_order.append(session_id)
        return self._caches[session_id]

    def _touch(self, session_id: str):
        if session_id in self._lru_order:
            self._lru_order.remove(session_id)
        self._lru_order.append(session_id)

    def reset_session(self, session_id: str):
        if session_id in self._caches:
            self._caches[session_id].reset()

    def delete_session(self, session_id: str):
        if session_id in self._caches:
            del self._caches[session_id]
        if session_id in self._lru_order:
            self._lru_order.remove(session_id)

    def total_memory_mb(self) -> float:
        return sum(c.memory_mb() for c in self._caches.values())

    def stats(self) -> Dict:
        return {
            "active_sessions": len(self._caches),
            "max_sessions": self.max_sessions,
            "total_memory_mb": round(self.total_memory_mb(), 1),
            "avg_utilization": round(
                sum(c.utilization() for c in self._caches.values()) / max(len(self._caches), 1), 3
            ),
        }