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