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