alephlm.py

"""AlephLM — the full craft, config-driven.

Trigram byte (or BPE) embedding -> pre-norm stack (CausalSDPA majority,
CausalSplatHUB at the configured depths) -> LayerNorm -> DualHead.

Toggle surface (the causal contribution ledger, run at every eval):
    forward(idx, disable_bank=True)   dispatched experts off (exact C6 null)
    forward(idx, disable_hub=True)    hub attention residuals skipped
    forward(idx, disable_head_aleph=True)  gamma path off
"""
from __future__ import annotations

from typing import NamedTuple, Optional

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.utils.checkpoint  # noqa: F401 — torch<=2.4 does NOT auto-import
# this submodule; without it the hub_ckpt path AttributeErrors at step 1
# (the recorded A40 landmine, 2026-08-06)


class LMOutput(NamedTuple):
    """Still a tuple — `logits, loss = model(x)` keeps working — but also
    HF-duck-typed (`out.logits`, `out.loss`) so frozen-trunk tooling like
    amoe-lora drives the model natively."""
    logits: torch.Tensor
    loss: Optional[torch.Tensor]

from .presets import AlephLMConfig
from .attention import CausalSDPA, CausalSplatHUB
from .bank import AnchoredBank
from .embedding import TrigramByteEmbedding, TokenEmbedding
from .head import DualHead
# model/fusion.py is imported lazily (only when a config asks for fusion),
# so a vendored copy of this file without it keeps loading unfused crafts.


class Block(nn.Module):
    def __init__(self, cfg: AlephLMConfig, layer_idx: int):
        super().__init__()
        d = cfg.d_model
        self.is_hub = layer_idx in cfg.hub_layers
        self.ckpt_mode = getattr(cfg, "hub_ckpt", 0)
        self.n1 = nn.LayerNorm(d)
        self.n2 = nn.LayerNorm(d)
        if self.is_hub:
            self.attn = CausalSplatHUB(d, cfg.hub_K, cfg.hub_D, cfg.tau,
                                       chunk=cfg.hub_chunk,
                                       n_const=getattr(cfg, "hub_const", 1))
        else:
            self.attn = CausalSDPA(d, cfg.n_heads)
        self.bank = AnchoredBank(d, cfg.bank_experts, cfg.bank_ff, cfg.tau,
                                 cfg.gate_init)

    def forward(self, x, disable_bank=False, disable_hub=False):
        # hub_ckpt (2026-08-26): recompute-in-backward for the heavy
        # branches — at v2 scale the retained scan tensors alone exceed a
        # 95GB card (measured OOM). Training-path only; eval/decode and
        # the census replay (which calls attn/bank directly) are untouched.
        ckpt = self.ckpt_mode and self.training and torch.is_grad_enabled()
        if not (disable_hub and self.is_hub):
            if ckpt:
                x = x + torch.utils.checkpoint.checkpoint(
                    lambda t: self.attn(t), self.n1(x), use_reentrant=False)
            else:
                x = x + self.attn(self.n1(x))
        if ckpt and self.ckpt_mode >= 2 and not disable_bank:
            return x + torch.utils.checkpoint.checkpoint(
                lambda t: self.bank(t), self.n2(x), use_reentrant=False)
        return x + self.bank(self.n2(x), disable_dispatch=disable_bank)

    def prefill(self, x):
        a, cache = self.attn.prefill(self.n1(x))
        x = x + a
        return x + self.bank(self.n2(x)), cache

    def step(self, x_t, cache):
        x_t = x_t + self.attn.step(self.n1(x_t), cache)
        return x_t + self.bank(self.n2(x_t))


