Merge branch 'master' into delta_attention_scan

test/unit/test_attention.py

@@ -1,6 +1,6 @@
 import unittest
 import numpy as np
-from tinygrad import Tensor, dtypes
+from tinygrad import Tensor, dtypes, nn
 from tinygrad.llm.model import (
   GatedDeltaNetBlock, SSMConfig, TransformerBlock, TransformerConfig,
   apply_rope as apply_rope_new, precompute_freqs_cis, pairwise_topk,
@@ -45,10 +45,10 @@ class TestGatedDeltaNetBlock(unittest.TestCase):
     return Tensor.linspace(start, stop, int(np.prod(shape)), dtype=dtypes.float32).reshape(*shape)
 
   def _make_config(self, **kwargs):
-    return TransformerConfig(**({"num_blocks":1, "dim":4, "hidden_dim":8, "n_heads":1, "n_kv_heads":1,
-                                 "norm_eps":1e-5, "vocab_size":32, "head_dim":4, "rope_theta":10000.0,
-                                 "rope_dim":4, "v_head_dim":4, "max_context":4, "ssm_layers":(True,),
-                                 "ssm":SSMConfig(conv_kernel=2, state_size=2, group_count=1, time_step_rank=1, inner_size=2)} | kwargs))
+    return TransformerConfig(**({"num_blocks":1, "dim":32, "hidden_dim":64, "n_heads":1, "n_kv_heads":1,
+                                 "norm_eps":1e-5, "vocab_size":32, "head_dim":32, "rope_theta":10000.0,
+                                 "rope_dim":32, "v_head_dim":32, "max_context":4, "ssm_layers":(True,),
+                                 "ssm":SSMConfig(conv_kernel=2, state_size=32, group_count=1, time_step_rank=1, inner_size=32)} | kwargs))
 
   def _make_block(self, config:TransformerConfig) -> GatedDeltaNetBlock:
     block = GatedDeltaNetBlock(config, config.ssm)
@@ -79,6 +79,10 @@ class TestGatedDeltaNetBlock(unittest.TestCase):
       recurrent_state = cache[:, conv_flat:].reshape(cache.shape[0], block.num_v_heads, block.head_v_dim, block.head_v_dim)
       return conv_state, recurrent_state
 
+  def _reset_state(self, block:GatedDeltaNetBlock):
+    Tensor.realize(block.conv_state.assign(block.conv_state.const_like(0)),
+                   block.recurrent_state.assign(block.recurrent_state.const_like(0)))
+
   def _linear_np(self, x:np.ndarray, weight:np.ndarray) -> np.ndarray:
     return x.astype(np.float32) @ weight.T.astype(np.float32)
 
@@ -86,7 +90,7 @@ class TestGatedDeltaNetBlock(unittest.TestCase):
     x_float = x.astype(np.float32)
     return (x_float / np.sqrt((x_float * x_float).mean(axis=-1, keepdims=True) + eps)) * weight.astype(np.float32)
 
-  def _normalize_np(self, x:np.ndarray, eps:float=1e-12) -> np.ndarray:
+  def _normalize_np(self, x:np.ndarray, eps:float=1e-6) -> np.ndarray:
     return x / np.maximum(np.sqrt((x * x).sum(axis=-1, keepdims=True)), eps)
 
   def _softplus_np(self, x:np.ndarray) -> np.ndarray:
@@ -148,6 +152,12 @@ class TestGatedDeltaNetBlock(unittest.TestCase):
     x = Tensor.linspace(-1.0, 1.0, 3 * config.dim, dtype=dtypes.float32).reshape(1, 3, config.dim)
 
     expected_outs, expected_conv, expected_recurrent = self._naive_attention(block, x)
+    out = self._run_attention(block, x, 0)
+    conv_state, recurrent_state = self._cache_views(block)
+    np.testing.assert_allclose(out, np.concatenate(expected_outs, axis=1), rtol=1e-3, atol=1e-3)
+    np.testing.assert_allclose(conv_state, expected_conv[-1], rtol=1e-3, atol=1e-3)
+    np.testing.assert_allclose(recurrent_state, expected_recurrent[-1], rtol=1e-3, atol=1e-3)
+    self._reset_state(block)
 
     for step in range(x.shape[1]):
       out = self._run_attention(block, x[:, step:step+1], step)
@@ -163,7 +173,7 @@ class TestGatedDeltaNetBlock(unittest.TestCase):
     prompt = Tensor.linspace(0.75, -0.75, 2 * config.dim, dtype=dtypes.float32).reshape(1, 2, config.dim)
 
     for i in range(warmup.shape[1]): self._run_attention(block, warmup[:, i:i+1], i)
-    Tensor.realize(*block._state_reset_ops())
+    self._reset_state(block)
     expected_outs, expected_conv, expected_recurrent = self._naive_attention(block, prompt)
 
     for step in range(prompt.shape[1]):
@@ -177,18 +187,41 @@ class TestGatedDeltaNetBlock(unittest.TestCase):
                                  err_msg=f"GatedDeltaNet reset recurrent cache mismatch at step {step}")
 
   def test_kda_channel_decay(self):
