# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Engram: n-gram hash lookups gated into the hyper-connection stream.

Port of the reference ``inference/engram.py`` + ``Engram`` /
``ParallelEngramEmbedding`` from ``inference/model.py`` (DeepSeek V4.1
checkpoint layout). Engram modules live on the backbone layers listed in
``engram_layer_ids`` only.

Two pieces of cross-forward state are needed because vLLM streams tokens
chunk-by-chunk while an n-gram at position ``p`` needs the token ids at
``p-1..p-3``:

- ``token_map``: token id -> compressed vocab id, built once from the
  model's tokenizer at init (deterministic; asserted against
  ``engram_compressed_vocab_size``).
- ``hash_cache``: one int32 slot per KV slot of the first local layer's
  sliding-window cache, holding the compressed id (or DEAD) of the token
  last written to that slot. Slots are stable per (request, position) —
  the block table pins a position to a physical slot, prefix-cache hits
  reuse both the physical blocks and the identical token ids, and
  spec-decode rollbacks rewrite the same slots — so lookbacks read back
  exactly what the owning request wrote. Lookback depth (3) is far inside
  the sliding window (128), so window eviction never frees a block a
  live lookback still needs.

  Slots are not part of the KV cache, so KV loaded from another instance
  (P/D, offload connectors) leaves them unwritten. The runner therefore
  passes ``lookback_token_ids``, the ids just before each request's chunk
  start, which take precedence over the slots. The V2 runner reads them
  from its device-resident token history and needs no slot cache; the V1
  runner's CPU token table holds placeholders for generated tokens under
  async scheduling, so it passes prompt positions only and keeps the slot
  cache for the rest.
