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