Skip to main content

memra_kv/
lib.rs

1//! memra-kv — the dual KV/recurrent cache, extracted (Phase D, ARCHITECTURE-H100.md §5).
2//!
3//! Moved VERBATIM from memra-engine/src/cache.rs behind the `KvDev` seam: the cache only
4//! ever needed 7 device ops (alloc/copy/set), so the trait is that surface and nothing
5//! more. The append/dequant KERNELS stay in the engine fatbins — this crate owns the
6//! structure, sizing math, and the KV format policy (env-selected, shared by the engine's
7//! fatbin router and every cache consumer). memra-engine re-exports this as `cache` so
8//! call sites are unchanged.
9
10// ---------------- KV format policy (env-selected; moved from memra-engine) ----------------
11
12/// Env-selected KV cache formats (MEMRA_KV_K / MEMRA_KV_V). The engine's flash-fatbin router
13/// and the cache sizing below MUST agree — both read this one function.
14pub fn kv_cache_formats() -> (&'static str, &'static str) {
15    static F: std::sync::OnceLock<(&'static str, &'static str)> = std::sync::OnceLock::new();
16    *F.get_or_init(|| {
17        let k = match std::env::var("MEMRA_KV_K").as_deref() {
18            Ok("fp8") => "fp8",
19            Ok("q8_0") | Ok("") | Err(_) => "q8_0",
20            Ok(o) => panic!("MEMRA_KV_K={o} unsupported (q8_0 | fp8)"),
21        };
22        let v = match std::env::var("MEMRA_KV_V").as_deref() {
23            Ok("q4_0") => "q4_0",
24            Ok("fp8") => "fp8",
25            Ok("q5_1") | Ok("") | Err(_) => "q5_1",
26            Ok(o) => panic!("MEMRA_KV_V={o} unsupported (q5_1 | q4_0 | fp8)"),
27        };
28        if (k, v) != ("q8_0", "q5_1") {
29            eprintln!("[memra] KV cache format: K={k} V={v} (non-default — new numeric config)");
30        }
31        (k, v)
32    })
33}
34
35/// Per-32-element block bytes for the selected (K, V) formats.
36pub fn kv_blk_bytes() -> (usize, usize) {
37    let (k, v) = kv_cache_formats();
38    let kb = match k {
39        "fp8" => 32,
40        _ => 34,
41    };
42    let vb = match v {
43        "q4_0" => 18,
44        "fp8" => 32,
45        _ => 24,
46    };
47    (kb, vb)
48}
49
50/// Exact allocation geometry for one rank of a tensor-parallel KV sidecar.
51///
52/// The context-linear coefficient and fixed allocation bytes are shared by the CUDA allocator
53/// and serving admission. Keeping both consumers on this function prevents a new KV format or
54/// rank width from making the pre-admit estimate disagree with the buffers allocated at decode.
55#[derive(Clone, Copy, Debug, PartialEq, Eq)]
56pub struct TpKvRankAllocationShape {
57    pub kv_dim_k: usize,
58    pub kv_dim_v: usize,
59    pub k_token_bytes: usize,
60    pub v_token_bytes: usize,
61    pub fixed_bytes: usize,
62}
63
64impl TpKvRankAllocationShape {
65    pub fn bytes_per_token(self) -> usize {
66        self.k_token_bytes.saturating_add(self.v_token_bytes)
67    }
68
69    pub fn allocation_bytes(self, capacity: usize) -> usize {
70        self.bytes_per_token()
71            .saturating_mul(capacity)
72            .saturating_add(self.fixed_bytes)
73    }
74}
75
76#[allow(clippy::manual_is_multiple_of)] // allow: divisor is runtime-derived; the modulo form keeps a zero divisor loud (a panic), where is_multiple_of would return false silently
77pub fn tp_kv_rank_allocation_shape(
78    kv_dim_k: usize,
79    kv_dim_v: usize,
80    ranks: usize,
81) -> Result<TpKvRankAllocationShape, String> {
82    if ranks == 0 || kv_dim_k == 0 || kv_dim_v == 0 {
83        return Err(format!(
84            "TP KV dimensions and rank count must be nonzero: k={kv_dim_k} v={kv_dim_v} \
85             ranks={ranks}"
86        ));
87    }
88    if kv_dim_k % ranks != 0 || kv_dim_v % ranks != 0 {
89        return Err(format!(
90            "TP KV dimensions k={kv_dim_k} v={kv_dim_v} are not divisible by TP={ranks}"
91        ));
92    }
93    let local_k = kv_dim_k / ranks;
94    let local_v = kv_dim_v / ranks;
95    if !local_k.is_multiple_of(32) || !local_v.is_multiple_of(32) {
96        return Err(format!(
97            "TP KV local dimensions k={local_k} v={local_v} must be 32-aligned"
98        ));
99    }
100    let (k_block_bytes, v_block_bytes) = kv_blk_bytes();
101    let k_token_bytes = (local_k / 32)
102        .checked_mul(k_block_bytes)
103        .ok_or("TP KV K token-byte overflow")?;
104    let v_token_bytes = (local_v / 32)
105        .checked_mul(v_block_bytes)
106        .ok_or("TP KV V token-byte overflow")?;
107    Ok(TpKvRankAllocationShape {
108        kv_dim_k: local_k,
109        kv_dim_v: local_v,
110        k_token_bytes,
111        v_token_bytes,
112        // Two CUDA byte planes retain their existing 8-byte tail pads, plus one i32 length
113        // mirror. Allocator alignment remains visible through the device pool high-water.
114        fixed_bytes: 8 + 8 + std::mem::size_of::<i32>(),
115    })
116}
117
118/// FP8-GLOBALS switch (MEMRA_GEMMA_GKV, default ON): gemma global (hd512) layers keep
119/// their KV in e4m3 — the dequant-latency arc (HANDOVER). Windowed layers stay q8_0/q5_1.
120pub fn gkv_on() -> bool {
121    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
122    *ON.get_or_init(|| {
123        std::env::var("MEMRA_GEMMA_GKV")
124            .map(|v| v != "0")
125            .unwrap_or(true)
126    })
127}
128
129/// FP8-WINDOWED switch (MEMRA_GEMMA_WKV; serving-mode default): SPEC serving (MEMRA_DRAFT
130/// set) -> OFF, plain -> ON — the acceptance-vs-depth record lives on the engine-side
131/// history of `Engine::wkv_on` (git). Explicit env always wins.
132pub fn wkv_on() -> bool {
133    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
134    *ON.get_or_init(|| {
135        std::env::var("MEMRA_GEMMA_WKV")
136            .map(|v| v != "0")
137            .unwrap_or_else(|_| std::env::var("MEMRA_DRAFT").is_err())
138    })
139}
140
141/// Per-model FP8-KV door (-1 = unset → env/default off; 0 = off; 1 = on). Set at qwen
142/// model load: the 2026-07-12 arc closed per-model — 9B +0.7-4% scaling with depth,
143/// 27B flat (weight-bound), 35B −2% (fp8 format-gates its v3 dp4a lane off). Explicit
144/// MEMRA_KV_FP8 wins. 9B adoption attempt REVERTED by measurement 2026-07-29 (−1% at 12k
145/// on the then-current build) — loaders currently store 0.
146pub static KV_FP8_FORCE: std::sync::atomic::AtomicI8 = std::sync::atomic::AtomicI8::new(-1);
147
148/// QWEN FP8-KV switch (MEMRA_KV_FP8 explicit; else the per-model KV_FP8_FORCE door set
149/// at model load; else OFF). Non-gemma full-attn layers hold e4m3 K/V via the kf8vf8
150/// module.
151pub fn kv_fp8_on() -> bool {
152    static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
153    if let Some(v) = *ENV.get_or_init(|| std::env::var("MEMRA_KV_FP8").ok().map(|v| v == "1")) {
154        return v;
155    }
156    matches!(KV_FP8_FORCE.load(std::sync::atomic::Ordering::Relaxed), 1)
157}
158
159/// Step35 SWA ring. Default OFF unless the loader arms the step37 serving default (owner flip
160/// 2026-08-27: the ring frees 16.4 GB on card0 at the natural 262144 context with identical
161/// throughput and ids, and the W8 doors OOM there without it). Architecture-scoped by its call
162/// sites: Gemma4's row-0-addressed window kernels cannot consume a rebased ring view, which is
163/// why the default arms per loaded family rather than globally. `MEMRA_SWA_RING=1` forces ON,
164/// `=0` is the kill switch either way.
165static SWA_RING_DEFAULT: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
166
167pub fn set_swa_ring_default(on: bool) {
168    SWA_RING_DEFAULT.store(on, std::sync::atomic::Ordering::Relaxed);
169}
170
171pub fn swa_ring_on() -> bool {
172    static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
173    match *ENV.get_or_init(|| match std::env::var("MEMRA_SWA_RING").ok().as_deref() {
174        Some("1") => Some(true),
175        Some("0") => Some(false),
176        _ => None,
177    }) {
178        Some(forced) => forced,
179        None => SWA_RING_DEFAULT.load(std::sync::atomic::Ordering::Relaxed),
180    }
181}
182
183/// With the SWA-ring door open, `prime_chunk_tokens` caps every legal chunk at this bound. The
184/// ring carries one whole maximum-size prime chunk in addition to the reader's window.
185pub const PRIME_CHUNK_MAX_TOKENS: usize = 4096;
186const SWA_VIEW_ALIGNMENT_ROWS: usize = 32;
187
188/// Physical rows required by the Step35 SWA reader contract. Prime starts at
189/// `(base_len - (window - 1)) & !31`, so at most 31 masked rows precede the live window.
190pub fn swa_ring_rows(window: usize, max_ctx: usize) -> usize {
191    // window + max prime chunk + REWIND HEADROOM + alignment. append_plan requires
192    // keep_rows + append_rows <= rows, and keep_rows is now window + SWA_REWIND_SLACK_ROWS, so the
193    // headroom has to be in `rows` or a full-size prime chunk stops fitting. Costs
194    // SWA_REWIND_SLACK_ROWS rows per ring-backed plane.
195    max_ctx.min(
196        window + PRIME_CHUNK_MAX_TOKENS + SWA_REWIND_SLACK_ROWS + (SWA_VIEW_ALIGNMENT_ROWS - 1),
197    )
198}
199
200/// Rows a ring-backed plane keeps BELOW the aligned window start so a backward rewind stays legal.
201///
202/// This is HEADROOM THE RING IS SIZED FOR, not slack scavenged from it. The original geometry
203/// (window + prime chunk + 31) left exactly one alignment block spare once a full prime chunk had
204/// to fit, and one block only covers a rewind shallower than 32 rows. Clamping a deeper request up
205/// to `base` instead makes the append legal while leaving the attention window pointing below rows
206/// the ring no longer holds — which produced all-NaN head logits and seed hiddens at pos 8661
207/// rather than an error. A ring that cannot serve the rewinds its own callers perform is
208/// undersized; the fix is to size it, not to keep redistributing 32 rows.
209pub const SWA_REWIND_SLACK_ROWS: usize = 512;
210
211/// The retain a ring-backed append must request: the aligned window start for `first_row`, minus
212/// the rewind slack, but NEVER below what the ring still holds.
213///
214/// The clamp is the part that took three attempts to find. Asking for slack unconditionally makes
215/// the REWIND legal and then breaks the very next APPEND: after a rewind, `first_row` moves back
216/// while `base` does not, so the ideal retain falls under `base` and append_plan refuses it
217/// ("SWA ring lapped required rows (base 4128, retain 4096, len 4669)"). Rows below `base` are
218/// gone and, being older than the window, are not needed — so clamping up to `base` is both legal
219/// and correct. Slack is an optimisation the ring grants when it can, never a demand.
220pub fn swa_retain_from(first_row: usize, window: usize, base: usize) -> usize {
221    let ideal = first_row
222        .saturating_sub(window.saturating_sub(1))
223        .saturating_sub(SWA_REWIND_SLACK_ROWS)
224        & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
225    ideal.max(base)
226}
227
228/// WORKING-SET rows the DSA indexer TAIL RING books by default.
229///
230/// IT IS NOT A BOUND ON ANYTHING, and that is the correction (lane/glm53-ring-sizing,
231/// 2026-08-28). The ring first shipped sized as `prime_chunk_bound + 1024`, i.e. against the
232/// largest `t` a CHUNKED prefill could hand one call. glm5_next, the one architecture this ring
233/// exists for, is eager-only (`ResidualTopology::HyperConnections` refuses every batched and
234/// speculative entry point) and primes MONOLITHICALLY: `prime_cache_hyper` never calls
235/// `prime_chunk_ranges`, so its per-call `t` is the whole prompt and its only ceiling is the
236/// admission limit itself. A ring that bounds `t` is therefore a ring of `max_ctx` rows, which is
237/// the flat plane, which is no ring at all. The bench box measured what the wrong bound cost:
238/// 4630 usable prompt tokens inside a configured `MEMRA_CTX=8192`, against 7300 with the ring off
239/// on the same binary (research/glm53-flash-bringup-20260827/rebaseline-and-surface-20260828,
240/// receipts 13 and 14).
241///
242/// So `t` was removed from the requirement instead of the constant being raised.
243/// `mla_kpool_indices` DRAINS the ring inside the call: it appends what fits, builds the pool keys
244/// that frees, and continues, so a call of any `t` is served by a ring of any size. The only
245/// correctness floor left is ONE POOL, enforced by the engine because the state plan does not
246/// carry `pool`; everything above it is working set, and the ring can never again be the reason a
247/// prompt is refused.
248///
249/// 5120 is chosen, not derived. It is what the flag already books, so every banked memory number
250/// stays exactly true (1M: 13.5 GiB to 1.56 GiB over 12 MLA layers), it is one nominal 4096-token
251/// prime chunk plus slack so a CHUNKED architecture drains in exactly one iteration and pays zero
252/// extra launches, and at 5 MiB per layer it is noise against the 1 GiB per layer the ring
253/// deletes. A monolithic 1M prime drains in about 205 iterations per MLA layer, two kernel
254/// launches each, against a prefill of a million tokens.
255pub const INDEX_RING_WORKING_ROWS: usize = 5120;
256
257/// Physical rows of the DSA k-pool indexer state plane when it is a TAIL RING, or `None` to keep
258/// the flat `max_ctx`-row plane. PURE: the env read is [`index_ring_rows`].
259///
260/// `explicit` is a parsed `MEMRA_DSA_INDEX_RING`: `Some(0)` disables the ring, `Some(n)` pins the
261/// row budget (gates use a tiny one to reach the wrap in a micro fixture), `None` books the
262/// working-set default. There is deliberately NO per-call `t` input any more: see
263/// [`INDEX_RING_WORKING_ROWS`]. A ring that is not SHORTER than the flat plane is pointless, so
264/// `rows >= max_ctx` collapses to `None` and the short-context sessions that dominate the test
265/// suite keep byte-for-byte their old allocation.
266pub fn index_ring_rows_for(explicit: Option<usize>, max_ctx: usize) -> Option<usize> {
267    let rows = match explicit {
268        Some(0) => return None,
269        Some(n) => n,
270        None => INDEX_RING_WORKING_ROWS,
271    };
272    (rows > 0 && rows < max_ctx).then_some(rows)
273}
274
275/// PHYSICAL rows the SHIPPED DEFAULT derivation books for `max_ctx`: no `MEMRA_DSA_INDEX_RING`
276/// override. Pure, so the sizing gate can assert on it without racing another test's environment.
277/// This is the one function the sizing gate calls.
278pub fn index_ring_default_rows(max_ctx: usize) -> Option<usize> {
279    index_ring_rows_for(None, max_ctx)
280}
281
282/// Rows of `remaining` the indexer may append to the tail ring BEFORE the pool-key build has to
283/// drain it, or `None` when the rows this call still owes an unbuilt pool are already lapped.
284///
285/// `ring` is the EFFECTIVE ring (a multiple of `pool`; `0` is the flat plane), `pools_ready` the
286/// pools whose keys are already resident, `cur` the absolute row the next append lands on, and
287/// `remaining` the rows of this call not yet appended.
288///
289/// THE WHOLE SAFETY ARGUMENT, in one window. The plane has exactly one writer
290/// (`Engine::mla_index_append`) and exactly one reader (`Engine::mla_kpool_pool_keys`), and a row
291/// is read exactly once, by the pool-key build of the pool it belongs to. So the rows that must be
292/// live at any instant are `[pools_ready * pool, cur + take)`: everything below has already been
293/// read and is dead, everything above is not written yet. `take` is whatever is left of the ring
294/// after the carry-over `live = cur - pools_ready * pool`, and the caller drains and comes back.
295///
296/// PROGRESS is guaranteed by `ring >= pool`, which the engine enforces separately, because a build
297/// leaves `live = cur mod pool < pool` behind. The one input that can still make this `None` is a
298/// `pools_ready` that sits further than `ring` below `cur`: a rewind that reduced the cache without
299/// clamping `index_pools_ready`, or a pool-key plane reallocation. Those rows are genuinely gone
300/// and no amount of draining brings them back, so it refuses.
301#[allow(clippy::manual_is_multiple_of)] // allow: divisor is runtime-derived; the modulo form keeps a zero divisor loud (a panic), where is_multiple_of would return false silently
302pub fn index_ring_take(
303    ring: usize,
304    pool: usize,
305    pools_ready: usize,
306    cur: usize,
307    remaining: usize,
308) -> Option<usize> {
309    if ring == 0 {
310        return Some(remaining);
311    }
312    debug_assert!(
313        ring % pool == 0,
314        "the effective ring is a whole number of pools"
315    );
316    let live = cur.checked_sub(pools_ready.saturating_mul(pool))?;
317    (ring > live).then(|| (ring - live).min(remaining))
318}
319
320/// DSA k-pool indexer TAIL RING sizing (`MEMRA_DSA_INDEX_RING`, default ON, see docs/FLAGS.md).
321/// Unparseable values are treated as unset. `MEMRA_PRIME_CHUNK` is NO LONGER READ HERE: the ring
322/// is drained inside the call, so no prefill chunk discipline can size it or break it.
323pub fn index_ring_rows(max_ctx: usize) -> Option<usize> {
324    let explicit = std::env::var("MEMRA_DSA_INDEX_RING")
325        .ok()
326        .and_then(|v| v.trim().parse::<usize>().ok());
327    index_ring_rows_for(explicit, max_ctx)
328}
329
330#[derive(Debug, Clone, Copy, PartialEq, Eq)]
331pub struct KvRing {
332    rows: usize,
333    window: usize,
334    base: usize,
335}
336
337#[derive(Debug, Clone, Copy, PartialEq, Eq)]
338pub enum KvRingAppend {
339    Contiguous {
340        write_row: usize,
341    },
342    Rebase {
343        src_row: usize,
344        keep_rows: usize,
345        new_base: usize,
346        write_row: usize,
347    },
348}
349
350impl KvRing {
351    pub fn new(rows: usize, window: usize) -> Self {
352        assert!(window > 0 && rows > 0, "invalid SWA ring geometry");
353        Self {
354            rows,
355            window,
356            base: 0,
357        }
358    }
359
360    pub fn rows(&self) -> usize {
361        self.rows
362    }
363    pub fn base(&self) -> usize {
364        self.base
365    }
366    pub fn window(&self) -> usize {
367        self.window
368    }
369
370    /// Plan a contiguous physical append. When the tail would wrap, retain the caller's exact
371    /// aligned read prefix at row zero; the following read remains one contiguous CUDA view.
372    pub fn append_plan(
373        &self,
374        len: usize,
375        retain_from: usize,
376        append_rows: usize,
377    ) -> Result<KvRingAppend, String> {
378        if len < self.base || retain_from < self.base || retain_from > len {
379            return Err(format!(
380                "SWA ring lapped required rows (base {}, retain {retain_from}, len {len})",
381                self.base
382            ));
383        }
384        let used = len - self.base;
385        if used > self.rows {
386            return Err(format!(
387                "SWA ring state exceeds capacity ({used} > {})",
388                self.rows
389            ));
390        }
391        if used.saturating_add(append_rows) <= self.rows {
392            return Ok(KvRingAppend::Contiguous {
393                write_row: used % self.rows,
394            });
395        }
396
397        let keep_rows = len - retain_from;
398        if keep_rows.saturating_add(append_rows) > self.rows {
399            return Err(format!(
400                "SWA ring append does not fit (keep {keep_rows} + append {append_rows} > {})",
401                self.rows
402            ));
403        }
404        Ok(KvRingAppend::Rebase {
405            src_row: retain_from - self.base,
406            keep_rows,
407            new_base: retain_from,
408            write_row: keep_rows,
409        })
410    }
411
412    pub fn apply_rebase(&mut self, new_base: usize) {
413        debug_assert!(new_base >= self.base);
414        self.base = new_base;
415    }
416
417    pub fn physical_range(
418        &self,
419        start: usize,
420        end: usize,
421    ) -> Result<std::ops::Range<usize>, String> {
422        if start < self.base || end < start || end - self.base > self.rows {
423            return Err(format!(
424                "SWA ring view [{start},{end}) is outside resident [{},{})",
425                self.base,
426                self.base + self.rows
427            ));
428        }
429        let start_row = (start - self.base) % self.rows;
430        let len = end - start;
431        debug_assert!(
432            start_row + len <= self.rows,
433            "ring view must be contiguous after rebase"
434        );
435        Ok(start_row..start_row + len)
436    }
437
438    /// A rewind is usable only when the next aligned Step35 window view is still resident.
439    pub fn can_rewind_to(&self, len: usize) -> bool {
440        let raw = len.saturating_sub(self.window - 1);
441        let view_start = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
442        view_start >= self.base
443    }
444
445    /// Plan a checkpoint restore into a FRESH ring (base 0): the aligned live window ending at
446    /// absolute `len`, as (new_base, source physical rows). `len` is the ABSOLUTE stream length
447    /// the checkpoint recorded — for a lapped ring it exceeds the physical row count, so a
448    /// restore that copies `len` rows from row zero is an out-of-bounds device slice (the
449    /// 2026-08-29 warm-turn-at-40k GPU-worker panic). Refuses when this ring no longer holds
450    /// the window (checkpoint lapped: the caller must full re-prime).
451    pub fn restore_plan(&self, len: usize) -> Result<(usize, std::ops::Range<usize>), String> {
452        let raw = len.saturating_sub(self.window.saturating_sub(1));
453        let new_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
454        let physical = self.physical_range(new_base, len)?;
455        Ok((new_base, physical))
456    }
457}
458
459// ---------------- the device seam ----------------
460
461/// The 7 device ops the cache needs — nothing more. Implemented by the engine (and by
462/// any future backend); all ops are stream-ordered on the implementor's worker stream.
463pub trait KvDev {
464    fn zeros(&self, n: usize) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
465    fn uninit(&self, n: usize) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
466    fn alloc_u8(&self, n: usize) -> Result<CudaSlice<u8>, Box<dyn std::error::Error>>;
467    fn htod_i32(&self, v: &[i32]) -> Result<CudaSlice<i32>, Box<dyn std::error::Error>>;
468    fn clone_dtod(
469        &self,
470        src: &CudaSlice<f32>,
471    ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
472    fn copy_into(
473        &self,
474        dst: &mut CudaSlice<f32>,
475        off: usize,
476        src: &CudaSlice<f32>,
477        len: usize,
478    ) -> Result<(), Box<dyn std::error::Error>>;
479    /// D2D copy with an offset on BOTH sides. `copy_into` always reads the source from 0, which
480    /// cannot express "copy this window OUT of a tail ring" — the shape the latent-plane
481    /// snapshot/restore needs (lane/glm5-prefix-latent, 2026-08-30).
482    fn copy_range_into(
483        &self,
484        dst: &mut CudaSlice<f32>,
485        dst_off: usize,
486        src: &CudaSlice<f32>,
487        src_off: usize,
488        len: usize,
489    ) -> Result<(), Box<dyn std::error::Error>>;
490    fn set_i32_one(&self, d: &mut CudaSlice<i32>, v: i32)
491    -> Result<(), Box<dyn std::error::Error>>;
492}
493
494use cudarc::driver::CudaSlice;
495use memra_gguf::config::{LayerKind, ModelConfig};
496use memra_gguf::model_plan::{ModelPlan, ResidualTopology, StatePlan};
497
498/// Per-full-attn-layer growing KV cache, resident on GPU. QUANTIZED (KVQUANT-PLAN §B):
499/// K stored q8_0 (34 B/32 elem), V stored q5_1 (24 B/32 elem). Per-token byte layout keeps the
500/// [token, kv_head, dim] element order so a 32-block never straddles a head (assert head_dim%32==0).
501/// Element-within-token index = kv_head*head_dim + d; block = idx/32; lane = idx%32.
502pub struct KvLayer {
503    pub k: CudaSlice<u8>,   // q8_0 packed, capacity max_ctx*k_tok_bytes
504    pub v: CudaSlice<u8>,   // q5_1 packed, capacity max_ctx*v_tok_bytes
505    pub kv_dim_k: usize,    // head_dim_k * n_head_kv  (K elements per token)
506    pub kv_dim_v: usize,    // head_dim_v * n_head_kv  (V elements per token)
507    pub k_tok_bytes: usize, // (kv_dim_k/32)*34
508    pub v_tok_bytes: usize, // (kv_dim_v/32)*24
509    pub len: usize,
510    /// Step35 SWA physical-row state. `len` remains absolute; `None` keeps the original flat
511    /// `[0, max_ctx)` addressing contract.
512    pub ring: Option<KvRing>,
513    /// Device-resident mirror of `len` (CUDA-GRAPH-PLAN Phase 2). Holds the KV write SLOT for the
514    /// append-dc kernel (old len, before this step's append); after `inc_seqlen` it holds the new
515    /// len == t_kv for fa_decode_dc. Kept in lock-step with the host `len`. i32[1].
516    pub len_d: CudaSlice<i32>,
517    /// Device-resident mirror of `ring.base()` (physical row of logical row 0) for the WINDOWED
518    /// device-counter draft arm (`append_kv_quantized_dcw` / `fa_decode_dcw`): the kernels derive
519    /// the SWA view as {lstart = max(0, len - window); physical = row - base} entirely from
520    /// device state, so a captured draft chain replays with zero per-token node updates. Armed
521    /// only on ring-backed draft-scratch planes (step35); `None` keeps the plain `_dc` contract
522    /// (base 0) and costs nothing. The ONE writer is the rebase arm of `prepare_kv_append`
523    /// (rebases are host-side, outside any captured region); rewinds move `len`/`len_d` only,
524    /// never `base`, so no other site touches it. i32[1].
525    pub base_d: Option<CudaSlice<i32>>,
526}
527
528impl KvLayer {
529    pub fn physical_rows(
530        &self,
531        start: usize,
532        end: usize,
533    ) -> Result<std::ops::Range<usize>, String> {
534        match &self.ring {
535            Some(ring) => ring.physical_range(start, end),
536            None => Ok(start..end),
537        }
538    }
539}
540
541/// Per-MLA-layer latent KV plane (DESIGN.md §3.2). ONE row per token, `width` elements wide,
542/// where `width` == `StatePlan::LatentKvCache { width }` == kv_lora_rank + rope_head_dim:
543///   row = [ rmsnorm(c_kv) : kv_rank | rope(k_pe) : d_rope ]
544/// There is NO V plane — V is the FIRST `kv_rank` elements of the SAME row, and every query
545/// head streams that one row (MQA). NoPE models (glm5_next, rope_head_dim 0) have width ==
546/// kv_rank and no k_pe tail.
547///
548/// f32, UNQUANTIZED, deliberately: increment 4 is the correctness arm and its gate is maxdiff
549/// against the `memra_engine::mla` f32 oracle, whose `c_kv` is f32. DESIGN.md §3.2's eventual
550/// q8_0 latent row (576 = 18 blocks, V view boundary 512 = 16 blocks, both on a 32-element
551/// boundary) is a later increment; quantizing here would fork the plane from the oracle it is
552/// gated against. The `% 32 == 0` KVQUANT constraint therefore does NOT apply to this plane.
553pub struct LatentKvLayer {
554    /// [max_ctx * width] f32, row-major by token.
555    pub rows: CudaSlice<f32>,
556    pub width: usize,
557    pub len: usize,
558    /// Device mirror of `len`, kept in lock-step exactly like `KvLayer::len_d`.
559    pub len_d: CudaSlice<i32>,
560    /// DSA k-pool indexer state, [max_ctx * index_width] f32 row-major by token:
561    ///   row = [ k_norm(wk(x)) : index_head_dim | index_kpool_compress_gate(x) : index_head_dim ]
562    /// `None` when the layer declares `index_width == 0` (no k-pool indexer). The reference's
563    /// `past_key_values.update_indexer` carries the same two channels; its third (a per-token
564    /// validity flag) is DELIBERATELY absent — this cache is single-sequence and unpadded, so
565    /// every row below `len` is valid and pooling starts at token 0, which is exactly the scope
566    /// `memra_reference::kpool_allowed_tokens` documents for itself. A batched/padded arm needs
567    /// that channel back and its own gate.
568    ///
569    /// `len` above is authoritative for BOTH planes: they are appended in the same call and
570    /// must never carry independent lengths.
571    ///
572    /// MEMORY — the TAIL RING, and it SHIPPED (`index_ring_rows`, `MEMRA_DSA_INDEX_RING`).
573    /// Flat, this plane is `2 * index_head_dim` = 256 f32 = 1 KiB per token per layer, i.e.
574    /// **12 GiB (12.88 GB)** over glm5_next's 12 MLA layers at 1M — larger than the latent
575    /// plane's share of the same budget is comfortable with. Two reductions were considered:
576    ///   * **f16/bf16 rows — DECLINED.** The rows feed the pool-key softmax, whose output feeds
577    ///     the ReLU score, whose ties the selection order depends on. Halving the mantissa moves
578    ///     scores, and moved scores move which pools win a tie — the one thing the gates forbid.
579    ///     It would need its own selection-parity gate at serving scale before it could ship, and
580    ///     it buys 6 GiB where the option below buys 11.94.
581    ///   * **A tail ring — the real answer, and the one implemented.** With `index_pool_keys`
582    ///     resident (below), a row of this plane is read exactly once: by the pool-key build of
583    ///     the pool it belongs to. Every row under `index_pools_ready * pool` is therefore
584    ///     PROVABLY DEAD — this cache has exactly ONE in-call reader of the plane
585    ///     (`Engine::mla_kpool_pool_keys`) and ONE writer (`Engine::mla_index_append`), and
586    ///     `CacheSnapshot` does not carry latent planes at all, so nothing else can observe a
587    ///     lapped row. (`snapshot_plane`, lane/glm5-prefix-latent, is a second reader BETWEEN
588    ///     calls, and it reads only the LIVE tail window `[index_pools_ready * pool, len)` —
589    ///     the liveness argument is unchanged.) The plane only has to hold the incomplete tail
590    ///     plus whatever slice of the current call is in flight, so a ring of `R` rows with `R`
591    ///     a multiple of `pool` (which keeps each pool contiguous mod `R`) replaces 12 GiB with
592    ///     60 MiB, EXACTLY: same rows, same kernel, different addresses, zero numeric cost,
593    ///     gated by `gpu_kpool_tail_ring_wraps_and_matches_the_flat_plane`.
594    ///     Net before: 12 GiB here + 1.5 GiB of pool keys. Net after: 1.56 GiB, an 8.7x cut,
595    ///     because a pool key is `index_head_dim` f32 per `pool` tokens = 32 f32/token against 256.
596    ///
597    /// `R` DOES NOT BOUND THE PER-CALL `t`, and getting that wrong is what shipped a regression
598    /// (lane/glm53-ring-sizing, 2026-08-28). The first cut sized `R` against the largest `t` a
599    /// CHUNKED prefill could hand one call; glm5_next primes MONOLITHICALLY, so its `t` is the
600    /// whole prompt and the guard refused every prompt past `R`: 4630 usable tokens inside a
601    /// configured 8192. `mla_kpool_indices` now DRAINS the ring inside the call, appending what
602    /// fits and building the pool keys that frees, so `R` is a working-set choice
603    /// ([`INDEX_RING_WORKING_ROWS`]) with a floor of one pool and nothing else.
604    ///
605    /// The indexer's `pool` is the one input the state plan does NOT carry, the same gap that
606    /// makes `index_pool_keys` a lazy allocation below. So `pool` is not used to size the ring:
607    /// the allocator books `index_ring_rows` PHYSICAL rows and the engine rounds that DOWN to a
608    /// multiple of `pool` on first use, so the effective ring is always `>= rows - pool + 1`.
609    pub index_rows: Option<CudaSlice<f32>>,
610    pub index_width: usize,
611    /// PHYSICAL rows of `index_rows` when it is a tail ring; `None` when the plane is flat
612    /// (`max_ctx` rows, absolute row addressing). The EFFECTIVE ring is this rounded down to a
613    /// multiple of the indexer's `pool`, computed by the engine — see the field doc above.
614    pub index_ring_rows: Option<usize>,
615    /// RESIDENT DSA pool-key plane, `[max_ctx / pool * index_head_dim]` f32 row-major by pool.
616    ///
617    /// LAYOUT: `index_pool_keys[p * d + c]` is channel `c` of the collapsed key of pool `p`, i.e.
618    /// of cache rows `[p * pool, (p + 1) * pool)`. `d` is `index_width / 2` (the indexer's head
619    /// dim); `pool` comes from the layer's `MlaIndexerGeom` and is NOT in the state plan, which is
620    /// why this buffer is allocated on FIRST USE by the engine rather than by the allocator below.
621    ///
622    /// INVALIDATION RULE, and it is the whole point: a pool's key is a function of exactly its own
623    /// `pool` rows of `index_rows` plus the layer's constant `kpool_ape`. `index_rows` is
624    /// APPEND-ONLY — a row is written once, when its token is appended, and never rewritten — so
625    /// once a pool's LAST row lands the key is FINAL and is never recomputed. `index_pools_ready`
626    /// is how many leading pools hold such final keys; each call builds only
627    /// `[index_pools_ready, len / pool)` and then advances it. The incomplete tail is NOT a pool
628    /// and has no key: rows `[len / pool * pool, len)` reach the query through the selection
629    /// kernel's `always_tail` append, recomputed every call.
630    ///
631    /// The rule therefore has exactly ONE trigger: if `len` ever DECREASES (a rewind that
632    /// overwrites already-pooled rows), `index_pools_ready` must be clamped to `len / pool` by the
633    /// same code that shortens `len`. Use `truncate_index_pool_keys` for that. Today `len` is
634    /// written in two places (`HybridModel::mla_attn_cached`, and `restore_plane` on a FRESH
635    /// layer) and only ever grows — `Cache::rollback` does not touch the latent planes at all —
636    /// so no caller needs the clamp yet; `mla_kpool_indices` asserts the invariant on every call
637    /// so a future rewind that forgets it fails loudly instead of selecting against stale keys,
638    /// and `snapshot_plane`/`validate_restore` assert it at both prefix-cache seams.
639    pub index_pool_keys: Option<CudaSlice<f32>>,
640    /// Pools `[0, index_pools_ready)` of `index_pool_keys` hold FINAL keys.
641    pub index_pools_ready: usize,
642    /// RESIDENT copy of the indexer's `pool` (tokens per k-pool), the one geometry input the
643    /// state plan does NOT carry (the same gap that makes `index_pool_keys` a lazy allocation).
644    /// `0` until the engine's first indexer call writes it (`mla_attn_cached`, which refuses a
645    /// nonzero value that disagrees with the loaded geometry rather than overwriting it). The
646    /// latent-plane snapshot/restore path (lane/glm5-prefix-latent, 2026-08-30) reads it to
647    /// address the tail ring and size the restored key plane; it refuses to capture a plane
648    /// whose pool is still unknown.
649    pub index_pool: usize,
650}
651
652impl LatentKvLayer {
653    /// Shorten the resident pool-key plane to what `len` still justifies. Call from any path that
654    /// REDUCES `len`; pools at or above `len / pool` may have been built over rows the rewind is
655    /// about to overwrite, so their keys are no longer final.
656    pub fn truncate_index_pool_keys(&mut self, pool: usize) {
657        if pool == 0 {
658            return;
659        }
660        self.index_pools_ready = self.index_pools_ready.min(self.len / pool);
661    }
662}
663
664/// Physical row of absolute row `abs` in an indexer state plane. `ring_rows == 0` is the flat
665/// plane (absolute addressing); otherwise the EFFECTIVE ring is `ring_rows` rounded down to a
666/// whole number of pools, exactly the engine's own rounding (`mla_kpool_indices`), because a
667/// ring that is not a multiple of `pool` would split a pool across the wrap.
668pub fn index_plane_physical_row(ring_rows: usize, pool: usize, abs: usize) -> usize {
669    if ring_rows == 0 {
670        return abs;
671    }
672    debug_assert!(pool > 0, "index plane addressing requires a known pool");
673    let effective = ring_rows / pool * pool;
674    debug_assert!(effective > 0, "the effective ring holds at least one pool");
675    abs % effective
676}
677
678/// One MLA/DSA layer's captured latent-plane state: everything `mla_attn_cached` +
679/// `mla_kpool_indices` need to continue as if the destination session had primed the prefix
680/// itself (lane/glm5-prefix-latent, 2026-08-30; design in
681/// research/glm5-prefix-latent-20260830/DESIGN.md).
682///
683/// The three asymmetries against an ordinary `PrefixPlane`, and how each is carried:
684///   * `rows` is deliberately UNQUANTIZED f32 (the maxdiff oracle depends on the f32 plane), so
685///     the copy is f32-for-f32 — no quantization program is introduced at the snapshot seam.
686///   * `index_rows` is a TAIL RING whose rows below `index_pools_ready * pool` are OVERWRITTEN
687///     by design, so "the index plane" is not copyable and not rebuildable: the snapshot carries
688///     the DERIVED keys (final by the append-only invariant, bit-identical to a rebuild) plus
689///     the `len % pool` still-live tail rows (`index_tail`, at most `pool - 1` rows).
690///   * `index_pool_keys` / `index_pools_ready` carry the append-only finality invariant, so the
691///     capture asserts `index_pools_ready == len / pool` (every call boundary leaves the drain
692///     there) and the restore re-establishes both, keeping the engine's residency tripwire and
693///     `index_ring_take` arithmetic blind to the fact that a restore happened.
694pub struct LatentPlaneSnapshot {
695    /// Rows `[0..len)` of the latent plane, `len * width` f32.
696    pub rows: CudaSlice<f32>,
697    pub width: usize,
698    pub len: usize,
699    /// `0` = the layer has no indexer state plane (and every `index_*` field below is empty).
700    pub index_width: usize,
701    /// The indexer's pool size at capture (`LatentKvLayer::index_pool`); `0` iff no indexer.
702    pub index_pool: usize,
703    /// The live tail-ring rows `[index_pools_ready * pool, len)`, `(len % pool) * index_width`
704    /// f32; `None` when the boundary is pool-aligned.
705    pub index_tail: Option<CudaSlice<f32>>,
706    /// The FINAL pool keys `[0..index_pools_ready * d)`, d = `index_width / 2`; `None` when no
707    /// pool has completed.
708    pub index_pool_keys: Option<CudaSlice<f32>>,
709    pub index_pools_ready: usize,
710}
711
712impl LatentPlaneSnapshot {
713    /// Device bytes this snapshot holds, for the prefix cache's byte ledger. The defective
714    /// pre-lane entry cost ZERO bytes per token; this is the honest bill.
715    pub fn bytes(&self) -> usize {
716        let tail = self.index_tail.as_ref().map_or(0, CudaSlice::len);
717        let keys = self.index_pool_keys.as_ref().map_or(0, CudaSlice::len);
718        (self.rows.len() + tail + keys) * std::mem::size_of::<f32>()
719    }
720}
721
722/// The generation-destroyed slice of one latent layer's BOUNDARY state, captured EAGERLY at a
723/// spec session's prompt boundary (lane/glm5-prefix-latent2, 2026-09-01) so a DEFERRED prefix
724/// publication can be completed later against the live plane:
725///   * the latent `rows` and the FINAL pool keys are append-only BELOW the boundary for the
726///     session's lifetime (the glm5 verify rollback truncates back to the accepted length,
727///     never below the prime boundary), so `snapshot_plane_at` slices them from the LIVE
728///     layer at publish time — no eager copy of the big planes;
729///   * the incomplete tail-ring rows are read-once and OVERWRITTEN by the very next pool
730///     build, so they travel HERE or the boundary is unrecoverable by publish time (the KDA
731///     conv/ssm half of the same problem rides the sibling `CacheSnapshot`).
732pub struct LatentTailCapture {
733    /// Boundary length (== capture pos) — the row count the deferred publisher slices.
734    pub len: usize,
735    /// Latent width at capture; the publish-time slice validates it against the live layer.
736    pub width: usize,
737    pub index_width: usize,
738    /// The indexer's pool size at capture (`0` iff no indexer plane).
739    pub index_pool: usize,
740    /// `len / pool` at the boundary (the capture asserts the drain invariant, same as
741    /// `snapshot_plane`).
742    pub index_pools_ready: usize,
743    /// The live tail-ring rows `[pools_ready * pool, len)` at the boundary,
744    /// `(len % pool) * index_width` f32; `None` when the boundary is pool-aligned (or the
745    /// layer has no indexer).
746    pub index_tail: Option<CudaSlice<f32>>,
747}
748
749impl LatentTailCapture {
750    /// Device bytes held eagerly (the tail only — the big planes are sliced at publish).
751    pub fn bytes(&self) -> usize {
752        self.index_tail.as_ref().map_or(0, CudaSlice::len) * std::mem::size_of::<f32>()
753    }
754}
755
756impl LatentKvLayer {
757    /// Deep-copy this layer's latent-plane state OUT of a live session cache. Stream-ordered on
758    /// the implementor's worker stream, like every other prefix-capture copy. Errors instead of
759    /// capturing anything a restore could not make whole:
760    ///   * `len == 0` (the caller records an unexecuted layer as absent instead),
761    ///   * an indexer plane whose `pool` was never resolved,
762    ///   * `index_pools_ready != len / pool` — a capture off a drained call boundary would
763    ///     publish keys that are behind or ahead of their rows (the finality invariant).
764    pub fn snapshot_plane(
765        &self,
766        e: &impl KvDev,
767    ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
768        let (len, width) = (self.len, self.width);
769        if len == 0 {
770            return Err("latent snapshot at len 0 (record the layer as absent instead)".into());
771        }
772        if self.rows.len() < len * width {
773            return Err(format!(
774                "latent plane holds {} f32 but len {len} x width {width} requires {}",
775                self.rows.len(),
776                len * width,
777            )
778            .into());
779        }
780        let mut rows = e.uninit(len * width)?;
781        e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
782        if self.index_width == 0 {
783            return Ok(LatentPlaneSnapshot {
784                rows,
785                width,
786                len,
787                index_width: 0,
788                index_pool: 0,
789                index_tail: None,
790                index_pool_keys: None,
791                index_pools_ready: 0,
792            });
793        }
794        let pool = self.index_pool;
795        if pool == 0 {
796            return Err(format!(
797                "latent snapshot: index plane (width {}) has an unresolved pool — no indexer \
798                 call ran against this layer, so its derived state cannot be validated",
799                self.index_width,
800            )
801            .into());
802        }
803        let d = self.index_width / 2;
804        let pools_ready = self.index_pools_ready;
805        if pools_ready != len / pool {
806            return Err(format!(
807                "latent snapshot: index_pools_ready {pools_ready} != len/pool {} (len {len}, \
808                 pool {pool}); a capture must sit at a drained call boundary or its keys \
809                 violate the append-only finality invariant",
810                len / pool,
811            )
812            .into());
813        }
814        let index_pool_keys = if pools_ready > 0 {
815            let src = self
816                .index_pool_keys
817                .as_ref()
818                .ok_or("latent snapshot: pools are ready but the resident key plane is gone")?;
819            if src.len() < pools_ready * d {
820                return Err(format!(
821                    "latent snapshot: resident key plane holds {} f32 but {pools_ready} pools \
822                     x d {d} require {}",
823                    src.len(),
824                    pools_ready * d,
825                )
826                .into());
827            }
828            let mut keys = e.uninit(pools_ready * d)?;
829            e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
830            Some(keys)
831        } else {
832            None
833        };
834        let tail_rows = len - pools_ready * pool;
835        let index_tail = if tail_rows > 0 {
836            let src = self
837                .index_rows
838                .as_ref()
839                .ok_or("latent snapshot: index_width > 0 but the state plane is gone")?;
840            let ring = self.index_ring_rows.unwrap_or(0);
841            // The tail starts pool-aligned and is shorter than one pool, and the effective ring
842            // is a whole number of pools, so the window is contiguous in ring and flat layouts.
843            let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
844            let want = (phys + tail_rows) * self.index_width;
845            if src.len() < want {
846                return Err(format!(
847                    "latent snapshot: index plane holds {} f32 but the live tail window \
848                     requires {want}",
849                    src.len(),
850                )
851                .into());
852            }
853            let mut tail = e.uninit(tail_rows * self.index_width)?;
854            e.copy_range_into(
855                &mut tail,
856                0,
857                src,
858                phys * self.index_width,
859                tail_rows * self.index_width,
860            )?;
861            Some(tail)
862        } else {
863            None
864        };
865        Ok(LatentPlaneSnapshot {
866            rows,
867            width,
868            len,
869            index_width: self.index_width,
870            index_pool: pool,
871            index_tail,
872            index_pool_keys,
873            index_pools_ready: pools_ready,
874        })
875    }
876
877    /// EAGER half of the deferred boundary capture (doc on [`LatentTailCapture`]): copy out
878    /// only what generation will destroy — the incomplete tail-ring rows — plus the boundary
879    /// metadata the publish-time slice validates against. Same preconditions as
880    /// `snapshot_plane` (len > 0, resolved pool, the pools-ready drain invariant); the big
881    /// planes are NOT copied here.
882    pub fn snapshot_tail(
883        &self,
884        e: &impl KvDev,
885    ) -> Result<LatentTailCapture, Box<dyn std::error::Error>> {
886        let (len, width) = (self.len, self.width);
887        if len == 0 {
888            return Err("latent tail capture at len 0 (record the layer as absent instead)".into());
889        }
890        if self.index_width == 0 {
891            return Ok(LatentTailCapture {
892                len,
893                width,
894                index_width: 0,
895                index_pool: 0,
896                index_pools_ready: 0,
897                index_tail: None,
898            });
899        }
900        let pool = self.index_pool;
901        if pool == 0 {
902            return Err(format!(
903                "latent tail capture: index plane (width {}) has an unresolved pool — no \
904                 indexer call ran against this layer, so its derived state cannot be validated",
905                self.index_width,
906            )
907            .into());
908        }
909        let pools_ready = self.index_pools_ready;
910        if pools_ready != len / pool {
911            return Err(format!(
912                "latent tail capture: index_pools_ready {pools_ready} != len/pool {} (len \
913                 {len}, pool {pool}); a capture must sit at a drained call boundary",
914                len / pool,
915            )
916            .into());
917        }
918        let tail_rows = len - pools_ready * pool;
919        let index_tail = if tail_rows > 0 {
920            let src = self
921                .index_rows
922                .as_ref()
923                .ok_or("latent tail capture: index_width > 0 but the state plane is gone")?;
924            let ring = self.index_ring_rows.unwrap_or(0);
925            let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
926            let want = (phys + tail_rows) * self.index_width;
927            if src.len() < want {
928                return Err(format!(
929                    "latent tail capture: index plane holds {} f32 but the live tail window \
930                     requires {want}",
931                    src.len(),
932                )
933                .into());
934            }
935            let mut tail = e.uninit(tail_rows * self.index_width)?;
936            e.copy_range_into(
937                &mut tail,
938                0,
939                src,
940                phys * self.index_width,
941                tail_rows * self.index_width,
942            )?;
943            Some(tail)
944        } else {
945            None
946        };
947        Ok(LatentTailCapture {
948            len,
949            width,
950            index_width: self.index_width,
951            index_pool: pool,
952            index_pools_ready: pools_ready,
953            index_tail,
954        })
955    }
956
957    /// DEFERRED half of the boundary capture: complete a [`LatentPlaneSnapshot`] at the
958    /// captured boundary by slicing the append-only planes (`rows` `[0..cap.len)`, FINAL pool
959    /// keys `[0..cap.index_pools_ready * d)`) from the LIVE layer and moving the eagerly
960    /// captured tail in. Every disagreement between the capture and the live layer refuses —
961    /// a publication is an optimization and must never publish planes it cannot prove are the
962    /// boundary's (the append-only-below-boundary invariant is what makes the slice legal:
963    /// the glm5 verify rollback truncates to the accepted length, never below the prime
964    /// boundary, and pool keys are final the instant their last row lands).
965    pub fn snapshot_plane_at(
966        &self,
967        e: &impl KvDev,
968        cap: LatentTailCapture,
969    ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
970        let (len, width) = (cap.len, cap.width);
971        if len == 0 {
972            return Err("latent boundary publish at len 0".into());
973        }
974        if width != self.width {
975            return Err(format!(
976                "latent boundary publish: captured width {width} != live width {}",
977                self.width,
978            )
979            .into());
980        }
981        if self.len < len {
982            return Err(format!(
983                "latent boundary publish: live len {} < boundary {len} — the plane was \
984                 truncated below the capture boundary",
985                self.len,
986            )
987            .into());
988        }
989        if self.rows.len() < len * width {
990            return Err(format!(
991                "latent boundary publish: live plane holds {} f32 but boundary {len} x width \
992                 {width} requires {}",
993                self.rows.len(),
994                len * width,
995            )
996            .into());
997        }
998        let mut rows = e.uninit(len * width)?;
999        e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
1000        if cap.index_width != self.index_width {
1001            return Err(format!(
1002                "latent boundary publish: captured index_width {} != live {}",
1003                cap.index_width, self.index_width,
1004            )
1005            .into());
1006        }
1007        if cap.index_width == 0 {
1008            return Ok(LatentPlaneSnapshot {
1009                rows,
1010                width,
1011                len,
1012                index_width: 0,
1013                index_pool: 0,
1014                index_tail: None,
1015                index_pool_keys: None,
1016                index_pools_ready: 0,
1017            });
1018        }
1019        if cap.index_pool != self.index_pool {
1020            return Err(format!(
1021                "latent boundary publish: captured pool {} != live pool {}",
1022                cap.index_pool, self.index_pool,
1023            )
1024            .into());
1025        }
1026        let d = cap.index_width / 2;
1027        let pools_ready = cap.index_pools_ready;
1028        if self.index_pools_ready < pools_ready {
1029            return Err(format!(
1030                "latent boundary publish: live index_pools_ready {} < boundary {pools_ready} \
1031                 — the key plane was clamped below the capture boundary",
1032                self.index_pools_ready,
1033            )
1034            .into());
1035        }
1036        let index_pool_keys = if pools_ready > 0 {
1037            let src = self
1038                .index_pool_keys
1039                .as_ref()
1040                .ok_or("latent boundary publish: pools are ready but the key plane is gone")?;
1041            if src.len() < pools_ready * d {
1042                return Err(format!(
1043                    "latent boundary publish: key plane holds {} f32 but {pools_ready} pools \
1044                     x d {d} require {}",
1045                    src.len(),
1046                    pools_ready * d,
1047                )
1048                .into());
1049            }
1050            let mut keys = e.uninit(pools_ready * d)?;
1051            e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
1052            Some(keys)
1053        } else {
1054            None
1055        };
1056        Ok(LatentPlaneSnapshot {
1057            rows,
1058            width,
1059            len,
1060            index_width: cap.index_width,
1061            index_pool: cap.index_pool,
1062            index_tail: cap.index_tail,
1063            index_pool_keys,
1064            index_pools_ready: pools_ready,
1065        })
1066    }
1067
1068    /// Device-independent half of the restore preflight: every shape/identity/bounds check, no
1069    /// copies, so the caller can validate EVERY layer before the first byte moves (a malformed
1070    /// entry must never leave a half-restored cache for a fallback to consume).
1071    pub fn validate_restore(
1072        &self,
1073        snap: &LatentPlaneSnapshot,
1074        max_ctx: usize,
1075    ) -> Result<(), String> {
1076        if self.len != 0 {
1077            return Err("restore destination latent plane is not fresh".into());
1078        }
1079        if self.width != snap.width {
1080            return Err(format!(
1081                "snapshot width {} != destination width {}",
1082                snap.width, self.width,
1083            ));
1084        }
1085        if snap.len == 0 || snap.len > max_ctx {
1086            return Err(format!("snapshot len {} outside [1,{max_ctx}]", snap.len));
1087        }
1088        if snap.rows.len() < snap.len * snap.width {
1089            return Err(format!(
1090                "snapshot rows plane holds {} f32 but len {} x width {} requires {} \
1091                 (truncated capture)",
1092                snap.rows.len(),
1093                snap.len,
1094                snap.width,
1095                snap.len * snap.width,
1096            ));
1097        }
1098        if self.rows.len() < snap.len * self.width {
1099            return Err(format!(
1100                "destination latent plane holds {} f32 but the restore requires {}",
1101                self.rows.len(),
1102                snap.len * self.width,
1103            ));
1104        }
1105        if self.index_width != snap.index_width {
1106            return Err(format!(
1107                "snapshot index width {} != destination {}",
1108                snap.index_width, self.index_width,
1109            ));
1110        }
1111        if snap.index_width == 0 {
1112            return Ok(());
1113        }
1114        let pool = snap.index_pool;
1115        if pool == 0 {
1116            return Err("snapshot carries an index plane with an unresolved pool".into());
1117        }
1118        if self.index_pool != 0 && self.index_pool != pool {
1119            return Err(format!(
1120                "snapshot pool {pool} != destination resident pool {}",
1121                self.index_pool,
1122            ));
1123        }
1124        let d = snap.index_width / 2;
1125        if snap.index_pools_ready != snap.len / pool {
1126            return Err(format!(
1127                "snapshot index_pools_ready {} != len/pool {} (len {}, pool {pool}): the \
1128                 append-only finality invariant does not hold, so its keys are stale",
1129                snap.index_pools_ready,
1130                snap.len / pool,
1131                snap.len,
1132            ));
1133        }
1134        match (&snap.index_pool_keys, snap.index_pools_ready) {
1135            (Some(keys), ready @ 1..) => {
1136                if keys.len() < ready * d {
1137                    return Err(format!(
1138                        "snapshot key plane holds {} f32 but {ready} pools x d {d} require {}",
1139                        keys.len(),
1140                        ready * d,
1141                    ));
1142                }
1143            }
1144            (None, 0) => {}
1145            (Some(_), 0) => return Err("snapshot carries keys for zero ready pools".into()),
1146            (None, ready) => {
1147                return Err(format!(
1148                    "snapshot claims {ready} ready pools but carries no keys"
1149                ));
1150            }
1151        }
1152        let tail_rows = snap.len - snap.index_pools_ready * pool;
1153        match (&snap.index_tail, tail_rows) {
1154            (Some(tail), rows @ 1..) => {
1155                if tail.len() < rows * snap.index_width {
1156                    return Err(format!(
1157                        "snapshot tail holds {} f32 but {rows} rows x index width {} require {}",
1158                        tail.len(),
1159                        snap.index_width,
1160                        rows * snap.index_width,
1161                    ));
1162                }
1163            }
1164            (None, 0) => {}
1165            (Some(_), 0) => return Err("snapshot carries a tail at a pool-aligned boundary".into()),
1166            (None, rows) => {
1167                return Err(format!(
1168                    "snapshot owes {rows} live tail rows but carries none"
1169                ));
1170            }
1171        }
1172        if self.index_rows.is_none() {
1173            return Err("destination declares an index plane but allocated none".into());
1174        }
1175        if tail_rows > 0 {
1176            let ring = self.index_ring_rows.unwrap_or(0);
1177            let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
1178            let want = (phys + tail_rows) * self.index_width;
1179            let have = self.index_rows.as_ref().map_or(0, CudaSlice::len);
1180            if have < want {
1181                return Err(format!(
1182                    "destination index plane holds {have} f32 but the tail window requires \
1183                     {want}",
1184                ));
1185            }
1186        }
1187        Ok(())
1188    }
1189
1190    /// Deep-copy a snapshot INTO this freshly allocated layer: latent rows at `[0..len)`,
1191    /// `len` + device mirror, and (for indexer-bearing layers) the resident key plane sized to
1192    /// the SESSION's capacity — exactly the `capacity_tokens / pool * d` sizing
1193    /// `mla_kpool_indices` books, so the next call keeps it resident instead of reallocating
1194    /// (a reallocation resets `index_pools_ready` and, under the ring, the rows to rebuild the
1195    /// keys from are gone) — plus `index_pools_ready` and the live tail rows at their physical
1196    /// ring (or flat) addresses. Validation runs first; a shape error moves no bytes.
1197    pub fn restore_plane(
1198        &mut self,
1199        e: &impl KvDev,
1200        snap: &LatentPlaneSnapshot,
1201        max_ctx: usize,
1202    ) -> Result<(), Box<dyn std::error::Error>> {
1203        self.validate_restore(snap, max_ctx)?;
1204        e.copy_range_into(&mut self.rows, 0, &snap.rows, 0, snap.len * snap.width)?;
1205        if snap.index_width > 0 {
1206            let pool = snap.index_pool;
1207            let d = snap.index_width / 2;
1208            // `zeros`, not `uninit`: unbuilt key slots must not carry garbage a diagnostic
1209            // D2H could mistake for state. The engine only ever reads `[0..pools_ready * d)`.
1210            let mut keys = e.zeros(((max_ctx / pool) * d).max(1))?;
1211            if let Some(src) = &snap.index_pool_keys {
1212                e.copy_range_into(&mut keys, 0, src, 0, snap.index_pools_ready * d)?;
1213            }
1214            self.index_pool_keys = Some(keys);
1215            self.index_pools_ready = snap.index_pools_ready;
1216            self.index_pool = pool;
1217            if let Some(tail) = &snap.index_tail {
1218                let tail_rows = snap.len - snap.index_pools_ready * pool;
1219                let ring = self.index_ring_rows.unwrap_or(0);
1220                let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
1221                let dst = self
1222                    .index_rows
1223                    .as_mut()
1224                    .ok_or("destination index plane vanished after validation")?;
1225                e.copy_range_into(
1226                    dst,
1227                    phys * self.index_width,
1228                    tail,
1229                    0,
1230                    tail_rows * self.index_width,
1231                )?;
1232            }
1233        }
1234        self.len = snap.len;
1235        let len_i32 = i32::try_from(snap.len).map_err(|_| "latent length exceeds i32 mirror")?;
1236        e.set_i32_one(&mut self.len_d, len_i32)?;
1237        Ok(())
1238    }
1239}
1240
1241/// Per-linear-attn-layer fixed recurrent state.
1242/// conv_state and ssm_state are BOTH kept RESIDENT on GPU — the conv ring assemble + roll runs
1243/// on-device (conv_assemble_and_roll), so there is no per-step dtoh/htod for either.
1244pub struct RecurLayer {
1245    pub conv_state: CudaSlice<f32>, // GPU [conv_dim, d_conv-1] (channel c, tap j at c*pad + j)
1246    pub ssm_state: CudaSlice<f32>,  // GPU [d_state, d_state, num_v] transposed M[col][i]
1247    /// PERSISTENT second SSM-state buffer for the gdn-scan double buffer (DECODE DETERMINISM FIX).
1248    /// gdn_scan needs DISTINCT in/out state buffers. The old eager path allocated a fresh
1249    /// `state_scratch` via `e.uninit` every step and swapped its pointer into `ssm_state`; that
1250    /// per-step alloc/free churned the stream-ordered async pool, and the freed prior `ssm_state`
1251    /// block was recycled by the next step's scratch while a kernel referencing the swapped-in state
1252    /// was still in flight — a use-after-reuse that produced RUN-TO-RUN nondeterministic decode
1253    /// (two identical prompt primes diverged). We instead PING-PONG between two STABLE resident
1254    /// buffers (no per-step alloc/free, no pool churn): step writes into the spare, then swaps the
1255    /// two owned buffers in place. Stable pointers, identical math. Sized like `ssm_state`.
1256    pub ssm_state_alt: CudaSlice<f32>,
1257}
1258
1259pub struct ResidentTpKvCacheRank {
1260    k: CudaSlice<u8>,
1261    v: CudaSlice<u8>,
1262    len_d: CudaSlice<i32>,
1263    /// Physical row of LOGICAL row 0 after the last ring rebase (graph increment A: the
1264    /// windowed device-counter fa derives its view as {lstart = max(0, len - window);
1265    /// physical = lstart - base}). Host-written at rebase (rare) and at cache init; None
1266    /// until the graph door first arms it.
1267    base_d: Option<CudaSlice<i32>>,
1268}
1269
1270impl ResidentTpKvCacheRank {
1271    pub fn new(k: CudaSlice<u8>, v: CudaSlice<u8>, len_d: CudaSlice<i32>) -> Self {
1272        Self {
1273            k,
1274            v,
1275            len_d,
1276            base_d: None,
1277        }
1278    }
1279
1280    pub fn base_d(&self) -> Option<&CudaSlice<i32>> {
1281        self.base_d.as_ref()
1282    }
1283
1284    pub fn base_d_mut(&mut self) -> Option<&mut CudaSlice<i32>> {
1285        self.base_d.as_mut()
1286    }
1287
1288    pub fn arm_base_d(&mut self, buf: CudaSlice<i32>) {
1289        self.base_d = Some(buf);
1290    }
1291
1292    pub fn k(&self) -> &CudaSlice<u8> {
1293        &self.k
1294    }
1295
1296    pub fn v(&self) -> &CudaSlice<u8> {
1297        &self.v
1298    }
1299
1300    pub fn len_d(&self) -> &CudaSlice<i32> {
1301        &self.len_d
1302    }
1303
1304    pub fn k_mut(&mut self) -> &mut CudaSlice<u8> {
1305        &mut self.k
1306    }
1307
1308    pub fn v_mut(&mut self) -> &mut CudaSlice<u8> {
1309        &mut self.v
1310    }
1311
1312    pub fn planes_mut(&mut self) -> (&mut CudaSlice<u8>, &mut CudaSlice<u8>) {
1313        (&mut self.k, &mut self.v)
1314    }
1315
1316    /// Split-borrow for the dcw append: both planes mutably plus the device counters shared.
1317    #[allow(clippy::type_complexity)]
1318    pub fn planes_and_counters_mut(
1319        &mut self,
1320    ) -> (
1321        &mut CudaSlice<u8>,
1322        &mut CudaSlice<u8>,
1323        &CudaSlice<i32>,
1324        Option<&CudaSlice<i32>>,
1325    ) {
1326        (&mut self.k, &mut self.v, &self.len_d, self.base_d.as_ref())
1327    }
1328
1329    pub fn len_d_mut(&mut self) -> &mut CudaSlice<i32> {
1330        &mut self.len_d
1331    }
1332}
1333
1334#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1335pub struct TpKvTransaction {
1336    generation: u64,
1337    base_len: usize,
1338}
1339
1340impl TpKvTransaction {
1341    pub fn generation(self) -> u64 {
1342        self.generation
1343    }
1344
1345    pub fn base_len(self) -> usize {
1346        self.base_len
1347    }
1348}
1349
1350#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1351pub struct TpKvAppendPlan {
1352    transaction: TpKvTransaction,
1353    target: usize,
1354    write_row: usize,
1355    ring_append: Option<KvRingAppend>,
1356}
1357
1358impl TpKvAppendPlan {
1359    pub fn target(self) -> usize {
1360        self.target
1361    }
1362
1363    pub fn write_row(self) -> usize {
1364        self.write_row
1365    }
1366
1367    pub fn ring_append(self) -> Option<KvRingAppend> {
1368        self.ring_append
1369    }
1370}
1371
1372#[derive(Debug, PartialEq, Eq)]
1373pub struct TpKvGrowPlan {
1374    rows: usize,
1375    source_row: usize,
1376    copy_rows: usize,
1377    target_base: usize,
1378    k_bytes: usize,
1379    v_bytes: usize,
1380    source_capacity: usize,
1381    target_capacity: usize,
1382    ring_window: Option<usize>,
1383    target_physical_rows: usize,
1384    kv_dim_k: usize,
1385    kv_dim_v: usize,
1386    k_tok_bytes: usize,
1387    v_tok_bytes: usize,
1388    ranks: usize,
1389    next_generation: u64,
1390}
1391
1392impl TpKvGrowPlan {
1393    pub fn rows(&self) -> usize {
1394        self.rows
1395    }
1396
1397    pub fn source_row(&self) -> usize {
1398        self.source_row
1399    }
1400
1401    pub fn copy_rows(&self) -> usize {
1402        self.copy_rows
1403    }
1404
1405    pub fn k_bytes(&self) -> usize {
1406        self.k_bytes
1407    }
1408
1409    pub fn v_bytes(&self) -> usize {
1410        self.v_bytes
1411    }
1412}
1413
1414#[derive(Clone, Debug, PartialEq, Eq)]
1415struct TpKvTransactionState {
1416    committed_len: usize,
1417    staged_len: usize,
1418    next_generation: u64,
1419    active: Option<TpKvTransaction>,
1420}
1421
1422impl TpKvTransactionState {
1423    fn new() -> Self {
1424        Self {
1425            committed_len: 0,
1426            staged_len: 0,
1427            next_generation: 1,
1428            active: None,
1429        }
1430    }
1431
1432    fn begin(&mut self) -> Result<TpKvTransaction, String> {
1433        if let Some(active) = self.active {
1434            return Err(format!(
1435                "TP KV transaction generation {} is already active at base {}",
1436                active.generation, active.base_len
1437            ));
1438        }
1439        if self.staged_len != self.committed_len {
1440            return Err(format!(
1441                "TP KV cache is half-committed: staged {} != committed {}",
1442                self.staged_len, self.committed_len
1443            ));
1444        }
1445        let transaction = TpKvTransaction {
1446            generation: self.next_generation,
1447            base_len: self.committed_len,
1448        };
1449        self.next_generation = self
1450            .next_generation
1451            .checked_add(1)
1452            .ok_or("TP KV transaction generation overflow")?;
1453        self.active = Some(transaction);
1454        Ok(transaction)
1455    }
1456
1457    fn validate(&self, transaction: TpKvTransaction) -> Result<(), String> {
1458        if self.active != Some(transaction) {
1459            return Err(format!(
1460                "stale TP KV transaction generation {} at base {}",
1461                transaction.generation, transaction.base_len
1462            ));
1463        }
1464        if transaction.base_len != self.committed_len {
1465            return Err(format!(
1466                "TP KV transaction base {} != committed length {}",
1467                transaction.base_len, self.committed_len
1468            ));
1469        }
1470        Ok(())
1471    }
1472
1473    fn append_target(
1474        &self,
1475        transaction: TpKvTransaction,
1476        rows: usize,
1477        capacity: usize,
1478    ) -> Result<usize, String> {
1479        self.validate(transaction)?;
1480        if rows == 0 {
1481            return Err("TP KV append must contain at least one row".into());
1482        }
1483        let target = self
1484            .staged_len
1485            .checked_add(rows)
1486            .ok_or("TP KV staged length overflow")?;
1487        if target > capacity {
1488            return Err(format!(
1489                "TP KV append exceeds capacity: {target} > {capacity}"
1490            ));
1491        }
1492        Ok(target)
1493    }
1494
1495    fn publish_append(
1496        &mut self,
1497        transaction: TpKvTransaction,
1498        target: usize,
1499    ) -> Result<(), String> {
1500        self.validate(transaction)?;
1501        if target <= self.staged_len {
1502            return Err(format!(
1503                "TP KV append target {target} must exceed staged length {}",
1504                self.staged_len
1505            ));
1506        }
1507        self.staged_len = target;
1508        Ok(())
1509    }
1510
1511    fn commit_target(
1512        &self,
1513        transaction: TpKvTransaction,
1514        accepted_rows: usize,
1515    ) -> Result<usize, String> {
1516        self.validate(transaction)?;
1517        let staged_rows = self
1518            .staged_len
1519            .checked_sub(transaction.base_len)
1520            .ok_or("TP KV staged length precedes its transaction base")?;
1521        if accepted_rows > staged_rows {
1522            return Err(format!(
1523                "TP KV commit accepts {accepted_rows} rows from a {staged_rows}-row transaction"
1524            ));
1525        }
1526        transaction
1527            .base_len
1528            .checked_add(accepted_rows)
1529            .ok_or_else(|| "TP KV committed length overflow".to_string())
1530    }
1531
1532    fn publish_finalize(
1533        &mut self,
1534        transaction: TpKvTransaction,
1535        target: usize,
1536    ) -> Result<(), String> {
1537        self.validate(transaction)?;
1538        if target < transaction.base_len || target > self.staged_len {
1539            return Err(format!(
1540                "TP KV finalize target {target} outside transaction range {}..={}",
1541                transaction.base_len, self.staged_len
1542            ));
1543        }
1544        self.committed_len = target;
1545        self.staged_len = target;
1546        self.active = None;
1547        Ok(())
1548    }
1549
1550    fn rewind(&mut self, target: usize, capacity: usize) -> Result<(), String> {
1551        if target > capacity {
1552            return Err(format!(
1553                "TP KV rewind target {target} exceeds capacity {capacity}"
1554            ));
1555        }
1556        self.committed_len = target;
1557        self.staged_len = target;
1558        self.active = None;
1559        Ok(())
1560    }
1561}
1562
1563pub struct ResidentTpKvCache {
1564    ranks: Vec<ResidentTpKvCacheRank>,
1565    kv_dim_k: usize,
1566    kv_dim_v: usize,
1567    k_tok_bytes: usize,
1568    v_tok_bytes: usize,
1569    capacity: usize,
1570    ring: Option<KvRing>,
1571    state: TpKvTransactionState,
1572}
1573
1574impl ResidentTpKvCache {
1575    #[allow(clippy::too_many_arguments)]
1576    pub fn new(
1577        ranks: Vec<ResidentTpKvCacheRank>,
1578        kv_dim_k: usize,
1579        kv_dim_v: usize,
1580        k_tok_bytes: usize,
1581        v_tok_bytes: usize,
1582        capacity: usize,
1583    ) -> Self {
1584        Self::new_inner(
1585            ranks,
1586            kv_dim_k,
1587            kv_dim_v,
1588            k_tok_bytes,
1589            v_tok_bytes,
1590            capacity,
1591            None,
1592        )
1593    }
1594
1595    #[allow(clippy::too_many_arguments)]
1596    pub fn new_swa(
1597        ranks: Vec<ResidentTpKvCacheRank>,
1598        kv_dim_k: usize,
1599        kv_dim_v: usize,
1600        k_tok_bytes: usize,
1601        v_tok_bytes: usize,
1602        capacity: usize,
1603        window: usize,
1604    ) -> Self {
1605        Self::new_inner(
1606            ranks,
1607            kv_dim_k,
1608            kv_dim_v,
1609            k_tok_bytes,
1610            v_tok_bytes,
1611            capacity,
1612            Some(KvRing::new(swa_ring_rows(window, capacity), window)),
1613        )
1614    }
1615
1616    #[allow(clippy::too_many_arguments)]
1617    fn new_inner(
1618        ranks: Vec<ResidentTpKvCacheRank>,
1619        kv_dim_k: usize,
1620        kv_dim_v: usize,
1621        k_tok_bytes: usize,
1622        v_tok_bytes: usize,
1623        capacity: usize,
1624        ring: Option<KvRing>,
1625    ) -> Self {
1626        Self {
1627            ranks,
1628            kv_dim_k,
1629            kv_dim_v,
1630            k_tok_bytes,
1631            v_tok_bytes,
1632            capacity,
1633            ring,
1634            state: TpKvTransactionState::new(),
1635        }
1636    }
1637
1638    pub fn begin_transaction(&mut self) -> Result<TpKvTransaction, String> {
1639        self.state.begin()
1640    }
1641
1642    pub fn committed_len(&self) -> usize {
1643        self.state.committed_len
1644    }
1645
1646    pub fn staged_len(&self) -> usize {
1647        self.state.staged_len
1648    }
1649
1650    pub fn capacity(&self) -> usize {
1651        self.capacity
1652    }
1653
1654    pub fn physical_capacity(&self) -> usize {
1655        self.ring
1656            .as_ref()
1657            .map(KvRing::rows)
1658            .unwrap_or(self.capacity)
1659    }
1660
1661    pub fn ring_window(&self) -> Option<usize> {
1662        self.ring.as_ref().map(KvRing::window)
1663    }
1664
1665    pub fn ring_base(&self) -> Option<usize> {
1666        self.ring.as_ref().map(KvRing::base)
1667    }
1668
1669    pub fn physical_range(
1670        &self,
1671        start: usize,
1672        end: usize,
1673    ) -> Result<std::ops::Range<usize>, String> {
1674        match &self.ring {
1675            Some(ring) => ring.physical_range(start, end),
1676            None => {
1677                if end < start || end > self.capacity {
1678                    return Err(format!(
1679                        "TP KV linear view [{start},{end}) exceeds capacity {}",
1680                        self.capacity
1681                    ));
1682                }
1683                Ok(start..end)
1684            }
1685        }
1686    }
1687
1688    pub fn can_rewind_to(&self, target: usize) -> bool {
1689        target <= self.capacity
1690            && self
1691                .ring
1692                .as_ref()
1693                .is_none_or(|ring| ring.can_rewind_to(target))
1694    }
1695
1696    pub fn kv_dim_k(&self) -> usize {
1697        self.kv_dim_k
1698    }
1699
1700    pub fn kv_dim_v(&self) -> usize {
1701        self.kv_dim_v
1702    }
1703
1704    pub fn k_tok_bytes(&self) -> usize {
1705        self.k_tok_bytes
1706    }
1707
1708    pub fn v_tok_bytes(&self) -> usize {
1709        self.v_tok_bytes
1710    }
1711
1712    pub fn ranks_len(&self) -> usize {
1713        self.ranks.len()
1714    }
1715
1716    pub fn rank(&self, rank: usize) -> Option<&ResidentTpKvCacheRank> {
1717        self.ranks.get(rank)
1718    }
1719
1720    pub fn rank_mut(&mut self, rank: usize) -> Option<&mut ResidentTpKvCacheRank> {
1721        self.ranks.get_mut(rank)
1722    }
1723
1724    pub fn ranks(&self) -> &[ResidentTpKvCacheRank] {
1725        &self.ranks
1726    }
1727
1728    pub fn ranks_mut(&mut self) -> &mut [ResidentTpKvCacheRank] {
1729        &mut self.ranks
1730    }
1731
1732    pub fn prepare_grow(
1733        &self,
1734        target_capacity: usize,
1735        rows: usize,
1736    ) -> Result<TpKvGrowPlan, String> {
1737        if let Some(active) = self.state.active {
1738            return Err(format!(
1739                "TP KV grow refuses active transaction generation {} at base {}",
1740                active.generation, active.base_len
1741            ));
1742        }
1743        if self.state.staged_len != self.state.committed_len {
1744            return Err(format!(
1745                "TP KV grow requires quiescent state, got committed/staged={}/{}",
1746                self.state.committed_len, self.state.staged_len
1747            ));
1748        }
1749        if target_capacity <= self.capacity {
1750            return Err(format!(
1751                "TP KV grow target capacity {target_capacity} must exceed source capacity {}",
1752                self.capacity
1753            ));
1754        }
1755        if target_capacity > i32::MAX as usize {
1756            return Err(format!(
1757                "TP KV grow target capacity {target_capacity} exceeds i32 device mirrors"
1758            ));
1759        }
1760        if rows > self.state.committed_len {
1761            return Err(format!(
1762                "TP KV grow rows {rows} exceed committed length {}",
1763                self.state.committed_len
1764            ));
1765        }
1766        let (source_row, copy_rows, target_base, ring_window, target_physical_rows) =
1767            match &self.ring {
1768                Some(ring) => {
1769                    let raw = rows.saturating_sub(ring.window().saturating_sub(1));
1770                    let target_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1771                    let physical = ring.physical_range(target_base, rows)?;
1772                    (
1773                        physical.start,
1774                        physical.len(),
1775                        target_base,
1776                        Some(ring.window()),
1777                        swa_ring_rows(ring.window(), target_capacity),
1778                    )
1779                }
1780                None => (0, rows, 0, None, target_capacity),
1781            };
1782        let k_bytes = copy_rows
1783            .checked_mul(self.k_tok_bytes)
1784            .ok_or("TP KV grow K byte extent overflow")?;
1785        let v_bytes = copy_rows
1786            .checked_mul(self.v_tok_bytes)
1787            .ok_or("TP KV grow V byte extent overflow")?;
1788        Ok(TpKvGrowPlan {
1789            rows,
1790            source_row,
1791            copy_rows,
1792            target_base,
1793            k_bytes,
1794            v_bytes,
1795            source_capacity: self.capacity,
1796            target_capacity,
1797            ring_window,
1798            target_physical_rows,
1799            kv_dim_k: self.kv_dim_k,
1800            kv_dim_v: self.kv_dim_v,
1801            k_tok_bytes: self.k_tok_bytes,
1802            v_tok_bytes: self.v_tok_bytes,
1803            ranks: self.ranks.len(),
1804            next_generation: self.state.next_generation,
1805        })
1806    }
1807
1808    pub fn publish_grow(&mut self, plan: TpKvGrowPlan) -> Result<(), String> {
1809        if self.state != TpKvTransactionState::new() {
1810            return Err(format!(
1811                "TP KV grow target must be fresh, got committed/staged={}/{} active={}",
1812                self.state.committed_len,
1813                self.state.staged_len,
1814                self.state.active.is_some()
1815            ));
1816        }
1817        if self.capacity != plan.target_capacity
1818            || self.capacity <= plan.source_capacity
1819            || self.kv_dim_k != plan.kv_dim_k
1820            || self.kv_dim_v != plan.kv_dim_v
1821            || self.k_tok_bytes != plan.k_tok_bytes
1822            || self.v_tok_bytes != plan.v_tok_bytes
1823            || self.ranks.len() != plan.ranks
1824            || self.ring.as_ref().map(KvRing::window) != plan.ring_window
1825            || self.physical_capacity() != plan.target_physical_rows
1826        {
1827            return Err("TP KV grow target layout does not match its source plan".into());
1828        }
1829        if plan.rows > self.capacity {
1830            return Err(format!(
1831                "TP KV grow rows {} exceed target capacity {}",
1832                plan.rows, self.capacity
1833            ));
1834        }
1835        if let Some(ring) = self.ring.as_mut() {
1836            let mut target_ring = *ring;
1837            target_ring.apply_rebase(plan.target_base);
1838            if !target_ring.can_rewind_to(plan.rows) {
1839                return Err(format!(
1840                    "TP KV grow target ring base {} cannot expose committed length {}",
1841                    target_ring.base(),
1842                    plan.rows
1843                ));
1844            }
1845            *ring = target_ring;
1846        }
1847        self.state.committed_len = plan.rows;
1848        self.state.staged_len = plan.rows;
1849        self.state.next_generation = plan.next_generation;
1850        self.state.active = None;
1851        Ok(())
1852    }
1853
1854    pub fn prepare_append(
1855        &self,
1856        transaction: TpKvTransaction,
1857        rows: usize,
1858    ) -> Result<TpKvAppendPlan, String> {
1859        let target = self.state.append_target(transaction, rows, self.capacity)?;
1860        let ring_append = self
1861            .ring
1862            .as_ref()
1863            .map(|ring| {
1864                let staged_retain =
1865                    target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1866                let rollback_retain = transaction
1867                    .base_len
1868                    .saturating_sub(ring.window().saturating_sub(1))
1869                    & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1870                ring.append_plan(
1871                    self.state.staged_len,
1872                    staged_retain.min(rollback_retain),
1873                    rows,
1874                )
1875            })
1876            .transpose()?;
1877        let write_row = match ring_append {
1878            Some(KvRingAppend::Contiguous { write_row })
1879            | Some(KvRingAppend::Rebase { write_row, .. }) => write_row,
1880            None => self.state.staged_len,
1881        };
1882        Ok(TpKvAppendPlan {
1883            transaction,
1884            target,
1885            write_row,
1886            ring_append,
1887        })
1888    }
1889
1890    /// Read-only peek at the NEXT append's ring plan: (write_row, would_rebase). The dcw
1891    /// (device-counter) append path uses it to route rebase tokens through the full host
1892    /// path — the in-kernel row (len - base) is only valid for contiguous appends.
1893    pub fn peek_append_ring(&self, rows: usize) -> Result<(usize, bool), String> {
1894        let target = self.state.staged_len + rows;
1895        let plan = self
1896            .ring
1897            .as_ref()
1898            .map(|ring| {
1899                let staged_retain =
1900                    target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1901                let rollback_retain = self
1902                    .state
1903                    .staged_len
1904                    .saturating_sub(ring.window().saturating_sub(1))
1905                    & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1906                ring.append_plan(
1907                    self.state.staged_len,
1908                    staged_retain.min(rollback_retain),
1909                    rows,
1910                )
1911            })
1912            .transpose()?;
1913        Ok(match plan {
1914            Some(KvRingAppend::Contiguous { write_row }) => (write_row, false),
1915            Some(KvRingAppend::Rebase { write_row, .. }) => (write_row, true),
1916            None => (self.state.staged_len, false),
1917        })
1918    }
1919
1920    pub fn publish_append_rebase(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1921        self.state.validate(plan.transaction)?;
1922        match (self.ring.as_mut(), plan.ring_append) {
1923            (
1924                Some(ring),
1925                Some(KvRingAppend::Rebase {
1926                    new_base,
1927                    keep_rows,
1928                    ..
1929                }),
1930            ) => {
1931                if keep_rows > ring.rows() {
1932                    return Err(format!(
1933                        "TP KV ring rebase keeps {keep_rows} rows in {} physical rows",
1934                        ring.rows()
1935                    ));
1936                }
1937                let mut target_ring = *ring;
1938                target_ring.apply_rebase(new_base);
1939                if !target_ring.can_rewind_to(plan.transaction.base_len) {
1940                    return Err(format!(
1941                        "TP KV ring rebase to {new_base} laps transaction base {}",
1942                        plan.transaction.base_len
1943                    ));
1944                }
1945                *ring = target_ring;
1946                Ok(())
1947            }
1948            (Some(_), Some(KvRingAppend::Contiguous { .. })) | (None, None) => Ok(()),
1949            _ => Err("TP KV append plan does not match cache ring layout".into()),
1950        }
1951    }
1952
1953    pub fn publish_append_plan(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1954        if let Some(KvRingAppend::Rebase { new_base, .. }) = plan.ring_append {
1955            if self.ring.as_ref().map(KvRing::base) != Some(new_base) {
1956                return Err(format!(
1957                    "TP KV append rebase {new_base} was not published before its state"
1958                ));
1959            }
1960        }
1961        self.state.publish_append(plan.transaction, plan.target)
1962    }
1963
1964    pub fn publish_hydration(
1965        &mut self,
1966        logical_len: usize,
1967        resident_start: usize,
1968    ) -> Result<(), Box<dyn std::error::Error>> {
1969        if self.state != TpKvTransactionState::new() {
1970            return Err("TP KV hydration target must be fresh".into());
1971        }
1972        if resident_start > logical_len || logical_len > self.capacity {
1973            return Err(format!(
1974                "TP KV hydration range [{resident_start},{logical_len}) exceeds capacity {}",
1975                self.capacity
1976            )
1977            .into());
1978        }
1979        match self.ring.as_mut() {
1980            Some(ring) => {
1981                let rows = logical_len - resident_start;
1982                if rows > ring.rows() {
1983                    return Err(format!(
1984                        "TP KV hydration requires {rows} rows in a {}-row ring",
1985                        ring.rows()
1986                    )
1987                    .into());
1988                }
1989                let mut hydrated_ring = *ring;
1990                hydrated_ring.apply_rebase(resident_start);
1991                if !hydrated_ring.can_rewind_to(logical_len) {
1992                    return Err(format!(
1993                        "TP KV hydration base {resident_start} cannot expose logical length \
1994                         {logical_len}"
1995                    )
1996                    .into());
1997                }
1998                *ring = hydrated_ring;
1999            }
2000            None if resident_start != 0 => {
2001                return Err("linear TP KV hydration must start at absolute row zero".into());
2002            }
2003            None => {}
2004        }
2005        self.rewind_to(logical_len)
2006    }
2007
2008    pub fn append_target(
2009        &self,
2010        transaction: TpKvTransaction,
2011        rows: usize,
2012    ) -> Result<usize, String> {
2013        self.state.append_target(transaction, rows, self.capacity)
2014    }
2015
2016    pub fn publish_append(
2017        &mut self,
2018        transaction: TpKvTransaction,
2019        target: usize,
2020    ) -> Result<(), String> {
2021        self.state.publish_append(transaction, target)
2022    }
2023
2024    pub fn commit_target(
2025        &self,
2026        transaction: TpKvTransaction,
2027        accepted_rows: usize,
2028    ) -> Result<usize, String> {
2029        self.state.commit_target(transaction, accepted_rows)
2030    }
2031
2032    pub fn validate_transaction(&self, transaction: TpKvTransaction) -> Result<(), String> {
2033        self.state.validate(transaction)
2034    }
2035
2036    pub fn publish_finalize(
2037        &mut self,
2038        transaction: TpKvTransaction,
2039        target: usize,
2040    ) -> Result<(), String> {
2041        if !self.can_rewind_to(target) {
2042            return Err(format!(
2043                "TP KV finalize target {target} is outside the resident cache window/capacity"
2044            ));
2045        }
2046        self.state.publish_finalize(transaction, target)
2047    }
2048
2049    pub fn rewind_to(&mut self, target: usize) -> Result<(), Box<dyn std::error::Error>> {
2050        if !self.can_rewind_to(target) {
2051            return Err(format!(
2052                "TP KV rewind target {target} is outside the resident cache window/capacity"
2053            )
2054            .into());
2055        }
2056        let target_i32 =
2057            i32::try_from(target).map_err(|_| "TP KV length exceeds i32 device mirror")?;
2058        for rank in &mut self.ranks {
2059            let stream = rank.len_d.stream().clone();
2060            stream.memcpy_htod(&[target_i32], &mut rank.len_d)?;
2061        }
2062        self.state.rewind(target, self.capacity)?;
2063        Ok(())
2064    }
2065}
2066
2067pub struct Cache {
2068    pub kv: Vec<Option<KvLayer>>,
2069    pub recur: Vec<Option<RecurLayer>>,
2070    /// Per-layer MLA latent KV plane (`StatePlan::LatentKvCache`). `None` on every non-MLA
2071    /// layer, so `iter().flatten()` loops skip them the way they skip `kv`/`recur` holes.
2072    pub latent: Vec<Option<LatentKvLayer>>,
2073    /// Optional per-layer tensor-parallel KV planes. The ordinary owning-stage cache remains
2074    /// allocated as the rollback oracle until the distributed serving path is fully qualified.
2075    pub tp_kv: Vec<Option<ResidentTpKvCache>>,
2076    /// glm5 TP (`MEMRA_GLM5_TP`) per-layer, per-rank KDA state planes: `[rank 0 (root),
2077    /// rank 1, ...]` shard-geometry conv ring + ssm ping-pong, lazily hydrated by the
2078    /// engine's TP walk on first touch (the kpool-plane precedent). The canonical
2079    /// `recur[il]` planes stay allocated untouched (full-width; never read by the TP walk).
2080    /// `None` everywhere the seam is off. The prefix-cache snapshot seams REFUSE while any
2081    /// slot is live (per-rank planes are not carried by CacheSnapshot); the SPEC
2082    /// verify/rollback seam is WIRED for these planes since lane/glm5-composition
2083    /// (admitted behind MEMRA_GLM5_SPEC_TP, default OFF) — the snapshot refusal is now a
2084    /// live runtime guard, never dead code.
2085    pub glm5_tp_recur: Vec<Option<Vec<RecurLayer>>>,
2086    /// glm5 TP PEER replicas of the MLA latent+indexer plane (replicated deterministic
2087    /// compute: every rank appends identical bytes in the same calls), one per peer rank
2088    /// (`[i]` = rank `i + 1`). The canonical `latent[il]` IS the root replica. Lazily
2089    /// hydrated like the field above.
2090    pub glm5_tp_latent_peer: Vec<Option<Vec<LatentKvLayer>>>,
2091    pub pos: usize,
2092    pub max_ctx: usize,
2093    /// A failed multi-stage wave may have advanced only a prefix of layers/rows. Such state is
2094    /// not a legal rollback point and must never be retried or returned to a reuse pool.
2095    pub tainted: bool,
2096    /// BATCHED-TICK increment 2 component 3 (lean logits, 2026-08-01): device-side park of
2097    /// this session's LAST logits row. Device-sampled rows in the batched serving tick skip
2098    /// the [n_vocab] logits D2H entirely; the tick instead dtod-copies the row here (device
2099    /// bandwidth, ~µs) so the ONE consumer that truly needs the final row — the KV-reuse
2100    /// pool's park-at-retire (an empty-suffix resume samples from parked last_logits) —
2101    /// can D2H it once at retire. Lazily allocated on the first lean tick; None on every
2102    /// non-lean path (zero cost). Travels with the Cache into the reuse pool.
2103    pub last_logits_dev: Option<CudaSlice<f32>>,
2104    /// DFlash tap sink (dflash lane, 2026-07-13): when armed, the gemma4 verify/prime
2105    /// trunks copy the residual stream AFTER each tapped layer into `buf` rows
2106    /// ([t, n_taps*hidden] row-major — the drafter fc input layout). None on every
2107    /// non-dflash path (zero cost).
2108    pub dflash_taps: Option<DflashTapSink>,
2109    /// HC-contract tap sink (glm5 DFlash2 draft source, 2026-08-30): when armed, the
2110    /// HyperConnections prime/verify walks write the STREAM-MEAN (`hc_contract`) of each
2111    /// tapped layer's completed output into HOST rows — see [`HcTapSink`]. Host-resident by
2112    /// design: under a ppN split the tapped layers span stage devices, and the drafter
2113    /// consumes the rows on the head engine; a host sink makes the seam placement-invariant
2114    /// (the probe's capture seam was host-side too). None on every non-dflash2 path
2115    /// (zero cost: one Option check per layer).
2116    pub hc_taps: Option<HcTapSink>,
2117}
2118
2119/// The context-linear K/V layout for one full-attention layer. This is the single sizing source
2120/// used by both `Cache::new_inner` and `cache_bytes_per_token`: admission must never reimplement
2121/// Gemma's per-layer geometry or the active KV-format doors independently from the allocator.
2122#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2123enum FullAttentionClass {
2124    Ordinary,
2125    GemmaGlobal,
2126    GemmaWindowed,
2127}
2128
2129fn full_attention_class(plan: &ModelPlan, il: u32) -> FullAttentionClass {
2130    let layer = plan
2131        .layers
2132        .iter()
2133        .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2134        .find(|layer| layer.index == il)
2135        .unwrap_or_else(|| panic!("ModelPlan has no layer {il}"));
2136    if !matches!(layer.residual, ResidualTopology::Gemma { .. }) {
2137        return FullAttentionClass::Ordinary;
2138    }
2139    match layer.state {
2140        StatePlan::SlidingKvCache { .. } => FullAttentionClass::GemmaWindowed,
2141        StatePlan::KvCache { .. } => FullAttentionClass::GemmaGlobal,
2142        _ => panic!("Gemma layer {il} does not declare a KV-cache state"),
2143    }
2144}
2145
2146fn full_attention_kv_layout(
2147    cfg: &ModelConfig,
2148    plan: &ModelPlan,
2149    il: u32,
2150) -> (usize, usize, usize, usize) {
2151    debug_assert_eq!(cfg.layer_kind(il), LayerKind::FullAttention);
2152    let class = full_attention_class(plan, il);
2153    let n_head_kv = cfg.n_head_kv as usize;
2154    let (kv_dim_k, kv_dim_v) = match class {
2155        FullAttentionClass::GemmaGlobal | FullAttentionClass::GemmaWindowed => {
2156            let g = cfg
2157                .gemma4
2158                .as_ref()
2159                .expect("Gemma ModelPlan layer requires Gemma cache geometry");
2160            let hd = match class {
2161                FullAttentionClass::GemmaWindowed => g.key_length_swa,
2162                FullAttentionClass::GemmaGlobal => g.key_length_global,
2163                FullAttentionClass::Ordinary => unreachable!(),
2164            } as usize;
2165            // E4B ships a SCALAR head_count_kv (per-layer vec empty; scalar = 2 in
2166            // the gguf, landing in cfg.n_head_kv): kv_dim = hd * 2 for BOTH kinds —
2167            // swa 2x256 = 512, global 2x512 = 1024. The old fallback used
2168            // key_length_global (512) for both, which HALVED the global layers' K/V
2169            // (the attn writes wk.out_features = 1024 rows): every E4B global layer
2170            // stored/attended half its K/V and the batched append read row strides
2171            // wrong — THE cross-mode maxdiff-30 root (2026-07-12 bisect, il=5 slot-1
2172            // byte forensics). 26B/31B keep the per-layer vec.
2173            let d = match g.head_count_kv.get(il as usize) {
2174                Some(n) => hd * *n as usize,
2175                None => hd * n_head_kv,
2176            };
2177            (d, d)
2178        }
2179        FullAttentionClass::Ordinary => (
2180            cfg.head_dim_k as usize * n_head_kv,
2181            cfg.head_dim_v as usize * n_head_kv,
2182        ),
2183    };
2184    assert!(
2185        kv_dim_k % 32 == 0 && kv_dim_v % 32 == 0,
2186        "KVQUANT requires per-layer kv_dim_k%32==0 && kv_dim_v%32==0 \
2187         (layer {il}: k={kv_dim_k} v={kv_dim_v})"
2188    );
2189    let (kbb, vbb) = kv_blk_bytes();
2190    let g4_global_fp8 = gkv_on() && class == FullAttentionClass::GemmaGlobal;
2191    let g4_windowed_fp8 = wkv_on() && class == FullAttentionClass::GemmaWindowed;
2192    let qwen_fp8 = kv_fp8_on() && class == FullAttentionClass::Ordinary;
2193    let (kbb_l, vbb_l) = if g4_global_fp8 || g4_windowed_fp8 || qwen_fp8 {
2194        (32, 32)
2195    } else {
2196        (kbb, vbb)
2197    };
2198    (kv_dim_k, kv_dim_v, kbb_l, vbb_l)
2199}
2200
2201fn kv_plane_allocation_bytes(rows: usize, token_bytes: usize) -> usize {
2202    rows * token_bytes + 8
2203}
2204
2205/// Context-linear bytes allocated by one trunk cache token.
2206///
2207/// Fixed allocations (the 8-byte plane tail pads, `len_d`, recurrent state, and optional lazy
2208/// buffers) are deliberately excluded. Admission adds their measured high-water residual as a
2209/// request-independent activation term; multiplying this coefficient by the request's own
2210/// `ctx_cap` exactly mirrors the context-scaled allocations in `Cache::new_inner`.
2211pub fn cache_bytes_per_token(cfg: &ModelConfig) -> usize {
2212    cache_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2213}
2214
2215/// Context-linear cache bytes per token owned by layers in `[lo, hi)`. PP admission uses the
2216/// same layer ranges as `Cache::new_ppn`, so each device is charged for exactly the cache planes
2217/// it allocates rather than for the aggregate model geometry.
2218pub fn cache_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2219    let plan = ModelPlan::compile(cfg).expect("cache sizing requires a compilable ModelPlan");
2220    cache_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2221}
2222
2223pub fn cache_bytes_per_token_for_plan(
2224    cfg: &ModelConfig,
2225    plan: &ModelPlan,
2226    lo: usize,
2227    hi: usize,
2228) -> usize {
2229    assert!(
2230        lo <= hi && hi <= cfg.n_layer as usize,
2231        "cache layer range out of bounds"
2232    );
2233    let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2234    let full_attn: usize = (lo as u32..hi as u32)
2235        .filter(|&il| cfg.layer_kind(il) == LayerKind::FullAttention)
2236        .filter(|&il| shared == 0 || il < cfg.n_layer - shared)
2237        .map(|il| {
2238            let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, il);
2239            (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2240        })
2241        .sum();
2242    full_attn + latent_kv_bytes_per_token_for_plan(cfg, plan, lo, hi)
2243}
2244
2245/// Context-linear bytes per token owned by `StatePlan::LatentKvCache` layers in `[lo, hi)`,
2246/// mirroring `Cache::new_inner`'s latent arm plus the engine's lazy resident pool-key plane
2247/// (lane/glm5-gpf-workspace, 2026-08-30).
2248///
2249/// UNTIL THIS TERM EXISTED, glm5_next's admission coefficient was literally 0 B/token: the
2250/// per-token sum above matches `LayerKind::FullAttention` KV planes only, its 34 KDA layers are
2251/// `Recurrent` (correctly 0/token), and its 11 MLA layers are `LatentKvCache` — unmatched. The
2252/// 262k 2-card cell (`research/glm53-flash-bringup-20260827/262k-2card-20260830/`) banked the
2253/// resulting receipt line (`request cost: ... = 0 B/token x ctx + 155MB fixed`): admission
2254/// admitted prompts the device could never serve and the failure surface was a mid-stream
2255/// engine OOM. The prefix-latent lane named the same accounting hole.
2256///
2257/// Terms, each anchored on the allocation it mirrors:
2258///   * latent rows: `width` f32 per token per layer (`Cache::new_inner`,
2259///     `rows: e.zeros(max_ctx * width)` — eager, ctx-scaled).
2260///   * resident k-pool keys: `index_head_dim` f32 per POOL of tokens per layer
2261///     (`mla_kpool_indices`, lazy `capacity_pools * d` — ctx-scaled). `pool` is not in the
2262///     state plan; it comes from `cfg.glm5` (`index_kpool`). A latent plan without that config
2263///     charges pool = 1, which only ever over-reserves.
2264///   * the flat indexer state plane: `index_width` f32 per token per layer, charged ONLY when
2265///     the tail ring is explicitly disabled (`MEMRA_DSA_INDEX_RING=0` -> flat `max_ctx` rows).
2266///     With the ring on (default), the plane is a fixed working set
2267///     ([`INDEX_RING_WORKING_ROWS`]) and belongs to admission's fixed-residual class. (At
2268///     `max_ctx` below the ring rows the allocator also books a flat plane; that plane is
2269///     smaller than the ring's fixed bytes, so leaving it to the residual class only
2270///     under-counts a bounded, small amount.)
2271///
2272/// Every family whose plan compiles no `LatentKvCache` layer gets 0 from this function —
2273/// their coefficient is byte-identical to the pre-lane behavior.
2274pub fn latent_kv_bytes_per_token_for_plan(
2275    cfg: &ModelConfig,
2276    plan: &ModelPlan,
2277    lo: usize,
2278    hi: usize,
2279) -> usize {
2280    let ring_disabled = std::env::var("MEMRA_DSA_INDEX_RING")
2281        .ok()
2282        .and_then(|v| v.trim().parse::<usize>().ok())
2283        == Some(0);
2284    plan.layers
2285        .iter()
2286        .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2287        .map(|layer| match layer.state {
2288            StatePlan::LatentKvCache { width, index_width } => {
2289                let latent = width as usize * std::mem::size_of::<f32>();
2290                let index_width = index_width as usize;
2291                let pool = cfg
2292                    .glm5
2293                    .as_ref()
2294                    .map(|g| g.index_kpool as usize)
2295                    .filter(|&p| p > 0)
2296                    .unwrap_or(1);
2297                // One pool key of `index_head_dim = index_width / 2` f32 per `pool` tokens.
2298                let pool_keys = if index_width > 0 {
2299                    (index_width / 2) * std::mem::size_of::<f32>() / pool
2300                } else {
2301                    0
2302                };
2303                let flat_plane = if index_width > 0 && ring_disabled {
2304                    index_width * std::mem::size_of::<f32>()
2305                } else {
2306                    0
2307                };
2308                latent + pool_keys + flat_plane
2309            }
2310            _ => 0,
2311        })
2312        .sum()
2313}
2314
2315/// Portion of [`cache_bytes_per_token`] whose physical row count is capped by the Step35 SWA
2316/// ring. Zero with the flag off and for every non-Step35 architecture.
2317pub fn cache_ring_bytes_per_token(cfg: &ModelConfig) -> usize {
2318    cache_ring_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2319}
2320
2321/// Ring-capped portion of [`cache_bytes_per_token_for_layers`] for `[lo, hi)`.
2322pub fn cache_ring_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2323    assert!(
2324        lo <= hi && hi <= cfg.n_layer as usize,
2325        "cache layer range out of bounds"
2326    );
2327    let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2328        return 0;
2329    };
2330    cache_ring_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2331}
2332
2333pub fn cache_ring_bytes_per_token_for_plan(
2334    cfg: &ModelConfig,
2335    plan: &ModelPlan,
2336    lo: usize,
2337    hi: usize,
2338) -> usize {
2339    let total = plan.layers.len() + plan.mtp_blocks.len();
2340    assert!(
2341        lo <= hi && hi <= total,
2342        "cache plan layer range out of bounds"
2343    );
2344    if !swa_ring_on() {
2345        return 0;
2346    }
2347    let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2348    plan.layers
2349        .iter()
2350        .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2351        .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2352        .filter(|layer| {
2353            matches!(
2354                layer.state,
2355                memra_gguf::model_plan::StatePlan::SlidingKvCache { .. }
2356            )
2357        })
2358        .filter(|layer| shared == 0 || layer.index < cfg.n_layer - shared)
2359        .map(|layer| {
2360            let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, layer.index);
2361            (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2362        })
2363        .sum()
2364}
2365
2366/// Physical row cap shared by the Step35 SWA trunk and MTP scratch; zero when no ring is active.
2367pub fn cache_ring_row_cap(cfg: &ModelConfig) -> usize {
2368    let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2369        return 0;
2370    };
2371    cache_ring_row_cap_for_plan(&plan)
2372}
2373
2374pub fn cache_ring_row_cap_for_plan(plan: &memra_gguf::model_plan::ModelPlan) -> usize {
2375    if !swa_ring_on() {
2376        return 0;
2377    }
2378    plan.layers
2379        .iter()
2380        .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2381        .filter_map(|layer| match layer.state {
2382            memra_gguf::model_plan::StatePlan::SlidingKvCache { window, .. } => {
2383                Some(window as usize)
2384            }
2385            _ => None,
2386        })
2387        .map(|window| swa_ring_rows(window, usize::MAX))
2388        .max()
2389        .unwrap_or(0)
2390}
2391
2392/// See [`Cache::dflash_taps`]. Armed per forward by the dflash round (t = that forward's
2393/// row count); the trunk writes tap slot s of row r at buf[r*n_taps*hidden + s*hidden ..].
2394pub struct DflashTapSink {
2395    pub layer_ids: Vec<usize>,
2396    pub buf: CudaSlice<f32>,
2397    pub hidden: usize,
2398    pub t: usize,
2399    /// Row offset for writers that walk the buffer in windows (the qwen chunked prime):
2400    /// tap rows land at [base..base+t_chunk). Whole-buffer writers leave it 0.
2401    pub base: usize,
2402}
2403
2404/// See [`Cache::hc_taps`]. Armed per walk by the glm5 DFlash2 draft source; the hc trunk
2405/// writes the CONTRACTED (stream-mean) completed output of tapped layer `layer_ids[s]` for
2406/// walk row r at `rows[(base + r) * n_taps * hidden + s * hidden ..][..hidden]` — the
2407/// drafter fc's input layout, measured by the dflash2 probe's capture seam
2408/// (research/glm53-flash-bringup-20260827/dflash2-probe-20260829/: stream-mean of the
2409/// completed layer output == the SGLang glm5_next hc_contract aux-hidden definition).
2410pub struct HcTapSink {
2411    /// Plan layer indices whose COMPLETED output is tapped, in drafter fc slot order.
2412    pub layer_ids: Vec<usize>,
2413    /// Host rows, `[t, n_taps * hidden]` row-major.
2414    pub rows: Vec<f32>,
2415    pub hidden: usize,
2416    /// Total rows the sink covers.
2417    pub t: usize,
2418    /// Row offset of the CURRENT walk's row 0 (chunked primes set it per chunk; the verify
2419    /// walk leaves it 0).
2420    pub base: usize,
2421    /// ABSOLUTE position of sink row 0 (lane/glm5-prefix-latent2, 2026-09-01): a SUFFIX
2422    /// prime over a restored cache writes at `cache.pos`-derived bases starting at the
2423    /// restored boundary, while its sink covers only the suffix rows — the writer lands
2424    /// row r of a walk at sink row `base - origin + r`. Fresh-prompt sinks leave it 0
2425    /// (byte-identical indexing to before the field existed).
2426    pub origin: usize,
2427    /// DEVICE STAGING (lane/glm5-loop-port, 2026-08-30): one optional `[t * hidden]` buffer
2428    /// per tap slot, allocated lazily by the walk ON THE WRITING engine's device (under a
2429    /// ppN split each tapped layer belongs to exactly one stage, so a slot's buffer lives
2430    /// where its layer runs). When `device_stage` is set the trunk walk D2D-copies the
2431    /// contracted rows here instead of blocking on a mid-walk DtoH — the five in-walk host
2432    /// syncs the 3way window priced into the fixed round cost (map row #17) — and the
2433    /// round drains every slot into `rows` at its ONE post-walk sync point.
2434    pub dev: Vec<Option<CudaSlice<f32>>>,
2435    /// Arm device staging. Verify-round sinks set it; PRIME sinks stay host-staged BY
2436    /// DESIGN — a `[prompt, hidden]` per-slot device transient at 16k-prompt depth is
2437    /// ~1.3 GiB of VRAM the prime must not hold, and the prime's per-chunk DtoH amortizes
2438    /// over >= 256 rows (DFlash2 TTFT is near-constant already, 3way cell 4).
2439    pub device_stage: bool,
2440}
2441
2442impl HcTapSink {
2443    pub fn new(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2444        let n_taps = layer_ids.len();
2445        Self {
2446            layer_ids,
2447            rows: vec![0.0; t * n_taps * hidden],
2448            hidden,
2449            t,
2450            base: 0,
2451            origin: 0,
2452            dev: (0..n_taps).map(|_| None).collect(),
2453            device_stage: false,
2454        }
2455    }
2456
2457    /// Suffix-prime sink (doc on [`Self::origin`]): covers `t` rows whose first row sits at
2458    /// absolute position `origin` — the restored-boundary continuation shape.
2459    pub fn new_at(layer_ids: Vec<usize>, hidden: usize, t: usize, origin: usize) -> Self {
2460        Self {
2461            origin,
2462            ..Self::new(layer_ids, hidden, t)
2463        }
2464    }
2465
2466    /// Device-staged sink (doc on [`Self::device_stage`]): the walk stages tap rows on
2467    /// device and the consumer drains them post-walk in one sync.
2468    pub fn new_device_staged(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2469        Self {
2470            device_stage: true,
2471            ..Self::new(layer_ids, hidden, t)
2472        }
2473    }
2474}
2475
2476/// Snapshot of the dual cache taken BEFORE a spec-decode draft+verify round (MTP-PLAN §C/§D.4).
2477/// - Full-attn KV: only the per-layer `len` is recorded; rollback truncates (append-only,
2478///   position-addressed — no copy). C.1.
2479/// - Linear-attn conv/ssm: real device-to-device COPIES of the recurrent state, because those
2480///   buffers are mutated IN PLACE by the verify pass and have no position index to truncate. C.2.
2481///   (We alloc fresh + memcpy_dtod. NOTE, corrected memra-next#23: the parenthetical here used
2482///   to justify that with "CudaSlice::clone is an Arc refcount, NOT a buffer copy", which is
2483///   false in the LOCKED cudarc 0.19.8 — `Clone` is `try_clone().unwrap()` = alloc + D2D copy. The explicit
2484///   copy is still the right call here, for two reasons that are NOT aliasing: it is fallible
2485///   rather than panicking, and it places the copy on the calling engine's current stream instead
2486///   of the source slice's. Genuine aliasing needs an `Arc<CudaSlice<T>>`.)
2487///
2488/// IT COVERS TWO OF THE CACHE'S FOUR STATE PLANES, AND THAT IS A KNOWN HOLE
2489/// (lane/prefix-restore-toolcall, 2026-08-28). `Cache` also has `tp_kv` (recorded here as
2490/// `tp_kv_len`) and `latent`, and NOTHING in this struct or in `Cache::rollback` mentions
2491/// `latent`. A `StatePlan::LatentKvCache` layer keeps its FULL-ATTENTION history there, so
2492/// rolling back a latent-bearing cache moves `pos` while every MLA layer keeps its longer
2493/// `len`: the next tokens append past the boundary and attend stale rows. The identical
2494/// two-plane assumption in the server's `PrefixEntry` is what made a glm5_next prefix-cache
2495/// hit restore an EMPTY attention history while reporting `cached_tokens: N of N`, and it
2496/// fabricated instead of failing (research/prefix-restore-toolcall-20260828/).
2497///
2498/// Today nothing reaches it: `maybe_plain_checkpoint` refuses to arm on a latent-bearing
2499/// cache, and the spec rewind cannot fire because every latent model is EAGER-ONLY with no
2500/// drafter. IT BECOMES LIVE THE MOMENT A LATENT MODEL GETS A SPEC ARM. Growing latent
2501/// awareness here is not a symmetric addition: the rows are unquantized f32, `index_rows` is
2502/// a tail ring rather than a flat addressable plane, and `index_pool_keys` /
2503/// `index_pools_ready` carry an append-only finality invariant (`truncate_index_pool_keys`
2504/// exists precisely because a `len` that moves backwards invalidates them).
2505pub struct CacheSnapshot {
2506    pub kv_len: Vec<Option<usize>>, // per layer (Some for full-attn layers)
2507    pub tp_kv_len: Vec<Option<usize>>, // per layer (Some for TP full-attn layers)
2508    pub conv: Vec<Option<CudaSlice<f32>>>, // per layer (Some for linear-attn layers, D2D copy)
2509    pub ssm: Vec<Option<CudaSlice<f32>>>,
2510    pub pos: usize,
2511}
2512
2513impl Cache {
2514    pub fn ensure_usable(&self, path: &str) -> Result<(), Box<dyn std::error::Error>> {
2515        if self.tainted {
2516            return Err(format!(
2517                "{path}: cache was tainted by a failed pipeline wave and cannot be reused"
2518            )
2519            .into());
2520        }
2521        Ok(())
2522    }
2523
2524    pub fn mark_tainted(&mut self) {
2525        self.tainted = true;
2526        self.last_logits_dev = None;
2527        self.dflash_taps = None;
2528    }
2529
2530    /// Allocate GPU-resident caches sized by arch + max context.
2531    pub fn new(
2532        e: &impl KvDev,
2533        cfg: &ModelConfig,
2534        max_ctx: usize,
2535    ) -> Result<Self, Box<dyn std::error::Error>> {
2536        Self::new_inner(&|_| e, cfg, None, max_ctx)
2537    }
2538
2539    pub fn new_planned(
2540        e: &impl KvDev,
2541        cfg: &ModelConfig,
2542        plan: &memra_gguf::model_plan::ModelPlan,
2543        max_ctx: usize,
2544    ) -> Result<Self, Box<dyn std::error::Error>> {
2545        Self::new_inner(&|_| e, cfg, Some(plan), max_ctx)
2546    }
2547
2548    /// M1-PP2 increment 2 (stage-owned KV): layers [0, split) allocate through `dev0`,
2549    /// layers [split, n) through `dev1` — each pipeline stage's cache lives on the
2550    /// device that runs the stage. With dev0 == dev1 this is byte-for-byte `new`
2551    /// (the single-device plumbing gate). Sizing math is IDENTICAL either way.
2552    pub fn new_pp2(
2553        dev0: &dyn KvDev,
2554        dev1: &dyn KvDev,
2555        split: usize,
2556        cfg: &ModelConfig,
2557        max_ctx: usize,
2558    ) -> Result<Self, Box<dyn std::error::Error>> {
2559        Self::new_inner(
2560            &|il| if il < split { dev0 } else { dev1 },
2561            cfg,
2562            None,
2563            max_ctx,
2564        )
2565    }
2566
2567    /// M2 N-stage twin of `new_pp2`: `fence` is the stage map from `memra_engine::pp::
2568    /// pp_cuts` ([0, c1, .., n_trunk]); layer il allocates through the engine of the
2569    /// stage that runs it. Layers at/beyond the fence end (MTP/NextN blocks) allocate
2570    /// through the LAST stage. Sizing math is IDENTICAL to `new` — only the allocating
2571    /// device varies.
2572    pub fn new_ppn(
2573        devs: &[&dyn KvDev],
2574        fence: &[usize],
2575        cfg: &ModelConfig,
2576        max_ctx: usize,
2577    ) -> Result<Self, Box<dyn std::error::Error>> {
2578        assert_eq!(
2579            devs.len() + 1,
2580            fence.len(),
2581            "ppn cache: devs vs fence mismatch"
2582        );
2583        let pick = |il: usize| -> &dyn KvDev {
2584            let s = match fence[1..fence.len() - 1].binary_search(&il) {
2585                Ok(k) => k + 1,
2586                Err(k) => k,
2587            };
2588            devs[s.min(devs.len() - 1)]
2589        };
2590        Self::new_inner(&pick, cfg, None, max_ctx)
2591    }
2592
2593    pub fn new_ppn_planned(
2594        devs: &[&dyn KvDev],
2595        fence: &[usize],
2596        cfg: &ModelConfig,
2597        plan: &memra_gguf::model_plan::ModelPlan,
2598        max_ctx: usize,
2599    ) -> Result<Self, Box<dyn std::error::Error>> {
2600        assert_eq!(
2601            devs.len() + 1,
2602            fence.len(),
2603            "ppn cache: devs vs fence mismatch"
2604        );
2605        let pick = |il: usize| -> &dyn KvDev {
2606            let stage = match fence[1..fence.len() - 1].binary_search(&il) {
2607                Ok(index) => index + 1,
2608                Err(index) => index,
2609            };
2610            devs[stage.min(devs.len() - 1)]
2611        };
2612        Self::new_inner(&pick, cfg, Some(plan), max_ctx)
2613    }
2614
2615    /// Shared allocation walk: `pick(il)` supplies the device that OWNS layer il's
2616    /// cache state (always the same device outside the pp2 door).
2617    fn new_inner<'a>(
2618        pick: &dyn Fn(usize) -> &'a dyn KvDev,
2619        cfg: &ModelConfig,
2620        plan: Option<&memra_gguf::model_plan::ModelPlan>,
2621        max_ctx: usize,
2622    ) -> Result<Self, Box<dyn std::error::Error>> {
2623        let fallback_plan = if plan.is_none() {
2624            Some(ModelPlan::compile(cfg)?)
2625        } else {
2626            None
2627        };
2628        let plan = plan
2629            .or(fallback_plan.as_ref())
2630            .expect("cache allocation requires a ModelPlan");
2631        let n = cfg.n_layer as usize;
2632        let mut kv = Vec::with_capacity(n);
2633        let mut recur = Vec::with_capacity(n);
2634        let mut latent = Vec::with_capacity(n);
2635        let head_dim_k = cfg.head_dim_k as usize;
2636        let head_dim_v = cfg.head_dim_v as usize;
2637        for il in 0..cfg.n_layer {
2638            // stage-owned allocation (pp2): the device that runs this layer allocates it.
2639            let e = pick(il as usize);
2640            let layer = plan
2641                .layers
2642                .iter()
2643                .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2644                .find(|layer| layer.index == il)
2645                .ok_or_else(|| format!("cache ModelPlan has no layer {il}"))?;
2646            // E4B KV-SHARING: the trailing shared_kv_layers have no k/v of their own — they
2647            // attend an earlier layer's cache (hybrid_forward resolves the target). No KvLayer
2648            // here: any accidental use is a loud unwrap at bring-up, and rewind/len loops
2649            // (iter_mut().flatten()) skip None naturally.
2650            let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2651            if g4_shared > 0 && il >= cfg.n_layer - g4_shared {
2652                kv.push(None);
2653                recur.push(None);
2654                latent.push(None);
2655                continue;
2656            }
2657            match layer.state {
2658                StatePlan::KvCache { .. } | StatePlan::SlidingKvCache { .. } => {
2659                    // KVQUANT block constraint. Scoped to the QUANTIZED planes: it was a
2660                    // function-wide assert, which made any model whose cfg head dims are not
2661                    // 32-multiples unallocatable even when no layer owns a quantized plane —
2662                    // glm-dsa's latent row (kv_lora + rope) is exactly that shape.
2663                    assert!(
2664                        head_dim_k.is_multiple_of(32) && head_dim_v.is_multiple_of(32),
2665                        "KVQUANT requires head_dim_k%32==0 && head_dim_v%32==0 \
2666                         (layer {il}: k={head_dim_k} v={head_dim_v})"
2667                    );
2668                    // Gemma per-layer geometry and every KV-format door are resolved by the same
2669                    // helper admission uses for its analytic byte coefficient.
2670                    let (kv_dim_k, kv_dim_v, kbb_l, vbb_l) =
2671                        full_attention_kv_layout(cfg, plan, il);
2672                    let k_tok_bytes = (kv_dim_k / 32) * kbb_l;
2673                    let v_tok_bytes = (kv_dim_v / 32) * vbb_l;
2674                    let planned_window = plan
2675                        .layers
2676                        .iter()
2677                        .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2678                        .find(|layer| layer.index == il)
2679                        .and_then(|layer| match layer.state {
2680                            StatePlan::SlidingKvCache { window, .. } => Some(window),
2681                            _ => None,
2682                        });
2683                    let ring = if swa_ring_on() {
2684                        planned_window.map(|window| {
2685                            let window = window as usize;
2686                            KvRing::new(swa_ring_rows(window, max_ctx), window)
2687                        })
2688                    } else {
2689                        None
2690                    };
2691                    let alloc_rows = ring.as_ref().map(KvRing::rows).unwrap_or(max_ctx);
2692                    kv.push(Some(KvLayer {
2693                        // +8B tail pad: the v4 stage's aligned funnelshift window reads up to
2694                        // 4B past the final block (PR #3's finding, adopted pad-style — the
2695                        // expert-dot precedent; zero hot-loop branches, values discarded).
2696                        k: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, k_tok_bytes))?,
2697                        v: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, v_tok_bytes))?,
2698                        kv_dim_k,
2699                        kv_dim_v,
2700                        k_tok_bytes,
2701                        v_tok_bytes,
2702                        len: 0,
2703                        ring,
2704                        len_d: e.htod_i32(&[0])?,
2705                        base_d: None,
2706                    }));
2707                    recur.push(None);
2708                    latent.push(None);
2709                }
2710                StatePlan::Recurrent {
2711                    conv_width,
2712                    conv_kernel,
2713                    state_width,
2714                } => {
2715                    kv.push(None);
2716                    recur.push(Some(RecurLayer {
2717                        conv_state: e.zeros(
2718                            conv_width as usize * (conv_kernel as usize).saturating_sub(1),
2719                        )?,
2720                        ssm_state: e.zeros(state_width as usize)?,
2721                        ssm_state_alt: e.zeros(state_width as usize)?,
2722                    }));
2723                    latent.push(None);
2724                }
2725                StatePlan::LatentKvCache { width, index_width } => {
2726                    // ONE f32 row per token for the whole layer (MQA): no per-head planes, no
2727                    // V plane. `width` is the plan's own number, not re-derived here — the
2728                    // engine's MLA arm asserts it against the loaded `MlaGeom`.
2729                    let width = width as usize;
2730                    assert!(
2731                        width > 0,
2732                        "layer {il}: LatentKvCache width must be positive"
2733                    );
2734                    kv.push(None);
2735                    recur.push(None);
2736                    let index_width = index_width as usize;
2737                    // TAIL RING: the indexer plane is read exactly once per row, by its own
2738                    // pool's key build, so it only has to hold the incomplete tail plus one
2739                    // call's tokens. `None` keeps the flat `max_ctx`-row plane.
2740                    let index_ring = if index_width == 0 {
2741                        None
2742                    } else {
2743                        index_ring_rows(max_ctx)
2744                    };
2745                    let index_rows = match index_width {
2746                        0 => None,
2747                        w => Some(e.zeros(index_ring.unwrap_or(max_ctx) * w)?),
2748                    };
2749                    latent.push(Some(LatentKvLayer {
2750                        rows: e.zeros(max_ctx * width)?,
2751                        width,
2752                        len: 0,
2753                        len_d: e.htod_i32(&[0])?,
2754                        index_rows,
2755                        index_width,
2756                        index_ring_rows: index_ring,
2757                        // Sized from the indexer's `pool`, which the state plan does not carry;
2758                        // the engine allocates it the first time the layer selects.
2759                        index_pool_keys: None,
2760                        index_pools_ready: 0,
2761                        index_pool: 0,
2762                    }));
2763                }
2764                ref state => {
2765                    return Err(format!(
2766                        "native cache allocator has no implementation for layer {il} state {state:?}"
2767                    )
2768                    .into());
2769                }
2770            }
2771        }
2772        Ok(Cache {
2773            kv,
2774            recur,
2775            latent,
2776            tp_kv: (0..n).map(|_| None).collect(),
2777            glm5_tp_recur: (0..n).map(|_| None).collect(),
2778            glm5_tp_latent_peer: (0..n).map(|_| None).collect(),
2779            pos: 0,
2780            max_ctx,
2781            tainted: false,
2782            dflash_taps: None,
2783            hc_taps: None,
2784            last_logits_dev: None,
2785        })
2786    }
2787
2788    pub fn has_swa_ring(&self) -> bool {
2789        self.kv.iter().flatten().any(|layer| layer.ring.is_some())
2790            || self
2791                .tp_kv
2792                .iter()
2793                .flatten()
2794                .any(|layer| layer.ring_window().is_some())
2795    }
2796
2797    pub fn can_rollback(&self, snap: &CacheSnapshot, accept_len: usize) -> bool {
2798        let local = self
2799            .kv
2800            .iter()
2801            .zip(&snap.kv_len)
2802            .all(|(layer, saved)| match (layer, saved) {
2803                (Some(layer), Some(saved)) => layer
2804                    .ring
2805                    .as_ref()
2806                    .is_none_or(|ring| ring.can_rewind_to(saved + accept_len)),
2807                _ => true,
2808            });
2809        let tensor = self
2810            .tp_kv
2811            .iter()
2812            .zip(&snap.tp_kv_len)
2813            .all(|(layer, saved)| match (layer, saved) {
2814                (Some(layer), Some(saved)) => saved
2815                    .checked_add(accept_len)
2816                    .is_some_and(|target| layer.can_rewind_to(target)),
2817                _ => true,
2818            });
2819        local && tensor
2820    }
2821
2822    /// Snapshot the dual cache before a spec-decode draft+verify round (MTP-PLAN §C/§D.4).
2823    /// Records each full-attn `len` (cheap) and makes a REAL device copy of each linear-attn
2824    /// conv_state/ssm_state (a fresh alloc + memcpy_dtod — NOT an Arc clone).
2825    pub fn snapshot(&self, e: &impl KvDev) -> Result<CacheSnapshot, Box<dyn std::error::Error>> {
2826        // glm5 TP-2 state is per-rank and lives outside CacheSnapshot; a snapshot taken over
2827        // live TP planes would silently drop the peer's half. Spec (the only snapshot
2828        // consumer for this family) is co-refused with the TP door — hold that closed here.
2829        if self.glm5_tp_recur.iter().any(Option::is_some)
2830            || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2831        {
2832            return Err(
2833                "cache snapshot is unwired for glm5 TP rank state (MEMRA_GLM5_TP): \
2834                        per-rank planes are not carried by CacheSnapshot"
2835                    .into(),
2836            );
2837        }
2838        self.ensure_usable("cache snapshot")?;
2839        let n = self.kv.len();
2840        let mut kv_len = Vec::with_capacity(n);
2841        let mut tp_kv_len = Vec::with_capacity(n);
2842        let mut conv = Vec::with_capacity(n);
2843        let mut ssm = Vec::with_capacity(n);
2844        for il in 0..n {
2845            match &self.kv[il] {
2846                Some(kvl) => kv_len.push(Some(kvl.len)),
2847                None => kv_len.push(None),
2848            }
2849            tp_kv_len.push(
2850                self.tp_kv[il]
2851                    .as_ref()
2852                    .map(ResidentTpKvCache::committed_len),
2853            );
2854            match &self.recur[il] {
2855                Some(rl) => {
2856                    conv.push(Some(e.clone_dtod(&rl.conv_state)?));
2857                    ssm.push(Some(e.clone_dtod(&rl.ssm_state)?));
2858                }
2859                None => {
2860                    conv.push(None);
2861                    ssm.push(None);
2862                }
2863            }
2864        }
2865        Ok(CacheSnapshot {
2866            kv_len,
2867            tp_kv_len,
2868            conv,
2869            ssm,
2870            pos: self.pos,
2871        })
2872    }
2873
2874    /// PERSISTENT-BUFFER snapshot (spec-decode hot loop): refresh `snap` IN PLACE — same values as
2875    /// `snapshot()` but the conv/ssm device buffers are reused across rounds (D2D copy-into, ZERO
2876    /// allocations vs 2 fresh clones per linear layer per round). `snap` must come from a prior
2877    /// `snapshot()` of THIS cache (same layer shapes).
2878    pub fn snapshot_into(
2879        &self,
2880        e: &impl KvDev,
2881        snap: &mut CacheSnapshot,
2882    ) -> Result<(), Box<dyn std::error::Error>> {
2883        if self.glm5_tp_recur.iter().any(Option::is_some)
2884            || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2885        {
2886            return Err("cache snapshot_into is unwired for glm5 TP rank state \
2887                        (MEMRA_GLM5_TP): per-rank planes are not carried by CacheSnapshot"
2888                .into());
2889        }
2890        self.ensure_usable("cache snapshot refresh")?;
2891        let n = self.kv.len();
2892        for il in 0..n {
2893            snap.kv_len[il] = self.kv[il].as_ref().map(|kvl| kvl.len);
2894            snap.tp_kv_len[il] = self.tp_kv[il]
2895                .as_ref()
2896                .map(ResidentTpKvCache::committed_len);
2897            if let Some(rl) = &self.recur[il] {
2898                let dc = snap.conv[il]
2899                    .as_mut()
2900                    .expect("snapshot_into: shape mismatch (conv)");
2901                let ds = snap.ssm[il]
2902                    .as_mut()
2903                    .expect("snapshot_into: shape mismatch (ssm)");
2904                let (cn, sn) = (rl.conv_state.len(), rl.ssm_state.len());
2905                e.copy_into(dc, 0, &rl.conv_state, cn)?;
2906                e.copy_into(ds, 0, &rl.ssm_state, sn)?;
2907            }
2908        }
2909        snap.pos = self.pos;
2910        Ok(())
2911    }
2912
2913    /// Roll the cache back to exactly `snap.pos + accept_len` committed tokens (MTP-PLAN §C).
2914    /// - Full-attn KV (C.1): set len = snapshot_len + accept_len (truncate, no copy).
2915    /// - Linear-attn (C.2): RESTORE the snapshot conv/ssm (real D2D copy back into the resident
2916    ///   buffers). The caller must then REPLAY the `accept_len` committed tokens through the full
2917    ///   T=1 decode path to rebuild the recurrent state for those positions. We restore (not
2918    ///   replay here) because replay needs the model; this only resets state to the pre-round value.
2919    ///   `cache.pos` is set to `snap.pos` so the caller's replay advances it back to the commit point.
2920    pub fn rollback(
2921        &mut self,
2922        e: &impl KvDev,
2923        snap: &CacheSnapshot,
2924        accept_len: usize,
2925    ) -> Result<(), Box<dyn std::error::Error>> {
2926        self.ensure_usable("cache rollback")?;
2927        if !self.can_rollback(snap, accept_len) {
2928            return Err(
2929                "SWA ring rewind checkpoint has been lapped; full re-prime required".into(),
2930            );
2931        }
2932        for il in 0..self.kv.len() {
2933            if let (Some(kvl), Some(saved)) = (self.kv[il].as_mut(), snap.kv_len[il]) {
2934                kvl.len = saved + accept_len;
2935                // keep the device mirror in lock-step (CUDA-GRAPH-PLAN Phase 2). Set IN PLACE
2936                // (stable pointer): a fresh htod_i32 would reallocate len_d, but its old pointer is
2937                // baked into the captured decode graph's append/inc/fa_decode kernels — replacing it
2938                // strands the graph on a freed buffer (stale-pointer hazard). memcpy_htod in place.
2939                e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
2940            }
2941            if let (Some(kvl), Some(saved)) = (self.tp_kv[il].as_mut(), snap.tp_kv_len[il]) {
2942                kvl.rewind_to(saved + accept_len)?;
2943            }
2944            if let Some(rl) = self.recur[il].as_mut() {
2945                if let Some(c) = &snap.conv[il] {
2946                    e.copy_into(&mut rl.conv_state, 0, c, c.len())?;
2947                }
2948                if let Some(s) = &snap.ssm[il] {
2949                    e.copy_into(&mut rl.ssm_state, 0, s, s.len())?;
2950                }
2951            }
2952        }
2953        self.pos = snap.pos;
2954        Ok(())
2955    }
2956}
2957
2958#[cfg(test)]
2959mod tp_transaction_tests {
2960    use super::{
2961        Cache, INDEX_RING_WORKING_ROWS, KvRingAppend, ResidentTpKvCache, TpKvTransactionState,
2962        index_ring_default_rows, index_ring_rows_for, index_ring_take, tp_kv_rank_allocation_shape,
2963    };
2964
2965    /// glm5_next's declared k-pool width, from `crates/memra-gguf/src/model_packs/glm5_next/mod.rs`
2966    /// (`KpoolPlan { pool: 4, .. }`). The ONE architecture `MEMRA_DSA_INDEX_RING` exists for.
2967    const GLM5_NEXT_POOL: usize = 4;
2968    /// Packed indexer row: `2 * index_head_dim` (128) f32 = 1 KiB per token per MLA layer.
2969    const GLM5_NEXT_STATE_ROW_BYTES: usize = 2 * 128 * 4;
2970    /// What the tail ring costs per MLA layer, at EVERY configured context. 5 MiB against the
2971    /// 1 GiB per layer a flat plane costs at 1M.
2972    const RING_BYTES_PER_LAYER: usize = INDEX_RING_WORKING_ROWS * GLM5_NEXT_STATE_ROW_BYTES;
2973
2974    /// THE SIZING GATE (lane/glm53-ring-sizing, 2026-08-28).
2975    ///
2976    /// The regression this exists for, measured on the bench box three arms one env flag apart
2977    /// on the SAME binary (research/glm53-flash-bringup-20260827/rebaseline-and-surface-20260828,
2978    /// receipts 13 and 14): at `MEMRA_CTX=8192` the ring ON served at most 4630 prompt tokens,
2979    /// `MEMRA_DSA_INDEX_RING=0` served 7300, and the pre-ring binary served 7312. USABLE CONTEXT
2980    /// WAS A FRACTION OF CONFIGURED CONTEXT because the ring was sized against a chunked-prefill
2981    /// bound, and glm5_next primes MONOLITHICALLY (`prime_cache_hyper`, no `prime_chunk_ranges`),
2982    /// so its per-call `t` is the whole prompt.
2983    ///
2984    /// So the gate asserts the RATIO, never one number: for every configured context, a single
2985    /// monolithic prime of the WHOLE context must be admitted by the ring the shipped default
2986    /// derivation books for it. It runs the shipped admission rule (`index_ring_take`) in the
2987    /// shipped drain shape, so it fails exactly when the engine fails.
2988    ///
2989    /// And it asserts the ring is STILL A RING, at every one of those contexts: a "fix" that
2990    /// grows the plane back to `max_ctx` rows passes the acceptance half and is a silent revert
2991    /// of the 11.94 GiB this flag exists to free.
2992    #[test]
2993    fn the_derived_ring_serves_a_monolithic_prime_of_the_whole_configured_context() {
2994        // Two decades of context, and this model's NATIVE 1,048,576. A sizing that works at 8192
2995        // and breaks at 262144 is not a sizing.
2996        for max_ctx in [8192usize, 262_144, 1 << 20] {
2997            let rows = index_ring_default_rows(max_ctx).unwrap_or_else(|| {
2998                panic!("the ring must engage at max_ctx {max_ctx}: it is where the saving is")
2999            });
3000            // The engine rounds the booked rows DOWN to a multiple of `pool`, because the state
3001            // plan does not carry `pool` and the allocator cannot book a pool-aligned budget.
3002            let ring = rows / GLM5_NEXT_POOL * GLM5_NEXT_POOL;
3003
3004            // MONOLITHIC PRIME: one call, nothing resident, `t` = the whole configured context.
3005            let mut cur = 0usize;
3006            let mut pools_ready = 0usize;
3007            let mut steps = 0usize;
3008            while cur < max_ctx {
3009                let take = index_ring_take(ring, GLM5_NEXT_POOL, pools_ready, cur, max_ctx - cur)
3010                    .unwrap_or_else(|| {
3011                        panic!(
3012                            "MEMRA_CTX={max_ctx}: the {ring}-row ring refused a monolithic prime \
3013                         after {cur} of {max_ctx} tokens ({}% of the configured context). \
3014                         USABLE CONTEXT MUST BE AT LEAST CONFIGURED CONTEXT. This is the \
3015                         4630-of-8192 regression, in arithmetic.",
3016                            cur * 100 / max_ctx
3017                        )
3018                    });
3019                assert!(
3020                    take > 0,
3021                    "MEMRA_CTX={max_ctx}: the drain made no progress at row {cur}: a zero take \
3022                     is an infinite loop in the engine, not a refusal"
3023                );
3024                cur += take;
3025                pools_ready = cur / GLM5_NEXT_POOL;
3026                steps += 1;
3027                assert!(
3028                    steps <= max_ctx,
3029                    "MEMRA_CTX={max_ctx}: the drain did not terminate"
3030                );
3031            }
3032            assert_eq!(cur, max_ctx, "the whole prompt must be appended");
3033
3034            // STILL A RING, and the property is that the plane DOES NOT GROW WITH CONTEXT.
3035            // Per MLA layer, and glm5_next has 12 of them. A sizing "fix" that bought
3036            // acceptance by scaling the ring toward `max_ctx` is a silent revert of the
3037            // 11.94 GiB, and it passes the acceptance half above, so this is the half that
3038            // catches it. At 1M the flat plane is 1 GiB per layer and the ring is 5 MiB.
3039            let ring_bytes = rows * GLM5_NEXT_STATE_ROW_BYTES;
3040            let flat_bytes = max_ctx * GLM5_NEXT_STATE_ROW_BYTES;
3041            assert_eq!(
3042                ring_bytes, RING_BYTES_PER_LAYER,
3043                "MEMRA_CTX={max_ctx}: the ring books {rows} rows, not the context-independent \
3044                 {INDEX_RING_WORKING_ROWS}. A plane that tracks max_ctx is the flat plane \
3045                 wearing a modulus"
3046            );
3047            assert!(
3048                rows < max_ctx,
3049                "MEMRA_CTX={max_ctx}: a ring of {rows} rows is not shorter than the flat plane \
3050                 it replaces, so it would not engage at all"
3051            );
3052            // An ABSOLUTE cap, so that raising the working-set constant to buy acceptance fails
3053            // here too rather than moving `RING_BYTES_PER_LAYER` along with it. 16 MiB per layer
3054            // is 3x the shipped ring and still 64x under the flat plane at 1M.
3055            assert!(
3056                ring_bytes <= 16 << 20,
3057                "MEMRA_CTX={max_ctx}: {} MiB per MLA layer, over glm5_next's 12 of them. The ring \
3058                 exists to delete 11.94 GiB; a working set this large is not paying for itself",
3059                ring_bytes >> 20
3060            );
3061            println!(
3062                "MEMRA_CTX={max_ctx}: ring {rows} rows (effective {ring}), monolithic prime of \
3063                 {max_ctx} tokens admitted in {steps} drain step(s); plane {} MiB/layer vs flat \
3064                 {} MiB/layer",
3065                ring_bytes >> 20,
3066                flat_bytes >> 20
3067            );
3068        }
3069    }
3070
3071    fn empty_tp_cache(capacity: usize) -> ResidentTpKvCache {
3072        ResidentTpKvCache::new(Vec::new(), 128, 128, 136, 96, capacity)
3073    }
3074
3075    #[test]
3076    fn step_tp8_rank_allocation_matches_the_official_kv_geometry() {
3077        let shape = tp_kv_rank_allocation_shape(8 * 128, 8 * 128, 8).unwrap();
3078        assert_eq!((shape.kv_dim_k, shape.kv_dim_v), (128, 128));
3079        assert_eq!((shape.k_token_bytes, shape.v_token_bytes), (136, 96));
3080        assert_eq!(shape.bytes_per_token(), 232);
3081        assert_eq!(shape.fixed_bytes, 20);
3082        assert_eq!(shape.allocation_bytes(262_144), 232 * 262_144 + 20);
3083    }
3084
3085    #[test]
3086    fn tp_rank_allocation_refuses_non_divisible_and_non_block_aligned_shards() {
3087        assert!(tp_kv_rank_allocation_shape(1024, 1024, 3).is_err());
3088        assert!(tp_kv_rank_allocation_shape(1024, 1024, 64).is_err());
3089        assert!(tp_kv_rank_allocation_shape(0, 1024, 8).is_err());
3090    }
3091
3092    #[test]
3093    fn partial_commit_publishes_only_the_accepted_prefix() {
3094        let mut state = TpKvTransactionState::new();
3095        let transaction = state.begin().unwrap();
3096        let staged = state.append_target(transaction, 3, 8).unwrap();
3097        state.publish_append(transaction, staged).unwrap();
3098        assert_eq!(state.committed_len, 0);
3099        assert_eq!(state.staged_len, 3);
3100
3101        let committed = state.commit_target(transaction, 2).unwrap();
3102        state.publish_finalize(transaction, committed).unwrap();
3103        assert_eq!(state.committed_len, 2);
3104        assert_eq!(state.staged_len, 2);
3105        assert!(state.active.is_none());
3106        assert!(state.validate(transaction).is_err());
3107    }
3108
3109    #[test]
3110    fn rollback_restores_the_committed_boundary() {
3111        let mut state = TpKvTransactionState::new();
3112        let first = state.begin().unwrap();
3113        let staged = state.append_target(first, 1, 8).unwrap();
3114        state.publish_append(first, staged).unwrap();
3115        let committed = state.commit_target(first, 1).unwrap();
3116        state.publish_finalize(first, committed).unwrap();
3117
3118        let speculative = state.begin().unwrap();
3119        let staged = state.append_target(speculative, 2, 8).unwrap();
3120        state.publish_append(speculative, staged).unwrap();
3121        assert_eq!(state.committed_len, 1);
3122        assert_eq!(state.staged_len, 3);
3123        state
3124            .publish_finalize(speculative, speculative.base_len)
3125            .unwrap();
3126        assert_eq!(state.committed_len, 1);
3127        assert_eq!(state.staged_len, 1);
3128        assert!(state.validate(speculative).is_err());
3129    }
3130
3131    #[test]
3132    fn index_ring_sizing_is_pure_and_carries_no_per_call_t() {
3133        // Default derivation: the working-set constant, engaged only when it is actually SHORTER
3134        // than the flat plane it replaces.
3135        let rows = INDEX_RING_WORKING_ROWS;
3136        assert_eq!(index_ring_rows_for(None, 1 << 20), Some(rows));
3137        assert_eq!(index_ring_default_rows(1 << 20), Some(rows));
3138        // 4k context: the ring would be LONGER than the flat plane, so it does not engage and
3139        // the saving at that context is honestly zero.
3140        assert_eq!(index_ring_rows_for(None, 4096), None);
3141        assert_eq!(index_ring_rows_for(None, rows), None);
3142        assert_eq!(index_ring_rows_for(None, rows + 1), Some(rows));
3143
3144        // THE CORRECTION (lane/glm53-ring-sizing). The derivation reads no prefill chunk bound at
3145        // all now, so the SAME rows are booked at every context above the collapse point, and no
3146        // value of any other flag can move them. Under the old rule an assumed 4096-token chunk
3147        // sized the ring and a monolithic prime blew straight through it.
3148        for max_ctx in [8192usize, 262_144, 1 << 20] {
3149            assert_eq!(
3150                index_ring_rows_for(None, max_ctx),
3151                Some(INDEX_RING_WORKING_ROWS),
3152                "the derived ring must not vary with the configured context"
3153            );
3154        }
3155
3156        // The knob: 0 is the rollback seam, n pins the row budget (how the wraparound gate
3157        // reaches a wrap in a micro fixture).
3158        assert_eq!(index_ring_rows_for(Some(0), 1 << 20), None);
3159        assert_eq!(index_ring_rows_for(Some(16), 64), Some(16));
3160        assert_eq!(index_ring_rows_for(Some(64), 64), None);
3161    }
3162
3163    /// The admission rule itself, over the shapes the engine actually presents it.
3164    #[test]
3165    fn index_ring_take_drains_instead_of_bounding_the_call() {
3166        const POOL: usize = GLM5_NEXT_POOL;
3167        // A flat plane takes the whole call in one bite, whatever else is true.
3168        assert_eq!(index_ring_take(0, POOL, 0, 0, 1 << 20), Some(1 << 20));
3169        // Fresh monolithic prime over a ring 16 times shorter than the call: it takes the ring,
3170        // never more, and never refuses.
3171        assert_eq!(index_ring_take(64, POOL, 0, 0, 1024), Some(64));
3172        // Steady state after a build: the carry-over is under one pool, so the next bite is at
3173        // least `ring - pool + 1` and progress is guaranteed.
3174        for cur in 0..64usize {
3175            let ready = cur / POOL;
3176            let take = index_ring_take(64, POOL, ready, cur, 1024).expect("never lapses");
3177            assert!(
3178                (64 - POOL + 1..=64).contains(&take),
3179                "cur {cur}: take {take} outside the guaranteed progress band"
3180            );
3181        }
3182        // A call SHORTER than what fits is taken whole, so a decode step is one iteration.
3183        assert_eq!(index_ring_take(64, POOL, 4, 16, 1), Some(1));
3184        // The one surviving lapse: resident pool keys further than the ring behind the append.
3185        // A rewind that did not clamp `index_pools_ready`, or a pool-key reallocation.
3186        assert_eq!(index_ring_take(16, POOL, 0, 64, 1), None);
3187        assert_eq!(index_ring_take(16, POOL, 0, 16, 1), None);
3188        assert_eq!(index_ring_take(16, POOL, 0, 15, 1), Some(1));
3189    }
3190
3191    #[test]
3192    fn rejects_nested_stale_and_out_of_range_actions() {
3193        let mut state = TpKvTransactionState::new();
3194        let transaction = state.begin().unwrap();
3195        assert!(state.begin().is_err());
3196        assert!(state.append_target(transaction, 0, 2).is_err());
3197        assert!(state.append_target(transaction, 3, 2).is_err());
3198        let staged = state.append_target(transaction, 2, 2).unwrap();
3199        state.publish_append(transaction, staged).unwrap();
3200        assert!(state.commit_target(transaction, 3).is_err());
3201        state.publish_finalize(transaction, 0).unwrap();
3202        assert!(state.publish_append(transaction, 1).is_err());
3203    }
3204
3205    #[test]
3206    fn rewind_resets_visibility_and_invalidates_an_active_transaction() {
3207        let mut state = TpKvTransactionState::new();
3208        let transaction = state.begin().unwrap();
3209        let staged = state.append_target(transaction, 3, 8).unwrap();
3210        state.publish_append(transaction, staged).unwrap();
3211        state.rewind(1, 8).unwrap();
3212        assert_eq!(state.committed_len, 1);
3213        assert_eq!(state.staged_len, 1);
3214        assert!(state.active.is_none());
3215        assert!(state.validate(transaction).is_err());
3216        assert!(state.rewind(9, 8).is_err());
3217    }
3218
3219    #[test]
3220    fn grow_preserves_generation_and_publishes_only_the_checkpoint_prefix() {
3221        let mut source = empty_tp_cache(8);
3222        let first = source.begin_transaction().unwrap();
3223        let staged = source.append_target(first, 5).unwrap();
3224        source.publish_append(first, staged).unwrap();
3225        let committed = source.commit_target(first, 5).unwrap();
3226        source.publish_finalize(first, committed).unwrap();
3227
3228        let rolled_back = source.begin_transaction().unwrap();
3229        source
3230            .publish_finalize(rolled_back, rolled_back.base_len())
3231            .unwrap();
3232        let plan = source.prepare_grow(16, 3).unwrap();
3233        assert_eq!(plan.rows(), 3);
3234        assert_eq!(plan.k_bytes(), 3 * 136);
3235        assert_eq!(plan.v_bytes(), 3 * 96);
3236
3237        let mut target = empty_tp_cache(16);
3238        target.publish_grow(plan).unwrap();
3239        assert_eq!(target.committed_len(), 3);
3240        assert_eq!(target.staged_len(), 3);
3241        assert_eq!(target.capacity(), 16);
3242        let next = target.begin_transaction().unwrap();
3243        assert_eq!(next.generation(), rolled_back.generation() + 1);
3244        assert_eq!(next.base_len(), 3);
3245    }
3246
3247    #[test]
3248    fn grow_refuses_active_source_and_invalid_target_state_or_layout() {
3249        let mut active = empty_tp_cache(8);
3250        active.begin_transaction().unwrap();
3251        assert!(active.prepare_grow(16, 0).is_err());
3252
3253        let mut source = empty_tp_cache(8);
3254        source.rewind_to(5).unwrap();
3255        assert!(source.prepare_grow(8, 5).is_err());
3256        assert!(source.prepare_grow(16, 6).is_err());
3257        let plan = source.prepare_grow(16, 4).unwrap();
3258
3259        let mut wrong_layout = ResidentTpKvCache::new(Vec::new(), 128, 128, 144, 96, 16);
3260        assert!(wrong_layout.publish_grow(plan).is_err());
3261
3262        let plan = source.prepare_grow(16, 4).unwrap();
3263        let mut dirty_target = empty_tp_cache(16);
3264        dirty_target.rewind_to(1).unwrap();
3265        assert!(dirty_target.publish_grow(plan).is_err());
3266    }
3267
3268    #[test]
3269    fn swa_transaction_rebase_preserves_the_rollback_window() {
3270        let mut cache = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3271        assert_eq!(cache.physical_capacity(), 32 + 4096 + 512 + 31);
3272        // 8250 -> 8762: the extra alignment block moved the wrap point, and at 8250 this append is
3273        // now Contiguous — the test would keep passing while no longer exercising the rebase
3274        // it is named for. Every offset below moves by the same 32 rows; intent unchanged.
3275        cache.publish_hydration(8762, 4096).unwrap();
3276        assert_eq!(cache.ring_base(), Some(4096));
3277
3278        let transaction = cache.begin_transaction().unwrap();
3279        let plan = cache.prepare_append(transaction, 10).unwrap();
3280        assert_eq!(plan.target(), 8772);
3281        assert_eq!(plan.write_row(), 58);
3282        assert_eq!(
3283            plan.ring_append(),
3284            Some(KvRingAppend::Rebase {
3285                src_row: 4608,
3286                keep_rows: 58,
3287                new_base: 8704,
3288                write_row: 58,
3289            })
3290        );
3291        cache.publish_append_rebase(plan).unwrap();
3292        cache.publish_append_plan(plan).unwrap();
3293        assert_eq!(cache.ring_base(), Some(8704));
3294        assert_eq!(cache.physical_range(8740, 8772).unwrap(), 36..68);
3295
3296        let rollback = cache.commit_target(transaction, 0).unwrap();
3297        cache.publish_finalize(transaction, rollback).unwrap();
3298        assert_eq!((cache.committed_len(), cache.staged_len()), (8762, 8762));
3299        assert!(cache.rewind_to(8200).is_err());
3300    }
3301
3302    #[test]
3303    fn swa_grow_normalizes_only_the_live_prefix_and_preserves_generation() {
3304        let mut source = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3305        source.publish_hydration(8250, 4096).unwrap();
3306        let transaction = source.begin_transaction().unwrap();
3307        source
3308            .publish_finalize(transaction, transaction.base_len())
3309            .unwrap();
3310
3311        let plan = source.prepare_grow(20_000, 8250).unwrap();
3312        assert_eq!(plan.source_row(), 4096);
3313        assert_eq!(plan.copy_rows(), 58);
3314        assert_eq!(plan.k_bytes(), 58 * 136);
3315        assert_eq!(plan.v_bytes(), 58 * 96);
3316
3317        let mut target = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 20_000, 32);
3318        target.publish_grow(plan).unwrap();
3319        assert_eq!(target.ring_base(), Some(8192));
3320        assert_eq!((target.committed_len(), target.staged_len()), (8250, 8250));
3321        assert_eq!(target.physical_range(8192, 8250).unwrap(), 0..58);
3322        let next = target.begin_transaction().unwrap();
3323        assert_eq!(next.generation(), transaction.generation() + 1);
3324    }
3325
3326    #[test]
3327    fn cache_reports_a_materialized_distributed_swa_ring() {
3328        let mut cache = Cache {
3329            kv: Vec::new(),
3330            recur: Vec::new(),
3331            latent: Vec::new(),
3332            tp_kv: vec![None],
3333            glm5_tp_recur: vec![None],
3334            glm5_tp_latent_peer: vec![None],
3335            pos: 0,
3336            max_ctx: 10_000,
3337            tainted: false,
3338            dflash_taps: None,
3339            hc_taps: None,
3340            last_logits_dev: None,
3341        };
3342        assert!(!cache.has_swa_ring());
3343        cache.tp_kv[0] = Some(ResidentTpKvCache::new_swa(
3344            Vec::new(),
3345            128,
3346            128,
3347            136,
3348            96,
3349            10_000,
3350            512,
3351        ));
3352        assert!(cache.has_swa_ring());
3353    }
3354}
3355
3356#[cfg(test)]
3357mod swa_ring_tests {
3358    use super::{
3359        KvRing, KvRingAppend, PRIME_CHUNK_MAX_TOKENS, SWA_REWIND_SLACK_ROWS,
3360        SWA_VIEW_ALIGNMENT_ROWS, kv_plane_allocation_bytes, swa_retain_from, swa_ring_rows,
3361    };
3362
3363    #[test]
3364    fn allocation_rows_cover_window_max_prime_and_alignment_slack() {
3365        assert_eq!(swa_ring_rows(512, 262_144), 512 + 4096 + 512 + 31);
3366        assert_eq!(swa_ring_rows(512, 4096), 4096);
3367        assert_eq!(
3368            kv_plane_allocation_bytes(5151, 1088),
3369            5151 * 1088 + 8,
3370            "the Step35 session plane allocates ring rows plus the existing tail pad",
3371        );
3372    }
3373
3374    /// REGRESSION, the SWA-ring MTP lap (2026-08-28) — BOTH steps, which took three attempts to
3375    /// separate on hardware.
3376    ///
3377    /// Step 1, the REWIND. A rebase that retains exactly the window parks `base` at the newest
3378    /// legal value, so the next backward rewind — even by one token — floors an alignment block
3379    /// under it and is refused:
3380    ///   rewind_to=4638 window=512 base=4128 rows=4639 needed_view_start=4096 < base
3381    ///
3382    /// Step 2, the RE-APPEND, which a slack-only fix broke. After a legal rewind `first_row` moves
3383    /// back while `base` does not, so an unclamped ideal retain falls under `base` and the append
3384    /// itself is refused: "SWA ring lapped required rows (base 4128, retain 4096, len 4669)".
3385    /// Slack is something the ring GRANTS when it can, never something a caller may demand.
3386    #[test]
3387    fn retain_grants_rewind_slack_but_never_asks_below_base() {
3388        const WINDOW: usize = 512;
3389        let rows = swa_ring_rows(WINDOW, 262_144);
3390        let len = rows;
3391        let aligned = |pos: usize| (pos - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3392
3393        // step 1 — from base 0 the retain sits below the aligned window start, so a rewind of up
3394        // to a full alignment block survives the rebase.
3395        let retain = swa_retain_from(len, WINDOW, 0);
3396        assert!(retain <= aligned(len) - SWA_REWIND_SLACK_ROWS);
3397        let mut ring = KvRing::new(rows, WINDOW);
3398        ring.apply_rebase(retain);
3399        assert!(
3400            ring.can_rewind_to(len - 1),
3401            "a one-token rewind must survive the rebase"
3402        );
3403        assert!(ring.can_rewind_to(len - SWA_REWIND_SLACK_ROWS));
3404
3405        // ...and a full prime chunk still fits at that retention, which is why the ring grew.
3406        assert!(len - retain + PRIME_CHUNK_MAX_TOKENS <= rows);
3407
3408        // the headroom is REAL, not clamped away: every rewind within it is legal from a base
3409        // the ring was actually sized to keep. This is what the 32-row version could not do —
3410        // it clamped instead, leaving the window pointing below resident rows (all-NaN logits).
3411        for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3412            assert!(
3413                ring.can_rewind_to(len - depth),
3414                "a {depth}-row rewind must be resident, not clamped away",
3415            );
3416        }
3417
3418        // step 2 — the property that actually keeps this safe is NOT `retain >= base`, it is that
3419        // the attention WINDOW is fully resident: window_start >= base. The clamp to `base` is
3420        // correct exactly while that holds, and v3's NaN came from clamping with only 32 rows of
3421        // headroom, where a deeper rewind clamped into a window that ran below resident rows.
3422        // With the ring sized for SWA_REWIND_SLACK_ROWS, every rewind inside the headroom keeps a
3423        // complete window — so the clamp is safe by construction rather than by luck.
3424        let base = ring.base();
3425        for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3426            let window_start = (len - depth - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3427            assert!(
3428                window_start >= base,
3429                "after a {depth}-row rewind the window starts at {window_start}, below base \
3430                 {base} — clamping here would serve rows the ring no longer holds (the pos-8661 \
3431                 all-NaN case)",
3432            );
3433            assert!(swa_retain_from(len - depth, WINDOW, base) >= base);
3434        }
3435
3436        // and one row past the headroom the window DOES run below base — the case that must stay
3437        // refused rather than clamped, which is what can_rewind_to enforces.
3438        let past = len - (SWA_REWIND_SLACK_ROWS + WINDOW);
3439        let past_start = (past - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3440        assert!(
3441            past_start < base,
3442            "beyond the headroom the window must fall below base"
3443        );
3444        assert!(
3445            !ring.can_rewind_to(past),
3446            "and can_rewind_to must refuse it"
3447        );
3448    }
3449
3450    #[test]
3451    fn ring_matches_flat_bytes_before_wrap() {
3452        let ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3453        let flat: Vec<u32> = (0..1024).collect();
3454        let mut physical = vec![u32::MAX; ring.rows()];
3455        let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, flat.len()).unwrap()
3456        else {
3457            panic!("first append unexpectedly wrapped")
3458        };
3459        physical[write_row..write_row + flat.len()].copy_from_slice(&flat);
3460        let view = ring.physical_range(0, flat.len()).unwrap();
3461        assert_eq!(&physical[view], flat.as_slice());
3462    }
3463
3464    #[test]
3465    fn wrap_rebases_the_exact_aligned_prime_view() {
3466        let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3467        let flat: Vec<u32> = (0..8192).collect();
3468        let mut physical = vec![u32::MAX; ring.rows()];
3469        let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, 4096).unwrap() else {
3470            panic!("first prime chunk unexpectedly wrapped")
3471        };
3472        physical[write_row..write_row + 4096].copy_from_slice(&flat[..4096]);
3473
3474        let off = (4096usize - (512 - 1)) & !31usize;
3475        let KvRingAppend::Rebase {
3476            src_row,
3477            keep_rows,
3478            new_base,
3479            write_row,
3480        } = ring.append_plan(4096, off, 4096).unwrap()
3481        else {
3482            panic!("second prime chunk did not wrap")
3483        };
3484        let retained = physical[src_row..src_row + keep_rows].to_vec();
3485        physical[..keep_rows].copy_from_slice(&retained);
3486        ring.apply_rebase(new_base);
3487        physical[write_row..write_row + 4096].copy_from_slice(&flat[4096..8192]);
3488
3489        let view = ring.physical_range(off, 8192).unwrap();
3490        assert_eq!(&physical[view], &flat[off..8192]);
3491        assert_eq!(ring.base(), off);
3492    }
3493
3494    #[test]
3495    fn rewind_declines_once_the_required_window_was_lapped() {
3496        let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3497        let KvRingAppend::Rebase { new_base, .. } = ring.append_plan(4096, 3584, 4096).unwrap()
3498        else {
3499            panic!("expected wrap")
3500        };
3501        ring.apply_rebase(new_base);
3502        assert!(ring.can_rewind_to(4095));
3503        assert!(!ring.can_rewind_to(4094));
3504        assert!(!ring.can_rewind_to(0));
3505    }
3506
3507    /// The 2026-08-29 warm-turn-at-40k panic: a checkpoint on a LAPPED ring records an absolute
3508    /// `len` far past the physical rows, and a flat `len`-row restore is an out-of-bounds device
3509    /// slice. The plan must hand back only the aligned live window plus the base to rebase a
3510    /// fresh target to — and refuse once the source ring no longer holds that window.
3511    #[test]
3512    fn restore_plan_copies_the_window_not_the_absolute_length() {
3513        let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3514        // Before any wrap: the plan is exactly the flat prefix.
3515        let (base, phys) = ring.restore_plan(400).unwrap();
3516        assert_eq!((base, phys), (0, 0..400));
3517
3518        // Lap the ring far past its physical capacity (a 40k-token session), the way a real
3519        // prime does: 4096-row chunks, rebasing whenever the tail would wrap.
3520        let mut live = 0usize;
3521        while live < 40_960 {
3522            let retain = swa_retain_from(live, 512, ring.base());
3523            if let KvRingAppend::Rebase { new_base, .. } =
3524                ring.append_plan(live, retain, 4096).unwrap()
3525            {
3526                ring.apply_rebase(new_base);
3527            }
3528            live += 4096;
3529        }
3530        assert!(ring.base() > 0, "a 40k walk must have lapped the ring");
3531        let (base, phys) = ring.restore_plan(live).unwrap();
3532        assert_eq!(base, (live - (512 - 1)) & !31usize);
3533        assert!(
3534            base >= ring.base(),
3535            "the plan must stay above the ring floor"
3536        );
3537        assert_eq!(phys.len(), live - base);
3538        assert!(
3539            phys.end <= ring.rows(),
3540            "the copy must fit the physical buffer ({} rows), got {:?}",
3541            ring.rows(),
3542            phys
3543        );
3544
3545        // A checkpoint from before the rebase is gone: refuse, never slice.
3546        assert!(ring.restore_plan(400).is_err());
3547    }
3548}