class AlephLM(nn.Module):
    def __init__(self, cfg: AlephLMConfig):
        super().__init__()
        self.cfg = cfg
        if cfg.tokenizer == "byte-trigram":
            assert cfg.vocab_size == 256, "byte crafts use vocab 256"
            self.embed = TrigramByteEmbedding(cfg.d_model, cfg.context)
            tied = None
        else:
            self.embed = TokenEmbedding(cfg.vocab_size, cfg.d_model, cfg.context)
            tied = self.embed.emb.weight if cfg.tie_embeddings else None
        self.blocks = nn.ModuleList(
            Block(cfg, i) for i in range(cfg.n_layers))
        self.nf = nn.LayerNorm(cfg.d_model)
        self.head = DualHead(cfg.d_model, cfg.vocab_size, cfg.head_K,
                             cfg.head_D, cfg.tau, tied_weight=tied)
        # v3: weak-token fusion (model/fusion.py). None keeps the byte-
        # resolution trunk verbatim; otherwise the middle blocks run over
        # units and the front/back blocks stay at byte resolution.
        spec = getattr(cfg, "fusion", None)
        if spec:
            # VENDORED: the live package reads
            #     from .fusion import Fusion
            # transformers' remote-code loader resolves every relative
            # import to a flat file beside this one, so written that way
            # it demands a fusion.py here and refuses to load the model.
            # This craft is fusion-free (its config carries no `fusion`
            # key, so this branch is never taken); fusion belongs to the
            # training side and itself imports the data subpackage, which
            # a flat package cannot carry. Deferred and resolved by name.
            from importlib import import_module
            Fusion = import_module("geolip.alephllm.model.fusion").Fusion
            self.fusion = Fusion(spec, cfg.d_model)
        else:
            self.fusion = None
        if self.fusion is not None:
            assert self.fusion.k_lo + self.fusion.k_hi <= cfg.n_layers, \
                "fusion: k_lo + k_hi must not exceed n_layers"

    def _ranges(self):
        L = len(self.blocks)
        lo, hi = self.fusion.k_lo, L - self.fusion.k_hi
        return self.blocks[:lo], self.blocks[lo:hi], self.blocks[hi:]

    def _trunk(self, x, idx, disable_bank=False, disable_hub=False):
        if self.fusion is None:
            for b in self.blocks:
                x = b(x, disable_bank=disable_bank, disable_hub=disable_hub)
            return x
        front, middle, back = self._ranges()
        plan = self.fusion.plan(idx)
        for b in front:
            x = b(x, disable_bank=disable_bank, disable_hub=disable_hub)
        if len(middle):
            u = plan.pool(x)
            for b in middle:
                u = b(u, disable_bank=disable_bank, disable_hub=disable_hub)
            x = x + plan.unpool(u, self.fusion.null)
        else:
            x = x + self.fusion.null.to(x.dtype)
        for b in back:
            x = b(x, disable_bank=disable_bank, disable_hub=disable_hub)
        return x

    def forward(self, idx=None, targets=None, disable_bank=False,
                disable_hub=False, disable_head_aleph=False,
                input_ids=None, labels=None, attention_mask=None):
        """HF-style aliases are accepted so frozen-trunk tooling drives the
        model unchanged, WITH HF semantics: `labels` are same-position and
        shifted internally (logits[:-1] vs labels[1:]); `targets` are the
        package's own pre-shifted convention and used as-is. attention_mask
        is deliberately ignored: under causal attention with right-padding
        and -100 label masking, pads can never influence a scored position."""
        if idx is None:
            idx = input_ids
        x = self.embed(idx)
        x = self._trunk(x, idx, disable_bank=disable_bank, disable_hub=disable_hub)
        h = self.nf(x)
        logits = self.head(h, disable_aleph=disable_head_aleph)
        if targets is not None:                     # pre-shifted (ours)
            loss = F.cross_entropy(
                logits.reshape(-1, logits.shape[-1]).float(),
                targets.reshape(-1), ignore_index=-100)
        elif labels is not None:                    # HF: shift internally
            loss = F.cross_entropy(
                logits[:, :-1].reshape(-1, logits.shape[-1]).float(),
                labels[:, 1:].reshape(-1), ignore_index=-100)
        else:
            return LMOutput(logits, None)
        return LMOutput(logits, loss)

    # ---------------------------------------------------- incremental decode
    @torch.no_grad()
    def prefill(self, idx):
        """Run the prompt once, return (last-position logits, decode cache).
        The cache carries per-layer attention state, the trigram history
        bytes, and the absolute position cursor."""
        self.eval()
        if self.fusion is not None:
            return self._prefill_fused(idx)
        from .embedding import PAD_ROW
        caches = []
        x = self.embed(idx)
        for b in self.blocks:
            x, c = b.prefill(x)
            caches.append(c)
        h = self.nf(x)
        logits = self.head(h[:, -1:])
        n = idx.shape[1]
        prev2 = idx[:, -2] if n >= 2 else torch.full_like(idx[:, -1], PAD_ROW)
        return logits, {"layers": caches, "t": n,
                        "prev1": idx[:, -1], "prev2": prev2}

    def _embed_step(self, next_id, cache, t):
        if isinstance(self.embed, TrigramByteEmbedding):
            e = (self.embed.emb0(next_id) + self.embed.emb1(cache["prev1"])
                 + self.embed.emb2(cache["prev2"])).unsqueeze(1) \
                + self.embed.pos[:, t:t + 1]
            cache["prev3"] = cache.get("prev2")
            cache["prev2"] = cache["prev1"]
            cache["prev1"] = next_id
        else:
            e = self.embed.emb(next_id).unsqueeze(1) + self.embed.pos[:, t:t + 1]
        return e

    @torch.no_grad()
    def decode_step(self, next_id, cache):
        """One token through the cached path. next_id: (B,) or (B,1)."""
        next_id = next_id.reshape(-1)
        t = cache["t"]
        assert t < self.cfg.context, "decode exceeded the position table"
        if self.fusion is not None:
            return self._decode_step_fused(next_id, cache, t)
        x = self._embed_step(next_id, cache, t)
        for b, c in zip(self.blocks, cache["layers"]):
            x = b.step(x, c)
        cache["t"] = t + 1
        return self.head(self.nf(x))

    # ------------------------------------------------- fused decode path
    @torch.no_grad()
    def _prefill_fused(self, idx):
        """The prompt through the hourglass: front blocks at byte
        resolution (batched), the middle over each row's COMPLETED units
        (the last unit stays open until a later byte starts the next), the
        back blocks at byte resolution. The middle's caches are merged
        across rows (the hub's prefix state has one shape per row)."""
        from .embedding import PAD_ROW
        from .governor import raw_block
        # VENDORED: from .fusion import cat_caches, zero_cache_rows (see __init__)
        from importlib import import_module
        _f = import_module("geolip.alephllm.model.fusion")
        cat_caches, zero_cache_rows = _f.cat_caches, _f.zero_cache_rows
        front, middle, back = self._ranges()
        for b in middle:
            assert raw_block(b).is_hub, "fused decode needs hub blocks in the middle"
        B, T = idx.shape
        plan = self.fusion.plan(idx)
        x = self.embed(idx)
        front_c = []
        for b in front:
            x, c = b.prefill(x)
            front_c.append(c)
        x_front = x
        u_all = plan.pool(x_front)                          # (B, J, d) incl. the open unit
        n_closed = plan.n_units - 1
        d = x.shape[-1]
        v_pad = torch.zeros(B, plan.J, d, device=x.device, dtype=x.dtype)
        mid_rows = []
        for r in range(B):
            nc = int(n_closed[r])
            u = u_all[r:r + 1, :max(nc, 1)]
            row_c = []
            for b in middle:
                u, c = b.prefill(u)
                row_c.append(c)
            if nc > 0:
                v_pad[r, :nc] = u[0]
            mid_rows.append(row_c)
        mid_c = [cat_caches([row[i] for row in mid_rows]) for i in range(len(middle))]
        empty = n_closed == 0
        if bool(empty.any()):
            mid_c = [zero_cache_rows(c, empty) for c in mid_c]
        g = plan.unpool(v_pad, self.fusion.null)              # (B, T, d)
        x = x_front + g
        back_c = []
        for b in back:
            x, c = b.prefill(x)
            back_c.append(c)
        logits = self.head(self.nf(x)[:, -1:])
        pad = lambda k: idx[:, -k] if T >= k else torch.full_like(idx[:, -1], PAD_ROW)  # noqa: E731
        return logits, {"front": front_c, "mid": mid_c, "back": back_c, "t": T,
                        "prev1": idx[:, -1], "prev2": pad(2), "prev3": pad(3),
                        "front_last": x_front[:, -1], "g": g[:, -1]}

    @torch.no_grad()
    def _decode_step_fused(self, next_id, cache, t):
        from .embedding import PAD_ROW
        # VENDORED: from .fusion import snapshot, restore_rows (see __init__)
        from importlib import import_module
        _f = import_module("geolip.alephllm.model.fusion")
        snapshot, restore_rows = _f.snapshot, _f.restore_rows
        front, middle, back = self._ranges()
        prev1, prev2, prev3 = cache["prev1"], cache["prev2"], cache["prev3"]
        st = self.fusion.starts_step(next_id, prev1, prev2, prev3, PAD_ROW)   # (B,) closes the open unit
        if bool(st.any()) and len(middle):
            u = cache["front_last"].unsqueeze(1)
            for b, c in zip(middle, cache["mid"]):
                old = snapshot(c)
                u = b.step(u, c)
                restore_rows(c, old, st)
            cache["g"] = torch.where(st.unsqueeze(-1), u[:, 0], cache["g"])
        x = self._embed_step(next_id, cache, t)
        for b, c in zip(front, cache["front"]):
            x = b.step(x, c)
        cache["front_last"] = x[:, 0]
        x = x + cache["g"].unsqueeze(1)
        for b, c in zip(back, cache["back"]):
            x = b.step(x, c)
        cache["t"] = t + 1
        return self.head(self.nf(x))

    @staticmethod
    def _sample(logits, temperature, top_p):
        logits = logits[:, -1].float()
        if temperature <= 0.02:
            return logits.argmax(-1, keepdim=True)
        probs = F.softmax(logits / temperature, dim=-1)
        sp, si = probs.sort(dim=-1, descending=True)
        keep = (sp.cumsum(-1) - sp) < top_p
        keep[..., :1] = True   # top-1 always survives: top_p<=0 must never
        sp = sp * keep         # yield an all-zero row (CUDA multinomial on
        return si.gather(-1, torch.multinomial(  # zeros poisons the context)
            sp / sp.sum(-1, keepdim=True), 1))

    @torch.no_grad()
    def generate(self, idx, max_new: int = 128, temperature: float = 1.0,
                 top_p: float = 0.95, use_cache: bool = True):
        """Cached decode while the sequence fits the position table; any
        remainder (long prompts, fills past the context) continues through
        the sliding-window parallel path — the hub's constant-size state
        cannot evict, so sliding continuation must recompute."""
        self.eval()
        ctx = self.cfg.context
        if use_cache and idx.shape[1] < ctx and max_new > 0:
            n_cached = min(max_new, ctx - idx.shape[1])
            logits, cache = self.prefill(idx)
            for i in range(n_cached):
                nxt = self._sample(logits, temperature, top_p)
                idx = torch.cat([idx, nxt], dim=1)
                if i + 1 < n_cached:
                    logits = self.decode_step(nxt, cache)
            max_new -= n_cached
        for _ in range(max_new):
            logits, _ = self(idx[:, -ctx:])
            nxt = self._sample(logits, temperature, top_p)
            idx = torch.cat([idx, nxt], dim=1)
        return idx

    def compile_hubs(self, **compile_kw):
        """Opt-in: torch.compile each hub's attention (measured 4.0x over
        eager at ctx 2048 with the fused forward, parity 1.4e-06 vs the
        naive oracle; MHA-parity wall-clock by ctx 8192). Call AFTER
        loading weights, and do NOT save a checkpoint while compiled (the
        OptimizedModule wrapper prefixes state_dict keys). TRAINING use is
        gated on a grad-parity + throughput check on the training
        hardware — the speed verdict was no-grad forward only."""
        import torch as _torch
        from .governor import raw_block
        for wrap in self.blocks:
            blk = raw_block(wrap)
            if blk.is_hub:
                blk.attn = _torch.compile(blk.attn, **compile_kw)
        return self

    def param_count(self) -> int:
        seen, total = set(), 0
        for p in self.parameters():
            if id(p) not in seen:
                seen.add(id(p))
                total += p.numel()
        return total