-    config = self._make_config(n_heads=2, ssm=SSMConfig(conv_kernel=2, state_size=2, group_count=2, time_step_rank=2, inner_size=4, kda=True))
-    block, x = GatedDeltaNetBlock(config, config.ssm), Tensor([[[1., 2., 0., 0.]]])
-    # f_b(f_a(x)) = [1, 2, 3, 4]
+    config = self._make_config(dim=4, hidden_dim=8, n_heads=2, head_dim=4, rope_dim=4, v_head_dim=4,
+      ssm=SSMConfig(conv_kernel=2, state_size=2, group_count=2, time_step_rank=2, inner_size=4, kda=True))
+    block, x = GatedDeltaNetBlock(config, config.ssm), Tensor([[[1., 2., 0., 0.], [2., 1., 0., 0.]]])
     block.ssm_f_a.weight = Tensor([[1., 0., 0., 0.], [0., 1., 0., 0.]])
     block.ssm_f_b.weight = Tensor([[1., 0.], [0., 1.], [1., 1.], [2., 1.]])
     block._init_state(x)
     initial_state = Tensor.arange(8, dtype=dtypes.float32).reshape(1, 2, 2, 2)
     block.recurrent_state.assign(initial_state).realize()
     block.ssm_a = Tensor([[-1.], [-1.]])
-    block._attention(x, 0).realize()
-    alpha = np.exp(-self._softplus_np(np.arange(1, 5)).reshape(1, 2, 1, 2))
-    np.testing.assert_allclose(block.recurrent_state.numpy(), initial_state.numpy() * alpha, rtol=1e-5, atol=1e-5)
+    block._attention(x, x.shape[1]).realize()
+    alpha = np.exp(-self._softplus_np(np.array([[1, 2, 3, 4], [2, 1, 3, 5]])).reshape(2, 2, 2)).prod(0)
+    np.testing.assert_allclose(block.recurrent_state.numpy(), initial_state.numpy() * alpha[..., None], rtol=1e-5, atol=1e-5)
+
+  def test_kda_prefill_matches_decode(self):
+    config = self._make_config(ssm=SSMConfig(conv_kernel=2, state_size=32, group_count=1, time_step_rank=1, inner_size=32, kda=True))
+    block = GatedDeltaNetBlock(config, config.ssm)
+    for p in nn.state.get_parameters(block):
+      p.replace(self._tensor_linspace(-0.05, 0.05, p.shape) if len(p.shape) > 1 else self._tensor_linspace(0.05, 0.1, p.shape))
+    x = self._tensor_linspace(-0.5, 0.5, (1, 3, config.dim))
+    prefill = self._run_attention(block, x, 0)
+    prefill_conv, prefill_recurrent = self._cache_views(block)
+    self._reset_state(block)
+    decode = np.concatenate([self._run_attention(block, x[:, i:i+1], i) for i in range(3)], axis=1)
+    decode_conv, decode_recurrent = self._cache_views(block)
+    np.testing.assert_allclose(prefill, decode, rtol=1e-3, atol=1e-3)
+    np.testing.assert_allclose(prefill_conv, decode_conv, rtol=1e-3, atol=1e-3)
+    np.testing.assert_allclose(prefill_recurrent, decode_recurrent, rtol=1e-3, atol=1e-3)
+
+  def test_start_zero_resets_realized_state(self):
+    config, x = self._make_config(max_context=3), self._tensor_linspace(-1, 1, (1, 3, 32))
+    block = self._make_block(config)
+    self._run_attention(block, x, 0)
+    restarted = self._run_attention(block, x[:, :2], 0)
+    fresh = self._run_attention(self._make_block(config), x[:, :2], 0)
+    np.testing.assert_allclose(restarted, fresh, rtol=1e-3, atol=1e-3)
 
 class TestPairwiseTopk(unittest.TestCase):
   def test_basic_topk(self):