"""

import weakref

import numpy as np
import torch
from torch import nn

# --- Tech2Wild/Kai 2026-09-10: disk-backed Engram tables for DGX Spark (UMA) ---
import json as _kai_json
import os as _kai_os
import struct as _kai_struct
from concurrent.futures import ThreadPoolExecutor as _KaiPool

from vllm.config import VllmConfig, get_current_vllm_config
from vllm.distributed import (
    get_tensor_model_parallel_rank,
    get_tensor_model_parallel_world_size,
    tensor_model_parallel_all_gather,
)
from vllm.logger import init_logger
from vllm.model_executor.layers.linear import ReplicatedLinear
from vllm.model_executor.layers.quantization import QuantizationConfig
from vllm.model_executor.utils import set_weight_attrs
from vllm.triton_utils import tl, triton
from vllm.utils.platform_utils import is_uva_available
from vllm.utils.torch_utils import get_accelerator_view_from_cpu_tensor

logger = init_logger(__name__)

# Cache value for tokens that take no part in an n-gram (image spans).
DEAD_ID = -1


def _is_prime(n: int) -> bool:
    """Deterministic Miller-Rabin for n < 2**32 (avoids a sympy import)."""
    if n < 2:
        return False
    for p in (2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37):
        if n % p == 0:
            return n == p
    d = n - 1
    r = 0
    while d % 2 == 0:
        d //= 2
        r += 1
    for a in (2, 7, 61):
        x = pow(a, d, n)
        if x in (1, n - 1):
            continue
        for _ in range(r - 1):
            x = x * x % n
            if x == n - 1:
                break
        else:
            return False
    return True


def find_next_prime(start: int, seen_primes: set[int]) -> int:
    """The smallest prime above `start` that has not been handed out yet."""
    candidate = start + 1
    while not _is_prime(candidate) or candidate in seen_primes:
        candidate += 1
    return candidate


def build_compressed_token_map(tokenizer) -> tuple[list[int], int]:
    """Map every token id onto a smaller id space where tokens that normalize
    alike collapse together.

    N-grams are hashed over these compressed ids, so " The", "the" and "THE"
    all hash the same way. The compressed size matters beyond bounds checking:
    every hash multiplier is derived from it.
    """
    from tokenizers import Regex, normalizers

    # A private-use char, so a token that is exactly one space survives
    # Strip() instead of collapsing to the empty string and merging with
    # unrelated tokens.
    sentinel = "\ue000"
    normalizer = normalizers.Sequence(
        [
            normalizers.NFKC(),
            normalizers.NFD(),
            normalizers.StripAccents(),
            normalizers.Lowercase(),
            normalizers.Replace(Regex(r"[ \t\r\n]+"), " "),
            normalizers.Replace(Regex(r"^ $"), sentinel),
            normalizers.Strip(),
            normalizers.Replace(sentinel, " "),
        ]
    )

    # The raw Rust tokenizer, matching what training decodes with
    # (no clean_up_tokenization_spaces).
    backend = tokenizer.backend_tokenizer
    key_to_new: dict[str, int] = {}
    lookup = [0] * len(tokenizer)
    for token_id in range(len(tokenizer)):
        text = backend.decode([token_id], skip_special_tokens=False)
        if "\ufffd" in text:
            # A partial UTF-8 byte token: nothing to normalize, so key it
            # by its raw form.
            key = backend.id_to_token(token_id)
        else:
            normalized = normalizer.normalize_str(text)
            key = normalized if normalized else text

        new_id = key_to_new.get(key)
        if new_id is None:
            new_id = len(key_to_new)
            key_to_new[key] = new_id
        lookup[token_id] = new_id

    return lookup, len(key_to_new)


def compute_hash_multipliers(
    layer_ids: tuple[int, ...], max_ngram_size: int, compressed_vocab_size: int
) -> torch.Tensor:
    """One multiplier per (layer, lookback), from a per-layer RNG so layers
    hash differently. Kept odd and bounded so `token_id * multiplier` cannot
    overflow int64.
    """
    max_long = np.iinfo(np.int64).max
    multiplier_bound = max(1, (max_long // compressed_vocab_size) // 2)
    rows = []
    for layer_id in layer_ids:
        generator = np.random.default_rng(10007 * layer_id)
        values = generator.integers(
            low=0,
            high=multiplier_bound,
            size=(max_ngram_size,),
            dtype=np.int64,
        )
        rows.append(torch.tensor(values * 2 + 1))
    return torch.stack(rows)


class EngramLayout:
    """Bucket layout of the n-gram hash tables.

    A position is hashed as `max_ngram_size - 1` n-grams (2-gram .. max), each
    split over `n_heads` heads. Every (n-gram size, head) pair owns its own
    prime-sized bucket range in the layer's table; the primes are drawn in
    order and never reused, which keeps the ranges disjoint.
    """

    def __init__(self, config) -> None:
        self.layer_ids: tuple[int, ...] = tuple(config.engram_layer_ids)
        self.num_embeddings: tuple[int, ...] = tuple(config.engram_num_embeddings)
        self.max_ngram_size: int = config.engram_max_ngram_size
        self.n_heads: int = config.engram_n_heads
        self.head_dim: int = config.engram_head_dim
        self.compressed_vocab_size: int = config.engram_compressed_vocab_size
        self.pad_token_id: int = config.engram_pad_token_id
        assert len(self.layer_ids) == len(self.num_embeddings)

        primes = []
        seen: set[int] = set()
        for _ in self.layer_ids:
            per_ngram = []
            for _ in range(self.max_ngram_size - 1):
                sizes, current = [], config.engram_vocab_size - 1
                for _ in range(self.n_heads):
                    current = find_next_prime(current, seen)
                    seen.add(current)
                    sizes.append(current)
                per_ngram.append(tuple(sizes))
            primes.append(tuple(per_ngram))
        self.primes: tuple[tuple[tuple[int, ...], ...], ...] = tuple(primes)
        self.n_hash_cols = (self.max_ngram_size - 1) * self.n_heads
        flat = [[p for per_ngram in layer for p in per_ngram] for layer in primes]
        offsets = [np.cumsum([0, *sizes[:-1]]) for sizes in flat]
        self.offsets = torch.tensor(np.array(offsets))  # [n_layers, n_hash_cols]

    @classmethod
    def from_config(cls, config) -> "EngramLayout | None":
        if not getattr(config, "engram_layer_ids", None):
            return None
        return cls(config)


@triton.jit(do_not_specialize=["num_tokens"])
def _write_hash_cache_kernel(
    input_ids,
    token_map,
    dead_mask,
    slot_mapping,
    cache,
    num_tokens,
    input_stride,
    mask_stride,
    slot_stride,
    BLOCK_SIZE: tl.constexpr,
    dead_id,
):
    token_idx = tl.program_id(0) * BLOCK_SIZE + tl.arange(0, BLOCK_SIZE)
    slot = tl.load(
        slot_mapping + token_idx * slot_stride, token_idx < num_tokens, other=-1
    ).to(tl.int64)
    valid = (token_idx < num_tokens) & (slot >= 0)
    token = tl.load(input_ids + token_idx * input_stride, valid, other=0)
    value = tl.load(token_map + token, valid, other=0)
    dead = tl.load(dead_mask + token_idx * mask_stride, valid, other=False)
    value = tl.where(dead, dead_id, value)
    tl.store(cache + slot, value, valid)


@triton.jit(
    do_not_specialize=[
        "num_tokens",
        "num_slots",
        "num_query_rows",
        "num_table_rows",
        "max_blocks",
    ]
)
def _hash_ids_kernel(
    input_ids,
    token_map,
    dead_mask,
    positions,
    block_table,
    query_start_loc,
    multipliers,
    primes,
    offsets,
    cache,
    lookback_token_ids,
    lookback_dead_mask,
    output,
    num_tokens,
    num_slots,
    pad_id,
    input_stride,
    mask_stride,
    position_stride,
    table_stride,
    table_col_stride,
    query_stride,
    num_query_rows,
    num_table_rows,
    max_blocks,
    cache_block_size,
    MAX_NGRAM: tl.constexpr,
    num_heads,
    BLOCK_T: tl.constexpr,
    BLOCK_H: tl.constexpr,
    dead_id,
    lookback_depth,
    lookback_row_stride,
    lookback_col_stride,
    lookback_mask_row_stride,
    lookback_mask_col_stride,
):
    token = tl.program_id(0) * BLOCK_T + tl.arange(0, BLOCK_T)
    layer = tl.program_id(1)
    num_layers = tl.num_programs(1)
    valid = token < num_tokens
    # Upper bound in query_start_loc[1:], including repeated padding boundaries.
    lo = tl.full((BLOCK_T,), 0, tl.int32)
    hi = tl.full((BLOCK_T,), num_query_rows, tl.int32)
    while tl.sum((lo < hi).to(tl.int32), 0) > 0:
        mid = (lo + hi) // 2
        end = tl.load(
            query_start_loc + (mid + 1) * query_stride,
            lo < hi,
            other=0,
        )
        right = token >= end
        active = lo < hi
        lo = tl.where(active & right, mid + 1, lo)
        hi = tl.where(active & ~right, mid, hi)
    req = tl.minimum(lo, num_query_rows - 1).to(tl.int64)
    chunk_idx = tl.load(query_start_loc + req * query_stride)
    chunk_idx = tl.minimum(chunk_idx, num_tokens - 1).to(tl.int64)
    chunk_start = tl.load(positions + chunk_idx * position_stride)
    position = tl.load(positions + token * position_stride, valid, other=0).to(tl.int64)
    head = tl.arange(0, BLOCK_H)
    blocked = tl.full((BLOCK_T,), False, tl.int1)
    rolling = tl.full((BLOCK_T,), 0, tl.int64)
    for shift in tl.static_range(MAX_NGRAM):
        lookback = position - shift
        in_batch = lookback >= chunk_start
        batch_idx = tl.maximum(token - shift, 0)
        batch_token = tl.load(
            input_ids + batch_idx * input_stride, valid & in_batch, other=0
        )
        batch_source = tl.load(token_map + batch_token, valid & in_batch, other=0)
        batch_dead = tl.load(
            dead_mask + batch_idx * mask_stride, valid & in_batch, other=False
        )
        batch_source = tl.where(batch_dead, dead_id, batch_source)

        col = chunk_start - 1 - lookback
        in_window = valid & ~in_batch & (col >= 0) & (col < lookback_depth)
        col = tl.minimum(tl.maximum(col, 0), lookback_depth - 1)
        window_token = tl.load(
            lookback_token_ids + req * lookback_row_stride + col * lookback_col_stride,
            in_window,
            other=-1,
        )
        known = in_window & (window_token >= 0)
        window_source = tl.load(token_map + window_token, known, other=0)
        window_dead = tl.load(
            lookback_dead_mask
            + req * lookback_mask_row_stride
            + col * lookback_mask_col_stride,
            known,
            other=False,
        )
        window_source = tl.where(window_dead, dead_id, window_source)

        if cache is not None:
            clamped = tl.minimum(
                tl.maximum(lookback, 0), max_blocks * cache_block_size - 1
            )
            block_row = tl.minimum(req, num_table_rows - 1)
            needs_cache = valid & ~in_batch & ~known
            block = tl.load(
                block_table
                + block_row * table_stride
                + (clamped // cache_block_size) * table_col_stride,
                needs_cache,
                other=0,
            ).to(tl.int64)
            slot = tl.minimum(
                tl.maximum(block * cache_block_size + clamped % cache_block_size, 0),
                num_slots - 1,
            )
            fallback = tl.load(cache + slot, needs_cache, other=0)
        else:
            fallback = tl.full((BLOCK_T,), pad_id, tl.int32)
        source = tl.where(
            in_batch, batch_source, tl.where(known, window_source, fallback)
        ).to(tl.int64)
        blocked |= (lookback < 0) | (source == dead_id)
        value = tl.where(blocked, pad_id, source)
        multiplier = tl.load(multipliers + layer * MAX_NGRAM + shift)
        rolling ^= value * multiplier
        if shift > 0:
            col = (shift - 1) * num_heads + head
            param_offset = layer * (MAX_NGRAM - 1) * num_heads + col
            prime = tl.load(primes + param_offset, head < num_heads, other=1)
            offset = tl.load(offsets + param_offset, head < num_heads, other=0)
            hashed = rolling[:, None] % prime[None, :] + offset[None, :]
            out_offset = (token.to(tl.int64) * num_layers + layer)[:, None] * (
                (MAX_NGRAM - 1) * num_heads
            ) + col[None, :]
            tl.store(output + out_offset, hashed, valid[:, None] & (head < num_heads))


class NgramHashState(nn.Module):
    """Maps each position to the hash ids of the n-grams ending there.

    Stateless on the V2 runner, which supplies every lookback token id. On
    the V1 runner it also keeps `hash_cache`, the slot-keyed rolling store
    of compressed ids (see module docstring), for generated tokens.
    """

    def __init__(
        self,
        vllm_config: VllmConfig,
        layout: EngramLayout,
        swa_cache_module: nn.Module,
    ) -> None:
        super().__init__()
        self.layout = layout
        self.swa_cache_module = swa_cache_module
        self.block_size: int = swa_cache_module.block_size
        self.lookback_depth: int = layout.max_ngram_size - 1
        self.use_slot_cache: bool = not vllm_config.use_v2_model_runner
        self._cache: torch.Tensor | None = None
        self._kv_cache_ref: weakref.ReferenceType[torch.Tensor] | None = None

        model_config = vllm_config.model_config
        from transformers import AutoTokenizer

        tokenizer = AutoTokenizer.from_pretrained(
            model_config.tokenizer,
            trust_remote_code=model_config.trust_remote_code,
            revision=model_config.revision,
        )
        token_map, vocab_size = build_compressed_token_map(tokenizer)
        if vocab_size != layout.compressed_vocab_size:
            raise ValueError(
                f"Compressed vocab size mismatch: built {vocab_size} from the "
                f"tokenizer, config expects {layout.compressed_vocab_size}; "
                "every hash multiplier derives from it, so the engram tables "
                "would be silently rehashed."
            )
        self.pad_id = token_map[layout.pad_token_id]
        multipliers = compute_hash_multipliers(
            layout.layer_ids, layout.max_ngram_size, vocab_size
        )
        self.register_buffer(
            "token_map", torch.tensor(token_map, dtype=torch.int32), persistent=False
        )
        self.register_buffer("primes", torch.tensor(layout.primes), persistent=False)
        self.register_buffer("offsets", layout.offsets, persistent=False)
        self.register_buffer("multipliers", multipliers, persistent=False)
        logger.info(
            "Built engram token map (%d -> %d ids) for layers %s",
            len(token_map),
            vocab_size,
            layout.layer_ids,
        )

    def ensure_cache(self) -> bool:
        """Lazily size the slot-keyed cache from the bound SWA KV cache.

        Returns False while the KV cache is unbound (profile run); the caller
        skips engram hashing then. Without the slot cache only that check
        remains.
        """
        kv_cache = self.swa_cache_module.kv_cache
        if kv_cache.numel() == 0:
            self._cache = None
            self._kv_cache_ref = None
            return False
        if not self.use_slot_cache:
            return True
        if self._kv_cache_ref is not None and self._kv_cache_ref() is kv_cache:
            return True
        # Graph memory profiling binds a temporary, smaller KV cache first.
        # Rebinding must discard its hash history without retaining KV storage.
        self._cache = torch.zeros(
            kv_cache.shape[0] * self.block_size,
            dtype=torch.int32,
            device=kv_cache.device,
        )
        self._kv_cache_ref = weakref.ref(kv_cache)
        return True

    def forward(
        self,
        input_ids: torch.Tensor,
        positions: torch.Tensor,
        query_start_loc: torch.Tensor,
        dead_mask: torch.Tensor,
        lookback_token_ids: torch.Tensor,
        lookback_dead_mask: torch.Tensor,
        slot_mapping: torch.Tensor | None,
        block_table: torch.Tensor | None,
    ) -> torch.Tensor:
        """Compute [tokens, layers, hash columns] int32 n-gram hashes.

        History comes from the current chunk, then the runner's lookback
        window, then the optional V1 slot cache. V2 needs only one launch.
        """
        cache = self._cache if self.use_slot_cache else None
        num_tokens = input_ids.shape[0]
        num_layers, max_ngram = self.multipliers.shape
        num_heads = self.primes.shape[-1]
        output = input_ids.new_empty(
            (num_tokens, num_layers, (max_ngram - 1) * num_heads), dtype=torch.int32
        )
        if num_tokens == 0:
            return output
        if self.use_slot_cache:
            assert cache is not None and slot_mapping is not None
            assert block_table is not None
            # Finish writes before other thread blocks read fallback history.
            _write_hash_cache_kernel[(triton.cdiv(num_tokens, 256),)](
                input_ids,
                self.token_map,
                dead_mask,
                slot_mapping,
                cache,
                num_tokens,
                input_ids.stride(0),
                dead_mask.stride(0),
                slot_mapping.stride(0),
                256,
                DEAD_ID,
            )
        _hash_ids_kernel[(triton.cdiv(num_tokens, 32), num_layers)](
            input_ids,
            self.token_map,
            dead_mask,
            positions,
            block_table,
            query_start_loc,
            self.multipliers,
            self.primes,
            self.offsets,
            cache,
            lookback_token_ids,
            lookback_dead_mask,
            output,
            num_tokens,
            cache.shape[0] if cache is not None else 0,
            self.pad_id,
            input_stride=input_ids.stride(0),
            mask_stride=dead_mask.stride(0),
            position_stride=positions.stride(0),
            table_stride=block_table.stride(0) if block_table is not None else 0,
            table_col_stride=block_table.stride(1) if block_table is not None else 0,
            query_stride=query_start_loc.stride(0),
            num_query_rows=query_start_loc.numel() - 1,
            num_table_rows=block_table.shape[0] if block_table is not None else 0,
            max_blocks=block_table.shape[1] if block_table is not None else 0,
            cache_block_size=self.block_size,
            MAX_NGRAM=max_ngram,
            num_heads=num_heads,
            BLOCK_T=32,
            BLOCK_H=triton.next_power_of_2(num_heads),
            dead_id=DEAD_ID,
            lookback_depth=lookback_token_ids.shape[1],
            lookback_row_stride=lookback_token_ids.stride(0),
            lookback_col_stride=lookback_token_ids.stride(1),
            lookback_mask_row_stride=lookback_dead_mask.stride(0),
            lookback_mask_col_stride=lookback_dead_mask.stride(1),
            num_warps=4,
        )
        return output


def _engram_head_shard_weight_loader(
    param: torch.nn.Parameter, loaded_weight: torch.Tensor
) -> None:
    """Load this rank's complete head buckets. ue8m0 scales arrive as
    float8_e8m0fnu; keep the raw bytes (the param stores uint8)."""
    part_rows = param.shape[0]
    if loaded_weight.dtype == torch.float8_e8m0fnu:
        loaded_weight = loaded_weight.view(torch.uint8)
    shard = loaded_weight.narrow(0, param.engram_vocab_start, part_rows)
    assert shard.shape == param.shape, (
        f"engram shard {tuple(shard.shape)} does not fit param {tuple(param.shape)}"
    )
    param.data.copy_(shard)


@triton.jit
def _engram_lookup_kernel(
    weight,
    scales,
    ids,
    out,
    vocab_start,
    vocab_end,
    num_rows,
    ids_stride_t,
    ids_stride_h,
    HEAD_START: tl.constexpr,
    LOCAL_HEADS: tl.constexpr,
    TOTAL_HEADS: tl.constexpr,
    DIM: tl.constexpr,
    QUANT_BLOCK: tl.constexpr,
    BLOCK_R: tl.constexpr,
    GRID: tl.constexpr,
):
    """Gather fp8 rows, apply their ue8m0 block scales, write bf16.

    Only this rank's heads are read; padded heads write zeros for all-gather.
    `weight`/`scales` may address pinned host memory through UVA.
    """
    cols = tl.arange(0, DIM)
    scale_cols = cols // QUANT_BLOCK
    for base in tl.range(tl.program_id(0) * BLOCK_R, num_rows, GRID * BLOCK_R):
        rows = base + tl.arange(0, BLOCK_R)
        valid = rows < num_rows
        head = HEAD_START + rows % LOCAL_HEADS
        token = (rows // LOCAL_HEADS).to(tl.int64)
        index = tl.load(
            ids + token * ids_stride_t + head * ids_stride_h,
            mask=valid & (head < TOTAL_HEADS),
            other=-1,
        ).to(tl.int64)
        owned = valid & (head < TOTAL_HEADS)
        owned &= (index >= vocab_start) & (index < vocab_end)
        local = tl.where(owned, index - vocab_start, 0)
        values = tl.load(
            weight + local[:, None] * DIM + cols[None, :],
            mask=owned[:, None],
            other=0.0,
        )
        scale = tl.load(
            scales + local[:, None] * (DIM // QUANT_BLOCK) + scale_cols[None, :],
            mask=owned[:, None],
            other=0,
        )
        # ue8m0 is a power of two, so its byte *is* the fp32 exponent field.
        scale = (scale.to(tl.int32) << 23).to(tl.float32, bitcast=True)
        tl.store(
            out + rows[:, None] * DIM + cols[None, :],
            (values.to(tl.float32) * scale).to(tl.bfloat16),
            mask=valid[:, None],
        )


_DSV41_ENGRAM_DISK = _kai_os.environ.get("DSV41_ENGRAM_DISK", "0") == "1"

# --- Tech2Wild/Kai 2026-09-10 (CUDA graphs): one process-wide read pool shared
# by every DiskEngramTable. A step's reads for BOTH engram layers (weight and
# scale rows) go out as one batch, so decode latency is ~one pread instead of
# 24 serial ones (measured on Reddie: 7.7 ms serial vs 3.1 ms parallel, C1).
_KAI_THREADS = int(_kai_os.environ.get("DSV41_ENGRAM_DISK_THREADS", "32"))
_KAI_CHUNK = int(_kai_os.environ.get("DSV41_ENGRAM_DISK_CHUNK", "16"))
_KAI_POOL: _KaiPool | None = None
# Stage 6: gather rows through a read-only memory map (one NumPy fancy-index copy
# per chunk) instead of one preadv syscall per row.
_KAI_MMAP = _kai_os.environ.get("DSV41_ENGRAM_MMAP", "0") == "1"
# The gather releases the GIL, so more threads = more page faults in flight
# (3 of 4 ranks fault rows in over NFS; each cold row is one round trip).
_KAI_MMAP_THREADS = int(_kai_os.environ.get("DSV41_ENGRAM_MMAP_THREADS", "96"))
# Stage 8: hash-ahead prefetch of the next prefill chunk's rows into the page
# cache while the current chunk computes (the prompt is known up front).
_KAI_PREFETCH = _kai_os.environ.get("DSV41_ENGRAM_PREFETCH", "0") == "1"
_KAI_PREFETCH_MODE = _kai_os.environ.get("DSV41_ENGRAM_PREFETCH_MODE", "fadvise")  # fadvise | pread
_KAI_PREFETCH_THREADS = int(_kai_os.environ.get("DSV41_ENGRAM_PREFETCH_THREADS", "32"))
_KAI_PREFETCH_TASK_ROWS = 512
# Gathers smaller than this many unique rows (per table) keep the preadv path
# (0 = memmap for every size). Lets decode steps (72 rows) and prefill chunks
# (tens of thousands) use different paths.
_KAI_MMAP_MIN_ROWS = int(_kai_os.environ.get("DSV41_ENGRAM_MMAP_MIN_ROWS", "0"))
# v3 (window 8: 72 decode rows cost 5-6 ms = one NVMe latency per row, serialized by
# the per-task faults): gathers up to this many rows skip the pool -- first an async
# readahead of every row page (posix_fadvise WILLNEED, ~2 us each, all reads in
# flight together), then ONE fancy-index gather per table on the calling thread.
_KAI_MMAP_INLINE_ROWS = int(_kai_os.environ.get("DSV41_ENGRAM_MMAP_INLINE_ROWS", "2048"))


def _kai_pool() -> _KaiPool:
    global _KAI_POOL
    if _KAI_POOL is None:
        _KAI_POOL = _KaiPool(
            max_workers=_KAI_MMAP_THREADS if _KAI_MMAP else _KAI_THREADS,
            thread_name_prefix="engram-disk",
        )
    return _KAI_POOL


def _kai_prefetch_rows(table, rows: list) -> None:
    """Warm rank-local rows `rows` of `table` in the page cache: weight rows by
    exact byte range, scale rows by 4 KiB page (32 rows share one)."""
    if not rows:
        return
    dim, sb = table.dim, table.sb
    if _KAI_PREFETCH_MODE == "pread":
        buf = bytearray(dim)
        for r in rows:
            try:
                _kai_os.preadv(table.w_fd, [buf], table.w_off + r * dim)
            except OSError:
                pass
        sbuf = bytearray(4096)
        for p in {(table.s_off + r * sb) >> 12 for r in rows}:
            try:
                _kai_os.preadv(table.s_fd, [sbuf], p << 12)
            except OSError:
                pass
        return
    adv = _kai_os.POSIX_FADV_WILLNEED
    for r in rows:
        try:
            _kai_os.posix_fadvise(table.w_fd, table.w_off + r * dim, dim, adv)
        except OSError:
            pass
    for p in {(table.s_off + r * sb) >> 12 for r in rows}:
        try:
            _kai_os.posix_fadvise(table.s_fd, p << 12, 4096, adv)
        except OSError:
            pass


def _kai_map_file(path: str):
    """Read-only shared mapping of a whole shard with readahead DISABLED
    (MADV_RANDOM, the mmap twin of the preadv path's POSIX_FADV_RANDOM): a
    fault brings exactly the 4 KiB page of the row. Window 7 (2026-09-11)
    showed why: without it every cold-row fault pulled an NFS readahead window
    (rsize 1 MiB) — 9K prefill 14.7 s vs 8.1 s, code decode 53 vs 61 tok/s."""
    import mmap as _mmap
    import numpy as _np

    fd = _kai_os.open(path, _kai_os.O_RDONLY)
    try:
        mm = _mmap.mmap(fd, 0, access=_mmap.ACCESS_READ)
    finally:
        _kai_os.close(fd)
    try:
        mm.madvise(_mmap.MADV_RANDOM)
    except (AttributeError, OSError):  # pragma: no cover - old kernels
        pass
    return mm, _np.frombuffer(mm, dtype=_np.uint8)


def _kai_mmap_rows(view, rel, lo: int, hi: int, row_bytes: int, buf) -> None:
    """Copy rows rel[lo:hi] of the [num_rows, row_bytes] memmap view into buf
    (a flat uint8 memoryview). One fancy-index gather: C memcpy for resident
    pages, page faults for cold ones; the GIL is released inside NumPy."""
    import numpy as _np

    dst = _np.frombuffer(buf, dtype=_np.uint8)[lo * row_bytes : hi * row_bytes].reshape(
        hi - lo, row_bytes
    )
    _np.take(view, _np.asarray(rel[lo:hi], dtype=_np.int64), axis=0, out=dst)


def _kai_pread_rows(
    fd: int, base: int, rel: list, lo: int, hi: int, row_bytes: int, buf
) -> None:
    for i in range(lo, hi):
        off = base + rel[i] * row_bytes
        view = buf[i * row_bytes : (i + 1) * row_bytes]
        got = 0
        while got < row_bytes:
            n = _kai_os.preadv(fd, [view[got:]], off + got)
            if n <= 0:
                raise OSError("engram disk table: short read")
            got += n


def _kai_parallel_read(jobs: list) -> None:
    """jobs: [(fd, base, rel_list, row_bytes, memoryview)]. Every row of every job
    is in flight at once: a task carries ceil(total / threads) rows (1 row at
    decode sizes), capped at DSV41_ENGRAM_DISK_CHUNK for prefill-sized batches.
    Only the calling thread submits, so the shared pool cannot deadlock."""
    total = sum(len(job[2]) for job in jobs)
    if total == 0:
        return

    def _run(job, lo, hi):
        fd, base, rel, row_bytes, buf = job[:5]
        view = job[5] if len(job) > 5 else None
        if view is not None:
            _kai_mmap_rows(view, rel, lo, hi, row_bytes, buf)
        else:
            _kai_pread_rows(fd, base, rel, lo, hi, row_bytes, buf)

    if total == 1:
        for job in jobs:
            _run(job, 0, len(job[2]))
        return
    # Memmap gathers: spread the rows over every pool thread (cold rows fault
    # in sequentially within a task, so more tasks = more reads in flight).
    if any(len(job) > 5 and job[5] is not None for job in jobs):
        if total <= _KAI_MMAP_INLINE_ROWS and all(len(job) > 5 and job[5] is not None for job in jobs):
            adv = _kai_os.POSIX_FADV_WILLNEED
            for fd, base, rel, row_bytes, buf, view in jobs:
                for r in rel:
                    try:
                        _kai_os.posix_fadvise(fd, base + r * row_bytes, row_bytes, adv)
                    except OSError:
                        pass
            for fd, base, rel, row_bytes, buf, view in jobs:
                _kai_mmap_rows(view, rel, 0, len(rel), row_bytes, buf)
            return
        chunk = max(1, -(-total // _KAI_MMAP_THREADS))
    else:
        chunk = max(1, min(_KAI_CHUNK, -(-total // _KAI_THREADS)))
    pool = _kai_pool()
    futs = [
        pool.submit(_run, job, lo, min(lo + chunk, len(job[2])))
        for job in jobs
        for lo in range(0, len(job[2]), chunk)
    ]
    for fut in futs:
        fut.result()


def _dsv41_engram_local_dir(model_dir: str, layer_id: int, row_start: int, num_rows) -> str:
    """Tech2Wild 2026-09-10: node-local Engram rows. DSV41_ENGRAM_DIR points at a sparse copy
    of the Engram shards holding only this rank's rows at the original byte offsets, plus
    engram-local.json with the copied row range per layer (tools/engram_local.py). It is used
    only when that range covers this table's rows; otherwise the table reads model_dir as before,
    so a changed rank map can never read the copy's empty holes."""
    local = _kai_os.environ.get("DSV41_ENGRAM_DIR", "")
    if not local:
        return model_dir
    try:
        with open(_kai_os.path.join(local, "engram-local.json")) as f:
            lo, hi = _kai_json.load(f)["layers"][str(layer_id)]
    except Exception as e:  # noqa: BLE001
        logger.warning("DSV41_ENGRAM_DIR=%s unusable for layer %d (%s); reading %s",
                       local, layer_id, e, model_dir)
        return model_dir
    if num_rows is None or not (lo <= row_start and row_start + num_rows <= hi):
        logger.warning("DSV41_ENGRAM_DIR=%s holds layer %d rows [%d, %d), this rank needs "
                       "[%d, +%s); reading %s", local, layer_id, lo, hi, row_start, num_rows, model_dir)
        return model_dir
    logger.info("Engram layer %d rows [%d, %d) read from node-local %s",
                layer_id, row_start, row_start + num_rows, local)
    return local


class DiskEngramTable:
    """Tech2Wild/Kai 2026-09-10: read Engram rows straight from the safetensors
    shards with positional preads on a thread pool, so the per-rank Engram
    shard (~47 GiB at TP4) never occupies memory. Built for DGX Spark, where
    "pinned host memory" is the same 128 GB pool the GPU uses. Rows are
    dequantized on the CPU (fp8 e4m3 x ue8m0 block scales -> bf16) and copied
    into the GPU staging buffer. Enable with DSV41_ENGRAM_DISK=1.
    Lineage: our own PLE-on-disk patch for Qwen3.8-Flash-Next (2026-09-05).
    """

    def __init__(
        self,
        model_dir: str,
        layer_id: int,
        dim: int,
        block_size: int,
        row_start: int = 0,
        num_rows: int | None = None,
    ):
        model_dir = _dsv41_engram_local_dir(model_dir, layer_id, row_start, num_rows)
        idx_path = _kai_os.path.join(model_dir, "model.safetensors.index.json")
        with open(idx_path) as f:
            weight_map = _kai_json.load(f)["weight_map"]
        wname = f"layers.{layer_id}.engram.embed.weight"
        sname = f"layers.{layer_id}.engram.embed.scale"
        self.w_fd, self.w_off, self.w_shape = self._open(model_dir, weight_map[wname], wname)
        self.s_fd, self.s_off, self.s_shape = self._open(model_dir, weight_map[sname], sname)
        self._w_path = _kai_os.path.join(model_dir, weight_map[wname])
        self._s_path = _kai_os.path.join(model_dir, weight_map[sname])
        self.dim = dim
        self.sb = dim // block_size
        assert self.w_shape[1] == dim, (self.w_shape, dim)
        assert self.s_shape[1] == self.sb, (self.s_shape, self.sb)
        assert self.s_shape[0] == self.w_shape[0], (self.s_shape, self.w_shape)
        # The checkpoint holds the FULL table (every TP rank's hash heads) per layer,
        # while callers pass rank-local row ids (global - vocab_start), so reads must
        # start at this rank's first row.
        if num_rows is None:
            num_rows = self.w_shape[0] - row_start
        assert 0 <= row_start and row_start + num_rows <= self.w_shape[0], (
            row_start, num_rows, self.w_shape)
        self.w_off += row_start * dim
        self.s_off += row_start * self.sb
        self.row_start = row_start
        self.num_rows = num_rows
        # Stage 6: [num_rows, row_bytes] uint8 views of this rank's rows through a
        # read-only memory map of each shard (opened once; pages come and go
        # with the OS page cache, nothing is pinned).
        self.w_view = self.s_view = None
        if _KAI_MMAP:
            import numpy as _np

            self._w_mm, w_buf = _kai_map_file(self._w_path)
            self._s_mm, s_buf = _kai_map_file(self._s_path)
            self.w_view = _np.ndarray(
                (num_rows, dim), dtype=_np.uint8, buffer=w_buf, offset=self.w_off
            )
            self.s_view = _np.ndarray(
                (num_rows, self.sb), dtype=_np.uint8, buffer=s_buf, offset=self.s_off
            )
        self.threads = _KAI_THREADS
        self.chunk = _KAI_CHUNK
        self.pool = _kai_pool()  # shared by all tables (was one pool per table)
        logger.info(
            "Engram DISK mode: layer %d rows [%d, %d) read from %s (off=%d) and %s (off=%d); "
            "%d threads, chunk %d",
            layer_id, row_start, row_start + num_rows, weight_map[wname], self.w_off,
            weight_map[sname], self.s_off, self.threads, self.chunk,
        )

    @staticmethod
    def _open(model_dir: str, fname: str, tname: str):
        path = _kai_os.path.join(model_dir, fname)
        fd = _kai_os.open(path, _kai_os.O_RDONLY)
        try:
            _kai_os.posix_fadvise(fd, 0, 0, _kai_os.POSIX_FADV_RANDOM)
        except Exception:  # noqa: BLE001
            pass
        with open(path, "rb") as f:
            n = _kai_struct.unpack("<Q", f.read(8))[0]
            hdr = _kai_json.loads(f.read(n))
        meta = hdr[tname]
        start = meta["data_offsets"][0]
        return fd, 8 + n + start, tuple(meta["shape"])

    def _read_rows(self, fd: int, base: int, rel: list, row_bytes: int, buf) -> None:
        _kai_parallel_read([(fd, base, rel, row_bytes, buf)])

    def read_jobs(self, rel_l: list, w: torch.Tensor, s: torch.Tensor) -> list:
        """Read jobs for rank-local rows `rel_l` into uint8 w [R, dim], s [R, sb]."""
        if self.w_view is not None and len(rel_l) >= _KAI_MMAP_MIN_ROWS:
            return [
                (self.w_fd, self.w_off, rel_l, self.dim, memoryview(w.numpy()).cast("B"), self.w_view),
                (self.s_fd, self.s_off, rel_l, self.sb, memoryview(s.numpy()).cast("B"), self.s_view),
            ]
        return [
            (self.w_fd, self.w_off, rel_l, self.dim, memoryview(w.numpy()).cast("B")),
            (self.s_fd, self.s_off, rel_l, self.sb, memoryview(s.numpy()).cast("B")),
        ]

    def dequant(self, w: torch.Tensor, s: torch.Tensor) -> torch.Tensor:
        """fp8 e4m3 rows (as uint8) x ue8m0 block scales -> [R, dim] fp32."""
        R = w.shape[0]
        vals = w.view(torch.float8_e4m3fn).to(torch.float32).view(
            R, self.sb, self.dim // self.sb
        )
        # ue8m0 byte is the fp32 exponent field: 2^(e-127)
        scale = (s.to(torch.int32) << 23).view(torch.float32)
        return (vals * scale[:, :, None]).reshape(R, self.dim)

    def gather_dequant(self, rel: torch.Tensor, owned: torch.Tensor) -> torch.Tensor:
        """rel: [R] int64 CPU local row ids; owned: [R] bool. Returns [R, dim] bf16 CPU."""
        return gather_dequant_many([(self, rel, owned)])[0]


def gather_dequant_many(requests: list) -> list[torch.Tensor]:
    """Tech2Wild/Kai 2026-09-10: requests = [(DiskEngramTable, rel [R] int64 CPU,
    owned [R] bool)] -> one [R, dim] bf16 CPU tensor per request. Rows are
    de-duplicated per table (repeated n-grams, zero-filled dummy batches) and the
    reads of every table go out in ONE parallel batch. Same math as before."""
    plans, jobs = [], []
    for table, rel, owned in requests:
        uniq, inverse = torch.unique(rel, return_inverse=True)
        w = torch.empty((uniq.numel(), table.dim), dtype=torch.uint8)
        s = torch.empty((uniq.numel(), table.sb), dtype=torch.uint8)
        jobs += table.read_jobs(uniq.tolist(), w, s)
        plans.append((table, w, s, inverse, owned))
    _kai_parallel_read(jobs)
    outs = []
    for table, w, s, inverse, owned in plans:
        out = table.dequant(w, s)[inverse]
        out[~owned] = 0
        outs.append(out.to(torch.bfloat16))
    return outs


class ParallelEngramEmbedding(nn.Module):
    """The n-gram hash table, sharded by complete hash heads over TP ranks.
    Rows stay fp8 and are dequantized with ue8m0 per-32 scales on lookup.

    With `cpu_offload` the shard lives in pinned host memory and is read over
    UVA instead of HBM; the TP sharding is unchanged either way.
    """

    def __init__(
        self,
        num_embeddings: int,
        dim: int,
        head_sizes: tuple[int, ...],
        block_size: int = 32,
        cpu_offload: bool = False,
        model_dir: str | None = None,
        layer_id: int | None = None,
    ):
        super().__init__()
        self.disk: DiskEngramTable | None = None
        if _DSV41_ENGRAM_DISK:
            cpu_offload = False
        tp_size = get_tensor_model_parallel_world_size()
        tp_rank = get_tensor_model_parallel_rank()
        assert head_sizes and all(size > 0 for size in head_sizes)
        assert sum(head_sizes) <= num_embeddings
        if cpu_offload and not is_uva_available():
            raise RuntimeError("Engram CPU offload requires UVA support")
        self.num_embeddings = num_embeddings
        self.dim = dim
        self.block_size = block_size
        self.n_hash_cols = len(head_sizes)
        self.part_n_hash_cols = triton.cdiv(self.n_hash_cols, tp_size)
        self.head_start = tp_rank * self.part_n_hash_cols
        head_end = self.head_start + self.part_n_hash_cols
        self.vocab_start_idx = sum(head_sizes[: self.head_start])
        self.vocab_end_idx = sum(head_sizes[:head_end])
        self.part_num_embeddings = self.vocab_end_idx - self.vocab_start_idx
        self.tp_size = tp_size
        self.cpu_offload = cpu_offload
        self._views: tuple[torch.Tensor, torch.Tensor] | None = None
        self._view_src: tuple[int, int] | None = None
        self._num_sms = torch.cuda.get_device_properties(
            torch.accelerator.current_device_index()
        ).multi_processor_count

        # Explicit device: model init runs under a `torch.device("cuda")`
        # context, which would otherwise put the shard in HBM.
        kwargs = {"device": "cpu", "pin_memory": True} if cpu_offload else {}
        if _DSV41_ENGRAM_DISK:
            assert model_dir is not None and layer_id is not None
            self.disk = DiskEngramTable(
                model_dir, layer_id, dim, block_size,
                row_start=self.vocab_start_idx, num_rows=self.part_num_embeddings,
            )
            # 1-row placeholders as BUFFERS (not Parameters): the checkpoint
            # rows are skipped by the weights iterator and must not trip the
            # "weights not initialized" check.
            self.register_buffer("weight", torch.zeros(1, dim, dtype=torch.float8_e4m3fn, device="cpu"), persistent=False)
            self.register_buffer("weight_scale_inv", torch.zeros(1, dim // block_size, dtype=torch.uint8, device="cpu"), persistent=False)
            logger.info(
                "Engram table DISK-backed: %d rows x %d per rank stay on disk (%.2f GiB not allocated)",
                self.part_num_embeddings, dim,
                self.part_num_embeddings * (dim + dim // block_size) / 1024**3,
            )
            return
        self.weight = nn.Parameter(
            torch.empty(
                self.part_num_embeddings, dim, dtype=torch.float8_e4m3fn, **kwargs
            ),
            requires_grad=False,
        )
        self.weight_scale_inv = nn.Parameter(
            torch.empty(
                self.part_num_embeddings,
                dim // block_size,
                dtype=torch.uint8,
                **kwargs,
            ),
            requires_grad=False,
        )
        for param in (self.weight, self.weight_scale_inv):
            set_weight_attrs(
                param,
                {
                    "weight_loader": _engram_head_shard_weight_loader,
                    "engram_vocab_start": self.vocab_start_idx,
                },
            )
        if cpu_offload:
            logger.info(
                "Engram table offloaded to pinned host memory: %d rows x %d, "
                "%.2f GiB per rank",
                self.part_num_embeddings,
                dim,
                self.part_num_embeddings * (dim + dim // block_size) / 1024**3,
            )

    def _storage(self) -> tuple[torch.Tensor, torch.Tensor]:
        """Parameters when resident, else cached UVA views of the pinned shard.

        Rebuilt if anything swaps `.data`, so a stale device pointer cannot
        survive silently.
        """
        if not self.cpu_offload:
            return self.weight.data, self.weight_scale_inv.data
        src = (self.weight.data_ptr(), self.weight_scale_inv.data_ptr())
        if self._view_src != src:
            self._views = (
                get_accelerator_view_from_cpu_tensor(self.weight.data),
                get_accelerator_view_from_cpu_tensor(self.weight_scale_inv.data),
            )
            self._view_src = src
        assert self._views is not None
        return self._views

    def lookup(
        self, indices: torch.Tensor, out: torch.Tensor, background: bool = False
    ) -> None:
        """Look up local heads of [T, heads] into [T, local_heads, dim] bf16.

        `background` limits the grid to leave SMs for concurrent work.
        """
        rows = indices.shape[0] * self.part_n_hash_cols
        if not rows:
            return
        if self.disk is not None:
            self._disk_lookup(indices, out)
            return
        weight, scales = self._storage()
        # The table dwarfs TLB reach, so a persistent grid near the SM count
        # beats one program per row; halve it to leave SMs for the main stream.
        tiles = triton.cdiv(rows, 16)
        grid = min(tiles, self._num_sms // 2 if background else self._num_sms)
        _engram_lookup_kernel[(grid,)](
            weight,
            scales,
            indices,
            out,
            self.vocab_start_idx,
            self.vocab_end_idx,
            rows,
            indices.stride(0),
            indices.stride(1),
            HEAD_START=self.head_start,
            LOCAL_HEADS=self.part_n_hash_cols,
            TOTAL_HEADS=self.n_hash_cols,
            DIM=self.dim,
            QUANT_BLOCK=self.block_size,
            BLOCK_R=16,
            GRID=grid,
        )

    def disk_rel_owned(self, local: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
        """local: [T, <=L] int64 CPU hash ids of this rank's heads -> flat
        (rel [T*L] rank-local row ids, owned [T*L] bool). Padded heads (beyond
        TOTAL_HEADS) and ids outside this rank's range are not owned (zeros)."""
        T = local.shape[0]
        L = self.part_n_hash_cols
        if local.shape[1] < L:
            pad = torch.full((T, L - local.shape[1]), -1, dtype=torch.int64)
            local = torch.cat([local, pad], dim=1)
        rows = local.reshape(-1)
        owned = (rows >= self.vocab_start_idx) & (rows < self.vocab_end_idx)
        rel = torch.where(owned, rows - self.vocab_start_idx, torch.zeros_like(rows))
        return rel, owned

    def _disk_lookup(self, indices: torch.Tensor, out: torch.Tensor) -> None:
        """Host-side gather + dequant for DISK mode, then H2D into `out`
        ([T, local_heads, dim] bf16). Padded heads (beyond TOTAL_HEADS) and
        rows outside this rank's range write zeros, matching the kernel.

        In-forward fallback only (V1 runner / eager). Under the V2 runner the
        rows are staged by EngramDiskStager before the forward instead."""
        if torch.cuda.is_available() and torch.cuda.is_current_stream_capturing():
            raise RuntimeError(
                "Engram DISK lookup reached a CUDA-graph capture: it needs a host "
                "round trip. Stage rows before the forward (EngramDiskStager, V2 "
                "runner + patched nvidia/model_state.py) or run --enforce-eager."
            )
        T = indices.shape[0]
        L = self.part_n_hash_cols
        ids = indices.detach().to("cpu", dtype=torch.int64)
        head_end = min(self.head_start + L, self.n_hash_cols)
        rel, owned = self.disk_rel_owned(ids[:, self.head_start:head_end])
        assert self.disk is not None
        deq = self.disk.gather_dequant(rel, owned)
        out[:T].copy_(deq.view(T, L, self.dim))

    def forward(self, indices: torch.Tensor) -> torch.Tensor:
        """indices: [num_tokens, n_hash_cols] -> [num_tokens, n_hash_cols, dim]
        bf16, gathered from all TP shards."""
        out = torch.empty(
            (indices.shape[0], self.part_n_hash_cols, self.dim),
            dtype=torch.bfloat16,
            device=indices.device,
        )
        self.lookup(indices, out)
        if self.tp_size > 1:
            out = tensor_model_parallel_all_gather(out, dim=1)
            out = out[:, : self.n_hash_cols]
        return out


@triton.jit(do_not_specialize=["num_kv_tokens"])
def _fused_engram_post_wkv_kernel(
    hidden_states,
    kv,
    q_weight,
    k_weight,
    token_mask,
    output,
    num_kv_tokens,
    hidden_stride_t,
    hidden_stride_h,
    hidden_stride_d,
    kv_stride_t,
    kv_stride_d,
    q_stride_h,
    q_stride_d,
    k_stride_h,
    k_stride_d,
    mask_stride,
    output_stride_t,
    output_stride_h,
    output_stride_d,
    eps,
    clamp_value,
    DIM: tl.constexpr,
    HC_MULT: tl.constexpr,
    BLOCK_SIZE: tl.constexpr,
    HAS_MASK: tl.constexpr,
):
    program_idx = tl.program_id(0)
    token_idx = program_idx // HC_MULT
    hc_idx = program_idx % HC_MULT
    token_idx = token_idx.to(tl.int64)
    source_idx = token_idx
    source_valid = source_idx < num_kv_tokens

    dim_offsets = tl.arange(0, BLOCK_SIZE)
    dim_valid = dim_offsets < DIM
    hidden = tl.load(
        hidden_states
        + token_idx * hidden_stride_t
        + hc_idx * hidden_stride_h
        + dim_offsets * hidden_stride_d,
        mask=dim_valid,
        other=0.0,
    ).to(tl.float32)
    key = tl.load(
        kv + source_idx * kv_stride_t + (hc_idx * DIM + dim_offsets) * kv_stride_d,
        mask=source_valid & dim_valid,
        other=0.0,
    ).to(tl.float32)
    q = tl.load(
        q_weight + hc_idx * q_stride_h + dim_offsets * q_stride_d,
        mask=dim_valid,
        other=0.0,
    ).to(tl.float32)
    k = tl.load(
        k_weight + hc_idx * k_stride_h + dim_offsets * k_stride_d,
        mask=dim_valid,
        other=0.0,
    ).to(tl.float32)

    hidden_rms = tl.rsqrt(tl.sum(hidden * hidden, axis=0) / DIM + eps)
    key_rms = tl.rsqrt(tl.sum(key * key, axis=0) / DIM + eps)
    dot = tl.sum(hidden * q * k * key, axis=0)
    dot *= hidden_rms * key_rms * tl.rsqrt(DIM * 1.0)
    gate_input = tl.sqrt(tl.maximum(tl.abs(dot), clamp_value))
    gate_input = tl.where(dot < 0.0, -gate_input, gate_input)
    gate = tl.sigmoid(gate_input)
    if HAS_MASK:
        active = tl.load(
            token_mask + source_idx * mask_stride,
            mask=source_valid,
            other=0,
        )
        gate = tl.where(active, gate, 0.0)

    value = tl.load(
        kv + source_idx * kv_stride_t + (HC_MULT * DIM + dim_offsets) * kv_stride_d,
        mask=source_valid & dim_valid,
        other=0.0,
    ).to(tl.float32)
    tl.store(
        output
        + token_idx * output_stride_t
        + hc_idx * output_stride_h
        + dim_offsets * output_stride_d,
        hidden + gate * value,
        mask=dim_valid,
    )


@triton.jit(do_not_specialize=["num_tokens", "token_start", "num_elements"])
def _engram_sp_rows_kernel(
    gathered,
    output,
    num_tokens,
    token_start,
    num_elements,
    LOCAL_WIDTH: tl.constexpr,
    WIDTH: tl.constexpr,
    BLOCK_SIZE: tl.constexpr,
):
    offsets = tl.program_id(0).to(tl.int64) * BLOCK_SIZE + tl.arange(0, BLOCK_SIZE)
    tokens = token_start + offsets // WIDTH
    cols = offsets % WIDTH
    source = (cols // LOCAL_WIDTH * num_tokens + tokens) * LOCAL_WIDTH
    source += cols % LOCAL_WIDTH
    values = tl.load(
        gathered + source, (offsets < num_elements) & (tokens < num_tokens), other=0
    )
    tl.store(output + offsets, values, offsets < num_elements)


class Engram(nn.Module):
    """Writes an n-gram lookup into the residual stream, gated by how well it
    matches that stream.

    The hash ids fetch `n_hash_cols` rows; `wkv` turns them into one key per
    hc copy plus a shared value. The gate is a normalized dot product of the
    stream against the key, signed-sqrt'ed before the sigmoid (matching the
    training kernel).
    """

    def __init__(
        self,
        config,
        quant_config: QuantizationConfig | None,
        layout: EngramLayout,
        layer_hash_index: int,
        use_sequence_parallel: bool,
        prefix: str,
    ) -> None:
        super().__init__()
        self.layer_hash_index = layer_hash_index
        self.dim = config.hidden_size
        self.hc_mult = config.hc_mult
        self.eps = config.rms_norm_eps
        self.clamp_value = 1e-6
        self.use_sequence_parallel = use_sequence_parallel

        # Named ``embed_tokens`` so the checkpoint's ``engram.embed.weight``
        # survives the mapper's ``embed.weight`` -> ``embed_tokens.weight``
        # suffix rule.
        engram_config = get_current_vllm_config().engram_config
        self.embed_tokens = ParallelEngramEmbedding(
            layout.num_embeddings[layer_hash_index],
            layout.head_dim,
            tuple(size for order in layout.primes[layer_hash_index] for size in order),
            cpu_offload=engram_config.cpu_offload if engram_config else True,
            model_dir=get_current_vllm_config().model_config.model,
            layer_id=layout.layer_ids[layer_hash_index],
        )
        n_hash_cols = (layout.max_ngram_size - 1) * layout.n_heads
        self.wkv = ReplicatedLinear(
            n_hash_cols * layout.head_dim,
            self.dim * (self.hc_mult + 1),
            bias=False,
            quant_config=quant_config,
            return_bias=False,
            prefix=f"{prefix}.wkv",
        )
        self.q_weight = nn.Parameter(
            torch.empty(self.hc_mult, self.dim, dtype=torch.bfloat16),
            requires_grad=False,
        )
        self.k_weight = nn.Parameter(
            torch.empty(self.hc_mult, self.dim, dtype=torch.bfloat16),
            requires_grad=False,
        )

        max_tokens = get_current_vllm_config().scheduler_config.max_num_batched_tokens
        # Tech2Wild/Kai 2026-09-10 (CUDA graphs): under the V2 runner a DISK
        # table is staged by the model state (EngramDiskStager) BEFORE the
        # forward, so the forward holds no host round trip and can be captured
        # (FULL decode graphs included). The V1 runner keeps the in-forward path.
        self.prestage = self.embed_tokens.disk is not None and bool(
            getattr(get_current_vllm_config(), "use_v2_model_runner", False)
        )
        # Keep lookup results alive across breakable graph segments. Zeroed so
        # rows of cudagraph padding tokens (never staged) stay finite.
        self.staged_rows = torch.zeros(
            max_tokens,
            self.embed_tokens.part_n_hash_cols,
            layout.head_dim,
            dtype=torch.bfloat16,
        )

    def prepare_embeddings(self, hash_ids: torch.Tensor) -> None:
        """Gather this layer's rows on the main stream before decoder layers."""
        if self.prestage:
            # Already staged for this step by EngramDiskStager.stage().
            return
        self.embed_tokens.lookup(hash_ids, self.staged_rows[: hash_ids.shape[0]])

    def embed(self, hash_ids: torch.Tensor) -> torch.Tensor:
        """Gather heads, returning only local tokens when SP is enabled."""
        rows = self.staged_rows[: hash_ids.shape[0]]
        if self.embed_tokens.tp_size == 1:
            return rows
        if self.use_sequence_parallel:
            tp_size = self.embed_tokens.tp_size
            num_tokens, local_heads, dim = rows.shape
            gathered = tensor_model_parallel_all_gather(rows, dim=0)
            chunk = (num_tokens + tp_size - 1) // tp_size
            rows = rows.new_empty((chunk, self.embed_tokens.n_hash_cols, dim))
            _engram_sp_rows_kernel[(triton.cdiv(rows.numel(), 1024),)](
                gathered,
                rows,
                num_tokens,
                get_tensor_model_parallel_rank() * chunk,
                rows.numel(),
                local_heads * dim,
                self.embed_tokens.n_hash_cols * dim,
                BLOCK_SIZE=1024,
            )
            return rows
        rows = tensor_model_parallel_all_gather(rows, dim=1)
        return rows[:, : self.embed_tokens.n_hash_cols]

    def forward(
        self,
        hidden_states: torch.Tensor,
        hash_ids: torch.Tensor,
        token_mask: torch.Tensor | None = None,
    ) -> torch.Tensor:
        # Stage 7: phase "engram_fwd" of the prefill budget (inert unless enabled).
        from vllm.model_executor.kernels.attention.dsa.candidate_blocks import _PrefillTimer

        with _PrefillTimer("engram_fwd", int(hash_ids.shape[0])):
            return self._forward_impl(hidden_states, hash_ids, token_mask)

    def _forward_impl(
        self,
        hidden_states: torch.Tensor,
        hash_ids: torch.Tensor,
        token_mask: torch.Tensor | None = None,
    ) -> torch.Tensor:
        """hidden_states: [T, hc_mult, dim]; hash_ids: [T, n_hash_cols] (all
        tokens, pre sequence-parallel shard); token_mask: [T], False shuts
        the gate so those positions pass through untouched."""
        kv = self.wkv(self.embed(hash_ids).flatten(-2))
        num_kv_tokens = hash_ids.shape[0]
        assert token_mask is None or token_mask.shape == (num_kv_tokens,)
        if self.use_sequence_parallel:
            tp_size = get_tensor_model_parallel_world_size()
            tp_rank = get_tensor_model_parallel_rank()
            shard_size = (num_kv_tokens + tp_size - 1) // tp_size
            assert hidden_states.shape[0] == shard_size
            start = min(tp_rank * shard_size, num_kv_tokens)
            num_kv_tokens = min(shard_size, num_kv_tokens - start)
            if token_mask is not None:
                token_mask = token_mask[start : start + num_kv_tokens]

        num_tokens, hc_mult, dim = hidden_states.shape
        assert hc_mult == self.hc_mult and dim == self.dim
        assert kv.ndim == 2 and kv.shape[1] == (hc_mult + 1) * dim
        output = torch.empty_like(hidden_states)
        if num_tokens == 0:
            return output

        block_size = triton.next_power_of_2(dim)
        num_warps = 8 if block_size >= 2048 else 4
        mask = token_mask if token_mask is not None else hidden_states
        _fused_engram_post_wkv_kernel[(num_tokens * hc_mult,)](
            hidden_states,
            kv,
            self.q_weight,
            self.k_weight,
            mask,
            output,
            num_kv_tokens,
            hidden_states.stride(0),
            hidden_states.stride(1),
            hidden_states.stride(2),
            kv.stride(0),
            kv.stride(1),
            self.q_weight.stride(0),
            self.q_weight.stride(1),
            self.k_weight.stride(0),
            self.k_weight.stride(1),
            token_mask.stride(0) if token_mask is not None else 0,
            output.stride(0),
            output.stride(1),
            output.stride(2),
            self.eps,
            self.clamp_value,
            DIM=dim,
            HC_MULT=hc_mult,
            BLOCK_SIZE=block_size,
            HAS_MASK=token_mask is not None,
            num_warps=num_warps,
        )
        return output


class EngramDiskStager:
    """Tech2Wild/Kai 2026-09-10 (CUDA graphs): stage DISK-mode Engram rows
    OUTSIDE the model forward.

    The V2 model state calls :meth:`stage` from ``prepare_inputs`` once per
    step, after the runner has written this step's input ids, positions and
    query_start_loc (all on the current stream). It hashes the step on the GPU
    (same kernel, same inputs as the forward), copies only this rank's hash
    columns to pinned host memory, waits on ONE event, reads every row of every
    engram layer in one parallel batch, dequantizes on the CPU exactly like the
    in-forward path, and copies (async, from pinned memory) into each layer's
    persistent ``staged_rows``. The forward (eager, breakable PIECEWISE or FULL
    graph) only reads ``staged_rows``, so it never needs the host.
    """

    def __init__(self, hash_state: NgramHashState, engrams: list["Engram"]) -> None:
        assert engrams and all(e.embed_tokens.disk is not None for e in engrams)
        self.hash_state = hash_state
        self.engrams = sorted(engrams, key=lambda e: e.layer_hash_index)
        emb = self.engrams[0].embed_tokens
        self.local_heads = emb.part_n_hash_cols
        self.head_start = emb.head_start
        self.head_end = min(emb.head_start + emb.part_n_hash_cols, emb.n_hash_cols)
        self.dim = emb.dim
        self.max_tokens = self.engrams[0].staged_rows.shape[0]
        num_layers = hash_state.multipliers.shape[0]
        self.hash_host = torch.empty(
            (self.max_tokens, num_layers, self.head_end - self.head_start),
            dtype=torch.int32,
            device="cpu",
            pin_memory=True,
        )
        self.rows_host = [
            torch.empty(
                (self.max_tokens, self.local_heads, self.dim),
                dtype=torch.bfloat16,
                device="cpu",
                pin_memory=True,
            )
            for _ in self.engrams
        ]
        self.hashes_ready = torch.cuda.Event()
        self.num_staged = 0
        # DSV41_ENGRAM_TIMING=1: host wall-clock of stage() phases (diagnostic).
        self._timing = _kai_os.environ.get("DSV41_ENGRAM_TIMING", "0") == "1"
        # two buckets (decode: n <= 16 tokens, prefill: n > 16): sync, gather+dequant, tail, total, calls, rows, unique rows
        self._t_acc = [[0.0, 0.0, 0.0, 0.0, 0, 0, 0] for _ in range(2)]
        self._t_last_total_ms = 0.0  # stage 7 reads it after each timed prefill step
        # Stage 8: hash-ahead prefetch state (see _hash_ahead).
        self._prefetch = _KAI_PREFETCH
        if self._prefetch:
            self.hash_host_next = torch.empty_like(self.hash_host)
            self.hashes_next_ready = torch.cuda.Event()
            self.last_pos_host = torch.zeros(256, dtype=torch.int64, device="cpu", pin_memory=True)
            self._prefetch_pool = _KaiPool(
                max_workers=_KAI_PREFETCH_THREADS, thread_name_prefix="engram-prefetch"
            )
            self._prefetch_futs: list = []
            self._prefetch_stats = [0, 0, 0.0]  # lookaheads, rows, host seconds
            logger.info(
                "Engram DISK prefetch: hash-ahead of the next prefill chunk (%s, %d threads)",
                _KAI_PREFETCH_MODE, _KAI_PREFETCH_THREADS,
            )

        logger.info(
            "Engram DISK rows staged before the forward (graph-safe): %d layers, "
            "heads [%d, %d) of %d, up to %d tokens/step, %d read threads",
            len(self.engrams),
            self.head_start,
            self.head_end,
            emb.n_hash_cols,
            self.max_tokens,
            _KAI_THREADS,
        )

    def _hash_ahead(self, input_batch, req_states, positions, query_start_loc, lookback_token_ids) -> None:
        """Stage 8: for the batch's longest still-prefilling request, hash the
        NEXT chunk of its (known) prompt with the same kernel and warm this
        rank's rows in the page cache while the current step computes."""
        if torch.cuda.is_current_stream_capturing():
            return
        import time as _kai_time

        from .mm_preprocess import image_sentinel_mask

        t0 = _kai_time.perf_counter()
        n_reqs = int(input_batch.num_reqs)
        if n_reqs <= 0:
            return
        idx_np = input_batch.idx_mapping_np[:n_reqs]
        prompt_len = req_states.prompt_len.np[idx_np]
        next_start = self.last_pos_host[:n_reqs].numpy() + 1  # valid after hashes_ready.synchronize()
        remaining = prompt_len - next_start
        i = int(remaining.argmax())
        if remaining[i] <= 0:
            return
        req_idx = int(idx_np[i])
        s = int(next_start[i])
        L = int(min(remaining[i], self.max_tokens))
        tok = req_states.all_token_ids.gpu
        ids = tok[req_idx, s : s + L]
        pos = torch.arange(s, s + L, dtype=positions.dtype, device=positions.device)
        qsl = torch.tensor([0, L], dtype=query_start_loc.dtype, device=query_start_loc.device)
        depth = int(lookback_token_ids.shape[1])
        lb = torch.full((1, depth), -1, dtype=lookback_token_ids.dtype, device=lookback_token_ids.device)
        k = min(depth, s)
        if k > 0:
            lb[0, :k] = tok[req_idx, s - k : s].flip(0).to(lb.dtype)
        hashes = self.hash_state(
            ids, pos, qsl, image_sentinel_mask(ids), lb, image_sentinel_mask(lb), None, None
        )
        host = self.hash_host_next[:L]
        host.copy_(hashes[:, :, self.head_start : self.head_end], non_blocking=True)
        self.hashes_next_ready.record()
        self.hashes_next_ready.synchronize()
        self._prefetch_futs = [f for f in self._prefetch_futs if not f.done()]
        total_rows = 0
        for engram in self.engrams:
            local = host[:, engram.layer_hash_index, :].to(torch.int64)
            rel, owned = engram.embed_tokens.disk_rel_owned(local)
            rows = torch.unique(rel[owned]).tolist()
            total_rows += len(rows)
            table = engram.embed_tokens.disk
            for lo in range(0, len(rows), _KAI_PREFETCH_TASK_ROWS):
                self._prefetch_futs.append(
                    self._prefetch_pool.submit(
                        _kai_prefetch_rows, table, rows[lo : lo + _KAI_PREFETCH_TASK_ROWS]
                    )
                )
        st = self._prefetch_stats
        st[0] += 1
        st[1] += total_rows
        st[2] += _kai_time.perf_counter() - t0
        if st[0] % 16 == 0 or L >= 4096:
            logger.info(
                "Engram prefetch: next chunk %d tokens from %d (prompt %d), %d unique rows queued, "
                "host %.1f ms (%d lookaheads so far, avg %.0f rows, %.1f ms)",
                L, s, int(prompt_len[i]), total_rows, 1e3 * (_kai_time.perf_counter() - t0),
                st[0], st[1] / st[0], 1e3 * st[2] / st[0],
            )

    @torch.inference_mode()
    def stage(
        self,
        input_ids: torch.Tensor,
        positions: torch.Tensor,
        query_start_loc: torch.Tensor,
        lookback_token_ids: torch.Tensor,
        num_tokens: int,
        input_batch=None,
        req_states=None,
    ) -> int:
        """Stage rows for the first `num_tokens` (unpadded) tokens of this step.
        Returns the number staged; 0 while the KV cache is unbound (profiling),
        matching the forward, which skips engram then."""
        n = min(int(num_tokens), self.max_tokens)
        if n <= 0 or not self.hash_state.ensure_cache():
            return 0
        from .mm_preprocess import image_sentinel_mask

        timing = self._timing
        if timing:
            import time as _kai_time

            _t0 = _kai_time.perf_counter()

        ids = input_ids[:n]
        hashes = self.hash_state(
            ids,
            positions[:n],
            query_start_loc,
            image_sentinel_mask(ids),
            lookback_token_ids,
            image_sentinel_mask(lookback_token_ids),
            None,
            None,
        )
        host = self.hash_host[:n]
        host.copy_(hashes[:, :, self.head_start : self.head_end], non_blocking=True)
        prefetch = self._prefetch and input_batch is not None and req_states is not None
        if prefetch:
            # last position of every request in this step -> next chunk start
            n_reqs = int(input_batch.num_reqs)
            if n_reqs > self.last_pos_host.shape[0]:
                self.last_pos_host = torch.zeros(
                    n_reqs * 2, dtype=torch.int64, device="cpu", pin_memory=True
                )
            ends = torch.as_tensor(
                input_batch.query_start_loc_np[1 : n_reqs + 1].astype("int64") - 1,
                device=positions.device,
            )
            self.last_pos_host[:n_reqs].copy_(positions[ends], non_blocking=True)
        self.hashes_ready.record()
        # The one host sync per step. It also orders this step's rewrite of the
        # pinned buffers after the previous step's async H2D copies from them.
        self.hashes_ready.synchronize()
        if timing:
            _t1 = _kai_time.perf_counter()
        requests = []
        for engram in self.engrams:
            local = host[:, engram.layer_hash_index, :].to(torch.int64)
            rel, owned = engram.embed_tokens.disk_rel_owned(local)
            requests.append((engram.embed_tokens.disk, rel, owned))
        rows_per_layer = gather_dequant_many(requests)
        if timing:
            _t2 = _kai_time.perf_counter()
            _acc = self._t_acc[0 if n <= 16 else 1]
            _acc[0] += _t1 - _t0
            _acc[1] += _t2 - _t1
            _acc[4] += 1
            _rows = sum(int(r[1].numel()) for r in requests)
            _uniq = sum(int(torch.unique(r[1]).numel()) for r in requests)
            _acc[5] += _rows
            _acc[6] += _uniq
            # the H2D tail is timed by the caller-visible end of stage(): record
            # the total when the call returns (see the return sites below)
            self._t_mid = _t2
            self._t_last = (_rows, _uniq, _t2 - _t1)
        for engram, rows, buf in zip(self.engrams, rows_per_layer, self.rows_host):
            staged = buf[:n]
            staged.copy_(rows.view(n, self.local_heads, self.dim))
            engram.staged_rows[:n].copy_(staged, non_blocking=True)
        self.num_staged = n
        if timing:
            _t3 = _kai_time.perf_counter()
            _acc = self._t_acc[0 if n <= 16 else 1]
            _acc[2] += _t3 - self._t_mid
            _acc[3] += _t3 - _t0
            self._t_last_total_ms = 1e3 * (_t3 - _t0)
            # stage 6 defines _KAI_MMAP (memmap gather); resolved at call time
            _mode = (
                "mmap"
                if globals().get("_KAI_MMAP", False)
                and self._t_last[0] >= globals().get("_KAI_MMAP_MIN_ROWS", 0) * len(self.engrams)
                else "preadv"
            )
            if n >= 1024:
                logger.info(
                    "Engram prefill stage [%s]: %d tokens, %d rows (%d unique), "
                    "gather+dequant %.1f ms, total %.1f ms",
                    _mode, n, self._t_last[0], self._t_last[1],
                    1e3 * self._t_last[2], 1e3 * (_t3 - _t0),
                )
            if _acc[4] % (200 if n <= 16 else 20) == 0:
                logger.info(
                    "Engram stage timing [%s, %s]: total %.2f ms/step (sync wait %.2f, "
                    "disk+dequant %.2f, h2d+rest %.2f), %.1f rows/step (%.1f unique), "
                    "avg of %d steps",
                    _mode, "decode" if n <= 16 else "prefill",
                    1e3 * _acc[3] / _acc[4],
                    1e3 * _acc[0] / _acc[4],
                    1e3 * _acc[1] / _acc[4],
                    1e3 * _acc[2] / _acc[4],
                    _acc[5] / _acc[4],
                    _acc[6] / _acc[4],
                    _acc[4],
                )
                for i in range(7):
                    _acc[i] = 0
        if prefetch:
            try:
                self._hash_ahead(input_batch, req_states, positions, query_start_loc, lookback_token_ids)
            except Exception as e:  # noqa: BLE001 - prefetch is best effort
                logger.warning_once("Engram prefetch disabled after error: %r", e)
                self._prefetch = False
        return n