Skip to main content

cortiq_engine/
kv_cache.rs

1//! KV cache — per-layer, head-major storage.
2//!
3//! Layout: one contiguous `Vec<f32>` per KV head (`[pos × head_dim]`),
4//! so per-head attention reads a straight slice — no per-head gather
5//! copies per token. Dead GQA groups (all Q heads masked) store
6//! nothing at all: masked heads cost neither FLOPs nor memory.
7
8/// KV storage mode. `CMF_KV=q8` enables the q8_2f cache: an int8 row per
9/// (position, head) + an f32 scale per row + a per-channel scale field,
10/// frozen after WARMUP positions with retroactive requantization
11/// (D4: "KV-quant 2f"). Memory ×~3.7 smaller than f32.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum KvMode {
14    F32,
15    /// Quantized components: (K, V) — sensitivity diagnostics.
16    Q8 {
17        k: bool,
18        v: bool,
19    },
20}
21
22impl KvMode {
23    pub fn from_env() -> Self {
24        match std::env::var("CMF_KV").as_deref() {
25            Ok("q8") | Ok("q8_2f") => KvMode::Q8 { k: true, v: true },
26            Ok("q8k") => KvMode::Q8 { k: true, v: false },
27            Ok("q8v") => KvMode::Q8 { k: false, v: true },
28            _ => KvMode::F32,
29        }
30    }
31
32    fn quant_k(self) -> bool {
33        matches!(self, KvMode::Q8 { k: true, .. })
34    }
35
36    fn quant_v(self) -> bool {
37        matches!(self, KvMode::Q8 { v: true, .. })
38    }
39}
40
41/// Positions before freezing the per-channel field (2f): before — col ≡ 1,
42/// after — col = RMS over channels of the stored rows, old rows are requantized.
43const KV_COL_WARMUP: usize = 64;
44
45/// K-rows are quantized in groups of 32 channels (scale per group):
46/// attention logits are sensitive to the dot-product error, per-group scales
47/// localize it along RoPE bands (35B: +4.6% PPL with a per-row scale
48/// → target <1% with a per-group one). V — per-row scale (measured +0.56%).
49const KV_K_GROUP: usize = 32;
50
51/// Source of `LayerKvCache::generation`. Process-wide so two caches (or a
52/// cache and a restored clone of itself from another moment) never share a
53/// value by accident: a device mirror keyed by (generation, rows) can then tell
54/// "the rows I hold" from "the same number of different rows".
55static KV_GEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
56
57fn next_gen() -> u64 {
58    KV_GEN.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
59}
60
61/// Per-layer O(1) Nyström attention state (runtime `attn_type`
62/// override — spec §7 presence-driven pattern, no format change).
63///
64/// Collecting: the prompt pass still runs EXACT cache attention (the
65/// prefill outputs feed the residual stream, so they cannot be
66/// deferred) while the per-position rotated queries are buffered;
67/// `o1_seal()` then freezes landmarks + M from the full prompt, replays
68/// it into per-KV-group streaming states, and DROPS the full KV.
69/// Sealed: decode replaces cache attention with
70/// `NystromState::step_group()`.
71#[derive(Debug, Clone)]
72pub enum O1State {
73    Collecting {
74        m: usize,
75        w: usize,
76        sink: usize,
77        rect: crate::nystrom::O1Rect,
78        /// Optional completed-row barrier. `None` means the caller will
79        /// request a full-prompt seal; `Some(B)` keeps a short prompt exact
80        /// until the first skeleton-safe boundary B.
81        seal_at: Option<usize>,
82        /// Rotated post-norm queries, `[pos × num_heads × head_dim]`.
83        q_buf: Vec<f32>,
84    },
85    /// One state per KV GROUP, each holding its group's Q heads. The
86    /// exact window / sinks / K̃ are stored once per group (every Q head
87    /// of the group reads the same k/v rows); only the far field, Q̃ and
88    /// M — the query-dependent pieces — stay per Q head. See
89    /// `NystromState` for which piece is which and why.
90    Sealed {
91        groups: Vec<crate::nystrom::NystromState>,
92    },
93}
94
95/// KV cache for a single layer, head-major.
96#[derive(Debug, Clone)]
97pub struct LayerKvCache {
98    pub mode: KvMode,
99    /// Per-KV-head keys: `k[h]` is `[seq_len × head_dim]` (empty if head is dead).
100    k: Vec<Vec<f32>>,
101    /// Per-KV-head values, same layout.
102    v: Vec<Vec<f32>>,
103    /// q8 storage (mode == Q8_2F): int8 rows + f32 scale per row.
104    kq: Vec<Vec<i8>>,
105    ks: Vec<Vec<f32>>,
106    vq: Vec<Vec<i8>>,
107    vs: Vec<Vec<f32>>,
108    /// Per-channel scale fields per head [head_dim]; empty until frozen.
109    kcol: Vec<Vec<f32>>,
110    vcol: Vec<Vec<f32>>,
111    /// Accumulated attention mass per stored position: importance of a
112    /// position is how much probability mass reads it.
113    imp: Vec<f32>,
114    /// Rows stored (grows once per token, dead heads included). Stored
115    /// row 0 is absolute position `base`, so this equals the context depth
116    /// only while `base == 0`; [`Self::pos_len`] is the depth. Every attend
117    /// indexes rows, never positions (eviction made this a row count long
118    /// before `trim_window` did).
119    pub seq_len: usize,
120    /// Absolute position of stored row 0. Only `trim_window` advances it;
121    /// 0 on every layer that was never trimmed.
122    base: usize,
123    /// The attend window of a layer that keeps only its tail
124    /// (`trim_window`), set by the first call. A property of the layer,
125    /// not of the conversation: `clear()` keeps it. Such a layer bounds
126    /// itself, so the cache-wide eviction leaves it alone.
127    tail: Option<usize>,
128    /// Storage generation: a fresh value on every mutation other than
129    /// `append` and `truncate_last` (clear, trim, eviction, wire import,
130    /// resets). A device mirror that copies these rows keys its validity on
131    /// (generation, rows): a trimmed layer's row count is periodic (576..639 for a
132    /// 512 window), so the count alone can match a different set of rows.
133    /// Appends and rollbacks show up in the count, and the speculative
134    /// paths re-point their mirrors themselves, so they keep the value.
135    generation: u64,
136    pub num_kv_heads: usize,
137    pub head_dim: usize,
138    /// Linear-core recurrent state S (vmf_phase), f64; empty on full layers.
139    pub linear_state: Vec<f32>,
140    /// Tentative lane-2 state during speculative verify.
141    pub linear_scratch: Vec<f32>,
142    /// The legacy cache wire has no operator tag. Delta operators therefore
143    /// bind this layer to a fail-closed boundary until a versioned wire
144    /// schema can carry the linear-core identity.
145    linear_wire_allowed: bool,
146    /// O(1) Nyström override (None = plain cache attention).
147    pub o1: Option<O1State>,
148    /// A deferred seal failure is terminal for the current request. The
149    /// attention functions return only a hidden vector, so the pipeline
150    /// consumes this side channel at its next forward boundary.
151    o1_error: Option<String>,
152    /// Rows the next seal keeps exact, ascending.
153    pub(crate) o1_keep: Vec<usize>,
154    /// The pipeline chooses the exact rows over every far-field layer once all of
155    /// them hold the boundary's rows (`o1_choose_exact_rows`), and seals then: the
156    /// layer does not convert itself on reaching the boundary (only before a row
157    /// past it would be appended).
158    pub(crate) o1_await_keep: bool,
159    /// One row in this many is in the query trace.
160    o1_q_stride: usize,
161    /// The geometry of the last `o1_begin`.
162    o1_params: Option<(usize, usize, usize, crate::nystrom::O1Rect)>,
163    /// Set when a collecting state actually becomes sealed. Pipeline owns
164    /// the epoch bump and consumes this bit after a complete forward.
165    o1_transitioned: bool,
166    /// Learned per-Q-head attention-sink logits of this layer (gpt-oss /
167    /// MiMo-V2 `self_attn.sinks`, one f32 per Q head). The sink is an
168    /// extra softmax column with no value: it joins the max and the
169    /// denominator of every head's softmax and so lets a head attend to
170    /// "nothing". These are WEIGHTS, not sequence state — `clear()` and
171    /// the wire import keep them. None = an ordinary softmax.
172    pub sinks: Option<Vec<f32>>,
173    /// Natively bounded anchor (`swa_sink_v1`): a fixed-size ring
174    /// installed from the header at load, never per prompt. A layer that
175    /// carries it stores NOTHING per position (`k`/`v` stay empty).
176    pub bounded: Option<crate::bounded::BoundedState>,
177    /// Which state record this layer exchanges on the wire (v2).
178    pub wire_kind: WireKind,
179    /// This layer's index in the stack (the wire header names it).
180    pub wire_layer: u32,
181    /// hash64 of the model's operator identity
182    /// (`ModelArch::linear_core_identity` JSON); 0 = no operator record.
183    pub wire_identity: u64,
184    /// `kv_abs_max`'s running answer: max |x| over the f32 K rows and over
185    /// the V rows `[0, kv_amax_rows)` of every head (+inf once one was inf
186    /// or NaN).
187    kv_amax: [f32; 2],
188    kv_amax_rows: usize,
189}
190
191/// State-record kind of the versioned cache wire (`export_wire` v2).
192#[derive(Debug, Clone, Copy, PartialEq, Eq)]
193#[repr(u8)]
194pub enum WireKind {
195    /// Per-position K/V (+ importance) — the legacy body.
196    Full = 0,
197    /// Recurrent state vector (S + conv ring), f32.
198    Linear = 1,
199    /// Bounded anchor: insert counter + ring K/V.
200    Bounded = 2,
201    /// A full layer that stores only its tail (`trim_window`): `base` as
202    /// u64, then the `Full` body of the stored rows. Sent only when
203    /// `base > 0`, so an untrimmed record is byte-identical to `Full`; a
204    /// peer that predates it refuses it as an unknown kind instead of
205    /// reading the tail as a whole history.
206    FullTail = 3,
207}
208
209impl WireKind {
210    fn from_u8(v: u8) -> Option<Self> {
211        match v {
212            0 => Some(WireKind::Full),
213            1 => Some(WireKind::Linear),
214            2 => Some(WireKind::Bounded),
215            3 => Some(WireKind::FullTail),
216            _ => None,
217        }
218    }
219}
220
221/// Magic of the versioned state wire.
222pub const WIRE_MAGIC: &[u8; 4] = b"CMFS";
223/// Current wire version.
224pub const WIRE_VERSION: u32 = 2;
225
226/// A layer's exact rows, for the choice of those a seal keeps.
227pub(crate) struct GroupRows<'a> {
228    k: &'a [Vec<f32>],
229    v: &'a [Vec<f32>],
230    hd: usize,
231}
232
233impl crate::dory::LayerRows for GroupRows<'_> {
234    fn key(&self, kh: usize, r: usize) -> &[f32] {
235        &self.k[kh][r * self.hd..(r + 1) * self.hd]
236    }
237    fn value(&self, kh: usize, r: usize) -> &[f32] {
238        &self.v[kh][r * self.hd..(r + 1) * self.hd]
239    }
240}
241
242impl LayerKvCache {
243    pub(crate) fn o1_set_q_stride(&mut self, stride: usize) {
244        self.o1_q_stride = stride.max(1);
245    }
246
247    /// The whole query trace at once, `[rows / stride][num_heads][hd]`.
248    pub(crate) fn o1_set_q_trace(&mut self, q: Vec<f32>) {
249        if let Some(O1State::Collecting { q_buf, .. }) = &mut self.o1 {
250            *q_buf = q;
251        }
252    }
253
254    /// Back to collecting over rows kept in host memory, `[rows][nkv][hd]`: the next
255    /// seal builds this layer's far field anew from them.
256    pub(crate) fn o1_recollect(&mut self, k: &[f32], v: &[f32], q: Vec<f32>, stride: usize, rows: usize) -> bool {
257        let Some(O1State::Sealed { .. }) = &self.o1 else { return false };
258        let (hd, nk) = (self.head_dim, self.num_kv_heads);
259        if self.o1_params.is_none() {
260            return false;
261        }
262        let (m, w, sink, rect) = self.o1_params.unwrap();
263        for g in 0..nk {
264            let mut kg = Vec::with_capacity(rows * hd);
265            let mut vg = Vec::with_capacity(rows * hd);
266            for r in 0..rows {
267                kg.extend_from_slice(&k[(r * nk + g) * hd..(r * nk + g + 1) * hd]);
268                vg.extend_from_slice(&v[(r * nk + g) * hd..(r * nk + g + 1) * hd]);
269            }
270            self.k[g] = kg;
271            self.v[g] = vg;
272        }
273        self.seq_len = rows;
274        self.o1_q_stride = stride.max(1);
275        self.o1_keep.clear();
276        self.o1 = Some(O1State::Collecting { m, w, sink, rect, seal_at: Some(rows), q_buf: q });
277        true
278    }
279
280    pub(crate) fn o1_is_collecting(&self) -> bool {
281        matches!(self.o1, Some(O1State::Collecting { .. }))
282    }
283
284    pub(crate) fn group_rows(&self) -> GroupRows<'_> {
285        GroupRows { k: &self.k, v: &self.v, hd: self.head_dim }
286    }
287
288    pub fn new(num_kv_heads: usize, head_dim: usize) -> Self {
289        Self {
290            sinks: None,
291            mode: KvMode::from_env(),
292            k: vec![Vec::new(); num_kv_heads],
293            v: vec![Vec::new(); num_kv_heads],
294            kq: vec![Vec::new(); num_kv_heads],
295            ks: vec![Vec::new(); num_kv_heads],
296            vq: vec![Vec::new(); num_kv_heads],
297            vs: vec![Vec::new(); num_kv_heads],
298            kcol: vec![Vec::new(); num_kv_heads],
299            vcol: vec![Vec::new(); num_kv_heads],
300            imp: Vec::new(),
301            seq_len: 0,
302            base: 0,
303            tail: None,
304            generation: next_gen(),
305            num_kv_heads,
306            head_dim,
307            linear_state: Vec::new(),
308            linear_scratch: Vec::new(),
309            linear_wire_allowed: true,
310            o1: None,
311            o1_error: None,
312            o1_keep: Vec::new(),
313            o1_await_keep: false,
314            o1_q_stride: 1,
315            o1_params: None,
316            o1_transitioned: false,
317            bounded: None,
318            wire_kind: WireKind::Full,
319            wire_layer: 0,
320            wire_identity: 0,
321            kv_amax: [0.0; 2],
322            kv_amax_rows: 0,
323        }
324    }
325
326    /// (max |k|, max |v|) over every K and over every V entry the cache
327    /// stores in f32, +inf when one is inf or NaN: what the wgpu prefill
328    /// compares with f16's range before its matrix-unit attention casts
329    /// these rows to f16. Lazy — it scans only the rows appended since the
330    /// last call; every other change to the rows (truncate, evict, clear,
331    /// import, an o1 seal) moves the scanned mark back
332    /// (`kv_rows_changed`), and a scan from the first row starts afresh.
333    pub(crate) fn kv_abs_max(&mut self) -> (f32, f32) {
334        let hd = self.head_dim.max(1);
335        let rows = self.kv_rows();
336        if rows < self.kv_amax_rows {
337            self.kv_amax_rows = 0;
338        }
339        if self.kv_amax_rows == 0 {
340            self.kv_amax = [0.0; 2];
341        }
342        let from = self.kv_amax_rows * hd;
343        for (m, heads) in self.kv_amax.iter_mut().zip([&self.k, &self.v]) {
344            for x in heads.iter().filter(|x| x.len() > from) {
345                *m = m.max(crate::gpu::abs_max_or_inf(&x[from..]));
346            }
347        }
348        self.kv_amax_rows = rows;
349        (self.kv_amax[0], self.kv_amax[1])
350    }
351
352    /// `kv_abs_max` for a caller that took the maxima of the rows it just
353    /// appended itself: `chunk` = (from, upto, max |k|, max |v|) over the
354    /// rows `[from, upto)`. Folded in without a scan when the cache holds
355    /// exactly `upto` rows and every row before `from` was scanned; else
356    /// the plain scan.
357    pub(crate) fn kv_abs_max_after(&mut self, chunk: (usize, usize, f32, f32)) -> (f32, f32) {
358        let (from, upto, km, vm) = chunk;
359        if self.kv_amax_rows != from || self.kv_rows() != upto || from > upto {
360            return self.kv_abs_max();
361        }
362        if from == 0 {
363            self.kv_amax = [0.0; 2];
364        }
365        self.kv_amax = [self.kv_amax[0].max(km), self.kv_amax[1].max(vm)];
366        self.kv_amax_rows = upto;
367        (self.kv_amax[0], self.kv_amax[1])
368    }
369
370    /// Rows any head stores.
371    fn kv_rows(&self) -> usize {
372        let hd = self.head_dim.max(1);
373        self.k
374            .iter()
375            .chain(&self.v)
376            .map(|x| x.len() / hd)
377            .max()
378            .unwrap_or(0)
379    }
380
381    /// Rows from `first` on may no longer be the ones `kv_abs_max` scanned.
382    fn kv_rows_changed(&mut self, first: usize) {
383        self.kv_amax_rows = self.kv_amax_rows.min(first);
384    }
385
386    // ── Natively bounded anchor (swa_sink_v1) ──
387
388    /// Give this layer its fixed-size ring (`[kvh][window][hd]` K and V).
389    /// Called once at load from the header; the record is zeroed on
390    /// `clear()` and never reallocated.
391    pub fn install_bounded(&mut self, window: usize) {
392        self.bounded = Some(crate::bounded::BoundedState::new(
393            self.num_kv_heads,
394            self.head_dim,
395            window,
396        ));
397        self.wire_kind = WireKind::Bounded;
398    }
399
400    /// One position of the bounded operator: insert the raw `k, v`
401    /// (`[kvh][hd]`) into slot `t mod W`, then attend every Q head of
402    /// `q` (`[nh][hd]`, raw) over sinks ∪ window into `out` (`[nh][hd]`).
403    /// Nothing is appended per position.
404    #[allow(clippy::too_many_arguments)]
405    pub fn bounded_step(
406        &mut self,
407        q: &[f32],
408        k: &[f32],
409        v: &[f32],
410        w: &crate::bounded::BoundedWeights,
411        rope: &crate::bounded::BoundedRope,
412        scale: f32,
413        num_heads: usize,
414        out: &mut [f32],
415    ) {
416        let st = self
417            .bounded
418            .as_mut()
419            .expect("bounded_step on a layer without an installed ring");
420        st.insert(k, v);
421        st.attend(q, num_heads, &w.sink_k, &w.sink_v, w.sink, &w.group_windows, rope, scale, out);
422        // Honest context depth for the memory/seq report — nothing is
423        // stored per position.
424        self.seq_len += 1;
425    }
426
427    /// Bytes of the bounded ring (0 on other layers).
428    pub fn bounded_state_bytes(&self) -> usize {
429        self.bounded.as_ref().map(|b| b.state_bytes()).unwrap_or(0)
430    }
431
432    /// Bit-for-bit copy of the ring for speculation (None on other layers).
433    pub fn bounded_snapshot(&self) -> Option<crate::bounded::BoundedSnapshot> {
434        self.bounded.as_ref().map(|b| b.snapshot())
435    }
436
437    /// Restore a ring snapshot taken on this layer; `seq_len` follows the
438    /// restored insert counter.
439    pub fn bounded_restore(&mut self, s: &crate::bounded::BoundedSnapshot) {
440        if let Some(b) = self.bounded.as_mut() {
441            b.restore(s);
442            self.seq_len = b.seen;
443            self.generation = next_gen();
444        }
445    }
446
447    /// Bind this layer to the legacy cache-wire policy. Delta layers must
448    /// refuse untagged state exchange rather than risk a plausible additive
449    /// interpretation on the peer.
450    pub fn set_linear_wire_allowed(&mut self, allowed: bool) {
451        self.linear_wire_allowed = allowed;
452    }
453
454    /// Discard tentative recurrent state after a speculative rejection or
455    /// any other path that abandons the lane-2 result.
456    pub fn discard_linear_scratch(&mut self) {
457        self.linear_scratch.clear();
458    }
459
460    /// Per-KV-head stored keys `[seq_len × head_dim]` (GPU token graph
461    /// sync): the stored tail, row 0 = position `base()`.
462    pub fn k_heads(&self) -> &[Vec<f32>] {
463        &self.k
464    }
465    /// Per-KV-head stored values `[seq_len × head_dim]`, row 0 = `base()`.
466    pub fn v_heads(&self) -> &[Vec<f32>] {
467        &self.v
468    }
469
470    // ── Sliding-window tail (trim_window) ──
471
472    /// Absolute position of stored row 0 (0 unless `trim_window` dropped
473    /// a prefix).
474    pub fn base(&self) -> usize {
475        self.base
476    }
477
478    /// Absolute context depth: the position the next append will hold.
479    pub fn pos_len(&self) -> usize {
480        self.base + self.seq_len
481    }
482
483    /// Storage generation (see the field): with the row count, the key a
484    /// device copy of these rows is valid under.
485    pub fn generation(&self) -> u64 {
486        self.generation
487    }
488
489    /// The window this layer trims to, once `trim_window` ran on it.
490    pub fn tail_window(&self) -> Option<usize> {
491        self.tail
492    }
493
494    /// Drop the rows a sliding-window layer can never read again. A query
495    /// at position p reads p−w+1..=p only, so once more than `2w` rows are
496    /// stored the front is drained down to `w + slack` rows, the cut
497    /// rounded DOWN to an absolute position that is a multiple of `align`
498    /// (so `w + slack ..= w + slack + align − 1` rows stay). Returns the
499    /// rows dropped.
500    ///
501    /// The rows that stay keep their order and values (`Vec::drain` of the
502    /// front), and every attend indexes rows relative to the newest one
503    /// (`first = rows − w`), so each later query sees the very rows it saw
504    /// untrimmed — bit for bit. `slack` rows past the window are what a
505    /// rollback (`truncate_last`) may take back without losing a row the
506    /// window needs. `align` keeps device GEMMs that reduce over the rows
507    /// from row 0 in fixed tiles on the tile grid they had untrimmed (the
508    /// dropped rows were whole all-zero tiles there).
509    ///
510    /// Callers trim only at a safe point: never between a call's appends
511    /// and its attend or importance accumulation (a prefill chunk appends
512    /// all its rows first and indexes them from the count it started at).
513    ///
514    /// No-op on an O(1) or bounded layer (they own their bound), and on a
515    /// q8 layer whose per-channel field is not frozen yet: the freeze
516    /// reads every stored row, so dropping one before it would change it.
517    pub fn trim_window(&mut self, w: usize, slack: usize, align: usize) -> usize {
518        self.trim_window_anchored(w, slack, align, usize::MAX)
519    }
520
521    /// `trim_window` that also keeps every row a rollback to `anchor −
522    /// slack` or later still reads (the cut never passes `anchor − w −
523    /// slack`). The pipeline anchors at the current turn's prompt end, so
524    /// the next chat turn can drop the whole answer however long it ran
525    /// (`Pipeline::rollback_prefix_len`); `usize::MAX` is the plain trim.
526    pub fn trim_window_anchored(
527        &mut self,
528        w: usize,
529        slack: usize,
530        align: usize,
531        anchor: usize,
532    ) -> usize {
533        if self.o1.is_some() || self.bounded.is_some() || w == 0 {
534            return 0;
535        }
536        self.tail = Some(w);
537        if self.seq_len <= 2 * w {
538            return 0;
539        }
540        if matches!(self.mode, KvMode::Q8 { .. })
541            && self.kcol.iter().all(Vec::is_empty)
542            && self.vcol.iter().all(Vec::is_empty)
543        {
544            return 0;
545        }
546        let align = align.max(1);
547        let keep_from = self
548            .pos_len()
549            .saturating_sub(w + slack)
550            .min(anchor.saturating_sub(w + slack));
551        let new_base = keep_from / align * align;
552        if new_base <= self.base {
553            return 0;
554        }
555        let d = new_base - self.base;
556        let hd = self.head_dim;
557        fn drop_front<T>(v: &mut Vec<T>, n: usize) {
558            let n = n.min(v.len());
559            v.drain(..n);
560        }
561        for h in 0..self.num_kv_heads {
562            drop_front(&mut self.k[h], d * hd);
563            drop_front(&mut self.v[h], d * hd);
564            drop_front(&mut self.kq[h], d * hd);
565            drop_front(&mut self.vq[h], d * hd);
566            drop_front(&mut self.ks[h], d * hd.div_ceil(KV_K_GROUP));
567            drop_front(&mut self.vs[h], d);
568        }
569        drop_front(&mut self.imp, d);
570        self.base = new_base;
571        self.seq_len -= d;
572        self.generation = next_gen();
573        // `kv_abs_max` counts its scanned mark from the old row 0. Moved
574        // with the rows, the next scan still starts at the first row it has
575        // not seen; left in place, rows appended after the trim up to the
576        // old mark would never be scanned. The maxima may keep a dropped
577        // row's value, which only over-states.
578        self.kv_amax_rows = self.kv_amax_rows.saturating_sub(d);
579        d
580    }
581
582    // ── O(1) Nyström override ──
583
584    /// Arm query collection for a fresh prompt pass (a cleared cache).
585    pub fn o1_begin(&mut self, m: usize, w: usize, sink: usize, rect: crate::nystrom::O1Rect) {
586        self.o1_begin_with_boundary(m, w, sink, rect, None);
587    }
588
589    /// Arm query collection with an optional completed-row seal barrier.
590    /// The barrier is deliberately part of the existing collecting state:
591    /// no second history or scheduler is introduced for short prompts.
592    pub(crate) fn o1_begin_with_boundary(
593        &mut self,
594        m: usize,
595        w: usize,
596        sink: usize,
597        rect: crate::nystrom::O1Rect,
598        seal_at: Option<usize>,
599    ) {
600        self.o1 = Some(O1State::Collecting {
601            m,
602            w,
603            sink,
604            rect,
605            seal_at,
606            q_buf: Vec::new(),
607        });
608        self.o1_keep.clear();
609        self.o1_await_keep = false;
610        self.o1_q_stride = 1;
611        self.o1_params = Some((m, w, sink, rect));
612        self.o1_error = None;
613        self.o1_transitioned = false;
614    }
615
616    /// Record one position's rotated queries (`[num_heads × head_dim]`)
617    /// during the exact prompt pass. No-op unless collecting — the hook
618    /// sits inside qwen_attention so every prefill flavor (sequential,
619    /// batched) feeds the same trace.
620    pub fn o1_push_q(&mut self, q_all: &[f32]) {
621        if let Some(O1State::Collecting { q_buf, .. }) = &mut self.o1 {
622            q_buf.extend_from_slice(q_all);
623        }
624    }
625
626    /// The rotated-query trace collected so far (`[rows × num_heads ×
627    /// head_dim]`), while collecting.
628    pub(crate) fn o1_q_trace(&self) -> Option<&[f32]> {
629        match &self.o1 {
630            Some(O1State::Collecting { q_buf, .. }) => Some(q_buf),
631            _ => None,
632        }
633    }
634
635    /// Move a collecting layer's deferred boundary (the seal diagnostics
636    /// collect past it, then seal at the row count they choose).
637    pub(crate) fn o1_set_boundary(&mut self, at: usize) {
638        if let Some(O1State::Collecting { seal_at, .. }) = &mut self.o1 {
639            *seal_at = Some(at);
640        }
641    }
642
643    pub fn o1_sealed(&self) -> bool {
644        matches!(self.o1, Some(O1State::Sealed { .. }))
645    }
646
647    /// Pending completed-row barrier, if any. A plain full-prompt seal has
648    /// no barrier until the caller asks to seal.
649    pub(crate) fn o1_pending_boundary(&self) -> Option<usize> {
650        match &self.o1 {
651            Some(O1State::Collecting { seal_at, .. }) => *seal_at,
652            _ => None,
653        }
654    }
655
656    /// Whether a batch of `count` exact rows would cross the deferred
657    /// boundary. This lets batched/pair callers split before row B rather
658    /// than appending exact KV past the point where conversion is required.
659    /// A batch that ends exactly on the boundary crosses nothing: all its
660    /// rows are exact, and the caller seals after it. `CMF_O1_WALK_BOUNDARY=1`
661    /// walks such a batch row by row as before.
662    pub(crate) fn o1_boundary_crossed_by(&self, count: usize) -> bool {
663        static WALK: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
664        let walk =
665            *WALK.get_or_init(|| std::env::var("CMF_O1_WALK_BOUNDARY").as_deref() == Ok("1"));
666        let Some(target) = self.o1_pending_boundary() else {
667            return false;
668        };
669        target <= self.seq_len
670            || self
671                .seq_len
672                .checked_add(count)
673                .map_or(true, |next| next > target || (walk && next == target))
674    }
675
676    pub(crate) fn take_o1_transition(&mut self) -> bool {
677        std::mem::take(&mut self.o1_transitioned)
678    }
679
680    pub(crate) fn take_o1_error(&self) -> Option<String> {
681        // Error observation is deliberately non-consuming.  The error is the
682        // append guard for this request; removing it would let a caller that
683        // ignored the returned Err resume ordinary KV growth after the
684        // bounded transition dropped its overlay.  `clear()` is the explicit
685        // reset boundary that clears the latch.
686        self.o1_error.clone()
687    }
688
689    /// Abort a malformed deferred transition after attention has already
690    /// produced its current row. Clearing the exact storage and dropping
691    /// the overlay makes the state unrecoverable by continued decode; the
692    /// pipeline then routes through its normal graph/cancel cleanup.
693    pub(crate) fn o1_abort(&mut self, err: String) {
694        self.k.iter_mut().for_each(Vec::clear);
695        self.v.iter_mut().for_each(Vec::clear);
696        self.kv_rows_changed(0);
697        self.kq.iter_mut().for_each(Vec::clear);
698        self.ks.iter_mut().for_each(Vec::clear);
699        self.vq.iter_mut().for_each(Vec::clear);
700        self.vs.iter_mut().for_each(Vec::clear);
701        self.kcol.iter_mut().for_each(Vec::clear);
702        self.vcol.iter_mut().for_each(Vec::clear);
703        self.imp.clear();
704        self.o1 = None;
705        self.seq_len = 0;
706        self.base = 0;
707        self.generation = next_gen();
708        self.o1_transitioned = false;
709        self.o1_error = Some(err);
710    }
711
712    /// Freeze the prompt into per-KV-group Nyström states and drop this
713    /// layer's full KV. Returns false while a short collecting layer is
714    /// below its deferred boundary; malformed prerequisites abort the
715    /// layer instead of silently resuming exact KV growth. The seal needs
716    /// f32 KV rows (`CMF_KV=q8` stores int8), every group densely stored, a
717    /// full q trace, and a GQA fan-out that actually divides.
718    pub fn o1_seal(&mut self, num_heads: usize) -> bool {
719        match self.o1_seal_checked(num_heads) {
720            Ok(sealed) => sealed,
721            Err(err) => {
722                tracing::error!("o1: seal aborted: {err}");
723                self.o1_abort(err);
724                false
725            }
726        }
727    }
728
729    /// Checked seal implementation. Validation happens while the collecting
730    /// state and full KV are still intact; only a valid completed boundary
731    /// is allowed to destructively convert them.
732    pub(crate) fn o1_seal_checked(&mut self, num_heads: usize) -> Result<bool, String> {
733        if let Some(err) = self.o1_error.clone() {
734            return Err(err);
735        }
736        // Idempotent: sealing a sealed (or plain) layer must not disturb its
737        // state, and a plain layer is not an O(1) participant.
738        if !matches!(self.o1, Some(O1State::Collecting { .. })) {
739            return Ok(self.o1_sealed());
740        }
741        let (m, w, sink, requested_boundary, q_len) = match &self.o1 {
742            Some(O1State::Collecting {
743                m,
744                w,
745                sink,
746                rect: _,
747                seal_at,
748                q_buf,
749            }) => (*m, *w, *sink, *seal_at, q_buf.len()),
750            _ => unreachable!("checked above"),
751        };
752        let floor = crate::nystrom::o1_deferred_boundary(w, sink)
753            .ok_or_else(|| "o1 seal: w + sink + slack + 1 overflow".to_string())?;
754        let target = requested_boundary.unwrap_or(floor).max(floor);
755        let t = self.seq_len;
756        if t < target {
757            if let Some(O1State::Collecting { seal_at, .. }) = &mut self.o1 {
758                if *seal_at != Some(target) {
759                    *seal_at = Some(target);
760                    tracing::info!(
761                        "o1 deferred seal: current rows={t}, boundary={target} (floor={floor})"
762                    );
763                }
764            }
765            return Ok(false);
766        }
767
768        let hd = self.head_dim;
769        if t == 0 {
770            return Err("o1 seal: cannot seal an empty layer".into());
771        }
772        if self.mode != KvMode::F32 {
773            return Err("o1 seal: requires dense F32 KV storage".into());
774        }
775        if self.num_kv_heads == 0 || num_heads == 0 || num_heads % self.num_kv_heads != 0 {
776            return Err(format!(
777                "o1 seal: invalid GQA geometry num_heads={num_heads} num_kv_heads={}",
778                self.num_kv_heads
779            ));
780        }
781        let hpk = num_heads / self.num_kv_heads;
782        let expected_k = t
783            .checked_mul(hd)
784            .ok_or_else(|| "o1 seal: KV row length overflow".to_string())?;
785        let stride = self.o1_q_stride;
786        let expected_q = t
787            .div_ceil(stride)
788            .checked_mul(hd)
789            .and_then(|v| v.checked_mul(num_heads))
790            .ok_or_else(|| "o1 seal: query trace length overflow".to_string())?;
791        if q_len != expected_q {
792            return Err(format!(
793                "o1 seal: query trace has {q_len} values, expected {expected_q}"
794            ));
795        }
796        if (0..self.num_kv_heads)
797            .any(|g| self.k[g].len() != expected_k || self.v[g].len() != expected_k)
798        {
799            return Err("o1 seal: KV heads are not densely populated".into());
800        }
801        if m < 4 || w == 0 {
802            return Err(format!("o1 seal: invalid geometry m={m} w={w}"));
803        }
804
805        let Some(O1State::Collecting {
806            m,
807            w,
808            sink,
809            rect,
810            q_buf,
811            ..
812        }) = self.o1.take()
813        else {
814            unreachable!("collecting state disappeared after validation");
815        };
816        // One KV group's state depends on that group's rows alone, so the
817        // groups seal on their own threads: the replay of every prompt row
818        // into the far field is the host's share of a turbo prompt (7 s of
819        // 33 for 16 000 tokens on an RTX 2000 Ada before this), and it ran
820        // one group after another. Each thread transposes its own group of
821        // the position-major query trace (the state wants each head's
822        // queries contiguous).
823        let wire_layer = self.wire_layer;
824        let keep = &self.o1_keep;
825        let (ks, vs) = (&self.k, &self.v);
826        let q_buf = &q_buf;
827        // the far rows' sums on the card when it can take them (CUDA): the host
828        // seals each group from its rows by index, without the replay, and the
829        // card sums the far rows
830        let t_seal0 = std::time::Instant::now();
831        let dev = crate::gpu::o1_far_sums_on() && std::env::var_os("CMF_O1_CAPTURE").is_none();
832        let order: Vec<u32> = if keep.is_empty() { (0..t as u32).collect() } else { crate::dory::seal_order(t, sink, keep) };
833        let order = &order;
834        type Sealed = (crate::nystrom::NystromState, Option<std::ops::Range<usize>>);
835        let seal_group = move |g: usize| -> Sealed {
836            let calib = crate::nystrom::O1Calib::load(wire_layer, g, hd, hpk);
837            let wg = crate::nystrom::wide_window(wire_layer, g).unwrap_or(w);
838            let mut st = crate::nystrom::NystromState::new_group(m, w, sink + keep.len(), hpk)
839                .with_window(wg)
840                .with_rect(rect)
841                .with_calib(calib);
842            if dev {
843                if let Some(range) = st.seal_gathered(&ks[g], &vs[g], q_buf, stride, num_heads, g * hpk, order, t, hd, hd) {
844                    return (st, Some(range));
845                }
846            }
847            let mut qh = vec![0.0f32; hpk * t * hd];
848            for hh in 0..hpk {
849                let h = g * hpk + hh;
850                for p in 0..t {
851                    let src = (p / stride * num_heads + h) * hd;
852                    let dst = (hh * t + p) * hd;
853                    qh[dst..dst + hd].copy_from_slice(&q_buf[src..src + hd]);
854                }
855            }
856            let qs: Vec<&[f32]> = (0..hpk)
857                .map(|hh| &qh[hh * t * hd..(hh + 1) * t * hd])
858                .collect();
859            crate::nystrom::capture_seal(wire_layer, g, t, hd, hpk, &qh, &ks[g], &vs[g]);
860            if keep.is_empty() {
861                st.prefill_group(&qs, &ks[g], &vs[g], t, hd, hd);
862            } else {
863                // The kept rows go right behind the sinks, and leave the far field.
864                let pick = |src: &[f32]| -> Vec<f32> {
865                    let mut out = Vec::with_capacity(t * hd);
866                    for &r in order {
867                        out.extend_from_slice(&src[r as usize * hd..(r as usize + 1) * hd]);
868                    }
869                    out
870                };
871                let qo: Vec<Vec<f32>> = qs.iter().map(|q| pick(q)).collect();
872                let qr: Vec<&[f32]> = qo.iter().map(|q| &q[..]).collect();
873                st.prefill_group(&qr, &pick(&ks[g]), &pick(&vs[g]), t, hd, hd);
874            }
875            (st, None)
876        };
877        let sealed: Vec<Sealed> = if self.num_kv_heads > 1 {
878            std::thread::scope(|s| {
879                let hs: Vec<_> = (0..self.num_kv_heads).map(|g| s.spawn(move || seal_group(g))).collect();
880                hs.into_iter().map(|h| h.join().expect("o1 seal: a group's thread panicked")).collect()
881            })
882        } else {
883            vec![seal_group(0)]
884        };
885        let t_groups = t_seal0.elapsed().as_secs_f64();
886        let (mut groups, fars): (Vec<_>, Vec<_>) = sealed.into_iter().unzip();
887        if fars.iter().any(Option::is_some) {
888            // every group's far rows on the card at once; the host's replay
889            // when the card declines (or the groups' windows differ)
890            let same = fars.iter().all(|f| f.is_some() && f == &fars[0]);
891            let (m_eff, _, scale) = groups[0].far_geometry();
892            let mut done = false;
893            if same && groups.iter().all(|g| g.far_geometry().0 == m_eff) {
894                let range = fars[0].clone().unwrap();
895                let mut qt = Vec::with_capacity(self.num_kv_heads * hpk * m_eff * hd);
896                for g in &groups {
897                    for h in 0..hpk {
898                        qt.extend_from_slice(g.far_q_tilde(h));
899                    }
900                }
901                let kf: Vec<&[f32]> = ks.iter().map(|k| &k[..]).collect();
902                let vf: Vec<&[f32]> = vs.iter().map(|v| &v[..]).collect();
903                if let Some((mz, th)) = crate::gpu::o1_far_sums(self.num_kv_heads, hpk, hd, m_eff, scale, &qt, &order[range.clone()], &kf, &vf) {
904                    for (gi, g) in groups.iter_mut().enumerate() {
905                        for h in 0..hpk {
906                            let hh = gi * hpk + h;
907                            let z = &mz[hh * 2 * m_eff..(hh + 1) * 2 * m_eff];
908                            g.set_far_sums(h, &z[..m_eff], &z[m_eff..], &th[hh * m_eff * hd..(hh + 1) * m_eff * hd], range.len());
909                        }
910                    }
911                    done = true;
912                }
913            }
914            if std::env::var("CMF_PREFILL_PROF").is_ok() {
915                eprintln!("o1 seal of a layer: groups {t_groups:.3} s, far sums {:.3} s ({})", t_seal0.elapsed().as_secs_f64() - t_groups, if done { "card" } else { "host" });
916            }
917            if !done {
918                std::thread::scope(|s| {
919                    for (gi, (g, f)) in groups.iter_mut().zip(&fars).enumerate() {
920                        if let Some(range) = f {
921                            let (k, v) = (&ks[gi][..], &vs[gi][..]);
922                            s.spawn(move || g.replay_far_idx(k, v, &order[range.clone()]));
923                        }
924                    }
925                });
926            }
927        }
928        // The states now carry everything decode needs — release the
929        // O(context) storage (this is the memory claim, not a cosmetic).
930        for h in 0..self.num_kv_heads {
931            self.k[h] = Vec::new();
932            self.v[h] = Vec::new();
933        }
934        self.kv_rows_changed(0);
935        self.imp = Vec::new();
936        self.o1 = Some(O1State::Sealed { groups });
937        self.o1_transitioned = true;
938        Ok(true)
939    }
940
941    /// One decode step on a sealed layer: per KV group, insert the
942    /// group's fresh (k, v) ONCE and read every Q head's attention
943    /// output. Returns `[num_heads × head_dim]`. Head h belongs to group
944    /// h/hpk, so a group's Q heads are contiguous in `q_all`/`out` —
945    /// same math as the shared KV row the exact path appends once.
946    /// Device views of the sealed o1 groups, or None when o1 is not
947    /// sealed on this layer (or any group is in the degenerate
948    /// exact-only mode the GPU path does not carry).
949    pub fn o1_views(&self) -> Option<Vec<crate::nystrom::O1DeviceView<'_>>> {
950        let Some(O1State::Sealed { groups }) = &self.o1 else {
951            return None;
952        };
953        let views: Vec<_> = groups.iter().map(|g| g.device_view()).collect();
954        if views.iter().any(|v| v.exact_only) {
955            return None;
956        }
957        Some(views)
958    }
959
960    pub fn o1_step(
961        &mut self,
962        q_all: &[f32],
963        k_new: &[f32],
964        v_new: &[f32],
965        num_heads: usize,
966    ) -> Vec<f32> {
967        let hd = self.head_dim;
968        let hpk = num_heads / self.num_kv_heads.max(1);
969        let mut out = vec![0.0f32; num_heads * hd];
970        let Some(O1State::Sealed { groups }) = &mut self.o1 else {
971            debug_assert!(false, "o1_step on an unsealed layer");
972            return out;
973        };
974        for (g, st) in groups.iter_mut().enumerate() {
975            let (lo, hi) = (g * hpk * hd, (g + 1) * hpk * hd);
976            st.step_group(
977                &q_all[lo..hi],
978                &k_new[g * hd..(g + 1) * hd],
979                &v_new[g * hd..(g + 1) * hd],
980                &mut out[lo..hi],
981            );
982        }
983        // Track the true context depth for the honest memory/seq report
984        // (nothing is stored per position — the state is O(1)).
985        self.seq_len += 1;
986        out
987    }
988
989    /// Bytes held by the O(1) override (query trace while collecting,
990    /// per-KV-group states once sealed).
991    pub fn o1_memory_bytes(&self) -> usize {
992        match &self.o1 {
993            Some(O1State::Collecting { q_buf, .. }) => q_buf.len() * std::mem::size_of::<f32>(),
994            Some(O1State::Sealed { groups }) => groups.iter().map(|s| s.memory_bytes()).sum(),
995            None => 0,
996        }
997    }
998
999    /// Quantize one row against the per-channel field (empty col = 1);
1000    /// `group` — elements per scale (the whole row or KV_K_GROUP).
1001    fn quant_row(row: &[f32], col: &[f32], q: &mut Vec<i8>, sc: &mut Vec<f32>, group: usize) {
1002        let mut resid = vec![0.0f32; row.len()];
1003        for (d, &x) in row.iter().enumerate() {
1004            resid[d] = if col.is_empty() { x } else { x / col[d] };
1005        }
1006        for g0 in (0..row.len()).step_by(group) {
1007            let g1 = (g0 + group).min(row.len());
1008            let mut absmax = 0.0f32;
1009            for &r in &resid[g0..g1] {
1010                absmax = absmax.max(r.abs());
1011            }
1012            let s = (absmax / 127.0).max(1e-12);
1013            sc.push(s);
1014            for &r in &resid[g0..g1] {
1015                q.push((r / s).round().clamp(-127.0, 127.0) as i8);
1016            }
1017        }
1018    }
1019
1020    /// Freeze the 2f field: col = RMS of channels over stored rows, old
1021    /// rows are requantized against the new field (once per conversation).
1022    fn freeze_cols(&mut self) {
1023        let hd = self.head_dim;
1024        let ngk = hd.div_ceil(KV_K_GROUP);
1025        for h in 0..self.num_kv_heads {
1026            for (qv, sv, colv, group) in [
1027                (
1028                    &mut self.kq[h],
1029                    &mut self.ks[h],
1030                    &mut self.kcol[h],
1031                    KV_K_GROUP,
1032                ),
1033                (&mut self.vq[h], &mut self.vs[h], &mut self.vcol[h], hd),
1034            ] {
1035                let spp = if group == hd { 1 } else { ngk }; // scales per position
1036                let n = sv.len() / spp;
1037                if n == 0 {
1038                    continue;
1039                }
1040                // Dequantize to f32, RMS over channels, requantize.
1041                let mut rows = vec![0.0f32; n * hd];
1042                for p in 0..n {
1043                    for d in 0..hd {
1044                        rows[p * hd + d] = qv[p * hd + d] as f32 * sv[p * spp + d / group];
1045                    }
1046                }
1047                let mut col = vec![0.0f32; hd];
1048                for p in 0..n {
1049                    for d in 0..hd {
1050                        col[d] += rows[p * hd + d] * rows[p * hd + d];
1051                    }
1052                }
1053                for c in col.iter_mut() {
1054                    *c = (*c / n as f32).sqrt().max(1e-6);
1055                }
1056                qv.clear();
1057                sv.clear();
1058                for p in 0..n {
1059                    Self::quant_row(&rows[p * hd..(p + 1) * hd], &col, qv, sv, group);
1060                }
1061                *colv = col;
1062            }
1063        }
1064    }
1065
1066    /// Append K/V for one position. `k_new`/`v_new` are
1067    /// `[num_kv_heads × head_dim]`; heads with `alive[h] == false` are
1068    /// skipped (their slices stay empty).
1069    /// Room for `rows` more rows of plain f32 storage, taken once: a prompt's
1070    /// rows coming back from a device mirror otherwise grow each head's
1071    /// vectors by doubling, a copy of everything stored so far each time.
1072    pub(crate) fn reserve_rows(&mut self, rows: usize) {
1073        if self.mode != KvMode::F32 {
1074            return;
1075        }
1076        let n = rows * self.head_dim;
1077        for h in 0..self.num_kv_heads {
1078            self.k[h].reserve(n);
1079            self.v[h].reserve(n);
1080        }
1081        self.imp.reserve(rows);
1082    }
1083
1084    pub fn append(&mut self, k_new: &[f32], v_new: &[f32], alive: &[bool]) {
1085        // A failed bounded transition is terminal until the sequence is
1086        // cleared. Do not let an ignored boolean/result resume plain KV
1087        // growth after the O(1) collector has aborted.
1088        if self.o1_error.is_some() {
1089            return;
1090        }
1091        debug_assert_eq!(k_new.len(), self.num_kv_heads * self.head_dim);
1092        debug_assert_eq!(v_new.len(), self.num_kv_heads * self.head_dim);
1093        // Freeze the 2f field AT THE START of append: only rows that
1094        // survived verify are visible (a rejected lane-2 draft does not
1095        // pollute the field — found in review), and the threshold uses >=
1096        // rather than strict equality (in small windows eviction may
1097        // oscillate across 64).
1098        if matches!(self.mode, KvMode::Q8 { .. })
1099            && self.seq_len >= KV_COL_WARMUP
1100            && self.kcol.iter().all(Vec::is_empty)
1101            && self.vcol.iter().all(Vec::is_empty)
1102        {
1103            self.freeze_cols();
1104        }
1105        for h in 0..self.num_kv_heads {
1106            if !alive.get(h).copied().unwrap_or(true) {
1107                continue;
1108            }
1109            let s = h * self.head_dim;
1110            if self.mode.quant_k() {
1111                Self::quant_row(
1112                    &k_new[s..s + self.head_dim],
1113                    &self.kcol[h],
1114                    &mut self.kq[h],
1115                    &mut self.ks[h],
1116                    KV_K_GROUP,
1117                );
1118            } else {
1119                self.k[h].extend_from_slice(&k_new[s..s + self.head_dim]);
1120            }
1121            if self.mode.quant_v() {
1122                Self::quant_row(
1123                    &v_new[s..s + self.head_dim],
1124                    &self.vcol[h],
1125                    &mut self.vq[h],
1126                    &mut self.vs[h],
1127                    self.head_dim,
1128                );
1129            } else {
1130                self.v[h].extend_from_slice(&v_new[s..s + self.head_dim]);
1131            }
1132        }
1133        self.imp.push(0.0);
1134        self.seq_len += 1;
1135    }
1136
1137    /// Per-head attention over its own storage: the f32 branch is
1138    /// bit-for-bit equal to attention_head() over slices; the q8 branch
1139    /// computes score = s_k·⟨q⊙col_k, k_q⟩ and the weighted sum of V in i8
1140    /// with f32 accumulation. Returns (output [head_dim], probs [stored]).
1141    pub fn attend(&self, q: &[f32], kv_head: usize) -> (Vec<f32>, Vec<f32>) {
1142        // Full-context layers only: a trimmed tail is not the whole row.
1143        debug_assert!(self.tail.is_none(), "attend() on a sliding-window tail");
1144        let hd = self.head_dim;
1145        if self.mode == KvMode::F32 {
1146            let stored = self.k[kv_head].len() / hd;
1147            return crate::attention::attention_head(
1148                q,
1149                &self.k[kv_head],
1150                &self.v[kv_head],
1151                hd,
1152                stored,
1153            );
1154        }
1155        let stored = self.head_len(kv_head);
1156        let scale = 1.0 / (hd as f32).sqrt();
1157        let mut scores = vec![0.0f32; stored];
1158        if self.mode.quant_k() {
1159            let (kq, ks) = (&self.kq[kv_head], &self.ks[kv_head]);
1160            // q ⊙ col_k — once per call.
1161            let kcol = &self.kcol[kv_head];
1162            let mut qc = vec![0.0f32; hd];
1163            for d in 0..hd {
1164                qc[d] = if kcol.is_empty() {
1165                    q[d]
1166                } else {
1167                    q[d] * kcol[d]
1168                };
1169            }
1170            let ng = hd.div_ceil(KV_K_GROUP);
1171            for p in 0..stored {
1172                let row = &kq[p * hd..(p + 1) * hd];
1173                // SAFETY: i8 and u8 share layout; dot_i8_f32 reads the
1174                // bytes back as i8.
1175                let row_u8 =
1176                    unsafe { std::slice::from_raw_parts(row.as_ptr() as *const u8, row.len()) };
1177                let mut dot = 0.0f32;
1178                for g in 0..ng {
1179                    let g0 = g * KV_K_GROUP;
1180                    let g1 = (g0 + KV_K_GROUP).min(hd);
1181                    dot +=
1182                        crate::qtensor::dot_i8_f32(&row_u8[g0..g1], &qc[g0..g1]) * ks[p * ng + g];
1183                }
1184                scores[p] = dot * scale;
1185            }
1186        } else {
1187            let k = &self.k[kv_head];
1188            for p in 0..stored {
1189                let row = &k[p * hd..(p + 1) * hd];
1190                scores[p] = crate::attention::dot_f32(q, row) * scale;
1191            }
1192        }
1193        let max_score = scores.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
1194        let mut sum = 0.0f32;
1195        for s in scores.iter_mut() {
1196            *s = (*s - max_score).exp();
1197            sum += *s;
1198        }
1199        if sum > 0.0 {
1200            for s in scores.iter_mut() {
1201                *s /= sum;
1202            }
1203        }
1204        let mut acc = vec![0.0f32; hd];
1205        if self.mode.quant_v() {
1206            let (vq, vs) = (&self.vq[kv_head], &self.vs[kv_head]);
1207            for p in 0..stored {
1208                let w = scores[p] * vs[p];
1209                if w.abs() < 1e-12 {
1210                    continue;
1211                }
1212                crate::qtensor::axpy_i8_f32(&mut acc, &vq[p * hd..(p + 1) * hd], w);
1213            }
1214            let vcol = &self.vcol[kv_head];
1215            if !vcol.is_empty() {
1216                for d in 0..hd {
1217                    acc[d] *= vcol[d];
1218                }
1219            }
1220        } else {
1221            let v = &self.v[kv_head];
1222            for p in 0..stored {
1223                let w = scores[p];
1224                if w.abs() < 1e-12 {
1225                    continue;
1226                }
1227                crate::attention::axpy_f32(&mut acc, &v[p * hd..(p + 1) * hd], w);
1228            }
1229        }
1230        (acc, scores)
1231    }
1232
1233    /// Grouped GQA attention: all Q-heads of one KV group in a single
1234    /// pass over the stored K rows and a single pass over the V rows
1235    /// (per-head `attend` re-read the shared group storage
1236    /// heads_per_kv times — roadmap §3 P1). Per-head score order,
1237    /// softmax and V accumulation are IDENTICAL to `attend`, so each
1238    /// head's output is bit-for-bit the same.
1239    ///
1240    /// `q_group`: `[n_heads_in_group × head_dim]` (global head order);
1241    /// `out`: same shape; `imp_acc[0..stored]` accumulates the probabilities
1242    /// of every head (attention importance), matching the caller's former loop.
1243    /// `scale` is the score scale (1/√hd unless the arch overrides);
1244    /// `first` is the earliest visible position — sliding-window layers
1245    /// pass `stored − window` so older rows get zero probability.
1246    /// `sinks` holds one learned sink logit per head of `q_group` (the
1247    /// caller slices `self.sinks` to the group's heads) or is empty for
1248    /// an ordinary softmax.
1249    #[allow(clippy::too_many_arguments)]
1250    pub fn attend_group(
1251        &self,
1252        q_group: &[f32],
1253        kv_head: usize,
1254        out: &mut [f32],
1255        imp_acc: &mut [f32],
1256        scale: f32,
1257        first: usize,
1258        softcap: f32,
1259        sinks: &[f32],
1260    ) {
1261        self.attend_group_upto(
1262            q_group,
1263            kv_head,
1264            out,
1265            imp_acc,
1266            scale,
1267            first,
1268            softcap,
1269            usize::MAX,
1270            sinks,
1271        )
1272    }
1273
1274    /// `attend_group` over the first `upto` stored rows only — what the
1275    /// same call saw when the cache held exactly `upto` rows. A prefill
1276    /// chunk appends all its rows first and then attends every position
1277    /// in parallel; position `i` passes `upto = s0 + i + 1`, which makes
1278    /// its result bit-identical to the sequential append-then-attend.
1279    ///
1280    /// Only the visible rows `[first, stored)` are scored, so a
1281    /// sliding-window decode step costs O(window), not O(context). The
1282    /// rows before `first` get probability exactly 0 — what the former
1283    /// −inf-filled score row produced (exp(−inf) = 0 adds nothing to the
1284    /// max, the sum, V or the importance), so the result is bit-identical
1285    /// to scoring the whole row.
1286    ///
1287    /// Learned sinks (`sinks` non-empty, one per head): the sink logit
1288    /// joins the softmax max and adds exp(sink − max) to the denominator;
1289    /// it has no value row. Identical to appending a value-less column,
1290    /// which is how gpt-oss and MiMo-V2 define it.
1291    #[allow(clippy::too_many_arguments)]
1292    pub fn attend_group_upto(
1293        &self,
1294        q_group: &[f32],
1295        kv_head: usize,
1296        out: &mut [f32],
1297        imp_acc: &mut [f32],
1298        scale: f32,
1299        first: usize,
1300        softcap: f32,
1301        upto: usize,
1302        sinks: &[f32],
1303    ) {
1304        let hd = self.head_dim;
1305        let nheads = q_group.len() / hd;
1306        debug_assert_eq!(out.len(), nheads * hd);
1307        assert!(
1308            sinks.is_empty() || sinks.len() == nheads,
1309            "attend_group: {} sink logits for {nheads} heads",
1310            sinks.len()
1311        );
1312        let stored = if self.mode == KvMode::F32 {
1313            self.k[kv_head].len() / hd
1314        } else {
1315            self.head_len(kv_head)
1316        }
1317        .min(upto);
1318        if stored == 0 {
1319            out.fill(0.0);
1320            return;
1321        }
1322        let first = first.min(stored.saturating_sub(1));
1323        // Visible span: row p ∈ [first, stored) lives at column p − first.
1324        let span = stored - first;
1325
1326        thread_local! {
1327            /// scores [nheads × span] — reused across layers/tokens.
1328            static GQA_SCORES: std::cell::RefCell<Vec<f32>> =
1329                const { std::cell::RefCell::new(Vec::new()) };
1330            /// q ⊙ col_k per head (q8 K mode).
1331            static GQA_QC: std::cell::RefCell<Vec<f32>> =
1332                const { std::cell::RefCell::new(Vec::new()) };
1333        }
1334
1335        GQA_SCORES.with(|sc| {
1336            let mut scores = sc.borrow_mut();
1337            scores.resize(nheads * span, 0.0);
1338
1339            // ── score pass: each stored K row is read ONCE for all heads.
1340            if self.mode.quant_k() {
1341                let (kq, ks) = (&self.kq[kv_head], &self.ks[kv_head]);
1342                let kcol = &self.kcol[kv_head];
1343                let ng = hd.div_ceil(KV_K_GROUP);
1344                GQA_QC.with(|qc| {
1345                    let mut qcb = qc.borrow_mut();
1346                    qcb.resize(nheads * hd, 0.0);
1347                    for h in 0..nheads {
1348                        for d in 0..hd {
1349                            let qv = q_group[h * hd + d];
1350                            qcb[h * hd + d] = if kcol.is_empty() { qv } else { qv * kcol[d] };
1351                        }
1352                    }
1353                    for p in first..stored {
1354                        let row = &kq[p * hd..(p + 1) * hd];
1355                        // SAFETY: i8 and u8 share layout; dot_i8_f32 reads
1356                        // the bytes back as i8.
1357                        let row_u8 = unsafe {
1358                            std::slice::from_raw_parts(row.as_ptr() as *const u8, row.len())
1359                        };
1360                        for h in 0..nheads {
1361                            let qch = &qcb[h * hd..(h + 1) * hd];
1362                            let mut dot = 0.0f32;
1363                            for g in 0..ng {
1364                                let g0 = g * KV_K_GROUP;
1365                                let g1 = (g0 + KV_K_GROUP).min(hd);
1366                                dot += crate::qtensor::dot_i8_f32(&row_u8[g0..g1], &qch[g0..g1])
1367                                    * ks[p * ng + g];
1368                            }
1369                            scores[h * span + (p - first)] = dot * scale;
1370                        }
1371                    }
1372                });
1373            } else {
1374                let k = &self.k[kv_head];
1375                for p in first..stored {
1376                    let row = &k[p * hd..(p + 1) * hd];
1377                    for h in 0..nheads {
1378                        scores[h * span + (p - first)] =
1379                            crate::attention::dot_f32(&q_group[h * hd..(h + 1) * hd], row) * scale;
1380                    }
1381                }
1382            }
1383
1384            // Gemma-2 attention-logit soft-capping: tanh-squash the
1385            // COMPUTED scores before the softmax (every scored row is in
1386            // the window; a learned sink is not a score and is not capped).
1387            if softcap > 0.0 {
1388                for v in scores.iter_mut() {
1389                    *v = softcap * (*v / softcap).tanh();
1390                }
1391            }
1392
1393            // ── per-head softmax (identical to attend / attention_head;
1394            // a sink joins the max and the denominator, never the rows).
1395            for h in 0..nheads {
1396                let s = &mut scores[h * span..(h + 1) * span];
1397                let row_max = s.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
1398                let sink = sinks.get(h).copied();
1399                let max_score = match sink {
1400                    Some(z) => row_max.max(z),
1401                    None => row_max,
1402                };
1403                let mut sum = 0.0f32;
1404                for v in s.iter_mut() {
1405                    *v = (*v - max_score).exp();
1406                    sum += *v;
1407                }
1408                if let Some(z) = sink {
1409                    sum += (z - max_score).exp();
1410                }
1411                if sum > 0.0 {
1412                    for v in s.iter_mut() {
1413                        *v /= sum;
1414                    }
1415                }
1416            }
1417
1418            // ── value pass: each stored V row is read ONCE for all heads.
1419            out.fill(0.0);
1420            if self.mode.quant_v() {
1421                let (vq, vs) = (&self.vq[kv_head], &self.vs[kv_head]);
1422                for p in first..stored {
1423                    let row = &vq[p * hd..(p + 1) * hd];
1424                    for h in 0..nheads {
1425                        let w = scores[h * span + (p - first)] * vs[p];
1426                        if w.abs() < 1e-12 {
1427                            continue;
1428                        }
1429                        crate::qtensor::axpy_i8_f32(&mut out[h * hd..(h + 1) * hd], row, w);
1430                    }
1431                }
1432                let vcol = &self.vcol[kv_head];
1433                if !vcol.is_empty() {
1434                    for h in 0..nheads {
1435                        for d in 0..hd {
1436                            out[h * hd + d] *= vcol[d];
1437                        }
1438                    }
1439                }
1440            } else {
1441                let v = &self.v[kv_head];
1442                for p in first..stored {
1443                    let row = &v[p * hd..(p + 1) * hd];
1444                    for h in 0..nheads {
1445                        let w = scores[h * span + (p - first)];
1446                        if w.abs() < 1e-12 {
1447                            continue;
1448                        }
1449                        crate::attention::axpy_f32(&mut out[h * hd..(h + 1) * hd], row, w);
1450                    }
1451                }
1452            }
1453
1454            // ── Attention-importance accumulation (Σ probs over heads), same
1455            // head order as the caller's former per-head loop. Rows before
1456            // `first` carry probability 0 and are left untouched.
1457            let n = imp_acc.len().min(stored);
1458            if n > first {
1459                for h in 0..nheads {
1460                    let s = &scores[h * span..(h + 1) * span];
1461                    for (dst, &p) in imp_acc[first..n].iter_mut().zip(s) {
1462                        *dst += p;
1463                    }
1464                }
1465            }
1466        });
1467    }
1468
1469    /// Batched causal attend for a prefill chunk (macOS, AArch64, x86-64):
1470    /// the cache already holds every chunk row (`s0` old + `b` new). Per
1471    /// KV group the scores GEMM `Q·Kᵀ` rides Accelerate (AMX) or the
1472    /// portable micro-GEMM (NEON, AVX2), the causal softmax zeroes the
1473    /// not-yet-visible tail so the `P·V` GEMM needs no mask, and
1474    /// attention importance takes the masked column sums. A sliding
1475    /// window bounds the key range of the whole chunk, so a windowed
1476    /// layer costs O(b·(b + w)) rather than O(b·n). Same math as the
1477    /// per-position attend; summation order differs (tolerance-class,
1478    /// like the projection GEMMs).
1479    #[cfg(any(target_arch = "aarch64", target_arch = "x86_64"))]
1480    #[allow(clippy::too_many_arguments)]
1481    pub fn attend_chunk(
1482        &mut self,
1483        q_all: &[f32],
1484        b: usize,
1485        s0: usize,
1486        nh: usize,
1487        heads_per_kv: usize,
1488        hd: usize,
1489        out: &mut [f32],
1490        pool: Option<&crate::pool::Pool>,
1491        scale: f32,
1492        window: Option<usize>,
1493    ) {
1494        let n = s0 + b;
1495        // A masking window: the earliest key any row of this chunk sees
1496        // is `s0 + 1 - w`, so the GEMM pair runs over the keys [lo0, n)
1497        // only — at 16 000 rows a 512-wide window is 32× less work than
1498        // the full causal range, and the zeroed columns contributed
1499        // nothing anyway.
1500        let lo0 = window.map_or(0, |w| (s0 + 1).saturating_sub(w)).min(n);
1501        let nk = n - lo0;
1502        struct SendPtr(*mut f32);
1503        unsafe impl Send for SendPtr {}
1504        unsafe impl Sync for SendPtr {}
1505        impl SendPtr {
1506            fn at(&self, i: usize) -> *mut f32 {
1507                // Method receiver keeps the closure capturing &SendPtr
1508                // (2021 disjoint capture would grab the raw field).
1509                unsafe { self.0.add(i) }
1510            }
1511        }
1512        thread_local! {
1513            static SCRATCH: std::cell::RefCell<(Vec<f32>, Vec<f32>, Vec<f32>, Vec<f32>)> =
1514                const { std::cell::RefCell::new((Vec::new(), Vec::new(), Vec::new(), Vec::new())) };
1515        }
1516        // The portable micro-GEMM (NEON, AVX2) pays dearly for the
1517        // gathered Bᵀ loads of the scores multiply — pack Kᵀ once per
1518        // (group, chunk) and hand it the sequential-B fast path instead.
1519        // Accelerate keeps the no-copy transposed call.
1520        let neon_gemm = cfg!(not(target_os = "macos"))
1521            || std::env::var("CMF_FORCE_NEON_GEMM")
1522                .map(|v| v == "1")
1523                .unwrap_or(false);
1524        SCRATCH.with(|s| {
1525            let mut s = s.borrow_mut();
1526            let (qpanel, scores, aopanel, ktpack) = &mut *s;
1527            // The whole KV-group attends in one GEMM pair: the group's
1528            // Q-heads stack head-major into one tall panel [hpk·b, hd]
1529            // (row hl·b + bi), so each layer costs 2 sgemm calls per
1530            // group instead of 2 per head — fat M keeps the AMX fed.
1531            let m = heads_per_kv * b;
1532            qpanel.resize(m * hd, 0.0);
1533            scores.resize(m * nk, 0.0);
1534            aopanel.resize(m * hd, 0.0);
1535            for g in 0..self.num_kv_heads {
1536                debug_assert_eq!(self.k[g].len(), n * hd);
1537                let kmat = &self.k[g][lo0 * hd..n * hd];
1538                let vmat = &self.v[g][lo0 * hd..n * hd];
1539                for hl in 0..heads_per_kv {
1540                    let hh = g * heads_per_kv + hl;
1541                    for bi in 0..b {
1542                        qpanel[(hl * b + bi) * hd..(hl * b + bi + 1) * hd]
1543                            .copy_from_slice(&q_all[bi * nh * hd + hh * hd..][..hd]);
1544                    }
1545                }
1546                if neon_gemm {
1547                    ktpack.resize(hd * nk, 0.0);
1548                    {
1549                        // Each worker transposes its own key rows into
1550                        // its own columns of Kᵀ.
1551                        let sp_kt = SendPtr(ktpack.as_mut_ptr());
1552                        let run = |start: usize, end: usize| {
1553                            for p in start..end {
1554                                let row = &kmat[p * hd..(p + 1) * hd];
1555                                for (d, &v) in row.iter().enumerate() {
1556                                    // SAFETY: column p belongs to this worker alone.
1557                                    unsafe { *sp_kt.at(d * nk + p) = v };
1558                                }
1559                            }
1560                        };
1561                        match pool {
1562                            Some(p) if nk >= 2048 => p.run_rows(nk, &run),
1563                            _ => run(0, nk),
1564                        }
1565                    }
1566                    // Accelerate threads its own GEMM; the micro-kernel
1567                    // splits the m rows across the pool instead.
1568                    let sp_q = SendPtr(qpanel.as_ptr() as *mut f32);
1569                    let sp_s = SendPtr(scores.as_mut_ptr());
1570                    let kt = &*ktpack;
1571                    let run = |start: usize, end: usize| {
1572                        if end > start {
1573                            // SAFETY: workers write disjoint score rows.
1574                            let a = unsafe {
1575                                std::slice::from_raw_parts(sp_q.at(start * hd), (end - start) * hd)
1576                            };
1577                            let c = unsafe {
1578                                std::slice::from_raw_parts_mut(
1579                                    sp_s.at(start * nk),
1580                                    (end - start) * nk,
1581                                )
1582                            };
1583                            crate::qtensor::micro_gemm_rm(
1584                                end - start,
1585                                nk,
1586                                hd,
1587                                scale,
1588                                a,
1589                                hd,
1590                                kt,
1591                                nk,
1592                                false,
1593                                c,
1594                                nk,
1595                            );
1596                        }
1597                    };
1598                    match pool {
1599                        Some(p) if m >= 64 => p.run_rows(m, &run),
1600                        _ => run(0, m),
1601                    }
1602                } else {
1603                    crate::qtensor::sgemm_rm(
1604                        m, nk, hd, scale, qpanel, hd, kmat, hd, true, scores, nk,
1605                    );
1606                }
1607                // Causal softmax, row-parallel (rows are disjoint).
1608                let sp = SendPtr(scores.as_mut_ptr());
1609                let run = |start: usize, end: usize| {
1610                    for r in start..end {
1611                        // Key indices relative to lo0, the first key of
1612                        // the GEMM range.
1613                        let allowed = s0 + (r % b) + 1 - lo0;
1614                        // Sliding-window layers see only the last W of
1615                        // the causal range; the zeroed head contributes
1616                        // nothing to P·V or attention importance.
1617                        let lo = window
1618                            .map(|w| (s0 + (r % b) + 1).saturating_sub(w) - lo0)
1619                            .unwrap_or(0);
1620                        // SAFETY: workers cover disjoint row ranges.
1621                        let row = unsafe { std::slice::from_raw_parts_mut(sp.at(r * nk), nk) };
1622                        crate::attention::softmax_row(&mut row[lo..allowed]);
1623                        row[..lo].fill(0.0);
1624                        row[allowed..].fill(0.0);
1625                    }
1626                };
1627                match pool {
1628                    Some(p) if m >= 64 => p.run_rows(m, &run),
1629                    _ => run(0, m),
1630                }
1631                // Attention importance: masked column sums (probs of the
1632                // zeroed tail contribute nothing, same as the CPU
1633                // per-position accumulate).
1634                let ni = self.imp.len().min(n);
1635                for r in 0..m {
1636                    let al = (s0 + (r % b) + 1).min(ni);
1637                    if al > lo0 {
1638                        for (dst, &p) in self.imp[lo0..al]
1639                            .iter_mut()
1640                            .zip(&scores[r * nk..r * nk + (al - lo0)])
1641                        {
1642                            *dst += p;
1643                        }
1644                    }
1645                }
1646                if neon_gemm {
1647                    let sp_s = SendPtr(scores.as_mut_ptr());
1648                    let sp_o = SendPtr(aopanel.as_mut_ptr());
1649                    let run = |start: usize, end: usize| {
1650                        if end > start {
1651                            // SAFETY: workers write disjoint output rows.
1652                            let a = unsafe {
1653                                std::slice::from_raw_parts(sp_s.at(start * nk), (end - start) * nk)
1654                            };
1655                            let c = unsafe {
1656                                std::slice::from_raw_parts_mut(
1657                                    sp_o.at(start * hd),
1658                                    (end - start) * hd,
1659                                )
1660                            };
1661                            crate::qtensor::micro_gemm_rm(
1662                                end - start,
1663                                hd,
1664                                nk,
1665                                1.0,
1666                                a,
1667                                nk,
1668                                vmat,
1669                                hd,
1670                                false,
1671                                c,
1672                                hd,
1673                            );
1674                        }
1675                    };
1676                    match pool {
1677                        Some(p) if m >= 64 => p.run_rows(m, &run),
1678                        _ => run(0, m),
1679                    }
1680                } else {
1681                    crate::qtensor::sgemm_rm(
1682                        m, hd, nk, 1.0, scores, nk, vmat, hd, false, aopanel, hd,
1683                    );
1684                }
1685                for hl in 0..heads_per_kv {
1686                    let hh = g * heads_per_kv + hl;
1687                    for bi in 0..b {
1688                        out[bi * nh * hd + hh * hd..][..hd]
1689                            .copy_from_slice(&aopanel[(hl * b + bi) * hd..(hl * b + bi + 1) * hd]);
1690                    }
1691                }
1692            }
1693        });
1694    }
1695
1696    /// Roll back the last `n_drop` positions (speculative-decode reject).
1697    pub fn truncate_last(&mut self, n_drop: usize) {
1698        self.discard_linear_scratch();
1699        let d = n_drop.min(self.seq_len);
1700        if let Some(b) = self.bounded.as_mut() {
1701            // The ring rolls back through its undo rows; nothing is
1702            // stored per position, so there is nothing else to drop.
1703            let rolled = b.rollback(d);
1704            if rolled < d {
1705                tracing::warn!(
1706                    "bounded anchor: rollback of {d} exceeds the undo depth ({rolled} restored)"
1707                );
1708            }
1709            self.seq_len = b.seen;
1710            return;
1711        }
1712        // A trimmed tail keeps `slack` rows past its window for exactly
1713        // this; a deeper rollback leaves the next query short of rows its
1714        // window reads (no Spark path rolls back more than 2).
1715        if let Some(w) = self.tail
1716            && self.base > 0
1717            && self.seq_len - d < w.saturating_sub(1)
1718        {
1719            tracing::warn!(
1720                "sliding tail: rollback of {d} leaves {} rows under the {w}-row window",
1721                self.seq_len - d
1722            );
1723        }
1724        let mut first_changed = usize::MAX;
1725        for h in 0..self.num_kv_heads {
1726            let keep = self.k[h].len().saturating_sub(d * self.head_dim);
1727            if !self.k[h].is_empty() || !self.v[h].is_empty() {
1728                first_changed = first_changed.min(keep / self.head_dim.max(1));
1729            }
1730            self.k[h].truncate(keep);
1731            self.v[h].truncate(keep);
1732            let ngk = self.head_dim.div_ceil(KV_K_GROUP);
1733            let keep_q = self.kq[h].len().saturating_sub(d * self.head_dim);
1734            self.kq[h].truncate(keep_q);
1735            let keep_vq = self.vq[h].len().saturating_sub(d * self.head_dim);
1736            self.vq[h].truncate(keep_vq);
1737            let keep_ks = self.ks[h].len().saturating_sub(d * ngk);
1738            self.ks[h].truncate(keep_ks);
1739            let keep_vs = self.vs[h].len().saturating_sub(d);
1740            self.vs[h].truncate(keep_vs);
1741        }
1742        self.imp.truncate(self.imp.len().saturating_sub(d));
1743        self.seq_len -= d;
1744        self.kv_rows_changed(first_changed);
1745    }
1746
1747    /// Accumulate attention mass per stored position (summed over heads).
1748    pub fn accumulate_imp(&mut self, probs: &[f32]) {
1749        for (dst, &p) in self.imp.iter_mut().zip(probs) {
1750            *dst += p;
1751        }
1752    }
1753
1754    /// Contiguous keys of one head: `[stored_len × head_dim]`.
1755    pub fn head_keys(&self, kv_head: usize) -> &[f32] {
1756        &self.k[kv_head]
1757    }
1758
1759    pub fn head_values(&self, kv_head: usize) -> &[f32] {
1760        &self.v[kv_head]
1761    }
1762
1763    /// Number of positions actually stored for a head (0 for dead heads).
1764    pub fn head_len(&self, kv_head: usize) -> usize {
1765        let ng = self.head_dim.div_ceil(KV_K_GROUP);
1766        (self.k[kv_head].len() / self.head_dim)
1767            .max(self.ks[kv_head].len() / ng)
1768            .max(self.vs[kv_head].len())
1769    }
1770
1771    /// Clear cache (e.g. on new conversation or task switch).
1772    pub fn clear(&mut self) {
1773        self.kv_rows_changed(0);
1774        for h in 0..self.num_kv_heads {
1775            self.k[h].clear();
1776            self.v[h].clear();
1777            self.kq[h].clear();
1778            self.ks[h].clear();
1779            self.vq[h].clear();
1780            self.vs[h].clear();
1781            self.kcol[h].clear();
1782            self.vcol[h].clear();
1783        }
1784        self.imp.clear();
1785        self.linear_state.clear();
1786        self.discard_linear_scratch();
1787        // Fresh conversation → the pipeline re-arms collection if the
1788        // layer is o1-flagged (landmarks are per-prompt, never reused).
1789        self.o1 = None;
1790        self.o1_error = None;
1791        self.o1_transitioned = false;
1792        // The bounded ring is zeroed in place: its size is a property of
1793        // the file, not of the conversation.
1794        if let Some(b) = self.bounded.as_mut() {
1795            b.clear();
1796        }
1797        self.seq_len = 0;
1798        // `tail` stays: the window is the layer's, not the conversation's.
1799        self.base = 0;
1800        self.generation = next_gen();
1801    }
1802
1803    /// Serialize this layer's state for the wire (versioned, v2): a
1804    /// fixed header `{magic "CMFS", version, operator identity hash64,
1805    /// layer, kind, f16 flag, position}` followed by one record whose
1806    /// shape the kind fixes — per-position K/V (+ importance) for a full
1807    /// layer, the recurrent vector for a linear layer, the insert counter
1808    /// + ring K/V for a bounded anchor. `f16` halves the K/V payloads and
1809    /// is the caller's explicit choice, exactly like the hidden-state
1810    /// wire; recurrent vectors stay f32 whatever the wire dtype (they are
1811    /// the ONLY state a linear layer has — rounding them rounds the whole
1812    /// history).
1813    ///
1814    /// REFUSES rather than travelling half-complete. A cache carrying
1815    /// frozen columns, a Nyström overlay or q8 storage holds state this
1816    /// format does not describe, and shipping the rest would land a
1817    /// plausible-looking cache that answers differently — the failure
1818    /// mode this whole format exists to avoid.
1819    pub fn export_wire(&self, f16: bool) -> Result<Vec<u8>, String> {
1820        if !matches!(self.mode, KvMode::F32) {
1821            return Err("kv export: only the F32 cache is described by this                         format (CMF_KV=q8 stores int8 rows and per-row scales)"
1822                .into());
1823        }
1824        if self.o1.is_some() {
1825            return Err("kv export: an O(1) Nyström overlay is not part of                         this format — the skeletons are irreversible and                         would have to travel with it"
1826                .into());
1827        }
1828        // Frozen columns only exist under q8 storage, which is refused
1829        // above. If one shows up under an F32 cache the format is lying
1830        // about something and the transfer must not proceed.
1831        if self.kcol.iter().any(|c| !c.is_empty()) || self.vcol.iter().any(|c| !c.is_empty()) {
1832            return Err(
1833                "kv export: frozen columns under an F32 cache — refusing to ship \
1834                        a state this format does not describe"
1835                    .into(),
1836            );
1837        }
1838        let mut out = Vec::with_capacity(self.memory_bytes() / if f16 { 2 } else { 1 } + 64);
1839        let u = |v: u32, o: &mut Vec<u8>| o.extend_from_slice(&v.to_le_bytes());
1840        out.extend_from_slice(WIRE_MAGIC);
1841        u(WIRE_VERSION, &mut out);
1842        out.extend_from_slice(&self.wire_identity.to_le_bytes());
1843        u(self.wire_layer, &mut out);
1844        // A trimmed full layer travels as its tail and says where it starts.
1845        let kind = match self.wire_kind {
1846            WireKind::Full if self.base > 0 => WireKind::FullTail,
1847            k => k,
1848        };
1849        out.push(kind as u8);
1850        out.push(u8::from(f16));
1851        out.extend_from_slice(&0u16.to_le_bytes());
1852        // The header carries the absolute depth (== seq_len untrimmed).
1853        out.extend_from_slice(&(self.pos_len() as u64).to_le_bytes());
1854        let push = |xs: &[f32], o: &mut Vec<u8>| {
1855            if f16 {
1856                for &x in xs {
1857                    o.extend_from_slice(&cortiq_core::quant::f32_to_f16(x).to_le_bytes());
1858                }
1859            } else {
1860                for &x in xs {
1861                    o.extend_from_slice(&x.to_le_bytes());
1862                }
1863            }
1864        };
1865        match kind {
1866            WireKind::Full => self.export_full_body(f16, &mut out),
1867            WireKind::FullTail => {
1868                out.extend_from_slice(&(self.base as u64).to_le_bytes());
1869                self.export_full_body(f16, &mut out);
1870            }
1871            WireKind::Linear => {
1872                u(self.num_kv_heads as u32, &mut out);
1873                u(self.head_dim as u32, &mut out);
1874                u(self.linear_state.len() as u32, &mut out);
1875                for &x in &self.linear_state {
1876                    out.extend_from_slice(&x.to_le_bytes());
1877                }
1878            }
1879            WireKind::Bounded => {
1880                let b = self
1881                    .bounded
1882                    .as_ref()
1883                    .ok_or("kv export: bounded wire kind without an installed ring")?;
1884                u(self.num_kv_heads as u32, &mut out);
1885                u(self.head_dim as u32, &mut out);
1886                u(b.window as u32, &mut out);
1887                u(b.len() as u32, &mut out);
1888                u(b.head() as u32, &mut out);
1889                push(&b.ring_k, &mut out);
1890                push(&b.ring_v, &mut out);
1891            }
1892        }
1893        Ok(out)
1894    }
1895
1896    /// The per-position record (the whole legacy wire): f16 flag,
1897    /// seq_len, geometry, recurrent vector, importance, per-head K/V.
1898    fn export_full_body(&self, f16: bool, out: &mut Vec<u8>) {
1899        let u = |v: u32, o: &mut Vec<u8>| o.extend_from_slice(&v.to_le_bytes());
1900        u(u8::from(f16) as u32, out);
1901        u(self.seq_len as u32, out);
1902        u(self.num_kv_heads as u32, out);
1903        u(self.head_dim as u32, out);
1904        u(self.linear_state.len() as u32, out);
1905        // Attention importance is ordinary state: every attention call
1906        // accumulates it and eviction reads it. Leaving it behind would
1907        // hand the far side a cache that forgets the RIGHT positions
1908        // later — a divergence that shows up only under pressure.
1909        u(self.imp.len() as u32, out);
1910        let push = |xs: &[f32], o: &mut Vec<u8>| {
1911            if f16 {
1912                for &x in xs {
1913                    o.extend_from_slice(&cortiq_core::quant::f32_to_f16(x).to_le_bytes());
1914                }
1915            } else {
1916                for &x in xs {
1917                    o.extend_from_slice(&x.to_le_bytes());
1918                }
1919            }
1920        };
1921        for &x in &self.linear_state {
1922            out.extend_from_slice(&x.to_le_bytes());
1923        }
1924        for &x in &self.imp {
1925            out.extend_from_slice(&x.to_le_bytes());
1926        }
1927        for h in 0..self.num_kv_heads {
1928            u(self.k[h].len() as u32, out);
1929            push(&self.k[h], out);
1930            u(self.v[h].len() as u32, out);
1931            push(&self.v[h], out);
1932        }
1933    }
1934
1935    /// Install a peer's state over this layer. The geometry must match the
1936    /// model both sides hold — it is checked, not assumed. Accepts the
1937    /// versioned wire (magic "CMFS") and, for full/linear layers, the old
1938    /// unversioned per-position body.
1939    pub fn import_wire(&mut self, buf: &[u8]) -> Result<(), String> {
1940        if buf.len() >= 4 && &buf[..4] == WIRE_MAGIC {
1941            return self.import_wire_v2(buf);
1942        }
1943        // Legacy (unversioned) wire: no operator tag travels with it.
1944        if !self.linear_wire_allowed {
1945            return Err(
1946                "kv import: Delta linear state cannot use the unversioned cache wire; refusing until the wire carries operator identity".into(),
1947            );
1948        }
1949        if self.bounded.is_some() {
1950            return Err(
1951                "kv import: a bounded anchor takes only the versioned wire (v2) — the \
1952                 unversioned body has no ring record"
1953                    .into(),
1954            );
1955        }
1956        let n = self.import_full_body(buf)?;
1957        if n != buf.len() {
1958            return Err(format!(
1959                "kv import: {} trailing byte(s) after the record",
1960                buf.len() - n
1961            ));
1962        }
1963        Ok(())
1964    }
1965
1966    fn import_wire_v2(&mut self, buf: &[u8]) -> Result<(), String> {
1967        let need = |n: usize, o: usize| -> Result<(), String> {
1968            if o + n > buf.len() {
1969                Err("kv import: truncated header".into())
1970            } else {
1971                Ok(())
1972            }
1973        };
1974        need(28, 0)?;
1975        let version = u32::from_le_bytes(buf[4..8].try_into().unwrap());
1976        if version != WIRE_VERSION {
1977            return Err(format!(
1978                "kv import: wire version {version}, this runtime speaks {WIRE_VERSION}"
1979            ));
1980        }
1981        let identity = u64::from_le_bytes(buf[8..16].try_into().unwrap());
1982        let layer = u32::from_le_bytes(buf[16..20].try_into().unwrap());
1983        let kind = WireKind::from_u8(buf[20])
1984            .ok_or_else(|| format!("kv import: unknown state kind {}", buf[20]))?;
1985        let f16 = buf[21] != 0;
1986        let position = u64::from_le_bytes(buf[24..32].try_into().unwrap()) as usize;
1987        if identity != self.wire_identity {
1988            return Err(format!(
1989                "kv import: peer operator identity {identity:016x} != mine {:016x} — \
1990                 the two sides do not hold the same operator",
1991                self.wire_identity
1992            ));
1993        }
1994        if layer != self.wire_layer {
1995            return Err(format!(
1996                "kv import: record is for layer {layer}, this is layer {}",
1997                self.wire_layer
1998            ));
1999        }
2000        // A full layer takes its trimmed tail too.
2001        let fits = kind == self.wire_kind
2002            || (kind == WireKind::FullTail && self.wire_kind == WireKind::Full);
2003        if !fits {
2004            return Err(format!(
2005                "kv import: record kind {kind:?} does not match this layer's {:?}",
2006                self.wire_kind
2007            ));
2008        }
2009        let mut o = 32usize;
2010        let u32_at = |o: &mut usize| -> Result<u32, String> {
2011            if *o + 4 > buf.len() {
2012                return Err("kv import: truncated record".into());
2013            }
2014            let v = u32::from_le_bytes(buf[*o..*o + 4].try_into().unwrap());
2015            *o += 4;
2016            Ok(v)
2017        };
2018        let need_payload = |n: usize, o: usize| -> Result<(), String> {
2019            if o + n > buf.len() {
2020                Err("kv import: truncated payload".into())
2021            } else {
2022                Ok(())
2023            }
2024        };
2025        match kind {
2026            WireKind::Full => {
2027                let n = self.import_full_body(&buf[o..])?;
2028                o += n;
2029                if self.seq_len != position {
2030                    return Err(format!(
2031                        "kv import: header position {position} != record seq_len {}",
2032                        self.seq_len
2033                    ));
2034                }
2035            }
2036            WireKind::FullTail => {
2037                need_payload(8, o)?;
2038                let base = u64::from_le_bytes(buf[o..o + 8].try_into().unwrap()) as usize;
2039                o += 8;
2040                let n = self.import_full_body(&buf[o..])?;
2041                o += n;
2042                if base.checked_add(self.seq_len) != Some(position) {
2043                    let rows = self.seq_len;
2044                    self.reset_per_position_storage();
2045                    self.seq_len = 0;
2046                    return Err(format!(
2047                        "kv import: tail base {base} + {rows} rows != header position {position}"
2048                    ));
2049                }
2050                self.base = base;
2051            }
2052            WireKind::Linear => {
2053                let heads = u32_at(&mut o)? as usize;
2054                let hd = u32_at(&mut o)? as usize;
2055                if heads != self.num_kv_heads || hd != self.head_dim {
2056                    return Err(format!(
2057                        "kv import: peer sent {heads}×{hd} per position, this layer is {}×{}",
2058                        self.num_kv_heads, self.head_dim
2059                    ));
2060                }
2061                let lin = u32_at(&mut o)? as usize;
2062                need_payload(lin * 4, o)?;
2063                self.linear_state = (0..lin)
2064                    .map(|i| f32::from_le_bytes(buf[o + i * 4..o + i * 4 + 4].try_into().unwrap()))
2065                    .collect();
2066                o += lin * 4;
2067                self.reset_per_position_storage();
2068                self.seq_len = position;
2069            }
2070            WireKind::Bounded => {
2071                let heads = u32_at(&mut o)? as usize;
2072                let hd = u32_at(&mut o)? as usize;
2073                let window = u32_at(&mut o)? as usize;
2074                let len = u32_at(&mut o)? as usize;
2075                let head = u32_at(&mut o)? as usize;
2076                let b = self
2077                    .bounded
2078                    .as_mut()
2079                    .ok_or("kv import: bounded record for a layer without a ring")?;
2080                if heads != b.num_kv_heads || hd != b.head_dim || window != b.window {
2081                    return Err(format!(
2082                        "kv import: bounded record {heads}×{window}×{hd} does not fit this \
2083                         layer's ring {}×{}×{}",
2084                        b.num_kv_heads, b.window, b.head_dim
2085                    ));
2086                }
2087                if len != position.min(window) || head != position % window {
2088                    return Err(format!(
2089                        "kv import: bounded record len/head {len}/{head} inconsistent with \
2090                         position {position} (window {window})"
2091                    ));
2092                }
2093                let n = b.ring_k.len();
2094                let w = if f16 { 2 } else { 4 };
2095                need_payload(2 * n * w, o)?;
2096                let read = |o: usize, dst: &mut [f32]| {
2097                    for (i, d) in dst.iter_mut().enumerate() {
2098                        let at = o + i * w;
2099                        *d = if f16 {
2100                            cortiq_core::quant::f16_to_f32(u16::from_le_bytes(
2101                                buf[at..at + 2].try_into().unwrap(),
2102                            ))
2103                        } else {
2104                            f32::from_le_bytes(buf[at..at + 4].try_into().unwrap())
2105                        };
2106                    }
2107                };
2108                read(o, &mut b.ring_k);
2109                o += n * w;
2110                read(o, &mut b.ring_v);
2111                o += n * w;
2112                b.seen = position;
2113                // The undo rows describe inserts this side never made.
2114                let snap = b.snapshot();
2115                b.restore(&snap);
2116                self.linear_state = Vec::new();
2117                self.reset_per_position_storage();
2118                self.seq_len = position;
2119            }
2120        }
2121        if o != buf.len() {
2122            return Err(format!(
2123                "kv import: {} trailing byte(s) after the record",
2124                buf.len() - o
2125            ));
2126        }
2127        Ok(())
2128    }
2129
2130    /// Empty every per-position store (keeps the recurrent vector and
2131    /// the bounded ring untouched).
2132    fn reset_per_position_storage(&mut self) {
2133        let heads = self.num_kv_heads;
2134        self.mode = KvMode::F32;
2135        self.k = vec![Vec::new(); heads];
2136        self.v = vec![Vec::new(); heads];
2137        self.kv_rows_changed(0);
2138        self.kq = vec![Vec::new(); heads];
2139        self.ks = vec![Vec::new(); heads];
2140        self.vq = vec![Vec::new(); heads];
2141        self.vs = vec![Vec::new(); heads];
2142        self.kcol = vec![Vec::new(); heads];
2143        self.vcol = vec![Vec::new(); heads];
2144        self.imp = Vec::new();
2145        self.discard_linear_scratch();
2146        self.o1 = None;
2147        self.o1_error = None;
2148        self.o1_transitioned = false;
2149        self.base = 0;
2150        self.generation = next_gen();
2151    }
2152
2153    /// Parse the per-position record (legacy wire body); returns the
2154    /// bytes consumed.
2155    fn import_full_body(&mut self, buf: &[u8]) -> Result<usize, String> {
2156        let mut o = 0usize;
2157        let u32_at = |o: &mut usize| -> Result<u32, String> {
2158            if *o + 4 > buf.len() {
2159                return Err("kv import: truncated header".into());
2160            }
2161            let v = u32::from_le_bytes(buf[*o..*o + 4].try_into().unwrap());
2162            *o += 4;
2163            Ok(v)
2164        };
2165        let f16 = u32_at(&mut o)? != 0;
2166        let seq_len = u32_at(&mut o)? as usize;
2167        let heads = u32_at(&mut o)? as usize;
2168        let hd = u32_at(&mut o)? as usize;
2169        let lin = u32_at(&mut o)? as usize;
2170        let nimp = u32_at(&mut o)? as usize;
2171        if heads != self.num_kv_heads || hd != self.head_dim {
2172            return Err(format!(
2173                "kv import: peer sent {heads}×{hd} per position, this layer is {}×{}",
2174                self.num_kv_heads, self.head_dim
2175            ));
2176        }
2177        let w = if f16 { 2 } else { 4 };
2178        let need = |n: usize, o: usize| -> Result<(), String> {
2179            if o + n > buf.len() {
2180                Err("kv import: truncated payload".into())
2181            } else {
2182                Ok(())
2183            }
2184        };
2185        need(lin * 4, o)?;
2186        self.linear_state = (0..lin)
2187            .map(|i| f32::from_le_bytes(buf[o + i * 4..o + i * 4 + 4].try_into().unwrap()))
2188            .collect();
2189        o += lin * 4;
2190        need(nimp * 4, o)?;
2191        let imp: Vec<f32> = (0..nimp)
2192            .map(|i| f32::from_le_bytes(buf[o + i * 4..o + i * 4 + 4].try_into().unwrap()))
2193            .collect();
2194        o += nimp * 4;
2195        let mut k: Vec<Vec<f32>> = Vec::with_capacity(heads);
2196        let mut v: Vec<Vec<f32>> = Vec::with_capacity(heads);
2197        for _ in 0..heads {
2198            for which in 0..2 {
2199                let n = u32_at(&mut o)? as usize;
2200                need(n * w, o)?;
2201                let xs: Vec<f32> = (0..n)
2202                    .map(|i| {
2203                        let at = o + i * w;
2204                        if f16 {
2205                            cortiq_core::quant::f16_to_f32(u16::from_le_bytes(
2206                                buf[at..at + 2].try_into().unwrap(),
2207                            ))
2208                        } else {
2209                            f32::from_le_bytes(buf[at..at + 4].try_into().unwrap())
2210                        }
2211                    })
2212                    .collect();
2213                o += n * w;
2214                if which == 0 { k.push(xs) } else { v.push(xs) }
2215            }
2216        }
2217        self.reset_per_position_storage();
2218        self.k = k;
2219        self.v = v;
2220        self.imp = imp;
2221        self.seq_len = seq_len;
2222        Ok(o)
2223    }
2224
2225    pub fn memory_bytes(&self) -> usize {
2226        let floats: usize = self.k.iter().map(Vec::len).sum::<usize>()
2227            + self.v.iter().map(Vec::len).sum::<usize>()
2228            + self.ks.iter().map(Vec::len).sum::<usize>()
2229            + self.vs.iter().map(Vec::len).sum::<usize>()
2230            + self.kcol.iter().map(Vec::len).sum::<usize>()
2231            + self.vcol.iter().map(Vec::len).sum::<usize>();
2232        let bytes: usize = self.kq.iter().map(Vec::len).sum::<usize>()
2233            + self.vq.iter().map(Vec::len).sum::<usize>();
2234        floats * std::mem::size_of::<f32>()
2235            + bytes
2236            // O(1) recurrent state of linear-core layers (vmf_phase/GDN):
2237            // constant in context, but real memory — the honest "KV+state"
2238            // line must count it (a pure-linear model reported 0 before).
2239            + self.linear_state.len() * std::mem::size_of::<f32>()
2240            // O(1) Nyström state (window + sinks + skeleton) — same
2241            // discipline: constant in context, but real memory.
2242            + self.o1_memory_bytes()
2243            // Natively bounded anchor: the ring is the whole state of the
2244            // layer, fixed by the header.
2245            + self.bounded_state_bytes()
2246    }
2247
2248    /// Drop oldest positions, keeping the last `keep_last`.
2249    fn evict(&mut self, keep_last: usize) {
2250        // A collecting o1 layer owns a still-needed exact prefix and query
2251        // trace.  Evicting it would lower the effective seal boundary while
2252        // leaving q_buf untouched, so conversion could never match its KV
2253        // rows.  A sealed layer stores nothing per position — the Nyström
2254        // state IS the eviction policy; resetting seq_len here would lie
2255        // about the context depth.  Both states therefore bypass ordinary
2256        // eviction until the transition or explicit reset completes.  A
2257        // bounded anchor likewise: the ring evicts itself every token.
2258        // A sliding-window tail (`trim_window`) bounds itself and must stay
2259        // the exact contiguous window its attends index.
2260        if self.o1.is_some()
2261            || self.bounded.is_some()
2262            || self.tail.is_some()
2263            || self.seq_len <= keep_last
2264        {
2265            return;
2266        }
2267        self.generation = next_gen();
2268        let drop = self.seq_len - keep_last;
2269        for h in 0..self.num_kv_heads {
2270            // Dead heads store fewer positions; drop proportionally.
2271            let stored = self.head_len(h);
2272            let d = drop.min(stored);
2273            let hd = self.head_dim;
2274            fn drop_front<T>(v: &mut Vec<T>, n: usize) {
2275                let n = n.min(v.len());
2276                v.drain(..n);
2277            }
2278            drop_front(&mut self.k[h], d * hd);
2279            drop_front(&mut self.v[h], d * hd);
2280            self.kv_rows_changed(0);
2281            drop_front(&mut self.kq[h], d * hd);
2282            drop_front(&mut self.vq[h], d * hd);
2283            drop_front(&mut self.ks[h], d * hd.div_ceil(KV_K_GROUP));
2284            drop_front(&mut self.vs[h], d);
2285        }
2286        let d = drop.min(self.imp.len());
2287        self.imp.drain(..d);
2288        self.seq_len = keep_last;
2289    }
2290
2291    /// Mass-based eviction: keep `sink` earliest positions (attention sinks),
2292    /// the `recent` latest, and fill the rest of the `keep_last` budget
2293    /// with the positions carrying the highest accumulated attention
2294    /// mass (vmfcore: PPL 8.342 vs 8.687 for recency-only, full 8.295).
2295    fn evict_born(&mut self, keep_last: usize, sink: usize, recent: usize) {
2296        if self.o1.is_some() || self.bounded.is_some() || self.tail.is_some() {
2297            // See evict(): collecting must retain the exact prefix as well as
2298            // sealed O(1) state must retain its own bounded representation;
2299            // a bounded anchor's ring is its own eviction, and a sliding
2300            // tail its own (a gather would make its rows non-contiguous).
2301            return;
2302        }
2303        let stored = self.imp.len();
2304        if stored <= keep_last {
2305            return;
2306        }
2307        self.generation = next_gen();
2308        // Budget discipline: sinks first, recents next, both clamped so
2309        // the total never exceeds keep_last.
2310        let sink_n = sink.min(keep_last);
2311        let recent_n = recent.min(keep_last - sink_n);
2312        let mut keep = vec![false; stored];
2313        for k in keep.iter_mut().take(sink_n) {
2314            *k = true;
2315        }
2316        for k in keep.iter_mut().skip(stored.saturating_sub(recent_n)) {
2317            *k = true;
2318        }
2319        let mut budget = keep_last.saturating_sub(keep.iter().filter(|&&x| x).count());
2320        // Highest accumulated mass first among the middle positions.
2321        let mut order: Vec<usize> = (0..stored).filter(|&i| !keep[i]).collect();
2322        order.sort_by(|&a, &b| {
2323            self.imp[b]
2324                .partial_cmp(&self.imp[a])
2325                .unwrap_or(std::cmp::Ordering::Equal)
2326        });
2327        for i in order {
2328            if budget == 0 {
2329                break;
2330            }
2331            keep[i] = true;
2332            budget -= 1;
2333        }
2334
2335        let kept: Vec<usize> = (0..stored).filter(|&i| keep[i]).collect();
2336        let hd = self.head_dim;
2337        fn gather<T: Copy>(src: &[T], kept: &[usize], step: usize) -> Vec<T> {
2338            let mut out = Vec::with_capacity(kept.len() * step);
2339            for &i in kept {
2340                out.extend_from_slice(&src[i * step..(i + 1) * step]);
2341            }
2342            out
2343        }
2344        // Each storage is gathered INDEPENDENTLY: in mixed modes
2345        // (q8k/q8v) K and V live in different storages — the paired branch
2346        // panicked (q8v) or silently left V uncompressed (q8k);
2347        // found by adversarial review, closed by regression tests.
2348        for h in 0..self.num_kv_heads {
2349            if !self.k[h].is_empty() {
2350                self.k[h] = gather(&self.k[h], &kept, hd);
2351                self.kv_rows_changed(0);
2352            }
2353            if !self.v[h].is_empty() {
2354                self.v[h] = gather(&self.v[h], &kept, hd);
2355                self.kv_rows_changed(0);
2356            }
2357            if !self.kq[h].is_empty() {
2358                self.kq[h] = gather(&self.kq[h], &kept, hd);
2359                self.ks[h] = gather(&self.ks[h], &kept, hd.div_ceil(KV_K_GROUP));
2360            }
2361            if !self.vq[h].is_empty() {
2362                self.vq[h] = gather(&self.vq[h], &kept, hd);
2363                self.vs[h] = gather(&self.vs[h], &kept, 1);
2364            }
2365        }
2366        self.imp = kept.iter().map(|&i| self.imp[i]).collect();
2367        self.seq_len = kept.len();
2368    }
2369}
2370
2371/// Eviction policy for a bounded cache.
2372#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2373pub enum EvictionPolicy {
2374    /// Sliding window: keep only the most recent positions.
2375    Recent,
2376    /// Mass-based eviction: sinks + recents + top accumulated attention mass.
2377    Born { sink: usize },
2378}
2379
2380/// Full KV cache for all layers.
2381#[derive(Debug)]
2382pub struct KvCache {
2383    pub layers: Vec<LayerKvCache>,
2384    pub max_seq_len: usize,
2385    pub policy: EvictionPolicy,
2386}
2387
2388impl KvCache {
2389    pub fn new(
2390        num_layers: usize,
2391        num_kv_heads: usize,
2392        head_dim: usize,
2393        max_seq_len: usize,
2394    ) -> Self {
2395        let layers = (0..num_layers)
2396            .map(|li| {
2397                let mut l = LayerKvCache::new(num_kv_heads, head_dim);
2398                l.wire_layer = li as u32;
2399                l
2400            })
2401            .collect();
2402        Self {
2403            layers,
2404            max_seq_len,
2405            policy: EvictionPolicy::Born { sink: 4 },
2406        }
2407    }
2408
2409    pub fn clear(&mut self) {
2410        for layer in &mut self.layers {
2411            layer.clear();
2412        }
2413    }
2414
2415    pub fn total_memory_bytes(&self) -> usize {
2416        self.layers.iter().map(|l| l.memory_bytes()).sum()
2417    }
2418
2419    /// Bytes owned by linear-core recurrent state (including the tentative
2420    /// speculative scratch).  This is reported separately from attention KV
2421    /// so a serving slot's O(1) capacity can be compared with its context
2422    /// cache without guessing from model geometry.
2423    pub fn recurrent_state_bytes(&self) -> usize {
2424        let floats: usize = self
2425            .layers
2426            .iter()
2427            .map(|l| l.linear_state.len() + l.linear_scratch.len())
2428            .sum();
2429        floats * std::mem::size_of::<f32>()
2430    }
2431
2432    /// Attention KV (or sealed O(1) attention state) bytes, excluding the
2433    /// linear recurrent vectors returned by [`recurrent_state_bytes`].
2434    pub fn attention_state_bytes(&self) -> usize {
2435        self.total_memory_bytes()
2436            .saturating_sub(self.recurrent_state_bytes())
2437    }
2438
2439    /// Current sequence length (max across layers — dead layers may lag):
2440    /// the absolute depth, which a trimmed sliding layer keeps in
2441    /// `pos_len()` while it stores only its tail.
2442    pub fn seq_len(&self) -> usize {
2443        self.layers.iter().map(|l| l.pos_len()).max().unwrap_or(0)
2444    }
2445
2446    /// Bytes owned by bounded-anchor rings (constant in context).
2447    pub fn bounded_state_bytes(&self) -> usize {
2448        self.layers.iter().map(|l| l.bounded_state_bytes()).sum()
2449    }
2450
2451    /// True when some layer with PER-POSITION storage reached the cap.
2452    /// Bounded anchors hold nothing per position and never need it: a
2453    /// model whose every layer is O(1) has no eviction cliff at all. A
2454    /// sliding tail (`trim_window`) bounds itself and is not counted.
2455    pub fn needs_eviction(&self) -> bool {
2456        self.layers
2457            .iter()
2458            .filter(|l| l.bounded.is_none() && l.tail.is_none())
2459            .map(|l| l.seq_len)
2460            .max()
2461            .unwrap_or(0)
2462            >= self.max_seq_len
2463    }
2464
2465    /// Evict down to `keep_last` positions according to the policy.
2466    pub fn evict(&mut self, keep_last: usize) {
2467        match self.policy {
2468            EvictionPolicy::Recent => {
2469                for layer in &mut self.layers {
2470                    layer.evict(keep_last);
2471                }
2472            }
2473            EvictionPolicy::Born { sink } => {
2474                let recent = (keep_last / 2).max(1);
2475                for layer in &mut self.layers {
2476                    layer.evict_born(keep_last, sink, recent);
2477                }
2478            }
2479        }
2480    }
2481}
2482
2483#[cfg(test)]
2484mod tests {
2485    use super::*;
2486
2487    /// The lazy K/V maximum the wgpu prefill's f16 guard reads: rows a
2488    /// truncation drops and later appends rewrite are scanned again, an
2489    /// inf or NaN reports +inf, `clear` starts afresh, and a caller's own
2490    /// maxima of the rows it just appended fold in only when they continue
2491    /// the scanned prefix.
2492    #[test]
2493    fn kv_abs_max_follows_appends_truncation_and_clear() {
2494        let (heads, hd) = (2usize, 4usize);
2495        let mut c = LayerKvCache::new(heads, hd);
2496        c.mode = KvMode::F32;
2497        let row = |x: f32| vec![x; heads * hd];
2498        for _ in 0..4 {
2499            c.append(&row(0.5), &row(-0.25), &[]);
2500        }
2501        assert_eq!(c.kv_abs_max(), (0.5, 0.25));
2502        c.truncate_last(2);
2503        c.append(&row(-9.0), &row(3.0), &[]);
2504        c.append(&row(0.5), &row(0.25), &[]);
2505        assert_eq!(c.kv_abs_max(), (9.0, 3.0), "rewritten rows were not rescanned");
2506        c.append(&row(f32::NAN), &row(0.0), &[]);
2507        assert_eq!(c.kv_abs_max().0, f32::INFINITY);
2508        c.clear();
2509        c.append(&row(1.0), &row(2.0), &[]);
2510        assert_eq!(c.kv_abs_max(), (1.0, 2.0));
2511        c.append(&row(4.0), &row(1.0), &[]);
2512        // Rows [1, 2) continue the scanned prefix: the caller's maxima.
2513        assert_eq!(c.kv_abs_max_after((1, 2, 4.0, 1.0)), (4.0, 2.0));
2514        c.append(&row(8.0), &row(1.0), &[]);
2515        // A stale `from` falls back to the scan.
2516        assert_eq!(c.kv_abs_max_after((0, 3, 0.0, 0.0)), (8.0, 2.0));
2517    }
2518
2519    /// A trim drops rows from the front: `kv_abs_max` must still scan every
2520    /// row appended after it, also when more are appended than were dropped
2521    /// before the next call.
2522    #[test]
2523    fn kv_abs_max_scans_rows_appended_after_a_trim() {
2524        let (heads, hd, w) = (1usize, 4usize, 8usize);
2525        let mut c = LayerKvCache::new(heads, hd);
2526        c.mode = KvMode::F32;
2527        let row = |x: f32| vec![x; heads * hd];
2528        for _ in 0..(2 * w + 4) {
2529            c.append(&row(0.5), &row(0.5), &[]);
2530        }
2531        assert_eq!(c.kv_abs_max(), (0.5, 0.5));
2532        let d = c.trim_window(w, 2, 2);
2533        assert!(d > 0, "the trim must drop rows for this test");
2534        // The first row after the trim is large, then more rows than were
2535        // dropped: without moving the mark, rows [stored - d, old mark) —
2536        // the large one among them — would go unscanned.
2537        let first_new = c.kv_rows();
2538        c.append(&row(-7.0), &row(6.0), &[]);
2539        for _ in 0..(d + 3) {
2540            c.append(&row(0.5), &row(0.5), &[]);
2541        }
2542        assert!(c.kv_rows() > 2 * w + 4 && first_new < 2 * w + 4);
2543        assert_eq!(c.kv_abs_max(), (7.0, 6.0));
2544    }
2545
2546    #[test]
2547    fn memory_breakdown_separates_recurrent_and_attention_state() {
2548        let mut cache = KvCache::new(1, 1, 4, 16);
2549        cache.layers[0].linear_state = vec![0.0; 8];
2550        cache.layers[0].linear_scratch = vec![0.0; 4];
2551        cache.layers[0].append(&[0.0; 4], &[1.0; 4], &[true]);
2552        let recurrent = cache.recurrent_state_bytes();
2553        assert_eq!(recurrent, 12 * std::mem::size_of::<f32>());
2554        assert_eq!(
2555            cache.attention_state_bytes() + recurrent,
2556            cache.total_memory_bytes()
2557        );
2558        assert!(cache.attention_state_bytes() > 0);
2559    }
2560
2561    #[test]
2562    fn wire_round_trip_reproduces_attention() {
2563        // The state has to arrive as state, not as something that looks
2564        // like it: the oracle is what the layer ANSWERS, not what it
2565        // stores. Same query, same output, bit for bit.
2566        let (heads, hd) = (2usize, 4usize);
2567        let mut a = LayerKvCache::new(heads, hd);
2568        for p in 0..5 {
2569            let k: Vec<f32> = (0..heads * hd)
2570                .map(|i| (p * 10 + i) as f32 * 0.031)
2571                .collect();
2572            let v: Vec<f32> = (0..heads * hd)
2573                .map(|i| (p * 7 + i) as f32 * -0.017)
2574                .collect();
2575            a.append(&k, &v, &[true, true]);
2576        }
2577        a.linear_state = vec![0.5, -0.25, 1.0];
2578        let q: Vec<f32> = (0..hd).map(|i| 0.1 * (i as f32 + 1.0)).collect();
2579
2580        let bytes = a.export_wire(false).expect("f32 cache exports");
2581        let mut b = LayerKvCache::new(heads, hd);
2582        b.linear_scratch = vec![9.0; 3];
2583        b.import_wire(&bytes).expect("import");
2584        assert!(
2585            b.linear_scratch.is_empty(),
2586            "import must discard tentative state"
2587        );
2588
2589        assert_eq!(b.seq_len, a.seq_len);
2590        assert_eq!(b.linear_state, a.linear_state);
2591        for h in 0..heads {
2592            let (oa, sa) = a.attend(&q, h);
2593            let (ob, sb) = b.attend(&q, h);
2594            assert_eq!(oa, ob, "head {h} attention output diverged");
2595            assert_eq!(sa, sb, "head {h} attention scores diverged");
2596        }
2597    }
2598
2599    #[test]
2600    fn wire_refuses_what_it_cannot_describe() {
2601        // A refusal is the feature: a cache whose extra state this format
2602        // does not carry must not travel looking complete.
2603        let mut c = LayerKvCache::new(1, 4);
2604        c.mode = KvMode::Q8 { k: true, v: true };
2605        let err = c.export_wire(false).unwrap_err();
2606        assert!(err.contains("F32"), "{err}");
2607    }
2608
2609    #[test]
2610    fn wire_refuses_unversioned_delta_state() {
2611        // The versioned wire carries the operator identity, so a delta
2612        // layer exports; only the OLD unversioned body is refused.
2613        let mut c = LayerKvCache::new(1, 4);
2614        c.set_linear_wire_allowed(false);
2615        let bytes = c.export_wire(false).expect("v2 export carries identity");
2616        assert_eq!(&bytes[..4], WIRE_MAGIC);
2617        let err = c.import_wire(&[0, 0, 0, 0]).unwrap_err();
2618        assert!(err.contains("operator identity"), "{err}");
2619    }
2620
2621    #[test]
2622    fn wire_v2_round_trips_linear_and_bounded_records() {
2623        // Linear record: the recurrent vector travels f32 whatever the
2624        // wire dtype and the per-position stores come back empty.
2625        let mut a = LayerKvCache::new(1, 4);
2626        a.wire_kind = WireKind::Linear;
2627        a.wire_identity = 0xC0FFEE;
2628        a.linear_state = vec![0.5, -0.25, 1.0, 3.5];
2629        a.seq_len = 9;
2630        let bytes = a.export_wire(true).unwrap();
2631        let mut b = LayerKvCache::new(1, 4);
2632        b.wire_kind = WireKind::Linear;
2633        b.wire_identity = 0xC0FFEE;
2634        b.import_wire(&bytes).unwrap();
2635        assert_eq!(b.linear_state, a.linear_state);
2636        assert_eq!(b.seq_len, 9);
2637        // Identity mismatch is a refusal, not a warning.
2638        let mut c = LayerKvCache::new(1, 4);
2639        c.wire_kind = WireKind::Linear;
2640        let err = c.import_wire(&bytes).unwrap_err();
2641        assert!(err.contains("operator identity"), "{err}");
2642
2643        // Bounded record, f32 and f16 rings.
2644        let (kvh, hd, w) = (2, 4, 8);
2645        let mut a = LayerKvCache::new(kvh, hd);
2646        a.install_bounded(w);
2647        let k: Vec<f32> = (0..kvh * hd).map(|i| i as f32 * 0.125).collect();
2648        for p in 0..11 {
2649            a.bounded.as_mut().unwrap().insert(&k, &k);
2650            a.seq_len = p + 1;
2651        }
2652        for f16 in [false, true] {
2653            let bytes = a.export_wire(f16).unwrap();
2654            let mut b = LayerKvCache::new(kvh, hd);
2655            b.install_bounded(w);
2656            b.import_wire(&bytes).unwrap();
2657            let (ra, rb) = (a.bounded.as_ref().unwrap(), b.bounded.as_ref().unwrap());
2658            assert_eq!(rb.seen, 11);
2659            assert_eq!(b.seq_len, 11);
2660            // 0.125 multiples are exact in f16, so both dtypes round-trip bit for bit.
2661            assert!(ra.same_state(rb), "f16={f16}");
2662            // A ring of another width refuses the record.
2663            let mut c = LayerKvCache::new(kvh, hd);
2664            c.install_bounded(w * 2);
2665            assert!(c.import_wire(&bytes).is_err());
2666        }
2667    }
2668
2669    #[test]
2670    fn wire_import_checks_geometry() {
2671        let a = LayerKvCache::new(2, 4);
2672        let bytes = a.export_wire(false).unwrap();
2673        let mut wrong = LayerKvCache::new(2, 8);
2674        let err = wrong.import_wire(&bytes).unwrap_err();
2675        assert!(err.contains("2×4"), "{err}");
2676    }
2677
2678    #[test]
2679    fn append_tracks_seq_len_and_layout() {
2680        let mut cache = LayerKvCache::new(4, 8);
2681        cache.mode = KvMode::F32;
2682        assert_eq!(cache.seq_len, 0);
2683
2684        let k: Vec<f32> = (0..32).map(|i| i as f32).collect();
2685        let v = vec![2.0f32; 32];
2686        cache.append(&k, &v, &[true; 4]);
2687
2688        assert_eq!(cache.seq_len, 1);
2689        assert_eq!(cache.head_len(0), 1);
2690        // head 1 slice is contiguous and equals its part of k_new
2691        assert_eq!(cache.head_keys(1), &k[8..16]);
2692        assert_eq!(cache.memory_bytes(), 256);
2693    }
2694
2695    #[test]
2696    fn dead_head_stores_nothing() {
2697        let mut cache = LayerKvCache::new(2, 4);
2698        cache.mode = KvMode::F32;
2699        let k = vec![1.0f32; 8];
2700        let v = vec![2.0f32; 8];
2701        cache.append(&k, &v, &[true, false]);
2702        cache.append(&k, &v, &[true, false]);
2703
2704        assert_eq!(cache.seq_len, 2);
2705        assert_eq!(cache.head_len(0), 2);
2706        assert_eq!(cache.head_len(1), 0, "dead head must not store KV");
2707        assert_eq!(cache.memory_bytes(), 2 * 2 * 4 * 4);
2708    }
2709
2710    #[test]
2711    fn eviction_keeps_recent() {
2712        let mut cache = KvCache::new(2, 4, 8, 10);
2713        cache.policy = EvictionPolicy::Recent;
2714        for l in &mut cache.layers {
2715            l.mode = KvMode::F32;
2716        }
2717        let k = vec![1.0f32; 32];
2718        let v = vec![2.0f32; 32];
2719        for _ in 0..8 {
2720            for layer in &mut cache.layers {
2721                layer.append(&k, &v, &[true; 4]);
2722            }
2723        }
2724        assert_eq!(cache.seq_len(), 8);
2725        assert!(!cache.needs_eviction());
2726
2727        cache.evict(4);
2728        assert_eq!(cache.seq_len(), 4);
2729        assert_eq!(cache.layers[0].head_len(0), 4);
2730    }
2731
2732    /// A batch that ends exactly on the deferred boundary holds only exact
2733    /// rows and must stay on the batched path; only a batch that would
2734    /// append a row past the boundary is split. (Walking the boundary chunk
2735    /// row by row sent the 27B's last prefix rows down another projection
2736    /// route than every other prefix row and flipped its O(1) generation.)
2737    #[test]
2738    fn o1_batch_ending_on_boundary_does_not_cross_it() {
2739        const B: usize = 19;
2740        let mut layer = KvCache::new(1, 1, 4, 64).layers.remove(0);
2741        layer.mode = KvMode::F32;
2742        layer.o1_begin_with_boundary(4, 8, 2, crate::nystrom::O1Rect::Aggregate, Some(B));
2743        for _ in 0..B - 3 {
2744            layer.o1_push_q(&[0.1; 8]);
2745            layer.append(&[0.2; 4], &[0.3; 4], &[]);
2746        }
2747        assert!(!layer.o1_boundary_crossed_by(1));
2748        assert!(
2749            !layer.o1_boundary_crossed_by(3),
2750            "ends on B: exact rows only"
2751        );
2752        assert!(layer.o1_boundary_crossed_by(4), "row B+1 must not be exact");
2753        for _ in 0..3 {
2754            layer.o1_push_q(&[0.1; 8]);
2755            layer.append(&[0.2; 4], &[0.3; 4], &[]);
2756        }
2757        assert!(layer.o1_boundary_crossed_by(1), "at B every row is past it");
2758    }
2759
2760    #[test]
2761    fn collecting_o1_eviction_retains_exact_storage_until_boundary() {
2762        const B: usize = 19;
2763        let q = vec![0.1f32; 8];
2764        let k = vec![0.2f32; 4];
2765        let v = vec![0.3f32; 4];
2766
2767        for policy in [EvictionPolicy::Recent, EvictionPolicy::Born { sink: 2 }] {
2768            let mut cache = KvCache::new(1, 1, 4, 6);
2769            cache.policy = policy;
2770            cache.layers[0].mode = KvMode::F32;
2771            cache.layers[0].o1_begin_with_boundary(
2772                4,
2773                8,
2774                2,
2775                crate::nystrom::O1Rect::Aggregate,
2776                Some(B),
2777            );
2778
2779            for pos in 0..B {
2780                {
2781                    let layer = &mut cache.layers[0];
2782                    layer.o1_push_q(&q);
2783                    layer.append(&k, &v, &[]);
2784                }
2785                if pos + 1 < B {
2786                    cache.evict(3);
2787                }
2788            }
2789
2790            let layer = &cache.layers[0];
2791            let rows = B * layer.head_dim;
2792            assert_eq!(layer.seq_len, B, "policy {policy:?} retained depth");
2793            assert_eq!(layer.k[0].len(), rows, "policy {policy:?} K rows");
2794            assert_eq!(layer.v[0].len(), rows, "policy {policy:?} V rows");
2795            assert!(
2796                layer.k[0].capacity() >= rows,
2797                "policy {policy:?} K capacity"
2798            );
2799            assert!(
2800                layer.v[0].capacity() >= rows,
2801                "policy {policy:?} V capacity"
2802            );
2803            let q_capacity = match layer.o1.as_ref() {
2804                Some(O1State::Collecting { q_buf, .. }) => q_buf.capacity(),
2805                other => panic!("policy {policy:?} changed state early: {other:?}"),
2806            };
2807            assert!(
2808                q_capacity >= B * 8,
2809                "policy {policy:?} Q capacity must cover the exact prefix"
2810            );
2811
2812            assert!(cache.layers[0].o1_seal_checked(2).unwrap());
2813            assert_eq!(cache.layers[0].k[0].capacity(), 0, "K released after seal");
2814            assert_eq!(cache.layers[0].v[0].capacity(), 0, "V released after seal");
2815        }
2816    }
2817
2818    #[test]
2819    fn truncate_rolls_back_speculative_positions() {
2820        let mut cache = LayerKvCache::new(2, 4);
2821        cache.mode = KvMode::F32;
2822        for pos in 0..5 {
2823            let k = vec![pos as f32; 8];
2824            let v = vec![pos as f32; 8];
2825            cache.append(&k, &v, &[true; 2]);
2826        }
2827        cache.truncate_last(2);
2828        assert_eq!(cache.seq_len, 3);
2829        assert_eq!(cache.head_len(0), 3);
2830        assert_eq!(cache.head_keys(0)[2 * 4], 2.0, "position 2 survives");
2831    }
2832
2833    /// q8_2f-attend ≈ f32-attend: 100 positions (crosses the field freeze
2834    /// at the 64th), pseudo-random vectors, relative tolerance of the
2835    /// int8 grid. Plus rollback and mass-based eviction on the q8 storage.
2836    #[test]
2837    fn q8_attend_matches_f32_within_grid() {
2838        let (heads, hd) = (2, 32);
2839        let mut f = LayerKvCache::new(heads, hd);
2840        f.mode = KvMode::F32;
2841        let mut q8 = LayerKvCache::new(heads, hd);
2842        q8.mode = KvMode::Q8 { k: true, v: true };
2843
2844        let synth = |p: usize, salt: usize| -> Vec<f32> {
2845            (0..heads * hd)
2846                .map(|i| {
2847                    let x = ((i * 31 + p * 17 + salt * 7 + 3) % 97) as f32 / 97.0 - 0.5;
2848                    // channel structure: even channels ×4 (checks the 2f field)
2849                    if i % 2 == 0 { x * 4.0 } else { x * 0.25 }
2850                })
2851                .collect()
2852        };
2853        for p in 0..100 {
2854            let k = synth(p, 1);
2855            let v = synth(p, 2);
2856            f.append(&k, &v, &[true; 2]);
2857            q8.append(&k, &v, &[true; 2]);
2858        }
2859        let q: Vec<f32> = (0..hd)
2860            .map(|i| ((i * 13 + 5) % 89) as f32 / 89.0 - 0.5)
2861            .collect();
2862        for g in 0..heads {
2863            let (of, pf) = f.attend(&q, g);
2864            let (o8, p8) = q8.attend(&q, g);
2865            let scale = of.iter().fold(0f32, |m, x| m.max(x.abs())).max(1e-6);
2866            for d in 0..hd {
2867                assert!(
2868                    (of[d] - o8[d]).abs() <= scale * 0.03 + 1e-3,
2869                    "g{g} d{d}: f32 {} vs q8 {}",
2870                    of[d],
2871                    o8[d]
2872                );
2873            }
2874            for p in 0..100 {
2875                assert!((pf[p] - p8[p]).abs() < 0.02, "prob p{p}");
2876            }
2877        }
2878        // rollback + eviction live on the q8 storage
2879        q8.truncate_last(30);
2880        assert_eq!(q8.head_len(0), 70);
2881        let imp: Vec<f32> = (0..70).map(|i| i as f32).collect();
2882        q8.accumulate_imp(&imp);
2883        q8.evict_born(20, 2, 8);
2884        assert_eq!(q8.head_len(0), 20);
2885        let (o, _) = q8.attend(&q, 0);
2886        assert!(o.iter().all(|x| x.is_finite()));
2887        // memory: q8 ≈ 1 byte/element + scale per row (vs 4 for f32)
2888        assert!(q8.memory_bytes() * 3 < f.memory_bytes());
2889    }
2890
2891    /// Grouped GQA attend must be bit-identical to per-head attend in
2892    /// every KV mode (it is the same math with rows streamed once).
2893    #[test]
2894    fn attend_group_equals_per_head_attend_bitexact() {
2895        let (kv_heads, hd, hpk) = (2usize, 32usize, 3usize); // 6 Q-heads
2896        for mode in [KvMode::F32, KvMode::Q8 { k: true, v: true }] {
2897            let mut c = LayerKvCache::new(kv_heads, hd);
2898            c.mode = mode;
2899            for p in 0..70 {
2900                let k: Vec<f32> = (0..kv_heads * hd)
2901                    .map(|i| ((i * 31 + p * 17 + 3) % 97) as f32 / 97.0 - 0.5)
2902                    .collect();
2903                let v: Vec<f32> = (0..kv_heads * hd)
2904                    .map(|i| ((i * 13 + p * 29 + 7) % 89) as f32 / 89.0 - 0.5)
2905                    .collect();
2906                c.append(&k, &v, &[true; 2]);
2907            }
2908            let q: Vec<f32> = (0..kv_heads * hpk * hd)
2909                .map(|i| ((i * 11 + 5) % 83) as f32 / 83.0 - 0.5)
2910                .collect();
2911            for g in 0..kv_heads {
2912                let span = g * hpk * hd..(g + 1) * hpk * hd;
2913                let mut out = vec![0f32; hpk * hd];
2914                let mut imp = vec![0f32; 70];
2915                c.attend_group(
2916                    &q[span.clone()],
2917                    g,
2918                    &mut out,
2919                    &mut imp,
2920                    1.0 / (hd as f32).sqrt(),
2921                    0,
2922                    0.0,
2923                    &[],
2924                );
2925                let mut imp_ref = vec![0f32; 70];
2926                for h in 0..hpk {
2927                    let qh = &q[span.start + h * hd..span.start + (h + 1) * hd];
2928                    let (o, probs) = c.attend(qh, g);
2929                    assert_eq!(
2930                        &out[h * hd..(h + 1) * hd],
2931                        &o[..],
2932                        "mode {mode:?} g{g} h{h}: grouped attend must be bit-identical"
2933                    );
2934                    for (dst, &p) in imp_ref.iter_mut().zip(&probs) {
2935                        *dst += p;
2936                    }
2937                }
2938                assert_eq!(
2939                    imp, imp_ref,
2940                    "mode {mode:?} g{g}: attention mass must match"
2941                );
2942            }
2943        }
2944    }
2945
2946    /// Learned sinks (gpt-oss / MiMo-V2): the grouped softmax must equal a
2947    /// softmax over the visible rows PLUS an explicit value-less sink
2948    /// column, computed independently in f64 — at position 0 (one stored
2949    /// row), mid-sequence, and with the window truncating the rows. The
2950    /// importance row must be the row probabilities (the sink's share is
2951    /// nobody's importance).
2952    #[test]
2953    fn sink_attend_matches_explicit_sink_column() {
2954        let (nkv, hd, hpk) = (2usize, 8usize, 3usize);
2955        let rows = 9usize;
2956        let mut c = LayerKvCache::new(nkv, hd);
2957        c.mode = KvMode::F32;
2958        let kv = |r: usize, i: usize, a: usize, m: usize| {
2959            (((r * a + i * 7 + 3) % m) as f32 / m as f32 - 0.5) * 2.0
2960        };
2961        let mut ks = Vec::new();
2962        let mut vs = Vec::new();
2963        for r in 0..rows {
2964            let k: Vec<f32> = (0..nkv * hd).map(|i| kv(r, i, 31, 97)).collect();
2965            let v: Vec<f32> = (0..nkv * hd).map(|i| kv(r, i, 17, 89)).collect();
2966            c.append(&k, &v, &[]);
2967            ks.push(k);
2968            vs.push(v);
2969        }
2970        let q: Vec<f32> = (0..nkv * hpk * hd)
2971            .map(|i| (((i * 11 + 5) % 83) as f32 / 83.0 - 0.5) * 3.0)
2972            .collect();
2973        // Mixed signs and one sink that dominates its head.
2974        let sinks = [0.7f32, -1.3, 2.5, 0.0, -4.0, 6.0];
2975        let scale = 1.0 / (hd as f32).sqrt();
2976        let mut checked = 0usize;
2977        for upto in [1usize, 2, 5, 9] {
2978            for window in [None, Some(3usize), Some(1)] {
2979                let first = window.map(|w| upto.saturating_sub(w)).unwrap_or(0);
2980                for g in 0..nkv {
2981                    let qg = &q[g * hpk * hd..(g + 1) * hpk * hd];
2982                    let sg = &sinks[g * hpk..(g + 1) * hpk];
2983                    let mut out = vec![0f32; hpk * hd];
2984                    let mut imp = vec![0f32; upto];
2985                    c.attend_group_upto(qg, g, &mut out, &mut imp, scale, first, 0.0, upto, sg);
2986                    let mut imp_ref = vec![0f64; upto];
2987                    for h in 0..hpk {
2988                        let qh = &qg[h * hd..(h + 1) * hd];
2989                        // Logits of the visible rows, then the sink column.
2990                        let mut z: Vec<f64> = (first..upto)
2991                            .map(|p| {
2992                                let k = &ks[p][g * hd..(g + 1) * hd];
2993                                qh.iter()
2994                                    .zip(k)
2995                                    .map(|(&a, &b)| a as f64 * b as f64)
2996                                    .sum::<f64>()
2997                                    * scale as f64
2998                            })
2999                            .collect();
3000                        z.push(sg[h] as f64);
3001                        let m = z.iter().cloned().fold(f64::NEG_INFINITY, f64::max);
3002                        let e: Vec<f64> = z.iter().map(|&x| (x - m).exp()).collect();
3003                        let s: f64 = e.iter().sum();
3004                        let p: Vec<f64> = e.iter().map(|&x| x / s).collect();
3005                        // The sink column carries a zero value vector.
3006                        for d in 0..hd {
3007                            let want: f64 = (first..upto)
3008                                .map(|r| p[r - first] * vs[r][g * hd + d] as f64)
3009                                .sum();
3010                            let got = out[h * hd + d] as f64;
3011                            assert!(
3012                                (got - want).abs() < 1e-6,
3013                                "upto {upto} window {window:?} g{g} h{h} d{d}: {got} vs {want}"
3014                            );
3015                        }
3016                        for r in first..upto {
3017                            imp_ref[r] += p[r - first];
3018                        }
3019                        checked += 1;
3020                    }
3021                    for r in 0..upto {
3022                        assert!(
3023                            (imp[r] as f64 - imp_ref[r]).abs() < 1e-6,
3024                            "imp upto {upto} window {window:?} g{g} row {r}: {} vs {}",
3025                            imp[r],
3026                            imp_ref[r]
3027                        );
3028                    }
3029                    // The sink takes real mass: rows sum below 1.
3030                    let row_mass: f32 = imp.iter().sum();
3031                    assert!(row_mass < hpk as f32, "sinks must absorb some mass");
3032                }
3033            }
3034        }
3035        assert_eq!(checked, 4 * 3 * nkv * hpk);
3036    }
3037
3038    /// Scoring only the window's rows must be bit-identical to the former
3039    /// whole-row scoring (−inf outside the window) — with and without a
3040    /// sink, output and importance. The reference is the plain per-head
3041    /// softmax over the same rows written out here.
3042    #[test]
3043    fn windowed_attend_equals_masked_full_row() {
3044        let (nkv, hd, hpk) = (1usize, 16usize, 2usize);
3045        let rows = 40usize;
3046        let mut c = LayerKvCache::new(nkv, hd);
3047        c.mode = KvMode::F32;
3048        for r in 0..rows {
3049            let k: Vec<f32> = (0..hd)
3050                .map(|i| ((r * 13 + i * 5) % 29) as f32 / 29.0 - 0.5)
3051                .collect();
3052            let v: Vec<f32> = (0..hd)
3053                .map(|i| ((r * 7 + i * 3) % 31) as f32 / 31.0 - 0.5)
3054                .collect();
3055            c.append(&k, &v, &[]);
3056        }
3057        let q: Vec<f32> = (0..hpk * hd)
3058            .map(|i| ((i * 19) % 23) as f32 / 23.0 - 0.5)
3059            .collect();
3060        let scale = 0.25f32;
3061        for w in [1usize, 7, 39, 40, 100] {
3062            let first = rows.saturating_sub(w);
3063            let mut out = vec![0f32; hpk * hd];
3064            let mut imp = vec![0f32; rows];
3065            c.attend_group(&q, 0, &mut out, &mut imp, scale, first, 0.0, &[]);
3066            // Reference: the historical full-row −inf-masked kernel.
3067            let mut out_ref = vec![0f32; hpk * hd];
3068            let mut imp_ref = vec![0f32; rows];
3069            for h in 0..hpk {
3070                let mut s = vec![f32::NEG_INFINITY; rows];
3071                for p in first..rows {
3072                    s[p] = crate::attention::dot_f32(
3073                        &q[h * hd..(h + 1) * hd],
3074                        &c.head_keys(0)[p * hd..(p + 1) * hd],
3075                    ) * scale;
3076                }
3077                let m = s.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
3078                let mut sum = 0f32;
3079                for v in s.iter_mut() {
3080                    *v = (*v - m).exp();
3081                    sum += *v;
3082                }
3083                for v in s.iter_mut() {
3084                    *v /= sum;
3085                }
3086                for p in first..rows {
3087                    if s[p].abs() < 1e-12 {
3088                        continue;
3089                    }
3090                    crate::attention::axpy_f32(
3091                        &mut out_ref[h * hd..(h + 1) * hd],
3092                        &c.head_values(0)[p * hd..(p + 1) * hd],
3093                        s[p],
3094                    );
3095                }
3096                for (d, &p) in imp_ref.iter_mut().zip(&s) {
3097                    *d += p;
3098                }
3099            }
3100            let bits = |v: &[f32]| v.iter().map(|x| x.to_bits()).collect::<Vec<_>>();
3101            assert_eq!(bits(&out), bits(&out_ref), "out window {w}");
3102            assert_eq!(bits(&imp), bits(&imp_ref), "imp window {w}");
3103        }
3104    }
3105
3106    /// Sinks are weights: a new conversation (clear) and a wire import
3107    /// must keep them.
3108    #[test]
3109    fn sinks_survive_clear_and_wire_import() {
3110        let mut c = LayerKvCache::new(1, 4);
3111        c.mode = KvMode::F32;
3112        c.sinks = Some(vec![0.5, -0.5]);
3113        c.append(&[1.0; 4], &[2.0; 4], &[]);
3114        let wire = c.export_wire(false).unwrap();
3115        c.clear();
3116        assert_eq!(c.sinks.as_deref(), Some(&[0.5f32, -0.5][..]));
3117        c.import_wire(&wire).unwrap();
3118        assert_eq!(c.sinks.as_deref(), Some(&[0.5f32, -0.5][..]));
3119        assert_eq!(c.seq_len, 1);
3120    }
3121
3122    /// Review regression: mass-based eviction in MIXED modes. q8v used to
3123    /// panic (gather over an empty v[h]), q8k silently left raw V
3124    /// uncompressed (stale rows under kept keys + memory leak).
3125    #[test]
3126    fn born_eviction_mixed_modes_stay_consistent() {
3127        for (mk, mv) in [(false, true), (true, false)] {
3128            let mut c = LayerKvCache::new(1, 4);
3129            c.mode = KvMode::Q8 { k: mk, v: mv };
3130            for p in 0..80 {
3131                let k = vec![p as f32 * 0.01; 4];
3132                let v = vec![p as f32; 4];
3133                c.append(&k, &v, &[true]);
3134            }
3135            let imp: Vec<f32> = (0..80).map(|i| i as f32).collect();
3136            c.accumulate_imp(&imp);
3137            let before = c.memory_bytes();
3138            c.evict_born(20, 4, 8); // q8v: used to panic here
3139            assert_eq!(c.head_len(0), 20, "k={mk} v={mv}");
3140            assert!(
3141                c.memory_bytes() < before / 2,
3142                "memory must shrink (k={mk} v={mv})"
3143            );
3144            // V rows match the kept set: the heaviest positions
3145            // (tail 60..79) must be present in the attend output.
3146            let (out, _) = c.attend(&[1.0, 1.0, 1.0, 1.0], 0);
3147            assert!(
3148                out[0] > 30.0,
3149                "V from the kept tail, not the stale head (k={mk} v={mv}, out {})",
3150                out[0]
3151            );
3152        }
3153    }
3154
3155    #[test]
3156    fn born_eviction_keeps_high_mass_position() {
3157        let mut cache = KvCache::new(1, 1, 2, 16);
3158        cache.policy = EvictionPolicy::Born { sink: 1 };
3159        for l in &mut cache.layers {
3160            l.mode = KvMode::F32;
3161        }
3162        let layer = &mut cache.layers[0];
3163        // 8 positions; keys carry the position index so we can verify
3164        // exactly which positions survive the gather.
3165        for pos in 0..8 {
3166            let k = vec![pos as f32; 2];
3167            let v = vec![pos as f32 + 100.0; 2];
3168            layer.append(&k, &v, &[true]);
3169        }
3170        // Position 3 carries the most attention mass.
3171        let mut imp = vec![0.05f32; 8];
3172        imp[3] = 5.0;
3173        layer.accumulate_imp(&imp);
3174
3175        cache.evict(4); // sink 1 + recent 2 + 1 top-mass slot
3176        let layer = &cache.layers[0];
3177        assert_eq!(layer.seq_len, 4);
3178        let kept_keys: Vec<f32> = (0..4).map(|i| layer.head_keys(0)[i * 2]).collect();
3179        assert_eq!(
3180            kept_keys,
3181            vec![0.0, 3.0, 6.0, 7.0],
3182            "kept = sink(0) + mass-top(3) + recent(6,7)"
3183        );
3184        // imp stays aligned with the gathered positions.
3185        assert_eq!(layer.head_len(0), 4);
3186    }
3187
3188    // ── Sliding-window tail (trim_window) ──
3189
3190    /// Deterministic row of `n` values in [-1, 1) for position `p`.
3191    fn swa_row(p: usize, n: usize, salt: usize) -> Vec<f32> {
3192        (0..n)
3193            .map(|i| ((p * 7919 + i * 104_729 + salt * 1_299_709) % 2003) as f32 / 1001.5 - 1.0)
3194            .collect()
3195    }
3196
3197    fn bits(v: &[f32]) -> Vec<u32> {
3198        v.iter().map(|x| x.to_bits()).collect()
3199    }
3200
3201    /// The stored tail is the untrimmed cache's rows from `base` on, in
3202    /// order, with the bounds the trigger promises — a dead head included.
3203    #[test]
3204    fn swa_trim_keeps_contiguous_tail() {
3205        let (nkv, hd, w, slack, align) = (2usize, 8usize, 6usize, 2usize, 4usize);
3206        let alive = [true, false];
3207        let mut full = LayerKvCache::new(nkv, hd);
3208        full.mode = KvMode::F32;
3209        let mut t = full.clone();
3210        let mut trims = 0;
3211        for p in 0..100 {
3212            let (k, v) = (swa_row(p, nkv * hd, 1), swa_row(p, nkv * hd, 2));
3213            full.append(&k, &v, &alive);
3214            t.append(&k, &v, &alive);
3215            if t.trim_window(w, slack, align) > 0 {
3216                trims += 1;
3217                assert!(
3218                    (w + slack..w + slack + align).contains(&t.seq_len),
3219                    "rows after a trim: {}",
3220                    t.seq_len
3221                );
3222            }
3223            assert!(t.seq_len <= 2 * w, "rows {} past the trigger", t.seq_len);
3224            assert_eq!(t.pos_len(), p + 1);
3225            assert_eq!(t.base() % align, 0);
3226            assert_eq!(t.base() + t.seq_len, full.seq_len);
3227            let b = t.base();
3228            assert_eq!(t.head_keys(0), &full.head_keys(0)[b * hd..]);
3229            assert_eq!(t.head_values(0), &full.head_values(0)[b * hd..]);
3230            assert!(t.head_keys(1).is_empty() && t.head_values(1).is_empty());
3231            assert_eq!(t.imp.len(), t.seq_len);
3232        }
3233        assert!(trims > 5, "only {trims} trims in 100 positions");
3234        assert_eq!(t.tail_window(), Some(w));
3235        assert_eq!(full.tail_window(), None);
3236    }
3237
3238    /// A tail trimmed under a turn anchor keeps every row the window at the
3239    /// anchor reads, however long the answer runs past it; rolling the whole
3240    /// answer back leaves exactly the untrimmed rows there, and the anchor
3241    /// moved on lets the plain trim drain them again.
3242    #[test]
3243    fn swa_anchored_trim_keeps_the_rollback_window() {
3244        let (nkv, hd, w, slack, align) = (1usize, 4usize, 6usize, 2usize, 4usize);
3245        let alive = [true];
3246        let mut full = LayerKvCache::new(nkv, hd);
3247        full.mode = KvMode::F32;
3248        let mut t = full.clone();
3249        let anchor = 40usize;
3250        for p in 0..100 {
3251            let (k, v) = (swa_row(p, nkv * hd, 1), swa_row(p, nkv * hd, 2));
3252            full.append(&k, &v, &alive);
3253            t.append(&k, &v, &alive);
3254            t.trim_window_anchored(w, slack, align, anchor);
3255            assert!(
3256                t.base() + w <= anchor - slack + 1,
3257                "base {} at {p}",
3258                t.base()
3259            );
3260        }
3261        let back = anchor - slack;
3262        t.truncate_last(t.pos_len() - back);
3263        full.truncate_last(full.pos_len() - back);
3264        assert_eq!(t.pos_len(), back);
3265        let b = t.base();
3266        assert!(t.seq_len >= w - 1);
3267        assert_eq!(t.head_keys(0), &full.head_keys(0)[b * hd..]);
3268        assert_eq!(t.head_values(0), &full.head_values(0)[b * hd..]);
3269        for p in back..back + 3 * w {
3270            let (k, v) = (swa_row(p, nkv * hd, 1), swa_row(p, nkv * hd, 2));
3271            t.append(&k, &v, &alive);
3272            t.trim_window_anchored(w, slack, align, usize::MAX);
3273        }
3274        assert!(t.seq_len <= 2 * w);
3275    }
3276
3277    /// Two caches fed the same rows, one trimmed: every windowed attend
3278    /// (`first = rows − w`) gives the same output and importance, bit for
3279    /// bit, under F32 and the q8 cache — the q8 field freezes at 64 rows,
3280    /// before the first trim.
3281    #[test]
3282    fn swa_trimmed_attend_equals_untrimmed() {
3283        let (nkv, hd, hpk) = (2usize, 16usize, 2usize);
3284        for mode in [KvMode::F32, KvMode::Q8 { k: true, v: true }] {
3285            for (w, slack, align) in [(40usize, 2usize, 4usize), (37, 5, 16), (64, 64, 64)] {
3286                let mut full = LayerKvCache::new(nkv, hd);
3287                full.mode = mode;
3288                let mut t = full.clone();
3289                let scale = 0.3f32;
3290                let mut trims = 0;
3291                for p in 0..400 {
3292                    let (k, v) = (swa_row(p, nkv * hd, 3), swa_row(p, nkv * hd, 4));
3293                    full.append(&k, &v, &[]);
3294                    t.append(&k, &v, &[]);
3295                    let q = swa_row(p, nkv * hpk * hd, 5);
3296                    for g in 0..nkv {
3297                        let qg = &q[g * hpk * hd..(g + 1) * hpk * hd];
3298                        let (nf, nt) = (full.head_len(g), t.head_len(g));
3299                        let (mut of, mut ot) = (vec![0f32; hpk * hd], vec![0f32; hpk * hd]);
3300                        let (mut imf, mut imt) = (vec![0f32; nf], vec![0f32; nt]);
3301                        full.attend_group(
3302                            qg,
3303                            g,
3304                            &mut of,
3305                            &mut imf,
3306                            scale,
3307                            nf.saturating_sub(w),
3308                            0.0,
3309                            &[],
3310                        );
3311                        t.attend_group(
3312                            qg,
3313                            g,
3314                            &mut ot,
3315                            &mut imt,
3316                            scale,
3317                            nt.saturating_sub(w),
3318                            0.0,
3319                            &[],
3320                        );
3321                        assert_eq!(bits(&of), bits(&ot), "{mode:?} w {w} pos {p} group {g}");
3322                        assert_eq!(bits(&imf[t.base()..]), bits(&imt), "{mode:?} w {w} pos {p}");
3323                        full.accumulate_imp(&imf);
3324                        t.accumulate_imp(&imt);
3325                    }
3326                    if t.trim_window(w, slack, align) > 0 {
3327                        trims += 1;
3328                    }
3329                    assert_eq!(bits(&full.imp[t.base()..]), bits(&t.imp));
3330                }
3331                assert!(trims > 0, "{mode:?} w {w}: never trimmed");
3332            }
3333        }
3334    }
3335
3336    /// A rollback no deeper than the slack, then new rows: still the
3337    /// untrimmed answer.
3338    #[test]
3339    fn swa_truncate_after_trim() {
3340        let (nkv, hd, w) = (1usize, 8usize, 10usize);
3341        let mut full = LayerKvCache::new(nkv, hd);
3342        full.mode = KvMode::F32;
3343        let mut t = full.clone();
3344        let mut p = 0usize;
3345        for round in 0..60 {
3346            for _ in 0..3 {
3347                let (k, v) = (swa_row(p, hd, 6), swa_row(p, hd, 7));
3348                full.append(&k, &v, &[]);
3349                t.append(&k, &v, &[]);
3350                p += 1;
3351            }
3352            t.trim_window(w, 2, 4);
3353            // Speculative-style rollback of 2, then replacement rows.
3354            full.truncate_last(2);
3355            t.truncate_last(2);
3356            p -= 2;
3357            assert_eq!(t.pos_len(), full.seq_len);
3358            for _ in 0..2 {
3359                let (k, v) = (swa_row(p, hd, 8 + round), swa_row(p, hd, 9 + round));
3360                full.append(&k, &v, &[]);
3361                t.append(&k, &v, &[]);
3362                p += 1;
3363            }
3364            let q = swa_row(p, hd, 10);
3365            let (nf, nt) = (full.head_len(0), t.head_len(0));
3366            let (mut of, mut ot) = (vec![0f32; hd], vec![0f32; hd]);
3367            let (mut imf, mut imt) = (vec![0f32; nf], vec![0f32; nt]);
3368            full.attend_group(&q, 0, &mut of, &mut imf, 0.5, nf - w.min(nf), 0.0, &[]);
3369            t.attend_group(&q, 0, &mut ot, &mut imt, 0.5, nt - w.min(nt), 0.0, &[]);
3370            assert_eq!(bits(&of), bits(&ot), "round {round}");
3371        }
3372        assert!(t.base() > 0);
3373    }
3374
3375    /// The cache-wide eviction leaves a sliding tail alone, and a tail
3376    /// does not count toward the eviction trigger.
3377    #[test]
3378    fn swa_evict_skips_tail() {
3379        for policy in [EvictionPolicy::Recent, EvictionPolicy::Born { sink: 1 }] {
3380            let mut c = KvCache::new(2, 1, 4, 30);
3381            c.policy = policy;
3382            for p in 0..25 {
3383                for l in &mut c.layers {
3384                    l.append(&swa_row(p, 4, 1), &swa_row(p, 4, 2), &[]);
3385                }
3386                c.layers[0].trim_window(8, 2, 4);
3387            }
3388            let t_rows = c.layers[0].seq_len;
3389            let t_base = c.layers[0].base();
3390            assert!(t_base > 0);
3391            assert!(!c.needs_eviction(), "25 rows < cap 30");
3392            for p in 25..30 {
3393                for l in &mut c.layers {
3394                    l.append(&swa_row(p, 4, 1), &swa_row(p, 4, 2), &[]);
3395                }
3396            }
3397            // Layer 0 (tail, 5 rows past its last trim) is not what triggers.
3398            assert!(c.needs_eviction());
3399            let gen0 = c.layers[0].generation();
3400            c.evict(10);
3401            assert_eq!(c.layers[0].seq_len, t_rows + 5, "{policy:?}: tail evicted");
3402            assert_eq!(c.layers[0].base(), t_base);
3403            assert_eq!(c.layers[0].generation(), gen0);
3404            assert_eq!(
3405                c.layers[1].seq_len, 10,
3406                "{policy:?}: full layer not evicted"
3407            );
3408            assert_eq!(c.seq_len(), 30, "absolute depth from the tail");
3409        }
3410    }
3411
3412    /// `generation` moves on every mutation a device copy cannot follow by its
3413    /// row count, and stays on appends and rollbacks.
3414    #[test]
3415    fn swa_gen_bumps() {
3416        let mut l = LayerKvCache::new(1, 4);
3417        l.mode = KvMode::F32;
3418        let g0 = l.generation();
3419        assert_ne!(
3420            LayerKvCache::new(1, 4).generation(),
3421            g0,
3422            "fresh caches differ"
3423        );
3424        for p in 0..20 {
3425            l.append(&swa_row(p, 4, 1), &swa_row(p, 4, 2), &[]);
3426        }
3427        assert_eq!(l.generation(), g0, "append");
3428        l.truncate_last(2);
3429        assert_eq!(l.generation(), g0, "truncate_last");
3430        assert_eq!(l.trim_window(10, 2, 4), 0, "18 rows ≤ 2w: no trim");
3431        assert_eq!(l.generation(), g0, "a no-op trim");
3432        for p in 18..21 {
3433            l.append(&swa_row(p, 4, 1), &swa_row(p, 4, 2), &[]);
3434        }
3435        assert!(l.trim_window(10, 2, 4) > 0);
3436        let g1 = l.generation();
3437        assert_ne!(g1, g0, "trim");
3438        let snap = l.clone();
3439        l.clear();
3440        let g2 = l.generation();
3441        assert_ne!(g2, g1, "clear");
3442        assert_eq!((l.base(), l.pos_len(), l.tail_window()), (0, 0, Some(10)));
3443        let mut e = LayerKvCache::new(1, 4);
3444        e.mode = KvMode::F32;
3445        for p in 0..12 {
3446            e.append(&swa_row(p, 4, 1), &swa_row(p, 4, 2), &[]);
3447        }
3448        let ge = e.generation();
3449        e.evict(20);
3450        assert_eq!(e.generation(), ge, "an eviction that does nothing");
3451        e.evict(6);
3452        assert_ne!(e.generation(), ge, "evict");
3453        let ge = e.generation();
3454        e.evict_born(3, 1, 1);
3455        assert_ne!(e.generation(), ge, "evict_born");
3456        let mut i = LayerKvCache::new(1, 4);
3457        let gi = i.generation();
3458        i.import_wire(&snap.export_wire(false).unwrap()).unwrap();
3459        assert_ne!(i.generation(), gi, "import");
3460        assert_ne!(
3461            i.generation(),
3462            snap.generation(),
3463            "an import is new storage"
3464        );
3465    }
3466
3467    /// A trimmed layer travels as `FullTail` and answers the same on the
3468    /// far side; an untrimmed one is still byte-for-byte `Full`; a layer
3469    /// with another kind refuses the tail.
3470    #[test]
3471    fn wire_full_tail_roundtrip() {
3472        let (nkv, hd, w) = (2usize, 4usize, 6usize);
3473        let mut a = LayerKvCache::new(nkv, hd);
3474        a.mode = KvMode::F32;
3475        for p in 0..5 {
3476            a.append(&swa_row(p, nkv * hd, 1), &swa_row(p, nkv * hd, 2), &[]);
3477        }
3478        let plain = a.export_wire(false).unwrap();
3479        assert_eq!(plain[20], WireKind::Full as u8, "untrimmed stays Full");
3480        assert_eq!(u64::from_le_bytes(plain[24..32].try_into().unwrap()), 5);
3481        for p in 5..40 {
3482            a.append(&swa_row(p, nkv * hd, 1), &swa_row(p, nkv * hd, 2), &[]);
3483            a.trim_window(w, 2, 4);
3484        }
3485        assert!(a.base() > 0);
3486        for f16 in [false, true] {
3487            let bytes = a.export_wire(f16).unwrap();
3488            assert_eq!(bytes[20], WireKind::FullTail as u8);
3489            assert_eq!(u64::from_le_bytes(bytes[24..32].try_into().unwrap()), 40);
3490            let mut b = LayerKvCache::new(nkv, hd);
3491            b.import_wire(&bytes).expect("tail import");
3492            assert_eq!(
3493                (b.base(), b.seq_len, b.pos_len()),
3494                (a.base(), a.seq_len, 40)
3495            );
3496            if !f16 {
3497                let q = swa_row(99, 2 * hd, 3);
3498                for g in 0..nkv {
3499                    let n = a.head_len(g);
3500                    let (mut oa, mut ob) = (vec![0f32; 2 * hd], vec![0f32; 2 * hd]);
3501                    let (mut ia, mut ib) = (vec![0f32; n], vec![0f32; n]);
3502                    a.attend_group(&q, g, &mut oa, &mut ia, 0.5, n - w, 0.0, &[]);
3503                    b.attend_group(&q, g, &mut ob, &mut ib, 0.5, n - w, 0.0, &[]);
3504                    assert_eq!(bits(&oa), bits(&ob));
3505                }
3506                // ... and a Full import over it puts row 0 back at 0.
3507                b.import_wire(&plain).unwrap();
3508                assert_eq!((b.base(), b.pos_len()), (0, 5));
3509            }
3510        }
3511        // A position that disagrees with base + rows is refused.
3512        let mut bad = a.export_wire(false).unwrap();
3513        bad[24..32].copy_from_slice(&41u64.to_le_bytes());
3514        assert!(LayerKvCache::new(nkv, hd).import_wire(&bad).is_err());
3515        // A bounded layer does not take a tail.
3516        let mut r = LayerKvCache::new(nkv, hd);
3517        r.install_bounded(8);
3518        let err = r.import_wire(&a.export_wire(false).unwrap()).unwrap_err();
3519        assert!(err.contains("does not match"), "{err}");
3520    }
3521}
3522
3523#[cfg(all(test, any(target_arch = "aarch64", target_arch = "x86_64")))]
3524mod chunk_attend_tests {
3525    use super::*;
3526
3527    /// The batched chunk attend against a double-precision softmax
3528    /// attention over the same cache rows: full causal, a window that
3529    /// masks (the GEMM pair then runs over the window's key range only)
3530    /// and a window wider than the context; attention importance too.
3531    /// On macOS this is the Accelerate arm (the NEON arm with
3532    /// CMF_FORCE_NEON_GEMM=1), elsewhere the NEON or AVX2 micro-GEMM.
3533    #[test]
3534    fn attend_chunk_matches_naive_attention() {
3535        let (nkv, hd, hpk) = (2usize, 32usize, 3usize);
3536        let nh = nkv * hpk;
3537        let (s0, b) = (40usize, 48usize);
3538        let n = s0 + b;
3539        let scale = 1.0 / (hd as f32).sqrt();
3540        for window in [None, Some(16usize), Some(200usize)] {
3541            let mut c = LayerKvCache::new(nkv, hd);
3542            c.mode = KvMode::F32;
3543            for p in 0..n {
3544                let k: Vec<f32> = (0..nkv * hd)
3545                    .map(|i| ((i * 31 + p * 17 + 3) % 97) as f32 / 97.0 - 0.5)
3546                    .collect();
3547                let v: Vec<f32> = (0..nkv * hd)
3548                    .map(|i| ((i * 13 + p * 29 + 7) % 89) as f32 / 89.0 - 0.5)
3549                    .collect();
3550                c.append(&k, &v, &[true; 2]);
3551            }
3552            let q: Vec<f32> = (0..b * nh * hd)
3553                .map(|i| ((i * 11 + 5) % 83) as f32 / 83.0 - 0.5)
3554                .collect();
3555            let imp0 = c.imp.clone();
3556            let mut out = vec![0f32; b * nh * hd];
3557            c.attend_chunk(&q, b, s0, nh, hpk, hd, &mut out, None, scale, window);
3558            let mut expect = vec![0f64; b * nh * hd];
3559            let mut imp_ref: Vec<f64> = imp0.iter().map(|&x| x as f64).collect();
3560            for bi in 0..b {
3561                let lim = s0 + bi + 1;
3562                let lo = window.map_or(0, |w| lim.saturating_sub(w));
3563                for hh in 0..nh {
3564                    let g = hh / hpk;
3565                    let qv = &q[bi * nh * hd + hh * hd..][..hd];
3566                    let mut sc: Vec<f64> = (lo..lim)
3567                        .map(|p| {
3568                            let kr = &c.k[g][p * hd..(p + 1) * hd];
3569                            qv.iter().zip(kr).map(|(&x, &y)| x as f64 * y as f64).sum::<f64>()
3570                                * scale as f64
3571                        })
3572                        .collect();
3573                    let mx = sc.iter().cloned().fold(f64::NEG_INFINITY, f64::max);
3574                    let mut z = 0f64;
3575                    for s in sc.iter_mut() {
3576                        *s = (*s - mx).exp();
3577                        z += *s;
3578                    }
3579                    for (idx, p) in (lo..lim).enumerate() {
3580                        let pr = sc[idx] / z;
3581                        if p < imp_ref.len() {
3582                            imp_ref[p] += pr;
3583                        }
3584                        let vr = &c.v[g][p * hd..(p + 1) * hd];
3585                        for d in 0..hd {
3586                            expect[bi * nh * hd + hh * hd + d] += pr * vr[d] as f64;
3587                        }
3588                    }
3589                }
3590            }
3591            for (i, (&x, &y)) in out.iter().zip(&expect).enumerate() {
3592                assert!(
3593                    (x as f64 - y).abs() <= 1e-4,
3594                    "window {window:?} out[{i}]: {x} vs {y}"
3595                );
3596            }
3597            assert_eq!(c.imp.len(), imp0.len(), "window {window:?}: importance length changed");
3598            for (p, (&x, &y)) in c.imp.iter().zip(&imp_ref).enumerate() {
3599                assert!(
3600                    (x as f64 - y).abs() <= 1e-3 * (1.0 + y.abs()),
3601                    "window {window:?} imp[{p}]: {x} vs {y}"
3602                );
3603            }
3604        }
3605    }
3606}