test/unit/test_llm_kernels.py

@@ -0,0 +1,67 @@
+import unittest
+import numpy as np
+from tinygrad import Tensor, UOp
+from tinygrad.llm.model import gated_delta_prefill
+
+def numpy_ref(q, k, v, beta, alpha, initial):
+  state, out = initial.copy(), np.empty_like(v)
+  for t in range(q.shape[2]):
+    av = alpha[:, :, t, :, None] if alpha.ndim == 4 else alpha[:, :, t, None, None]
+    sa = alpha[:, :, t] if alpha.ndim == 4 else alpha[:, :, t, None]
+    previous = state.copy()
+    delta = (v[:, :, t] - (previous*k[:, :, t, None]).sum(-1)*sa) * beta[:, :, t, None]
+    state = previous*av + delta[..., None]*k[:, :, t, None, :]
+    out[:, :, t] = (previous*q[:, :, t, None]).sum(-1)*sa + delta*(q[:, :, t]*k[:, :, t]).sum(-1, keepdims=True)
+  return out, state
+
+class TestGatedDeltaPrefill(unittest.TestCase):
+  def _make(self, B, H, T, V, K, alpha_4d=False, seed=42):
+    rng = np.random.default_rng(seed)
+    # normalize like the model does: with raw unit-norm keys the delta rule is stable, random keys make it diverge
+    q, k = (rng.normal(size=(B, H, T, K)).astype(np.float32) for _ in range(2))
+    k = k / np.maximum(np.sqrt((k*k).sum(-1, keepdims=True)), 1e-6)
+    v, beta = rng.normal(size=(B, H, T, V)).astype(np.float32), rng.uniform(size=(B, H, T)).astype(np.float32)
+    alpha = rng.uniform(0.9, 1, size=(B, H, T, V) if alpha_4d else (B, H, T)).astype(np.float32)
+    initial = rng.normal(size=(B, H, V, K)).astype(np.float32)
+    return q, k, v, beta, alpha, initial
+
+  def test_rectangular_state_and_row_decay(self):
+    q, k, v, beta, alpha, initial = self._make(1, 1, 3, 4, 32, alpha_4d=True)
+    expected_out, expected_state = numpy_ref(q, k, v, beta, alpha, initial)
+    state = Tensor(initial).contiguous().realize()
+    out = gated_delta_prefill(Tensor(q), Tensor(k), Tensor(v), Tensor(beta), Tensor(alpha), state).realize()
+    np.testing.assert_allclose(out.numpy(), expected_out, rtol=1e-4, atol=1e-4)
+    np.testing.assert_allclose(state.numpy(), expected_state, rtol=1e-4, atol=1e-4)
+
+  def test_prefill_matches_single_steps(self):
+    # one T=32 kernel call must match 32 sequential T=1 calls with in-place state
+    q, k, v, beta, alpha, initial = self._make(1, 4, 32, 128, 128)
+    state_a = Tensor(initial).contiguous().realize()
+    out_a = gated_delta_prefill(Tensor(q), Tensor(k), Tensor(v), Tensor(beta), Tensor(alpha), state_a).realize()
+    outs, state_b = [], Tensor(initial).contiguous().realize()
+    for t in range(32):
+      outs.append(gated_delta_prefill(Tensor(q[:, :, t:t+1]), Tensor(k[:, :, t:t+1]), Tensor(v[:, :, t:t+1]),
+                                      Tensor(beta[:, :, t:t+1]), Tensor(alpha[:, :, t:t+1]), state_b).realize())
+    np.testing.assert_allclose(out_a.numpy(), Tensor.stack(*outs, dim=2).squeeze(3).numpy(), rtol=1e-4, atol=1e-4)
+    np.testing.assert_allclose(state_a.numpy(), state_b.numpy(), rtol=1e-4, atol=1e-4)
+
+  def test_start_pos_zero_resets_state(self):
+    q, k, v, beta, alpha, initial = self._make(1, 2, 5, 8, 16)
+    # garbage state must be ignored when start_pos binds to 0
+    garbage = np.full_like(initial, 1.0e9)
+    def run(sp, init):
+      state = Tensor(init).contiguous().realize()
+      initial = Tensor(UOp.variable("start_pos", 0, 63).bind(sp)).eq(0)
+      return gated_delta_prefill(Tensor(q), Tensor(k), Tensor(v), Tensor(beta), Tensor(alpha), state, initial).realize(), state
+    out_reset, state_reset = run(0, garbage)
+    expected_out, expected_state = numpy_ref(q, k, v, beta, alpha, np.zeros_like(initial))
+    np.testing.assert_allclose(out_reset.numpy(), expected_out, rtol=1e-4, atol=1e-4)
+    np.testing.assert_allclose(state_reset.numpy(), expected_state, rtol=1e-4, atol=1e-4)
+    # nonzero start_pos must use the provided state
+    out_cont, state_cont = run(3, initial)
+    expected_out, expected_state = numpy_ref(q, k, v, beta, alpha, initial)
+    np.testing.assert_allclose(out_cont.numpy(), expected_out, rtol=1e-4, atol=1e-4)
+    np.testing.assert_allclose(state_cont.numpy(), expected_state, rtol=1e-4, atol=1e-4)
+
+if __name__ == "__main__":
+  unittest.main()

test/unit/test_llm_server.py

@@ -42,7 +42,8 @@ class TestTransformerGenerate(unittest.TestCase):
       return Tensor([[42]])
     with patch.object(Transformer, '__call__', mock_call):
       next(model.generate([1, 2, 3, 4, 5, 42, 10]))
-    self.assertEqual(calls, [((1, 1), V_START_POS.bind(5)), ((1, 1), V_START_POS.bind(6))])
+    # recurrent blocks prefill chunks like attention blocks: the 2 new tokens go through one chunked call
+    self.assertEqual(calls, [((1, V_TOKS.bind(2)), V_START_POS.bind(5))])
 
   def test_recurrent_divergent_prompt_restarts(self):
     model, calls = Transformer(TEST_CONFIG), []

tinygrad/codegen/gpudims.py

@@ -43,6 +43,14 @@ def add_gpudims(ctx:Renderer, s:UOp):
   s_topo = list(s.toposort())
   if any(x.op is Ops.SPECIAL for x in s_topo): return None
 
+  # renderers without local workgroups execute LOCAL/WARP ranges as sequential loops in the thread.
+  # this is only valid without cross-thread communication (BARRIER), local memory stays unsupported
+  if not ctx.has_local and any(r.op is Ops.RANGE and r.arg[-1] in (AxisType.LOCAL, AxisType.WARP) for r in s_topo):
+    if any(x.op is Ops.BARRIER or (x.op is Ops.BUFFER and x.addrspace is AddrSpace.LOCAL) for x in s_topo): return None
+    s = s.substitute({r: r.replace(arg=r.arg[0:-1]+(AxisType.LOOP,)) for r in s_topo if r.op is Ops.RANGE
+                      and r.arg[-1] in (AxisType.LOCAL, AxisType.WARP)})
+    s_topo = list(s.toposort())
+
   # get ranges
   all_ranges = {x.arg[0:-1]:x for x in s_topo if x.op is Ops.RANGE}
 

tinygrad/llm/model.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 import functools, itertools, pathlib
 from dataclasses import dataclass, replace
+from typing import cast
 from tinygrad import Tensor, nn, UOp, TinyJit, getenv, function, dtypes
+from tinygrad.dtype import AddrSpace
 from tinygrad.nn import Linear
 from tinygrad.llm.gguf import gguf_load
-from tinygrad.uop.ops import resolve
+from tinygrad.uop.ops import resolve, AxisType, KernelInfo, Ops, sint
 
 @functools.cache
 def precompute_freqs_cis(dim: int, end: int, theta: float = 10000.0, device:str|None=None) -> Tensor:
@@ -124,8 +126,6 @@ class FFNBlock:
 
   # given the token-prefix match, return how much cached state this block can still reuse
   def _reusable_prefix_len(self, prefix_len:int, cached_len:int) -> int: return prefix_len
-  # return writes that reset this block's state after a cache mismatch
-  def _state_reset_ops(self) -> list[Tensor]: return []
   def _init_state(self, x:Tensor): raise NotImplementedError
   def _attention(self, x:Tensor, start_pos:int|UOp) -> Tensor: raise NotImplementedError
 
@@ -238,6 +238,73 @@ class MLATransformerBlock(FFNBlock):
       self.cache_k = Tensor.empty(x.shape[0], 1, self.config.max_context, self.config.kv_lora_rank + self.config.rope_dim, device=x.device)
       self.freqs_cis = precompute_freqs_cis(self.config.rope_dim, self.config.max_context, self.config.rope_theta, device=x.device)
 
+def _tree_sum(xs:list[UOp]) -> UOp:
+  # balanced tree keeps the reduction depth at log2(n) (compilers can't reassociate floats, so this shape reaches the ALU)
+  if not xs: return UOp.const(0, dtypes.float32)
+  while len(xs) > 1: xs = [a+b for a, b in zip(xs[::2], xs[1::2])] + xs[2*(len(xs)//2):]
+  return xs[0]
+
+@functools.cache
+def _gated_delta_prefill_kernel(core:UOp, q:UOp, k:UOp, v:UOp, beta:UOp, alpha:UOp, state:UOp, kq:UOp,
+                                initial:UOp|None=None) -> UOp:
+  batch, heads, tokens, value_dim = cast(tuple[int, int, int, int], core.shape)
+  key_dim, alpha_dim = cast(int, q.shape[-1]), cast(int, alpha.shape[-1]) if len(alpha.shape) == 4 else 1
+  core, v = (x.reshape(batch*heads, tokens, value_dim) for x in (core, v))
+  q, k = (x.reshape(batch*heads, tokens, key_dim) for x in (q, k))
+  beta, kq = (x.reshape(batch*heads, tokens) for x in (beta, kq))
+  alpha, state = alpha.reshape(batch*heads, tokens, alpha_dim), state.reshape(batch*heads, value_dim, key_dim)
+  # parallel over (batch*head, state row): one thread owns one state row in registers across the sequential token loop.
+  # one block per (batch*head), rows are the LOCAL threads so k/q token loads broadcast within the block
+  # (on renderers without local workgroups, gpudims reruns the rows as a sequential in-thread loop)
+  bh = UOp.range(batch*heads, 0, AxisType.GLOBAL)
+  row = UOp.range(value_dim, 1, AxisType.LOCAL)
+  cols = tuple(range(key_dim))
+  current = UOp.placeholder((key_dim,), dtypes.float32, slot=0, addrspace=AddrSpace.REG)
+  # the state starts from zero when the (scalar bool) initial flag is set; otherwise it resumes from `state`
+  reset = None if initial is None else initial.reshape(1)[0].load()
+  current = current.after(UOp.group(*(current[col].store(state[bh, row, col].float() if reset is None else
+    reset.where(0, state[bh, row, col].float())) for col in cols)))
+  token = UOp.range(tokens, 3, AxisType.REDUCE)
+  previous = tuple(current.after(token)[col].load() for col in cols)
+  keys, queries = (tuple(x[bh, token, col].load() for col in cols) for x in (k, q))
+  av, bv = alpha[bh, token, row if alpha_dim > 1 else 0].load(), beta[bh, token].load()
+  state_k = _tree_sum([x*y for x, y in zip(previous, keys)])
+  state_q = _tree_sum([x*y for x, y in zip(previous, queries)])
+  delta = (v[bh, token, row].load() - state_k*av) * bv
+  step = UOp.group(core[bh, token, row].store(state_q*av + delta*kq[bh, token]),
+                   *(current[col].store(x*av + delta*y) for col, x, y in zip(cols, previous, keys))).end(token)
+  stores = (state[bh, row, col].store(current.after(step)[col].load().cast(state.dtype)) for col in cols)
+  return UOp.group(*stores).end(row, bh).sink(arg=KernelInfo(name="gated_delta_prefill", opts_to_apply=()))
+
+def gated_delta_prefill(q:Tensor, k:Tensor, v:Tensor, beta:Tensor, alpha:Tensor, state:Tensor, initial:Tensor|None=None) -> Tensor:
+  """Gated delta rule over `tokens` steps in one kernel, updating the recurrent state in place.
+
+  q, k: (batch, heads, tokens, key_dim). v: (batch, heads, tokens, value_dim). beta: (batch, heads, tokens).
+  alpha: (batch, heads, tokens) for head-wise decay, or (batch, heads, tokens, value_dim) for per-channel decay.
+  state: (batch, heads, value_dim, key_dim), updated in place. initial: scalar bool Tensor; state starts from zero when set.
+  `tokens` may be symbolic: the sequence is padded to its maximum size and masked (beta=0, alpha=1), so one
+  graph serves every chunk size. Decoding (tokens == 1) takes a static path without padding.
+  """
+  tokens:sint = q.shape[2]
+  batch, heads, _, key_dim = q.shape
+  value_dim = cast(int, v.shape[-1])
+  assert isinstance(key_dim, int), "key/value dims must be static"
+  assert q.shape == k.shape and v.shape[:3] == q.shape[:3] and beta.shape == (batch, heads, tokens)
+  assert alpha.shape in ((batch, heads, tokens), (batch, heads, tokens, value_dim))
+  assert state.shape == (batch, heads, value_dim, key_dim)
+  static = isinstance(tokens, int)
+  out_shape = v.shape
+  if not static:
+    # pad the variable-length sequence to its max size with no-op steps: beta=0 and alpha=1 leave the state untouched
+    q, k, v, beta = (x.pad_to(x.max_shape) for x in (q, k, v, beta))
+    alpha = alpha.pad_to(alpha.max_shape, value=1)
+    tokens = q.shape[2]
+  core, kq = Tensor.empty(batch, heads, tokens, value_dim), (q*k).sum(-1).contiguous()
+  state = state if state.uop.op is Ops.AFTER else state.contiguous()  # keep the AFTER chain of in-place state updates
+  srcs = (core, q.contiguous(), k.contiguous(), v.contiguous(), beta.contiguous(), alpha.contiguous(), state, kq)
+  out = Tensor.custom_kernel(*srcs, *(() if initial is None else (initial,)), fxn=_gated_delta_prefill_kernel)[0]
+  return (out if static else out[:, :, :out_shape[2]]).reshape(out_shape)
+
 class GatedDeltaNetBlock(FFNBlock):
   def __init__(self, config:TransformerConfig, ssm:SSMConfig):
     super().__init__(config)
@@ -260,45 +327,55 @@ class GatedDeltaNetBlock(FFNBlock):
 
   def _attention(self, x:Tensor, start_pos:int|UOp) -> Tensor:
     B, T, _ = x.shape
-    assert T == 1, "GatedDeltaNetBlock currently only supports T=1"
+    # bind ints to a variable so the reset flag stays a runtime value (it toggles when generation restarts at position 0)
+    start_pos = start_pos if isinstance(start_pos, UOp) else UOp.variable("start_pos", 0, self.config.max_context-1).bind(start_pos)
+    initial = Tensor(start_pos).eq(0)
     is_kda = hasattr(self, "ssm_g_a")
+    symbolic = isinstance(T, UOp)
+    T_pad = x.max_shape[1]  # symbolic chunks are padded to their max size: one graph serves every size
 
     # input processing
     x = x.half()
     out_gate = self.ssm_g_b(self.ssm_g_a(x)) if is_kda else self.attn_gate(x)
-    out_gate = out_gate.reshape(B, 1, self.num_v_heads, self.head_v_dim)
-    beta = self.ssm_beta(x).sigmoid().reshape(B, self.num_v_heads, 1, 1)
+    out_gate = out_gate.reshape(B, T, self.num_v_heads, self.head_v_dim)
+    beta = self.ssm_beta(x).sigmoid().reshape(B, T, self.num_v_heads)
     alpha = self.ssm_f_b(self.ssm_f_a(x)) if is_kda else self.ssm_alpha(x)
-    alpha = ((alpha.float() + self.ssm_dt["bias"]).softplus().reshape(B, self.num_v_heads, -1) *
-             self.ssm_a.reshape(1, self.num_v_heads, -1)).exp().unsqueeze(-2)
+    log_alpha = ((alpha.float() + self.ssm_dt["bias"]).softplus().reshape(B, T, self.num_v_heads, -1) * self.ssm_a).squeeze(-1) \
+      if is_kda else ((alpha.float() + self.ssm_dt["bias"]).softplus() * self.ssm_a).reshape(B, T, self.num_v_heads)
 
-    # qkv conv
-    conv_window = self.conv_state.cat(self.attn_qkv(x), dim=1)
-    conv_out = (conv_window * self.ssm_conv1d["weight"].T.unsqueeze(0)).sum(1).silu()
+    # qkv conv, conv_state is reset when starting from position 0
+    conv_state = initial.where(0, self.conv_state)
+    # assemble the conv window in a static-size buffer: [conv_state | qkv rows | zero-pad].
+    # padded steps are exact no-ops: beta=0 (delta rule off), log_alpha=0 (decay 1 after exp)
+    conv_window = Tensor.zeros(B, self.ssm_conv_kernel-1 + T_pad, self.conv_channels)
+    win = conv_window.uop
+    win = win.after(win[:, :self.ssm_conv_kernel-1].store(conv_state.cast(win.dtype).uop))
+    win = win.after(win[:, self.ssm_conv_kernel-1:self.ssm_conv_kernel-1+T].store(self.attn_qkv(x).cast(win.dtype).uop))
+    conv_window = Tensor(win)
+    # the last conv_kernel-1 columns of the window become the next conv state
+    conv_state_store = self.conv_state.uop.store(conv_window[:, T:T+self.ssm_conv_kernel-1].cast(self.conv_state.dtype).uop)
+
+    conv_out = functools.reduce(lambda a,b: a+b,
+      (conv_window[:, i:i+T_pad] * self.ssm_conv1d["weight"][:, i] for i in range(self.ssm_conv_kernel))).silu()
+    if symbolic:
+      out_gate = out_gate.pad_to((B, T_pad, self.num_v_heads, self.head_v_dim))
+      beta, log_alpha = beta.pad_to((B, T_pad, self.num_v_heads)), log_alpha.pad_to((B, T_pad, self.num_v_heads))
     q, k, v = conv_out.split([self.q_dim, self.q_dim, self.conv_channels - 2*self.q_dim], dim=-1)
-    q = q.reshape(B, self.num_k_heads, self.head_k_dim).normalize(dim=-1).repeat(1, self.num_v_heads//self.num_k_heads, 1)
-    k = k.reshape(B, self.num_k_heads, self.head_k_dim).normalize(dim=-1).repeat(1, self.num_v_heads//self.num_k_heads, 1)
-    v = v.reshape(B, self.num_v_heads, self.head_v_dim)
-    q, k, v = q.mul(self.head_k_dim**-0.5).unsqueeze(-1), k.unsqueeze(-1), v.unsqueeze(-1)
+    qk_eps = 1e-12 if is_kda else 1e-6
+    q, k = (z.reshape(B, T_pad, self.num_k_heads, self.head_k_dim).normalize(dim=-1, eps=qk_eps)
+            .repeat(1, 1, self.num_v_heads//self.num_k_heads, 1) for z in (q, k))
+    v = v.reshape(B, T_pad, self.num_v_heads, self.head_v_dim)
+    q, k, v, beta = [z.transpose(1, 2).float() for z in (q, k, v, beta)]
+    alpha = log_alpha.transpose(1, 2).exp()
 
-    # recurrent
-    recurrent_state = self.recurrent_state * alpha
-    recurrent_state = recurrent_state + ((v - recurrent_state@k) * beta)@k.transpose(-1, -2)
+    # recurrent: the conv and recurrent states are updated in place
+    state = Tensor(self.recurrent_state.uop.after(conv_state_store))
+    core = gated_delta_prefill(q * self.head_k_dim**-0.5, k, v, beta, alpha, state, initial).transpose(1, 2)
 
-    # store the updated state
-    conv_state_store = self.conv_state.uop.store(conv_window[:, 1:, :].cast(self.conv_state.dtype).uop)
-    recurrent_state_store = self.recurrent_state.uop.store(recurrent_state.cast(self.recurrent_state.dtype).uop)
-    recurrent_state = Tensor(self.recurrent_state.uop.after(recurrent_state_store, conv_state_store))
-
-    # output
-    core_attn_out = self.ssm_norm((recurrent_state@q).squeeze(-1).reshape(B, 1, self.num_v_heads, self.head_v_dim))
-    out_gate = out_gate.sigmoid() if is_kda else out_gate.silu()
-    return self.ssm_out((core_attn_out * out_gate).reshape(B, 1, -1).cast(x.dtype))
-
-  # recurrent state can't be partially reused after divergence, force a full rebuild
-  def _state_reset_ops(self):
-    return [self.conv_state.assign(self.conv_state.const_like(0)),
-            self.recurrent_state.assign(self.recurrent_state.const_like(0))] if hasattr(self, "conv_state") else []
+    # output; undo the padding before the output projection
+    z = (self.ssm_norm(core) * (out_gate.sigmoid() if is_kda else out_gate.silu())).cast(x.dtype).contiguous()
+    if symbolic: z = z[:, :T]
+    return self.ssm_out(z.reshape(B, T, -1))
 
   def _init_state(self, x):
     if not hasattr(self, "conv_state"):
@@ -429,7 +506,6 @@ class Transformer:
     return min(block._reusable_prefix_len(prefix_len, len(self._cached_tokens)) for block in self.blk)
 
   def generate(self, tokens:list[int], chunk_size:int=32, temperature:float=0.0):
-    if self.has_recurrent_block: chunk_size = 1
     v_start_pos = UOp.variable("start_pos", 0, self.max_context-1)
     v_toks = UOp.variable("toks", 1, chunk_size)
     # TODO: use UOp.variable for temperature once float variables are supported
@@ -438,7 +514,6 @@ class Transformer:
     t = Tensor(tokens + [0] * (self.max_context - len(tokens)), dtype="int32").reshape(1, self.max_context)
     # recompute start_pos from what's currently valid in the caches
     start_pos = self.get_start_pos(tokens)
-    if start_pos < len(self._cached_tokens) and (resets := [r for b in self.blk for r in b._state_reset_ops()]): Tensor.realize(*resets)
     out, prompt_len = None, len(tokens)
     while len(tokens) < self.max_context:
       n_toks = min(chunk_size, len(tokens) - start_pos)