memra_engine/dflash.rs
1//! DFlash block-diffusion drafter (DFLASH-BRINGUP-PLAN.md, 2026-07-13).
2//!
3//! 5-layer qwen3-class mini-transformer that drafts a 16-token block in ONE non-causal
4//! forward, conditioned on the TARGET's hidden states at 6 tapped layers (concatenated
5//! through `fc` + `hidden_norm`). No embed / lm_head of its own — the round reuses the
6//! target's. Reference: z-lab/dflash `dflash/model.py` (semantics frozen in the plan doc);
7//! oracle: tools/dflash_oracle.py -> /data/cache/dflash-oracle.npz.
8//!
9//! FIRST LIGHT = f32-resident weights + fresh full-context forward (no draft KV cache) —
10//! correctness vs the oracle, then the cache/quant/window arms land measurement-gated.
11
12use crate::Engine;
13use crate::model::GpuTensor;
14use cudarc::driver::CudaSlice;
15
16pub struct DflashCfg {
17 pub hidden: usize, // 5376
18 pub n_head: usize, // 64
19 pub n_kv: usize, // 8
20 pub head_dim: usize, // 128
21 pub n_ff: usize, // 10752
22 pub n_layer: usize, // 5
23 pub eps: f32, // 1e-6
24 pub rope_theta: f32, // 1e6
25 pub block_size: usize, // 16
26 pub mask_token_id: u32, // 4
27 pub target_layer_ids: Vec<usize>, // [1,12,23,35,46,57]
28 pub sliding_window: usize, // 2048
29 /// true = sliding_attention for that layer (4x true + 1x false on the 31B draft).
30 pub layer_sliding: Vec<bool>,
31 /// Checkpoint training-strategy census (`dspark_strategy_census` over the raw
32 /// config.json): true = a SpecForge DSPARK-strategy export (shifted labels, ALL rows
33 /// supervised — the q38 arm-a family). Keys the HARVEST DEFAULT strategy-keyed,
34 /// never env-keyed (owner-ratified 2026-08-20 after B1 confirmed H1 ×5;
35 /// DSPARK-POSTMORTEM-20260820.md B0 default-flip plan).
36 pub strategy_dspark: bool,
37 /// Explicit top-level `is_causal` from config.json (z-lab reference: an explicit
38 /// value OVERRIDES the per-layer-type default). The DFlash2 q38 checkpoint carries
39 /// `"is_causal": false` — every sliding layer is NON-causal with a symmetric
40 /// +/-2048 window (model.py `_attention_mask`). None = key absent (historical
41 /// exports; the windowless-assert arm keeps handling those byte-identically).
42 pub is_causal: Option<bool>,
43}
44
45pub struct DflashLayer {
46 pub wq: GpuTensor, // [nh*hd, hidden] row-major (out_f rows)
47 pub wk: GpuTensor, // [nkv*hd, hidden]
48 pub wv: GpuTensor, // [nkv*hd, hidden]
49 pub wo: GpuTensor, // [hidden, nh*hd]
50 pub w_gate: GpuTensor, // [n_ff, hidden]
51 pub w_up: GpuTensor, // [n_ff, hidden]
52 pub w_down: GpuTensor, // [hidden, n_ff]
53 pub ln_in: CudaSlice<f32>, // [hidden]
54 pub ln_post: CudaSlice<f32>, // [hidden]
55 pub q_norm: CudaSlice<f32>, // [hd]
56 pub k_norm: CudaSlice<f32>, // [hd]
57}
58
59pub struct DflashDraft {
60 pub cfg: DflashCfg,
61 pub layers: Vec<DflashLayer>,
62 pub fc: GpuTensor, // [hidden, n_taps*hidden]
63 pub hidden_norm: CudaSlice<f32>, // [hidden]
64 pub norm: CudaSlice<f32>, // [hidden]
65 /// DSpark semi-AR markov head (present in the repo-root checkpoint variant):
66 /// draft logits at position k get + W2(W1[prev_realized_token]) — left-to-right
67 /// within the block (the patch's _markov_semiar_sample_block semantics, greedy).
68 /// w1 = raw bf16 [V, rank] (row-gathered by device token id); w2 = q8_0 [rank->V].
69 pub markov: Option<MarkovHead>,
70 /// DSpark accept-rate head (trained with confidence loss). sglang's DSPARK planner
71 /// consumes it to SIZE VERIFY WINDOWS (cumprod survival — v0.5.16 headline; the
72 /// earlier "reference serving loop never consumes it" note matched SpecForge's
73 /// legacy spec_generate only). memra schedules with it under
74 /// `MEMRA_DSPARK_VT=confidence` (the H4 fix, DSPARK-POSTMORTEM-20260820.md:
75 /// per-round verify window from cumprod survival, `dspark_confidence_vt`) and
76 /// keeps it census+parity-only under the default ladder. Host-resident (5k floats).
77 pub confidence: Option<ConfidenceHead>,
78 /// YaRN rope (q38 arm-a inherits the target's rope_parameters: rope_type yarn,
79 /// factor 32, original 8192, beta 32/1). ff = per-dim divisors for rope_neox_ff
80 /// (effective inv_freq_j = base^(-2j/d)/ff[j] = the HF-yarn remapped frequency,
81 /// verified vs Qwen3RotaryEmbedding to 1.6e-7), mscale = attention_scaling
82 /// (0.1*ln(factor)+1) applied to q/k post-rope — cos/sin scaling distributes onto
83 /// the rotated vector exactly. None = plain rope (gemma/z-lab drafters).
84 pub rope_yarn: Option<(CudaSlice<f32>, f32)>,
85 /// DFlash2 head (z-lab `DFlash2DraftModel`, DFLASH2-EVAL-20260820.md): grouped
86 /// dynamic causal convs around EVERY sublayer + the candidate path selector that
87 /// replaces the markov chain. A DISTINCT semantic program from the DSpark head
88 /// (no-generic-support law): present iff config `architectures` names
89 /// `DFlash2DraftModel`, and then ALL 23 family tensors are REQUIRED — loading the
90 /// 58 backbone tensors alone computes an untrained model (the census trap).
91 pub dflash2: Option<Dflash2Head>,
92}
93
94/// One `GroupedDynamicCausalConv` module (reference model.py): a causal 2-tap
95/// depthwise conv over the BLOCK rows (block-local — row 0 zero-pads its missing
96/// predecessor; stateless across rounds), with per-position dynamic per-group
97/// coefficients projected from the module INPUT. `prepare` convolves the sublayer
98/// input with base_kernel[0] + dyn half 0; `finish` convolves the sublayer OUTPUT
99/// with base_kernel[1] + dyn half 1 (both dyn halves come from the SAME projection
100/// of the pre-conv input).
101pub struct Dflash2Conv {
102 /// base_kernel [2, k, hidden] flattened f32 (half-major: prepare then finish).
103 pub base: CudaSlice<f32>,
104 /// kernel_projection.weight [2*k*groups, hidden] (row layout = view(2, k, groups)).
105 pub proj: GpuTensor,
106}
107
108pub struct Dflash2Head {
109 pub attn_conv: Vec<Dflash2Conv>, // per layer
110 pub mlp_conv: Vec<Dflash2Conv>, // per layer
111 /// candidate_selector.hidden_projection.weight [rank, hidden].
112 pub hidden_proj: GpuTensor,
113 /// Codebooks [V, rank] raw bf16, HOST-resident: the walk gathers ~1+16 rows per
114 /// draft slot (~70KB/round) — host math beside the round's existing chain dtoh,
115 /// no device residency for 2x127MB tables. Checkpoint quirk: stored WITHOUT the
116 /// `.weight` suffix (reference from_pretrained installs a key_mapping).
117 pub pred_codebook: Vec<u8>,
118 pub succ_codebook: Vec<u8>,
119 pub rank: usize, // selector_rank 256
120 pub top_k: usize, // selector_top_k 16
121 pub conv_k: usize, // conv_kernel_size 2
122 pub group_size: usize, // conv_group_size 16
123 pub vocab: usize, // codebook rows (248320)
124}
125
126/// One bf16 codebook row -> f32 (exact widening).
127fn cb_row(cb: &[u8], tok: usize, rank: usize) -> Vec<f32> {
128 bf16_to_f32(&cb[tok * rank * 2..(tok + 1) * rank * 2])
129}
130
131/// Greedy selector walk (reference `CandidateSelector.select` at T=0): per draft
132/// slot p, score(k) = unary[p,k] + <pred_codebook[prev] .* hidden_proj_row[p],
133/// succ_codebook[cand[p,k]]>, argmax over the top-k candidate set; the CHOSEN
134/// candidate seeds the next slot (sequential — the chain is the semantics, not an
135/// optimization). Host math (~nd*k*rank fused ops per round) over host-resident bf16
136/// codebooks; ties break to the LOWEST k (torch argmax convention). Pure so the
137/// selector semantics are CPU-gateable.
138///
139/// `unary`/`cand`: [nd, top_k] row-major; `hproj`: [nd, rank] row-major.
140#[allow(clippy::too_many_arguments)]
141pub fn dflash2_walk_greedy(
142 pred_codebook: &[u8],
143 succ_codebook: &[u8],
144 vocab: usize,
145 rank: usize,
146 top_k: usize,
147 unary: &[f32],
148 cand: &[u32],
149 hproj: &[f32],
150 anchor: u32,
151 nd: usize,
152) -> Vec<u32> {
153 let (kk, r) = (top_k, rank);
154 assert_eq!(unary.len(), nd * kk, "walk: unary shape");
155 assert_eq!(cand.len(), nd * kk, "walk: candidate shape");
156 assert_eq!(hproj.len(), nd * r, "walk: hidden-projection shape");
157 let mut path = Vec::with_capacity(nd);
158 let mut prev = anchor;
159 for p in 0..nd {
160 assert!(
161 (prev as usize) < vocab,
162 "walk: predecessor token {prev} outside codebook vocab {vocab}"
163 );
164 let pr = cb_row(pred_codebook, prev as usize, r);
165 let hp = &hproj[p * r..(p + 1) * r];
166 // gate = pred_row .* hidden_proj (shared across the candidate set)
167 let gate: Vec<f32> = pr.iter().zip(hp).map(|(a, b)| a * b).collect();
168 let (mut best, mut bi) = (f32::NEG_INFINITY, 0usize);
169 for k in 0..kk {
170 let c = cand[p * kk + k] as usize;
171 assert!(c < vocab, "walk: candidate {c} outside codebook vocab");
172 let sr = cb_row(succ_codebook, c, r);
173 let mut s = unary[p * kk + k];
174 for j in 0..r {
175 s += gate[j] * sr[j];
176 }
177 if s > best {
178 best = s;
179 bi = k;
180 }
181 }
182 prev = cand[p * kk + bi];
183 path.push(prev);
184 }
185 path
186}
187
188impl Dflash2Head {
189 /// Greedy selector walk over this head's codebooks — see `dflash2_walk_greedy`.
190 pub fn walk_greedy(
191 &self,
192 unary: &[f32],
193 cand: &[u32],
194 hproj: &[f32],
195 anchor: u32,
196 nd: usize,
197 ) -> Vec<u32> {
198 dflash2_walk_greedy(
199 &self.pred_codebook,
200 &self.succ_codebook,
201 self.vocab,
202 self.rank,
203 self.top_k,
204 unary,
205 cand,
206 hproj,
207 anchor,
208 nd,
209 )
210 }
211
212 /// Sampled (T>0) selector walk — see `dflash2_walk_sampled`.
213 #[allow(clippy::too_many_arguments)]
214 pub fn walk_sampled(
215 &self,
216 unary: &[f32],
217 cand: &[u32],
218 hproj: &[f32],
219 anchor: u32,
220 nd: usize,
221 temp: f32,
222 uniforms: &mut dyn FnMut() -> f32,
223 ) -> (Vec<u32>, Vec<f32>, Vec<f32>) {
224 dflash2_walk_sampled(
225 &self.pred_codebook,
226 &self.succ_codebook,
227 self.vocab,
228 self.rank,
229 self.top_k,
230 unary,
231 cand,
232 hproj,
233 anchor,
234 nd,
235 temp,
236 uniforms,
237 )
238 }
239}
240
241/// AcceptRatePredictor: raw linear proj over [hidden ; markov_prev_embedding(rank)]
242/// (with_markov=true on the q38 arm-a export) — output is the PRE-sigmoid scalar.
243pub struct ConfidenceHead {
244 pub w: Vec<f32>, // [in_dim]
245 pub b: f32,
246 pub in_dim: usize,
247 pub with_markov: bool,
248}
249
250impl ConfidenceHead {
251 /// Host dot: the PRE-sigmoid accept score for one draft slot. `hidden` = the
252 /// drafter output row the slot is harvested from (the same row its logits use);
253 /// `emb` = the markov `w1` row of the slot's PREVIOUS chain token (required iff
254 /// `with_markov`) — the exact input contract the parity gate pins (prev ids =
255 /// `[anchor, chain[..nd-1]]`, dspark_q38_parity.rs stage 5).
256 pub fn raw_score(&self, hidden: &[f32], emb: Option<&[f32]>) -> f32 {
257 let mut acc = self.b;
258 for (w, x) in self.w.iter().zip(hidden) {
259 acc += w * x;
260 }
261 if self.with_markov {
262 let emb = emb.expect("with_markov confidence head scored without the markov embedding");
263 debug_assert_eq!(hidden.len() + emb.len(), self.in_dim);
264 for (w, x) in self.w[hidden.len()..].iter().zip(emb) {
265 acc += w * x;
266 }
267 } else {
268 debug_assert_eq!(hidden.len(), self.in_dim);
269 }
270 acc
271 }
272}
273
274pub struct MarkovHead {
275 pub w1_bf16: CudaSlice<u8>, // [V, rank] bf16 raw
276 pub w2: GpuTensor, // [rank -> V] q8_0
277 pub rank: usize,
278 pub vocab: usize,
279}
280
281/// Draft-row harvest convention for DFlash-family block drafters
282/// (darklanes research/deepseek-flash-20260818/DSPARK-POSTMORTEM-20260820.md).
283///
284/// The DFlash and DSpark SpecForge training strategies supervise DIFFERENT rows of the
285/// same `[anchor, MASK x b-1]` block, so the row -> trunk-position mapping is a property
286/// of the CHECKPOINT's training strategy, not of the loader:
287///
288/// - **Dflash** (mask-fill; z-lab dflash / SpecForge `OnlineDFlashModel`): row k is
289/// trained to predict the token AT position anchor+k — "Labels: same-position
290/// prediction", `weight_mask *= (pos_in_block > 0)` excludes the anchor row
291/// (SpecForge `specforge/algorithms/common/dflash_family_model.py:453-472`).
292/// Drafts = rows 1..b-1; the anchor row's output is untrained.
293/// - **Dspark** (shifted; SpecForge `OnlineDSparkModel`, `training.strategy: dspark` —
294/// the q38 arm-a export): row k is trained to predict the token at anchor+k+1, ALL
295/// rows supervised INCLUDING the anchor row (`label_offsets = arange(1,
296/// block_size+1)`, `dflash_family_model.py:816`). sglang's DSPARK worker — the stack
297/// every arm-a bank number was measured on — harvests gamma = block_size drafts with
298/// the anchor row's output as draft 1 (verified on the v0.5.17 eval-pin tag:
299/// `dspark_components/dspark_draft.py:248,260,318`; `dspark_config.py:269`).
300///
301/// Mismatching the convention verifies every slot against a position the row was never
302/// trained for — the q38 accept collapse (2.9 -> 1.43) in the postmortem.
303#[derive(Clone, Copy, PartialEq, Eq, Debug)]
304pub enum DsparkHarvest {
305 /// mask-fill: drafts = rows 1..b-1, row k fills position anchor+k.
306 Dflash,
307 /// shifted: drafts = rows 0..b-1, row k predicts position anchor+k+1.
308 Dspark,
309}
310
311impl DsparkHarvest {
312 /// The served resolution: explicit `MEMRA_DSPARK_HARVEST={dflash|dspark}` wins
313 /// (unknown values REFUSE loudly — a typo silently reverting the convention would
314 /// re-open the postmortem's misalignment); UNSET defers to the CHECKPOINT's own
315 /// training-strategy census — the owner-ratified default flip (2026-08-20, after
316 /// B1 confirmed H1 interleaved ×5 on serving-class hardware: accept 1.38→2.41
317 /// agentic / 1.53→3.66 math, E2E ALL EXACT both arms). Strategy-keyed, not
318 /// env-keyed, per the B0 plan: a DSPARK-strategy export harvests shifted
319 /// (all-rows), a mask-fill export keeps the historical dflash arm byte-identical.
320 pub fn resolve(cfg: &DflashCfg) -> Self {
321 Self::resolve_value(
322 std::env::var("MEMRA_DSPARK_HARVEST").ok().as_deref(),
323 cfg.strategy_dspark,
324 )
325 }
326
327 pub fn resolve_value(v: Option<&str>, strategy_dspark: bool) -> Self {
328 match v {
329 None | Some("") => {
330 if strategy_dspark {
331 DsparkHarvest::Dspark
332 } else {
333 DsparkHarvest::Dflash
334 }
335 }
336 set => Self::from_env_value(set),
337 }
338 }
339
340 /// ENV-ONLY parser (no checkpoint census): unset = `Dflash`, the historical arm.
341 /// Kept for the explicit-value path of [`Self::resolve_value`] and the seam tests;
342 /// round arms resolve through [`Self::resolve`] so the default stays strategy-keyed.
343 pub fn from_env_value(v: Option<&str>) -> Self {
344 match v {
345 None | Some("") | Some("dflash") => DsparkHarvest::Dflash,
346 Some("dspark") => DsparkHarvest::Dspark,
347 Some(other) => panic!(
348 "MEMRA_DSPARK_HARVEST={other}: unknown harvest convention (dflash|dspark); \
349 refusing — a wrong convention verifies every draft slot against a position \
350 the drafter row was not trained for (DSPARK-POSTMORTEM-20260820.md)"
351 ),
352 }
353 }
354
355 /// Resolve the harvest convention for a LOADED drafter — FAMILY-keyed first, then
356 /// STRATEGY-keyed (v0.100 train merge of the two ratified keyings):
357 /// - DFlash2 is a mask-fill-family drafter by construction (reference
358 /// `dflash_generate` harvests rows `1-verify_size:`; the card says "block size 8
359 /// (7 draft tokens per verification step)" — DFLASH2-EVAL-20260820.md §3). An env
360 /// value that CONTRADICTS the census REFUSES rather than silently re-keying the
361 /// round.
362 /// - Every other checkpoint rides [`Self::resolve_value`]: explicit env wins (typos
363 /// refuse loudly), unset defers to the checkpoint's own training-strategy census
364 /// (the owner-ratified 2026-08-20 default flip).
365 pub fn for_draft(draft: &DflashDraft) -> Self {
366 Self::for_family_value(
367 draft.dflash2.is_some(),
368 std::env::var("MEMRA_DSPARK_HARVEST").ok().as_deref(),
369 draft.cfg.strategy_dspark,
370 )
371 }
372
373 pub fn for_family_value(is_dflash2: bool, env: Option<&str>, strategy_dspark: bool) -> Self {
374 if is_dflash2 {
375 if env == Some("dspark") {
376 panic!(
377 "MEMRA_DSPARK_HARVEST=dspark with a DFlash2 checkpoint: DFlash2 \
378 is mask-fill (b-1 drafts, anchor row is not a draft — reference \
379 dflash_generate rows 1-verify_size:); the shifted harvest would \
380 verify every slot one position early. Refusing (census-keyed, \
381 not env-keyed)."
382 );
383 }
384 return DsparkHarvest::Dflash;
385 }
386 Self::resolve_value(env, strategy_dspark)
387 }
388
389 /// Manifest/serialized name (the oracle geometry manifest's `harvest` field).
390 pub fn name(self) -> &'static str {
391 match self {
392 DsparkHarvest::Dflash => "dflash",
393 DsparkHarvest::Dspark => "dspark",
394 }
395 }
396
397 pub fn from_name(v: &str) -> Option<Self> {
398 match v {
399 "dflash" => Some(DsparkHarvest::Dflash),
400 "dspark" => Some(DsparkHarvest::Dspark),
401 _ => None,
402 }
403 }
404
405 /// First drafter OUTPUT row consumed as a draft candidate.
406 pub fn first_row(self) -> usize {
407 match self {
408 DsparkHarvest::Dflash => 1,
409 DsparkHarvest::Dspark => 0,
410 }
411 }
412
413 /// Drafted tokens harvested per round from a `b`-row block.
414 pub fn n_drafts(self, b: usize) -> usize {
415 match self {
416 DsparkHarvest::Dflash => b - 1,
417 DsparkHarvest::Dspark => b,
418 }
419 }
420
421 /// The position offset (relative to the round anchor at the block's row 0) that
422 /// drafter output row `row` is TRAINED to predict under this convention.
423 pub fn trained_offset_of_row(self, row: usize) -> usize {
424 match self {
425 DsparkHarvest::Dflash => row,
426 DsparkHarvest::Dspark => row + 1,
427 }
428 }
429}
430
431/// Checkpoint training-strategy census over the raw config.json text (the loader's
432/// minimal-extractor idiom — no json dep in-tree). TRUE iff the export declares the
433/// DSPARK strategy: `architectures` naming a DSpark model class (`Qwen3DSparkModel`,
434/// the SpecForge OnlineDSparkModel export form) or `dflash_config.projector_type ==
435/// "dspark"`. z-lab / OnlineDFlashModel mask-fill exports carry neither signal. Pure,
436/// so the census is testable against config fragments without files.
437pub fn dspark_strategy_census(txt: &str) -> bool {
438 let arch = txt
439 .find("\"architectures\"")
440 .and_then(|i| {
441 let rest = &txt[i..];
442 let a = rest.find('[')?;
443 let b = rest.find(']')?;
444 Some(rest[a..b].contains("DSpark"))
445 })
446 .unwrap_or(false);
447 let proj = txt
448 .find("\"projector_type\"")
449 .map(|i| {
450 let rest = &txt[i..];
451 let after = rest.find(':').map(|c| &rest[c + 1..]).unwrap_or("");
452 after.trim_start().starts_with("\"dspark\"")
453 })
454 .unwrap_or(false);
455 arch || proj
456}
457
458/// Accepted-prefix length of a round's candidates against the trunk's verify argmaxes:
459/// `cand[0]` = the round anchor (already decided), `cand[1..]` = the drafts;
460/// `vam[j]` = the trunk's argmax prediction for position anchor+j+1. Returns m =
461/// number of accepted drafts (`cand[1..=m]` committed, `vam[m]` becomes the next
462/// anchor). Pure so the harvest-alignment fixture can exercise it CPU-side.
463pub fn dspark_accept_prefix(cand: &[u32], vam: &[u32], vt: usize) -> usize {
464 let mut m = 0usize;
465 while m < vt - 1 && cand[m + 1] == vam[m] {
466 m += 1;
467 }
468 m
469}
470
471/// Verify-window policy for the dspark round (H4, DSPARK-POSTMORTEM-20260820.md §3).
472///
473/// B2 measured the structural fork: the fixed full-block window (vt=8) buys 95–100%
474/// of the sglang accept bank but LOSES wall speed to the reactive ladder everywhere
475/// except math — at 0.2–0.5 slot rates, full-block verify pays 5–6 empty rows per
476/// round. The confidence policy is the mechanism both leading engines schedule with
477/// (sglang v0.5.16 `dspark_planner.py` cumprod survival; vLLM #47808): size EACH
478/// round's window from the drafter's own trained accept-rate head, so windows open
479/// on confident streaks (math/code) and shrink on bursty text without a 4-round
480/// ladder climb.
481#[derive(Clone, Copy, PartialEq, Debug)]
482pub enum DsparkVtPolicy {
483 /// The shipped reactive ladder: `vt = (m+2).clamp(3, vt_cap)` per round
484 /// (`MEMRA_DFLASH_ADAPT=0` pins vt at `vt_cap` = the fixed-window arm).
485 Ladder,
486 /// `MEMRA_DSPARK_VT=confidence`: per-round window from cumprod survival of the
487 /// confidence head's sigmoid scores, thresholded at `tau`
488 /// (`MEMRA_DSPARK_VT_TAU`, default 0.5). Raw sigmoid — no STS sidecar
489 /// calibration exists for this export; the postmortem names this the starting
490 /// policy.
491 Confidence { tau: f32 },
492 /// `MEMRA_DSPARK_VT=confidence-slot` (owner directive, 2026-08-20: "take only
493 /// high confidence offers"): submit only the longest draft PREFIX whose every
494 /// slot clears `tau` on its own sigmoid — the low-confidence tail never enters
495 /// verify. Same tau env. vs `Confidence`: if the head's per-row score is the
496 /// MARGINAL accept probability (it already sinks with depth), cumprod survival
497 /// double-counts the decay and over-truncates; if it is the CONDITIONAL,
498 /// per-slot under-truncates. Which statistic the q38 head emits is empirical —
499 /// both arms ride the A/B.
500 ConfidenceSlot { tau: f32 },
501}
502
503impl DsparkVtPolicy {
504 /// The served resolution: explicit `MEMRA_DSPARK_VT={ladder|confidence|
505 /// confidence-slot}` wins (unknown values REFUSE loudly — a typo silently
506 /// reverting the window policy would invalidate an A/B without a trace); UNSET
507 /// defaults to **`confidence-slot` at τ = `MEMRA_DSPARK_VT_TAU` (default 0.5)** —
508 /// the owner-ratified H4 flip (2026-08-20; cell 2's 4-arm A/B ×5 + cell 3's tau
509 /// ladder put the knee at τ=.5 for the slot arm: 94–98% of the fixed-8 accept bank
510 /// at wall ≥ the reactive ladder, exactness 11/11 ALL EXACT). Census-keyed per the
511 /// capacity-keyed-defaults law: a checkpoint WITHOUT an accept-rate head has no
512 /// signal to schedule with, so unset-env resolves to the ladder there (loudly, at
513 /// load) instead of panicking on a default; `MEMRA_DFLASH_ADAPT=0` (an explicit
514 /// fixed-window request) also keeps the ladder-family arm.
515 pub fn resolve(has_confidence_head: bool) -> Self {
516 Self::resolve_value(
517 std::env::var("MEMRA_DSPARK_VT").ok().as_deref(),
518 std::env::var("MEMRA_DSPARK_VT_TAU").ok().as_deref(),
519 std::env::var("MEMRA_DFLASH_ADAPT").ok().as_deref(),
520 has_confidence_head,
521 )
522 }
523
524 pub fn resolve_value(
525 vt: Option<&str>,
526 tau: Option<&str>,
527 adapt: Option<&str>,
528 has_confidence_head: bool,
529 ) -> Self {
530 match vt {
531 None | Some("") => {
532 if adapt == Some("0") || !has_confidence_head {
533 DsparkVtPolicy::Ladder
534 } else {
535 // The ratified default rides the SAME tau parse as the explicit
536 // arm (a bad MEMRA_DSPARK_VT_TAU refuses, never silently ignored).
537 Self::from_env_value(Some("confidence-slot"), tau, adapt)
538 }
539 }
540 set => Self::from_env_value(set, tau, adapt),
541 }
542 }
543
544 /// ENV-ONLY parser (no head census): unset = `Ladder`. Kept for the explicit-value
545 /// path of [`Self::resolve_value`] and the policy-gate tests; round arms resolve
546 /// through [`Self::resolve`] so the default stays head-census-keyed.
547 pub fn from_env_value(vt: Option<&str>, tau: Option<&str>, adapt: Option<&str>) -> Self {
548 match vt {
549 None | Some("") | Some("ladder") => DsparkVtPolicy::Ladder,
550 Some(mode @ ("confidence" | "confidence-slot")) => {
551 if adapt == Some("0") {
552 panic!(
553 "MEMRA_DSPARK_VT={mode} together with MEMRA_DFLASH_ADAPT=0 is \
554 contradictory (a pinned fixed window vs a per-round confidence \
555 window); unset one — refuse-on-ambiguity"
556 );
557 }
558 let tau = tau
559 .map(|t| {
560 t.parse::<f32>()
561 .unwrap_or_else(|_| panic!("MEMRA_DSPARK_VT_TAU={t}: not a float"))
562 })
563 .unwrap_or(0.5);
564 assert!(
565 tau > 0.0 && tau < 1.0,
566 "MEMRA_DSPARK_VT_TAU={tau}: confidence threshold must be in (0,1)"
567 );
568 if mode == "confidence" {
569 DsparkVtPolicy::Confidence { tau }
570 } else {
571 DsparkVtPolicy::ConfidenceSlot { tau }
572 }
573 }
574 Some(other) => panic!(
575 "MEMRA_DSPARK_VT={other}: unknown verify-window policy \
576 (ladder|confidence|confidence-slot); refusing — a wrong policy \
577 silently reverts the H4 arm (DSPARK-POSTMORTEM-20260820.md)"
578 ),
579 }
580 }
581
582 /// True for every head-scheduled arm (the loops gate the head requirement and
583 /// the embedding stash on this).
584 pub fn is_confidence(&self) -> bool {
585 !matches!(self, DsparkVtPolicy::Ladder)
586 }
587
588 /// Size this round's verify window from the head's pre-sigmoid slot scores.
589 /// `None` under the ladder (the caller keeps its carried vt).
590 pub fn size_window(&self, raws: &[f32], vt_cap: usize) -> Option<usize> {
591 match *self {
592 DsparkVtPolicy::Ladder => None,
593 DsparkVtPolicy::Confidence { tau } => Some(dspark_confidence_vt(raws, tau, vt_cap)),
594 DsparkVtPolicy::ConfidenceSlot { tau } => {
595 Some(dspark_slot_confidence_vt(raws, tau, vt_cap))
596 }
597 }
598 }
599}
600
601/// H4 window sizing (the sglang-planner/vLLM-#47808 mechanism, thresholded): `raws[k]`
602/// = the accept-rate head's PRE-sigmoid score for draft slot k+1; survival
603/// `S_k = prod_{j<=k} sigmoid(raws[j])`; the window keeps leading slots while
604/// `S_k >= tau`. Returns `vt` = 1 (anchor) + kept drafts, clamped to `[2, vt_cap]`:
605/// the draft forward is already paid, so at least one draft rides every verify — one
606/// extra verify row costs less than a guaranteed empty round. Pure, so the policy's
607/// knee is testable CPU-side like `dspark_accept_prefix`.
608pub fn dspark_confidence_vt(raws: &[f32], tau: f32, vt_cap: usize) -> usize {
609 let mut surv = 1.0f32;
610 let mut kept = 0usize;
611 for &r in raws {
612 surv *= 1.0 / (1.0 + (-r).exp());
613 if surv < tau {
614 break;
615 }
616 kept += 1;
617 }
618 (1 + kept).clamp(2, vt_cap.max(2))
619}
620
621/// Owner-directive arm (2026-08-20, "take only high confidence offers"): keep the
622/// longest draft PREFIX whose EVERY slot clears `tau` on its own sigmoid — truncate
623/// at the first sub-threshold slot, so the low-confidence tail (B2 measured 0.2–0.5
624/// slot rates at depth) never enters verify. Prefix truncation is forced by the
625/// accept rule anyway (`dspark_accept_prefix` stops at the first miss — a kept slot
626/// after a dropped one could never commit); the policy fork vs `dspark_confidence_vt`
627/// is only the stopping statistic (per-slot marginal vs cumulative survival). Same
628/// floor/cap contract.
629pub fn dspark_slot_confidence_vt(raws: &[f32], tau: f32, vt_cap: usize) -> usize {
630 let mut kept = 0usize;
631 for &r in raws {
632 let p = 1.0 / (1.0 + (-r).exp());
633 if p < tau {
634 break;
635 }
636 kept += 1;
637 }
638 (1 + kept).clamp(2, vt_cap.max(2))
639}
640
641// ================= SAMPLED ADMISSION (T>0) — lane/dspark-sampled-admission-20260820 =====
642// True rejection sampling for the dspark route (mystery A of DSPARK-POSTMORTEM-20260820):
643// draft slot j is DRAWN from a recorded proposal distribution q_j, the trunk's verify column
644// arbitrates with the Leviathan/Chen rule (accept x_j while u_j*q_j(x_j) < p_j(x_j); on
645// reject resample from norm(max(0, p-q)); on full accept the bonus ~ p at the last column),
646// so the committed stream's distribution equals trunk-only sampling from the FILTERED target
647// p — the same contract the frspec/MTP route ships (spec.rs sampled accept walk; kernels
648// oracled by sample_check). T==0/None keeps every greedy path byte-identical (the exactness
649// instrument and the kill-switch are the same code).
650//
651// Two proposal families, each recording the TRUE distribution its drafts were drawn from:
652// - Rows (dspark/dflash strategy checkpoints): per-slot FILTERED softmax of the draft-logits
653// row — markov-corrected in place when the head is present (the sglang DSPARK worker's
654// "chain rejection sampling over markov-corrected draft probs"), plain rows otherwise
655// (the z-lab reference's independent-row T>0 arm).
656// - Selector (DFlash2): the candidate-path selector's per-slot softmax over its top-k
657// candidate set at temperature ONLY — the reference applies no top-k/top-p to selector
658// scores (z-lab model.py `CandidateSelector.select`: `_sampling_probs(scores, temperature)`
659// with default filters) — with the candidate-set residual (`scatter_add_` of -q, clamped).
660
661/// Rejection-sampling prefix walk: accept draft j while `u_j * q_j < p_j` (strict, f64 —
662/// byte-identical to the frspec accept test). `p`/`q` are the FILTERED target/proposal
663/// probabilities of the drafted tokens; `u` the per-slot uniforms. Pure so the composition
664/// gate can pin the rule on CPU.
665pub fn rejection_accept_len(p: &[f32], q: &[f32], u: &[f32]) -> usize {
666 assert!(
667 q.len() >= p.len() && u.len() >= p.len(),
668 "accept walk shape"
669 );
670 let mut m = 0usize;
671 while m < p.len() && (u[m] as f64) * (q[m] as f64) < p[m] as f64 {
672 m += 1;
673 }
674 m
675}
676
677/// Sampled selector walk (reference `CandidateSelector.select`, temperature>0 arm): per
678/// draft slot the pair scores over the top-k candidate set become a softmax at `temp`
679/// (temperature ONLY — the reference passes no top-k/top-p here), one uniform draws the
680/// candidate (fixed-order CDF walk), and the CHOSEN candidate seeds the next slot exactly
681/// like the greedy chain. Returns (path, q_chosen[nd], q_rows[nd*top_k]) — q_rows are the
682/// recorded per-slot candidate probabilities (the residual's `scatter_add_` input), and
683/// q_chosen[j] == q_rows[j*top_k + chosen_j] is the accept-test q. Pure (uniforms injected)
684/// so the T->0 limit, the chain conditioning, and the recorded-q contract are CPU-gateable.
685#[allow(clippy::too_many_arguments)]
686pub fn dflash2_walk_sampled(
687 pred_codebook: &[u8],
688 succ_codebook: &[u8],
689 vocab: usize,
690 rank: usize,
691 top_k: usize,
692 unary: &[f32],
693 cand: &[u32],
694 hproj: &[f32],
695 anchor: u32,
696 nd: usize,
697 temp: f32,
698 uniforms: &mut dyn FnMut() -> f32,
699) -> (Vec<u32>, Vec<f32>, Vec<f32>) {
700 assert!(
701 temp > 0.0,
702 "sampled walk is the T>0 arm; T=0 is walk_greedy"
703 );
704 let (kk, r) = (top_k, rank);
705 assert_eq!(unary.len(), nd * kk, "walk: unary shape");
706 assert_eq!(cand.len(), nd * kk, "walk: candidate shape");
707 assert_eq!(hproj.len(), nd * r, "walk: hidden-projection shape");
708 let mut path = Vec::with_capacity(nd);
709 let mut q_chosen = Vec::with_capacity(nd);
710 let mut q_rows = Vec::with_capacity(nd * kk);
711 let mut prev = anchor;
712 for p in 0..nd {
713 assert!(
714 (prev as usize) < vocab,
715 "walk: predecessor token {prev} outside codebook vocab {vocab}"
716 );
717 let pr = cb_row(pred_codebook, prev as usize, r);
718 let hp = &hproj[p * r..(p + 1) * r];
719 let gate: Vec<f32> = pr.iter().zip(hp).map(|(a, b)| a * b).collect();
720 let mut scores = vec![0f32; kk];
721 for (k, s) in scores.iter_mut().enumerate() {
722 let c = cand[p * kk + k] as usize;
723 assert!(c < vocab, "walk: candidate {c} outside codebook vocab");
724 let sr = cb_row(succ_codebook, c, r);
725 let mut acc = unary[p * kk + k];
726 for j in 0..r {
727 acc += gate[j] * sr[j];
728 }
729 *s = acc;
730 }
731 // softmax over the candidate set at temp (f64 internals; recorded probs are the
732 // f32 values the CDF walk actually samples from — recorded q IS the proposal).
733 let mx = scores.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
734 let mut z = 0f64;
735 let ex: Vec<f64> = scores
736 .iter()
737 .map(|&s| {
738 let e0 = (((s - mx) / temp) as f64).exp();
739 z += e0;
740 e0
741 })
742 .collect();
743 let probs: Vec<f32> = ex.iter().map(|&e0| (e0 / z) as f32).collect();
744 let u = uniforms() as f64;
745 let mut acc = 0f64;
746 // fp-residue fallback (u >= f32-accumulated mass, ~2^-24 events): the max-prob
747 // candidate — never a zero-prob one (host_u01's range includes 1.0 exactly).
748 let mut bi = probs
749 .iter()
750 .enumerate()
751 .max_by(|a, b| a.1.total_cmp(b.1))
752 .map(|(k, _)| k)
753 .unwrap_or(0);
754 for (k, &pk) in probs.iter().enumerate() {
755 acc += pk as f64;
756 if u < acc {
757 bi = k;
758 break;
759 }
760 }
761 prev = cand[p * kk + bi];
762 path.push(prev);
763 q_chosen.push(probs[bi]);
764 q_rows.extend_from_slice(&probs);
765 }
766 (path, q_chosen, q_rows)
767}
768
769/// `dflash2_propose_sampled`'s wire: (path, q_chosen, candidate ids, q_rows).
770pub(crate) type Dflash2SampledProposal = (Vec<u32>, Vec<f32>, Vec<u32>, Vec<f32>);
771
772/// Per-round proposal record for the sampled dspark round — everything the rejection
773/// walk needs to evaluate the TRUE per-slot proposal distribution q.
774pub(crate) enum DsparkDraftSample {
775 /// q lives in the round's draft-logits buffer `dl` (markov-biased in place when the
776 /// head is armed); per-slot FILTERED stats retained device-contiguous for the accept
777 /// gather + host-mirrored for the reject-slot residual.
778 Rows {
779 th: CudaSlice<f32>, // [nd] filter thresholds (e-units), slot-indexed
780 z: CudaSlice<f32>, // [nd] renorm masses
781 stats: Vec<(f32, f32, f32)>, // host (mx, th, z) per slot
782 },
783 /// DFlash2 candidate-path selector: q is the recorded candidate-set distribution.
784 Selector {
785 cand: Vec<u32>, // [nd*top_k] candidate ids
786 q_rows: Vec<f32>, // [nd*top_k] per-slot candidate probs
787 q_chosen: Vec<f32>, // [nd] prob of the drawn candidate (accept-test q)
788 top_k: usize,
789 },
790}
791
792/// The sampled round's verify+accept: filtered p gathered from the trunk's verify logits
793/// (row j arbitrates draft `cand[j+1]` — the position mapping the greedy prefix walk uses),
794/// the rejection walk over host uniforms, then `next` = bonus (full accept: filtered-Gumbel
795/// from the LAST verify row with its OWN fresh stats — the sampfix-20260805 law: that row is
796/// one past the gathered set) or the residual sample at the reject slot (family-keyed q:
797/// full-row logits for Rows, sparse candidate-set probs for Selector). Returns (m, next) —
798/// the exact (accepted-drafts, next-anchor) contract of the greedy `dspark_accept_prefix` +
799/// `vam[m]` pair, so both round bodies commit identically downstream.
800///
801/// PENALIZED SAMPLED (lane/dspark-penalized-sampled-20260821): when the request carries
802/// non-identity penalties, the vt verify columns are materialized ONCE into a penalized
803/// copy where row j's Keskar pass runs over `pen_win ++ cand[1..=j]` (window-capped) —
804/// the tokens committed before position j ON EVERY PATH WHERE ROW j IS CONSULTED,
805/// same-round accepts included (row j is only read when drafts 1..j were all accepted,
806/// i.e. exactly when `cand[1..=j]` is the committed prefix; the bonus row vt-1 is only
807/// read on full accept, when all nq drafts are committed). Every p read — the batched
808/// stats+gather, the bonus draw, the reject-slot residual column — points at that buffer,
809/// so p is the true penalized per-state target and the committed stream equals plain
810/// penalized sampling (the composition gate's penalty fixtures, self-hit included).
811/// q stays the RECORDED proposal the drafts were actually drawn from (unpenalized):
812/// rejection sampling is unbiased for ANY proposal with `u·q(x) < p(x)` + residual
813/// `norm(max(0, p−q))`; penalizing q would only buy acceptance overlap and would cost an
814/// evolving-history pass inside the sync-free device chain. `pen_win` is the caller's
815/// session window ALREADY trimmed to `min(penalty_last_n, PEN_WINDOW_MAX)` (empty when
816/// penalties are off — the unpenalized path is byte-untouched).
817#[allow(clippy::too_many_arguments)]
818pub(crate) fn dspark_accept_sampled(
819 e: &Engine,
820 tlogits: &CudaSlice<f32>,
821 cand: &[u32],
822 vt: usize,
823 n_vocab: usize,
824 dl: &CudaSlice<f32>,
825 prop: &DsparkDraftSample,
826 sp: &crate::spec::SpecSampling,
827 pen_win: &[u32],
828 sctr: &mut u32,
829 uctr: &mut u32,
830) -> Result<(usize, u32), Box<dyn std::error::Error>> {
831 let nq = vt - 1; // drafts under this round's verify window
832 debug_assert!(nq >= 1 && cand.len() > nq, "sampled accept shape");
833 // --- penalized verify columns (identity penalties: no copy, no launch, raw tlogits) ---
834 let pen_on = sp.pen_on();
835 let ptl: Option<CudaSlice<f32>> = if pen_on {
836 let win = sp.penalty_last_n.min(crate::spec::PEN_WINDOW_MAX);
837 debug_assert!(pen_win.len() <= win, "pen_win must arrive pre-trimmed");
838 let mut hist: Vec<u32> = Vec::with_capacity(pen_win.len() + nq);
839 hist.extend_from_slice(pen_win);
840 hist.extend_from_slice(&cand[1..=nq]); // drafted tokens: row j reads the first j
841 let hd = e.htod_u32_v(&hist)?;
842 let mut buf = e.clone_dtod(tlogits)?;
843 e.penalize_logits_rows_inc(
844 &mut buf,
845 &hd,
846 pen_win.len(),
847 sp.penalty_repeat,
848 sp.penalty_freq,
849 sp.penalty_present,
850 n_vocab,
851 vt,
852 win,
853 )?;
854 Some(buf)
855 } else {
856 None
857 };
858 let p_src: &CudaSlice<f32> = ptl.as_ref().unwrap_or(tlogits);
859 // --- filtered p at the drafted tokens (one batched stats + gather over rows 0..nq-1) ---
860 let rows: Vec<i32> = (0..nq as i32).collect();
861 let ids: Vec<u32> = cand[1..=nq].to_vec();
862 let rowsd = e.htod_i32(&rows)?;
863 let idsd = e.htod_u32_v(&ids)?;
864 let (mut pth, mut pz, mut pmx) = (e.zeros(nq)?, e.zeros(nq)?, e.zeros(nq)?);
865 e.filter_stats(
866 p_src, n_vocab, &rowsd, &mut pth, &mut pz, &mut pmx, n_vocab, nq, sp.temp, sp.top_k,
867 sp.top_p, sp.min_p,
868 )?;
869 let mut pj_d = e.zeros(nq)?;
870 e.softmax_gather_filtered(
871 p_src, n_vocab, &idsd, &rowsd, &pth, &pz, &mut pj_d, n_vocab, nq, sp.temp,
872 )?;
873 let pj = e.dtoh(&pj_d)?;
874 let (pthv, pzv, pmxv) = (e.dtoh(&pth)?, e.dtoh(&pz)?, e.dtoh(&pmx)?);
875 // --- q at the drafted tokens (the recorded proposal distribution) ---
876 let qj: Vec<f32> = match prop {
877 DsparkDraftSample::Rows { th, z, .. } => {
878 // dl row j is draft j's (bias-corrected) logits row; th/z are slot-indexed, and
879 // rows 0..nq-1 index both the buffer rows and the stat pairs.
880 let mut qd = e.zeros(nq)?;
881 e.softmax_gather_filtered(
882 dl, n_vocab, &idsd, &rowsd, th, z, &mut qd, n_vocab, nq, sp.temp,
883 )?;
884 e.dtoh(&qd)?
885 }
886 DsparkDraftSample::Selector { q_chosen, .. } => q_chosen[..nq].to_vec(),
887 };
888 // --- the rejection walk ---
889 let mut us = Vec::with_capacity(nq);
890 for _ in 0..nq {
891 us.push(crate::spec::host_u01(sp.seed, *uctr));
892 *uctr = uctr.wrapping_add(1);
893 }
894 let m = rejection_accept_len(&pj[..nq], &qj[..nq], &us);
895 // --- next anchor: bonus or residual ---
896 let next = if m == nq {
897 // FULL ACCEPT: bonus ~ filtered p at verify row vt-1 — fresh stats for THIS row.
898 // Under penalties p_src row vt-1 carries the FULL drafted block in its window
899 // (all nq drafts are committed on this path — the "drafted token penalizes its
900 // own successor" case the composition gate's self-hit fixture pins).
901 let rows_l = e.htod_i32(&[(vt - 1) as i32])?;
902 let (mut th1, mut z1, mut mx1) = (e.zeros(1)?, e.zeros(1)?, e.zeros(1)?);
903 e.filter_stats(
904 p_src, n_vocab, &rows_l, &mut th1, &mut z1, &mut mx1, n_vocab, 1, sp.temp, sp.top_k,
905 sp.top_p, sp.min_p,
906 )?;
907 let mut pb = e.zeros(n_vocab)?;
908 e.gumbel_perturb_filtered_col(
909 p_src,
910 vt - 1,
911 &mut pb,
912 n_vocab,
913 sp.seed,
914 *sctr,
915 sp.temp,
916 &mx1,
917 &th1,
918 0,
919 )?;
920 *sctr = sctr.wrapping_add(1);
921 let td = e.argmax_token_device(&pb, n_vocab)?;
922 e.dtoh_u32_one(&td)?
923 } else {
924 // REJECT at slot m: token ~ norm(max(0, p_m - q_m)); p row m's stats come from the
925 // gathered set (rows 0..nq-1 cover every reject slot). Under penalties the column
926 // copy MUST come from p_src (the penalized buffer) — a raw-tlogits residual is the
927 // composition gate's "residual reads unpenalized p" tooth.
928 let mut col = e.zeros(n_vocab)?;
929 e.copy_view_into(
930 &mut col,
931 0,
932 &p_src.slice(m * n_vocab..(m + 1) * n_vocab),
933 n_vocab,
934 )?;
935 let p_stats = (pmxv[m], pthv[m], pzv[m]);
936 let mut tok_d = e.alloc_u32_zeroed(1)?;
937 let sc = *sctr;
938 *sctr = sctr.wrapping_add(1);
939 match prop {
940 DsparkDraftSample::Rows { stats, .. } => {
941 let mut qbuf = e.zeros(n_vocab)?;
942 e.copy_view_into(
943 &mut qbuf,
944 0,
945 &dl.slice(m * n_vocab..(m + 1) * n_vocab),
946 n_vocab,
947 )?;
948 e.residual_sample_filtered(
949 &col,
950 Some(&qbuf),
951 n_vocab,
952 sp.temp,
953 sp.seed,
954 sc,
955 p_stats,
956 stats[m],
957 &mut tok_d,
958 )?;
959 }
960 DsparkDraftSample::Selector {
961 cand: cids,
962 q_rows,
963 top_k,
964 ..
965 } => {
966 let k = *top_k;
967 let ids_m = e.htod_u32_v(&cids[m * k..(m + 1) * k])?;
968 let qs_m = e.htod(&q_rows[m * k..(m + 1) * k])?;
969 e.residual_sample_sparse_q(
970 &col, &ids_m, &qs_m, k, n_vocab, sp.temp, sp.seed, sc, p_stats, &mut tok_d,
971 )?;
972 }
973 }
974 e.dtoh_u32(&tok_d)?[0]
975 };
976 Ok((m, next))
977}
978
979/// Clip door for the DFlash2 windowed round attention (lane/dflash2-longctx, §10.6(c)).
980/// Default ON: the lo-clipped kernel — byte-identical output (kernel_check
981/// `sdpa_naive_w_lo`), O(window) key scan, and no T_kv*4-byte shared-mem launch bound, so
982/// the route survives past ~12k ctx (GATES-SMOKE-20260821 B2: DriverError(
983/// CUDA_ERROR_INVALID_VALUE) at ctx 16,571/30,157, last success 9,510).
984/// MEMRA_DFLASH2_SDPA_CLIP=0 = the legacy full-scan kernel byte-for-byte — the rollback
985/// seam and the long-ctx gate's crash-reproduction arm.
986fn dflash2_sdpa_clip_on() -> bool {
987 std::env::var("MEMRA_DFLASH2_SDPA_CLIP")
988 .map(|v| v != "0")
989 .unwrap_or(true)
990}
991
992/// The DFlash2 round attention over the non-causal symmetric window: one seam for both the
993/// first-light (`forward_block`) and cached (`forward_round`) arms, dispatching the clipped
994/// kernel unless the rollback door is thrown.
995#[allow(clippy::too_many_arguments)]
996fn d2_windowed_attn(
997 e: &Engine,
998 q: &CudaSlice<f32>,
999 k: &CudaSlice<f32>,
1000 v: &CudaSlice<f32>,
1001 attn: &mut CudaSlice<f32>,
1002 hd: usize,
1003 nh: usize,
1004 nkv: usize,
1005 t: usize,
1006 t_kv: usize,
1007 scale: f32,
1008 c: &DflashCfg,
1009) -> Result<(), Box<dyn std::error::Error>> {
1010 if dflash2_sdpa_clip_on() {
1011 e.sdpa_naive_w_lo(
1012 q,
1013 k,
1014 v,
1015 attn,
1016 hd,
1017 nh,
1018 nkv,
1019 t,
1020 t_kv,
1021 scale,
1022 false,
1023 c.sliding_window,
1024 )
1025 } else {
1026 e.sdpa_naive_w(
1027 q,
1028 k,
1029 v,
1030 attn,
1031 hd,
1032 nh,
1033 nkv,
1034 t,
1035 t_kv,
1036 scale,
1037 false,
1038 c.sliding_window,
1039 )
1040 }
1041}
1042
1043fn bf16_to_f32(bytes: &[u8]) -> Vec<f32> {
1044 bytes
1045 .chunks_exact(2)
1046 .map(|c| f32::from_bits((u16::from_le_bytes([c[0], c[1]]) as u32) << 16))
1047 .collect()
1048}
1049
1050/// Host q8_0 encode (ggml block layout: [d f16][32 x i8] = 34B/32 vals). The drafter's
1051/// weights ride the dp4a fast path at 1.6GB resident (bf16 3.1GB + the 31B trunk OOM'd
1052/// 24GB; f32 6.2GB worse). Drafter quantization moves ACCEPTANCE only — verify exactness
1053/// is structural.
1054fn encode_q8_0(vals: &[f32]) -> Vec<u8> {
1055 let mut out = Vec::with_capacity(vals.len() / 32 * 34);
1056 for blk in vals.chunks_exact(32) {
1057 let amax = blk.iter().fold(0f32, |a, v| a.max(v.abs()));
1058 let d = amax / 127.0;
1059 let id = if d > 0.0 { 1.0 / d } else { 0.0 };
1060 let dh = half_from_f32(d);
1061 out.extend_from_slice(&dh.to_le_bytes());
1062 for &v in blk {
1063 out.push(((v * id).round().clamp(-127.0, 127.0)) as i8 as u8);
1064 }
1065 }
1066 out
1067}
1068
1069/// Host q4_0 encode (ggml: [d f16][16B packed nibbles] = 18B/32 vals; q = round(v/d)+8,
1070/// d = amax/-7 sign trick NOT used — plain amax/7? ggml uses d = max/-8 .. follow ggml:
1071/// d = amax / -8 when the max is negative-dominant; reference quantize_row_q4_0: d =
1072/// max(|v|)/-8 signed-max form). Implemented to match ggml quantize_row_q4_0_ref.
1073fn encode_q4_0(vals: &[f32]) -> Vec<u8> {
1074 let mut out = Vec::with_capacity(vals.len() / 32 * 18);
1075 for blk in vals.chunks_exact(32) {
1076 // ggml ref: pick the value with the LARGEST |v| (keeping sign), d = that / -8
1077 let mut amax = 0f32;
1078 let mut mx = 0f32;
1079 for &v in blk {
1080 if v.abs() > amax {
1081 amax = v.abs();
1082 mx = v;
1083 }
1084 }
1085 let d = mx / -8.0;
1086 let id = if d != 0.0 { 1.0 / d } else { 0.0 };
1087 out.extend_from_slice(&half_from_f32(d).to_le_bytes());
1088 for j in 0..16 {
1089 let x0 = (blk[j] * id + 8.5).clamp(0.0, 15.0) as u8;
1090 let x1 = (blk[j + 16] * id + 8.5).clamp(0.0, 15.0) as u8;
1091 out.push(x0 | (x1 << 4));
1092 }
1093 }
1094 out
1095}
1096
1097fn half_from_f32(v: f32) -> u16 {
1098 // f32 -> IEEE f16 (round-to-nearest-even; range of q8_0 d values is tame)
1099 let b = v.to_bits();
1100 let sign = ((b >> 16) & 0x8000) as u16;
1101 let exp = ((b >> 23) & 0xff) as i32 - 127 + 15;
1102 let man = b & 0x7fffff;
1103 if exp <= 0 {
1104 return sign;
1105 } // flush tiny d to zero
1106 if exp >= 31 {
1107 return sign | 0x7c00;
1108 } // inf (unreachable for sane d)
1109 let mut h = sign | ((exp as u16) << 10) | ((man >> 13) as u16);
1110 // round to nearest even on the truncated 13 bits
1111 let rem = man & 0x1fff;
1112 if rem > 0x1000 || (rem == 0x1000 && (h & 1) == 1) {
1113 h += 1;
1114 }
1115 h
1116}
1117
1118impl DflashDraft {
1119 /// Load the backbone-only checkpoint dir (config.json + model.safetensors, bf16).
1120 /// Config scalars ride a minimal extractor (no json dep in-tree — HfConfig precedent).
1121 pub fn load(e: &Engine, dir: &std::path::Path) -> Result<Self, Box<dyn std::error::Error>> {
1122 let txt = std::fs::read_to_string(dir.join("config.json"))?;
1123 fn num(txt: &str, key: &str) -> Option<f64> {
1124 let i = txt.find(&format!("\"{key}\""))?;
1125 let rest = &txt[i..];
1126 let colon = rest.find(':')?;
1127 let val: String = rest[colon + 1..]
1128 .trim_start()
1129 .chars()
1130 .take_while(|c| {
1131 c.is_ascii_digit()
1132 || *c == '.'
1133 || *c == '-'
1134 || *c == 'e'
1135 || *c == 'E'
1136 || *c == '+'
1137 })
1138 .collect();
1139 val.parse().ok()
1140 }
1141 fn num_list(txt: &str, key: &str) -> Vec<usize> {
1142 let Some(i) = txt.find(&format!("\"{key}\"")) else {
1143 return Vec::new();
1144 };
1145 let rest = &txt[i..];
1146 let (Some(a), Some(b)) = (rest.find('['), rest.find(']')) else {
1147 return Vec::new();
1148 };
1149 rest[a + 1..b]
1150 .split(',')
1151 .filter_map(|s| s.trim().parse().ok())
1152 .collect()
1153 }
1154 /// Substring of the JSON OBJECT value of a top-level key (brace-balanced) —
1155 /// the explicit scoped parse the DFlash2 census demands: `dflash_config` and
1156 /// `rope_parameters` are nested objects, and finding their keys by global
1157 /// `txt.find` is luck, not a contract (DFLASH2-EVAL-20260820.md §5.1).
1158 fn scope<'a>(txt: &'a str, key: &str) -> Option<&'a str> {
1159 let i = txt.find(&format!("\"{key}\""))?;
1160 let rest = &txt[i..];
1161 let open = rest.find('{')?;
1162 let mut depth = 0usize;
1163 for (j, ch) in rest[open..].char_indices() {
1164 match ch {
1165 '{' => depth += 1,
1166 '}' => {
1167 depth -= 1;
1168 if depth == 0 {
1169 return Some(&rest[open..open + j + 1]);
1170 }
1171 }
1172 _ => {}
1173 }
1174 }
1175 None
1176 }
1177 // Family detection is the ARCHITECTURES string, not tensor presence: a DFlash2
1178 // checkpoint whose new tensors were stripped must REFUSE, not degrade into the
1179 // 58-tensor untrained program (DFLASH2-EVAL-20260820.md §3).
1180 let is_dflash2 = {
1181 let arch = scope_list(&txt, "architectures");
1182 arch.contains("DFlash2DraftModel")
1183 };
1184 fn scope_list(txt: &str, key: &str) -> String {
1185 let Some(i) = txt.find(&format!("\"{key}\"")) else {
1186 return String::new();
1187 };
1188 let rest = &txt[i..];
1189 match (rest.find('['), rest.find(']')) {
1190 (Some(a), Some(b)) if a < b => rest[a + 1..b].to_string(),
1191 _ => String::new(),
1192 }
1193 }
1194 // DFlash2 scalars parse from their OWN scopes; other families keep the
1195 // historical global-find behavior byte-identically.
1196 let d2_cfg_txt: Option<&str> = if is_dflash2 {
1197 Some(scope(&txt, "dflash_config").unwrap_or_else(|| {
1198 panic!("DFlash2DraftModel config.json has no dflash_config object — refusing")
1199 }))
1200 } else {
1201 None
1202 };
1203 let g = |k: &str| num(&txt, k).unwrap_or_else(|| panic!("config missing {k}")) as usize;
1204 let g2 = |k: &str| -> usize {
1205 let t = d2_cfg_txt.expect("dflash2 scope");
1206 num(t, k).unwrap_or_else(|| panic!("dflash_config missing {k} — refusing")) as usize
1207 };
1208 // layer_types order: count entries, mark sliding ones
1209 let layer_sliding: Vec<bool> = {
1210 let i = txt.find("\"layer_types\"").expect("layer_types");
1211 let rest = &txt[i..];
1212 let (a, b) = (rest.find('[').unwrap(), rest.find(']').unwrap());
1213 rest[a + 1..b]
1214 .split(',')
1215 .map(|s| s.contains("sliding_attention"))
1216 .collect()
1217 };
1218 // sliding_window is null on all-full-attention exports (q38 arm-a); the window
1219 // only constrains rounds when a sliding layer exists (reference: resolve_dflash_
1220 // attention_layout returns None when no layer slides).
1221 let sliding_window = if layer_sliding.iter().any(|&s| s) {
1222 g("sliding_window")
1223 } else {
1224 num(&txt, "sliding_window")
1225 .map(|v| v as usize)
1226 .unwrap_or(usize::MAX)
1227 };
1228 // Explicit top-level is_causal (z-lab reference: overrides the layer-type
1229 // default). Parsed as a bare bool; absent = None (historical arms unchanged).
1230 let is_causal = txt
1231 .find("\"is_causal\"")
1232 .and_then(|i| txt[i..].find(':').map(|c| i + c + 1))
1233 .map(|v| txt[v..].trim_start().starts_with("true"));
1234 let cfg = DflashCfg {
1235 hidden: g("hidden_size"),
1236 n_head: g("num_attention_heads"),
1237 n_kv: g("num_key_value_heads"),
1238 head_dim: g("head_dim"),
1239 n_ff: g("intermediate_size"),
1240 n_layer: g("num_hidden_layers"),
1241 eps: num(&txt, "rms_norm_eps").expect("rms_norm_eps") as f32,
1242 // DFlash2 (transformers-5 style): rope_theta lives in the nested
1243 // rope_parameters object — parse it from its scope, not by global find.
1244 rope_theta: if is_dflash2 {
1245 let rp = scope(&txt, "rope_parameters")
1246 .unwrap_or_else(|| panic!("DFlash2 config has no rope_parameters — refusing"));
1247 assert!(
1248 rp.contains("\"default\""),
1249 "DFlash2 rope_parameters rope_type is not \"default\" — the port \
1250 implements plain neox rope only; refusing ({rp})"
1251 );
1252 num(rp, "rope_theta").expect("rope_parameters.rope_theta") as f32
1253 } else {
1254 num(&txt, "rope_theta").expect("rope_theta") as f32
1255 },
1256 block_size: if is_dflash2 {
1257 g2("block_size")
1258 } else {
1259 g("block_size")
1260 },
1261 mask_token_id: if is_dflash2 {
1262 g2("mask_token_id")
1263 } else {
1264 g("mask_token_id")
1265 } as u32,
1266 target_layer_ids: if is_dflash2 {
1267 num_list(d2_cfg_txt.expect("dflash2 scope"), "target_layer_ids")
1268 } else {
1269 num_list(&txt, "target_layer_ids")
1270 },
1271 sliding_window,
1272 layer_sliding,
1273 strategy_dspark: dspark_strategy_census(&txt),
1274 is_causal,
1275 };
1276 if is_dflash2 {
1277 // The windowed round arm implements the reference's NON-causal symmetric
1278 // window only (config `is_causal: false` on the q38 DFlash2 export). A
1279 // causal DFlash2 variant is a different mask program — refuse it rather
1280 // than run the wrong one fluently.
1281 assert_eq!(
1282 cfg.is_causal,
1283 Some(false),
1284 "DFlash2 port requires explicit config is_causal=false \
1285 (non-causal symmetric sliding window); got {is_causal:?} — refusing"
1286 );
1287 assert!(
1288 cfg.layer_sliding.iter().all(|&s| s),
1289 "DFlash2 port expects all layers sliding_attention (q38 export); \
1290 got {:?} — refusing (unverified mask program)",
1291 cfg.layer_sliding
1292 );
1293 assert!(
1294 cfg.block_size <= cfg.sliding_window,
1295 "DFlash2 block {} exceeds the sliding window {} — the windowed SDPA \
1296 omits the future-side mask because block rows stay within the window",
1297 cfg.block_size,
1298 cfg.sliding_window
1299 );
1300 }
1301 let st = memra_gguf::safetensors::StModel::open(&dir.join("model.safetensors"))?;
1302 // 1D norm weights ride raw slices; 2D matmul weights ride GpuTensor::Float
1303 // (cuBLASLt f32 arm — the Stage-A numeric class, right for oracle parity).
1304 let up = |name: &str| -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
1305 let (_info, bytes) = st
1306 .raw(name)
1307 .ok_or_else(|| format!("missing tensor {name}"))?;
1308 Ok(e.htod(&bf16_to_f32(bytes))?)
1309 };
1310 // Precision policy (MEMRA_DFLASH_PREC seam): "q8" = all q8_0 (1.6GB, default);
1311 // "mixed" = bf16 attn+fc (the ctx-conditioning path) + q8_0 ffn (~2.2GB — fits the
1312 // ~2.8GB headroom beside the 31B trunk); "bf16" = all bf16 (parity runs, no target).
1313 let prec = std::env::var("MEMRA_DFLASH_PREC").unwrap_or_else(|_| "q8".into());
1314 let upw = |name: &str| -> Result<GpuTensor, Box<dyn std::error::Error>> {
1315 let (info, bytes) = st
1316 .raw(name)
1317 .ok_or_else(|| format!("missing tensor {name}"))?;
1318 let shape = info.ne(); // ggml order: ne[0]=in_f, ne[1]=out_f
1319 let in_f = shape[0] as usize;
1320 let is_ffn = name.contains(".mlp.");
1321 let bf16 = prec == "bf16"
1322 || (prec == "mixed" && !is_ffn)
1323 || (prec == "fc" && name == "fc.weight");
1324 if bf16 {
1325 return Ok(GpuTensor::FloatBf16 {
1326 data: e.upload_u8(bytes)?,
1327 ne: shape.to_vec(),
1328 });
1329 }
1330 let f32s = bf16_to_f32(bytes);
1331 if prec == "q4" {
1332 let q = encode_q4_0(&f32s);
1333 return Ok(GpuTensor::Quant {
1334 bytes: e.upload_u8(&q)?,
1335 qtype: crate::QT_Q4_0,
1336 row_bytes: in_f / 32 * 18,
1337 ne: shape.to_vec(),
1338 scale: 1.0,
1339 rp: false,
1340 #[cfg(memra_cutlass)]
1341 cutlass: None,
1342 fp8: None,
1343 blk: None,
1344 rp4: None,
1345 f16: None,
1346 });
1347 }
1348 let q = encode_q8_0(&f32s);
1349 Ok(GpuTensor::Quant {
1350 bytes: e.upload_u8(&q)?,
1351 qtype: crate::QT_Q8_0,
1352 row_bytes: in_f / 32 * 34,
1353 ne: shape.to_vec(),
1354 scale: 1.0,
1355 rp: false,
1356 #[cfg(memra_cutlass)]
1357 cutlass: None,
1358 fp8: None,
1359 blk: None,
1360 rp4: None,
1361 f16: None,
1362 })
1363 };
1364 let mut layers = Vec::with_capacity(cfg.n_layer);
1365 for i in 0..cfg.n_layer {
1366 let p = |s: &str| format!("layers.{i}.{s}");
1367 layers.push(DflashLayer {
1368 wq: upw(&p("self_attn.q_proj.weight"))?,
1369 wk: upw(&p("self_attn.k_proj.weight"))?,
1370 wv: upw(&p("self_attn.v_proj.weight"))?,
1371 wo: upw(&p("self_attn.o_proj.weight"))?,
1372 w_gate: upw(&p("mlp.gate_proj.weight"))?,
1373 w_up: upw(&p("mlp.up_proj.weight"))?,
1374 w_down: upw(&p("mlp.down_proj.weight"))?,
1375 ln_in: up(&p("input_layernorm.weight"))?,
1376 ln_post: up(&p("post_attention_layernorm.weight"))?,
1377 q_norm: up(&p("self_attn.q_norm.weight"))?,
1378 k_norm: up(&p("self_attn.k_norm.weight"))?,
1379 });
1380 }
1381 let markov = if let Some((info, bytes)) = st.raw("markov_head.markov_w1.weight") {
1382 let sh = info.ne(); // [rank, vocab] in ggml order (safetensors [V, rank] reversed)
1383 let (rank, vocab) = (sh[0] as usize, sh[1] as usize);
1384 let (i2, b2) = st
1385 .raw("markov_head.markov_w2.weight")
1386 .ok_or("markov_w2 missing beside markov_w1")?;
1387 // w2 follows the precision seam: bf16 for parity runs (the q8_0 encode is a
1388 // serving-size choice and would put quant error inside the markov-logits gate),
1389 // q8_0 otherwise (acceptance-only impact, like the trunk weights).
1390 let w2 = if prec == "bf16" {
1391 GpuTensor::FloatBf16 {
1392 data: e.upload_u8(b2)?,
1393 ne: i2.ne().to_vec(),
1394 }
1395 } else {
1396 let w2f = bf16_to_f32(b2);
1397 let w2q = encode_q8_0(&w2f);
1398 GpuTensor::Quant {
1399 bytes: e.upload_u8(&w2q)?,
1400 qtype: crate::QT_Q8_0,
1401 row_bytes: rank / 32 * 34,
1402 ne: vec![rank as u64, vocab as u64],
1403 scale: 1.0,
1404 rp: false,
1405 #[cfg(memra_cutlass)]
1406 cutlass: None,
1407 fp8: None,
1408 blk: None,
1409 rp4: None,
1410 f16: None,
1411 }
1412 };
1413 Some(MarkovHead {
1414 w1_bf16: e.upload_u8(bytes)?,
1415 w2,
1416 rank,
1417 vocab,
1418 })
1419 } else {
1420 None
1421 };
1422 let confidence = if let Some((info, bytes)) = st.raw("confidence_head.proj.weight") {
1423 let sh = info.ne(); // ggml order: ne[0]=in_dim, ne[1]=1
1424 let in_dim = sh[0] as usize;
1425 let (_bi, bb) = st
1426 .raw("confidence_head.proj.bias")
1427 .ok_or("confidence bias missing beside weight")?;
1428 let with_markov = markov
1429 .as_ref()
1430 .map(|m| in_dim == cfg.hidden + m.rank)
1431 .unwrap_or(false);
1432 if !with_markov && in_dim != cfg.hidden {
1433 panic!(
1434 "confidence_head in_dim {in_dim} matches neither hidden {} nor hidden+rank",
1435 cfg.hidden
1436 );
1437 }
1438 Some(ConfidenceHead {
1439 w: bf16_to_f32(bytes),
1440 b: bf16_to_f32(bb)[0],
1441 in_dim,
1442 with_markov,
1443 })
1444 } else {
1445 None
1446 };
1447 // ---- DFlash2 family tensors (DFLASH2-EVAL-20260820.md §2): 10 conv modules
1448 // (base_kernel + kernel_projection around attention AND mlp in EVERY layer) +
1449 // the candidate path selector (hidden_projection + two codebooks). REQUIRED
1450 // when the arch says DFlash2DraftModel: a missing tensor is a refusal (`?`),
1451 // never a degraded program.
1452 let dflash2 = if is_dflash2 {
1453 assert!(
1454 markov.is_none() && confidence.is_none(),
1455 "DFlash2 checkpoint carries markov/confidence tensors — no such \
1456 variant exists in the family (census refuses the ambiguity)"
1457 );
1458 let rank = g2("selector_rank");
1459 let top_k = g2("selector_top_k");
1460 let conv_k = g2("conv_kernel_size");
1461 let group_size = g2("conv_group_size");
1462 let groups = cfg.hidden / group_size;
1463 let load_conv = |name: &str| -> Result<Dflash2Conv, Box<dyn std::error::Error>> {
1464 let (bi, bb) = st
1465 .raw(&format!("{name}.base_kernel"))
1466 .ok_or_else(|| format!("DFlash2 census: missing {name}.base_kernel"))?;
1467 // safetensors [2, k, hidden] -> ggml ne reversed [hidden, k, 2]
1468 let bne = bi.ne();
1469 assert_eq!(
1470 (bne[0] as usize, bne[1] as usize, bne[2] as usize),
1471 (cfg.hidden, conv_k, 2),
1472 "{name}.base_kernel shape != [2, conv_kernel_size, hidden]"
1473 );
1474 let pname = format!("{name}.kernel_projection.weight");
1475 let (pi, _pb) = st
1476 .raw(&pname)
1477 .ok_or_else(|| format!("DFlash2 census: missing {pname}"))?;
1478 let pne = pi.ne(); // ggml: [in_f=hidden, out_f=2*k*groups]
1479 assert_eq!(
1480 (pne[0] as usize, pne[1] as usize),
1481 (cfg.hidden, 2 * conv_k * groups),
1482 "{pname} shape != [2*conv_kernel_size*groups, hidden]"
1483 );
1484 Ok(Dflash2Conv {
1485 base: e.htod(&bf16_to_f32(bb))?,
1486 proj: upw(&pname)?,
1487 })
1488 };
1489 let mut attn_conv = Vec::with_capacity(cfg.n_layer);
1490 let mut mlp_conv = Vec::with_capacity(cfg.n_layer);
1491 for i in 0..cfg.n_layer {
1492 attn_conv.push(load_conv(&format!("layers.{i}.attention_conv"))?);
1493 mlp_conv.push(load_conv(&format!("layers.{i}.mlp_conv"))?);
1494 }
1495 // Codebooks: stored WITHOUT `.weight` (checkpoint quirk; reference
1496 // from_pretrained maps the keys). Host-resident raw bf16.
1497 let cb = |name: &str| -> Result<(Vec<u8>, usize), Box<dyn std::error::Error>> {
1498 let (ci, cbytes) = st
1499 .raw(&format!("candidate_selector.{name}"))
1500 .ok_or_else(|| format!("DFlash2 census: missing candidate_selector.{name}"))?;
1501 let ne = ci.ne(); // ggml: [rank, V]
1502 assert_eq!(ne[0] as usize, rank, "candidate_selector.{name} rank");
1503 Ok((cbytes.to_vec(), ne[1] as usize))
1504 };
1505 let (pred_codebook, v1) = cb("predecessor_codebook")?;
1506 let (succ_codebook, v2) = cb("successor_codebook")?;
1507 assert_eq!(v1, v2, "codebook vocab mismatch");
1508 let hp_name = "candidate_selector.hidden_projection.weight";
1509 let (hi, _hb) = st
1510 .raw(hp_name)
1511 .ok_or_else(|| format!("DFlash2 census: missing {hp_name}"))?;
1512 assert_eq!(
1513 (hi.ne()[0] as usize, hi.ne()[1] as usize),
1514 (cfg.hidden, rank),
1515 "{hp_name} shape != [rank, hidden]"
1516 );
1517 Some(Dflash2Head {
1518 attn_conv,
1519 mlp_conv,
1520 hidden_proj: upw(hp_name)?,
1521 pred_codebook,
1522 succ_codebook,
1523 rank,
1524 top_k,
1525 conv_k,
1526 group_size,
1527 vocab: v1,
1528 })
1529 } else {
1530 None
1531 };
1532 // CENSUS GATE: every tensor in the export must be consumed by the map above.
1533 // DSpark-class checkpoints (markov head present) and DFlash2 checkpoints
1534 // REFUSE on unrecognized names — an unmapped tensor is a semantic program we
1535 // would silently drop (house law). Plain dflash checkpoints keep the
1536 // historical warn-only behavior.
1537 {
1538 let mut consumed: std::collections::HashSet<String> = std::collections::HashSet::new();
1539 for i in 0..cfg.n_layer {
1540 for s in [
1541 "self_attn.q_proj.weight",
1542 "self_attn.k_proj.weight",
1543 "self_attn.v_proj.weight",
1544 "self_attn.o_proj.weight",
1545 "self_attn.q_norm.weight",
1546 "self_attn.k_norm.weight",
1547 "input_layernorm.weight",
1548 "post_attention_layernorm.weight",
1549 "mlp.gate_proj.weight",
1550 "mlp.up_proj.weight",
1551 "mlp.down_proj.weight",
1552 ] {
1553 consumed.insert(format!("layers.{i}.{s}"));
1554 }
1555 if dflash2.is_some() {
1556 for s in [
1557 "attention_conv.base_kernel",
1558 "attention_conv.kernel_projection.weight",
1559 "mlp_conv.base_kernel",
1560 "mlp_conv.kernel_projection.weight",
1561 ] {
1562 consumed.insert(format!("layers.{i}.{s}"));
1563 }
1564 }
1565 }
1566 for s in [
1567 "fc.weight",
1568 "hidden_norm.weight",
1569 "norm.weight",
1570 "markov_head.markov_w1.weight",
1571 "markov_head.markov_w2.weight",
1572 "confidence_head.proj.weight",
1573 "confidence_head.proj.bias",
1574 ] {
1575 consumed.insert(s.into());
1576 }
1577 if dflash2.is_some() {
1578 for s in [
1579 "candidate_selector.hidden_projection.weight",
1580 "candidate_selector.predecessor_codebook",
1581 "candidate_selector.successor_codebook",
1582 ] {
1583 consumed.insert(s.into());
1584 }
1585 }
1586 let leftovers: Vec<&String> = st.names().filter(|n| !consumed.contains(*n)).collect();
1587 if !leftovers.is_empty() {
1588 if markov.is_some() || dflash2.is_some() {
1589 panic!("dspark/dflash2 census: unrecognized tensors {leftovers:?}");
1590 }
1591 eprintln!("[dflash census] unmapped tensors (ignored): {leftovers:?}");
1592 }
1593 }
1594 // YaRN rope from config rope_parameters (HF _compute_yarn_parameters, verified
1595 // numerically vs Qwen3RotaryEmbedding on the arm-a export).
1596 let rope_yarn =
1597 if txt.contains("\"rope_type\": \"yarn\"") || txt.contains("\"rope_type\":\"yarn\"") {
1598 let factor = num(&txt, "factor").expect("yarn factor") as f64;
1599 let orig = num(&txt, "original_max_position_embeddings").expect("yarn orig");
1600 let beta_fast = num(&txt, "beta_fast").expect("beta_fast");
1601 let beta_slow = num(&txt, "beta_slow").expect("beta_slow");
1602 let base = cfg.rope_theta as f64;
1603 let d = cfg.head_dim as f64;
1604 let corr =
1605 |r: f64| d * (orig / (r * 2.0 * std::f64::consts::PI)).ln() / (2.0 * base.ln());
1606 let low = corr(beta_fast).floor().max(0.0);
1607 let high = corr(beta_slow).ceil().min(d - 1.0);
1608 let half = cfg.head_dim / 2;
1609 let mut ff = Vec::with_capacity(half);
1610 for j in 0..half {
1611 let base_inv = base.powf(-2.0 * j as f64 / d);
1612 let ramp = (((j as f64) - low) / (high - low)).clamp(0.0, 1.0);
1613 let ex = 1.0 - ramp; // extrapolation share
1614 let yarn_inv = (base_inv / factor) * (1.0 - ex) + base_inv * ex;
1615 ff.push((base_inv / yarn_inv) as f32);
1616 }
1617 let mscale = (0.1 * factor.ln() + 1.0) as f32;
1618 Some((e.htod(&ff)?, mscale))
1619 } else {
1620 None
1621 };
1622 // Ratified-default receipts (capacity-keyed-defaults law: the active program is
1623 // NAMED at load, never inferred from silence). The boot output-sample gate greps
1624 // these two lines; a run whose log lacks them did not load this code.
1625 eprintln!(
1626 "[dspark] harvest={} (checkpoint census dflash2={} strategy_dspark={}, \
1627 MEMRA_DSPARK_HARVEST {})",
1628 DsparkHarvest::for_family_value(
1629 dflash2.is_some(),
1630 std::env::var("MEMRA_DSPARK_HARVEST").ok().as_deref(),
1631 cfg.strategy_dspark,
1632 )
1633 .name(),
1634 dflash2.is_some(),
1635 cfg.strategy_dspark,
1636 match std::env::var("MEMRA_DSPARK_HARVEST") {
1637 Ok(v) if !v.is_empty() => "set",
1638 _ => "unset",
1639 },
1640 );
1641 eprintln!(
1642 "[dspark] verify-window={:?} (accept-rate head {}, MEMRA_DSPARK_VT {})",
1643 DsparkVtPolicy::resolve(confidence.is_some()),
1644 if confidence.is_some() {
1645 "present"
1646 } else {
1647 "ABSENT -> ladder"
1648 },
1649 match std::env::var("MEMRA_DSPARK_VT") {
1650 Ok(v) if !v.is_empty() => "set",
1651 _ => "unset",
1652 },
1653 );
1654 Ok(Self {
1655 fc: upw("fc.weight")?,
1656 hidden_norm: up("hidden_norm.weight")?,
1657 norm: up("norm.weight")?,
1658 cfg,
1659 layers,
1660 markov,
1661 confidence,
1662 rope_yarn,
1663 dflash2,
1664 })
1665 }
1666
1667 /// Rope q or k rows in place: yarn (ff divisors + post-rope mscale) when the config
1668 /// carries it, plain neox otherwise. One primitive for all five drafter rope sites.
1669 fn rope_rows(
1670 &self,
1671 e: &Engine,
1672 x: &mut CudaSlice<f32>,
1673 pos_d: &CudaSlice<i32>,
1674 n_heads: usize,
1675 n_tokens: usize,
1676 ) -> Result<(), Box<dyn std::error::Error>> {
1677 let c = &self.cfg;
1678 match &self.rope_yarn {
1679 Some((ff, mscale)) => {
1680 e.rope_neox_ff(
1681 x,
1682 pos_d,
1683 c.head_dim,
1684 c.head_dim,
1685 n_heads,
1686 n_tokens,
1687 c.rope_theta,
1688 1.0,
1689 ff,
1690 )?;
1691 e.scale_inplace(x, *mscale, n_tokens * n_heads * c.head_dim)?;
1692 }
1693 None => {
1694 e.rope_neox(
1695 x,
1696 pos_d,
1697 c.head_dim,
1698 c.head_dim,
1699 n_heads,
1700 n_tokens,
1701 c.rope_theta,
1702 1.0,
1703 )?;
1704 }
1705 }
1706 Ok(())
1707 }
1708
1709 /// f32 GEMM helper via the engine Float arm (cuBLASLt): y[t, out_f].
1710 fn mm(
1711 &self,
1712 e: &Engine,
1713 w: &GpuTensor,
1714 x: &CudaSlice<f32>,
1715 t: usize,
1716 _in_f: usize,
1717 _out_f: usize,
1718 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
1719 Ok(e.matmul(w, x, t)?)
1720 }
1721
1722 /// DFlash2 conv `prepare` (reference GroupedDynamicCausalConv.prepare): projects
1723 /// the pre-conv rows to BOTH dynamic kernels, convolves the rows with base half 0
1724 /// + dyn half 0, and returns (convolved rows, the dyn projection) — `finish`
1725 /// reuses the SAME projection's half 1. Block-local causal shift (row 0 zero-pads).
1726 pub fn d2_conv_prepare(
1727 &self,
1728 e: &Engine,
1729 conv: &Dflash2Conv,
1730 xn: &CudaSlice<f32>,
1731 rows: usize,
1732 ) -> Result<(CudaSlice<f32>, CudaSlice<f32>), Box<dyn std::error::Error>> {
1733 let d2 = self
1734 .dflash2
1735 .as_ref()
1736 .expect("d2_conv on a non-dflash2 draft");
1737 let h = self.cfg.hidden;
1738 let groups = h / d2.group_size;
1739 let dyn_ = self.mm(e, &conv.proj, xn, rows, h, 2 * d2.conv_k * groups)?;
1740 let mut out = e.uninit(rows * h)?;
1741 e.dflash2_dynconv(
1742 xn,
1743 &dyn_,
1744 &conv.base,
1745 &mut out,
1746 rows,
1747 h,
1748 d2.group_size,
1749 d2.conv_k,
1750 0,
1751 )?;
1752 Ok((out, dyn_))
1753 }
1754
1755 /// DFlash2 conv `finish`: convolves the sublayer OUTPUT rows with base half 1 +
1756 /// dyn half 1 (dyn from the matching `prepare`).
1757 pub fn d2_conv_finish(
1758 &self,
1759 e: &Engine,
1760 conv: &Dflash2Conv,
1761 y: &CudaSlice<f32>,
1762 dyn_: &CudaSlice<f32>,
1763 rows: usize,
1764 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
1765 let d2 = self
1766 .dflash2
1767 .as_ref()
1768 .expect("d2_conv on a non-dflash2 draft");
1769 let h = self.cfg.hidden;
1770 let mut out = e.uninit(rows * h)?;
1771 e.dflash2_dynconv(
1772 y,
1773 dyn_,
1774 &conv.base,
1775 &mut out,
1776 rows,
1777 h,
1778 d2.group_size,
1779 d2.conv_k,
1780 1,
1781 )?;
1782 Ok(out)
1783 }
1784
1785 /// DFlash2 proposal (reference `DFlash2DraftModel.propose`, greedy arm): device
1786 /// top-k over the draft logits + the rank-`r` hidden projection, ONE small dtoh
1787 /// (~nd*(2k+rank) floats — the same per-round sync slot the markov chain's token
1788 /// readback occupies), then the host codebook walk. Returns the nd drafted tokens
1789 /// (mask-fill rows 1..b-1; the anchor row is not a draft).
1790 pub fn dflash2_propose_greedy(
1791 &self,
1792 e: &Engine,
1793 dl: &CudaSlice<f32>,
1794 rows: &CudaSlice<f32>,
1795 nd: usize,
1796 n_vocab: usize,
1797 anchor: u32,
1798 ) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
1799 let d2 = self
1800 .dflash2
1801 .as_ref()
1802 .expect("dflash2_propose on a non-dflash2 draft");
1803 assert!(
1804 n_vocab <= d2.vocab,
1805 "target head vocab {n_vocab} exceeds the selector codebooks ({})",
1806 d2.vocab
1807 );
1808 let (vals_d, idx_d) = e.topk_rows(dl, nd, n_vocab, d2.top_k)?;
1809 let hproj_d = e.matmul(&d2.hidden_proj, rows, nd)?;
1810 let unary = e.dtoh(&vals_d)?;
1811 let cand = e.dtoh_u32(&idx_d)?;
1812 let hproj = e.dtoh(&hproj_d)?;
1813 Ok(d2.walk_greedy(&unary, &cand, &hproj, anchor, nd))
1814 }
1815
1816 /// DFlash2 proposal, SAMPLED arm (reference `DFlash2DraftModel.propose` at T>0): same
1817 /// device top-k + hidden projection + one dtoh as the greedy arm, then the host
1818 /// candidate-set softmax walk (`dflash2_walk_sampled`) drawing one host-Philox uniform
1819 /// per slot from the session's `uctr` stream. Returns (path, q_chosen, cand, q_rows).
1820 #[allow(clippy::too_many_arguments)]
1821 pub(crate) fn dflash2_propose_sampled(
1822 &self,
1823 e: &Engine,
1824 dl: &CudaSlice<f32>,
1825 rows: &CudaSlice<f32>,
1826 nd: usize,
1827 n_vocab: usize,
1828 anchor: u32,
1829 temp: f32,
1830 seed: u64,
1831 uctr: &mut u32,
1832 ) -> Result<Dflash2SampledProposal, Box<dyn std::error::Error>> {
1833 let d2 = self
1834 .dflash2
1835 .as_ref()
1836 .expect("dflash2_propose on a non-dflash2 draft");
1837 assert!(
1838 n_vocab <= d2.vocab,
1839 "target head vocab {n_vocab} exceeds the selector codebooks ({})",
1840 d2.vocab
1841 );
1842 let (vals_d, idx_d) = e.topk_rows(dl, nd, n_vocab, d2.top_k)?;
1843 let hproj_d = e.matmul(&d2.hidden_proj, rows, nd)?;
1844 let unary = e.dtoh(&vals_d)?;
1845 let cand = e.dtoh_u32(&idx_d)?;
1846 let hproj = e.dtoh(&hproj_d)?;
1847 let mut draw = || {
1848 let u = crate::spec::host_u01(seed, *uctr);
1849 *uctr = uctr.wrapping_add(1);
1850 u
1851 };
1852 let (path, q_chosen, q_rows) =
1853 d2.walk_sampled(&unary, &cand, &hproj, anchor, nd, temp, &mut draw);
1854 Ok((path, q_chosen, cand, q_rows))
1855 }
1856
1857 /// Sampled draft chain for the Rows families (T>0 twin of the greedy markov chain):
1858 /// slot k gets the markov bias of the PREVIOUS chain token added in place (when the
1859 /// head is armed — the sglang DSPARK worker's markov-corrected draft probs), then ONE
1860 /// draw from the row's FILTERED softmax (filter_stats -> device-stat gumbel perturb ->
1861 /// argmax into the chain buffer — the frspec eager-chain composition, stats kept on
1862 /// device so the chain stays sync-free like the greedy arm). Without a markov head the
1863 /// rows sample independently (the z-lab reference's T>0 arm for plain DFlash). `dl` is
1864 /// biased IN PLACE and retained by the caller: it is the accept walk's q source.
1865 #[allow(clippy::too_many_arguments)]
1866 pub(crate) fn dspark_chain_sampled(
1867 &self,
1868 e: &Engine,
1869 dl: &mut CudaSlice<f32>,
1870 nd: usize,
1871 n_vocab: usize,
1872 anchor: u32,
1873 sp: &crate::spec::SpecSampling,
1874 sctr: &mut u32,
1875 // H4 confidence-policy stash (v0.100 train merge): Some = copy each slot's
1876 // markov prev-token embedding (the exact `w1` row the chain gathers) into a
1877 // [nd, rank] buffer — the same d2d stash the greedy chain carries, so the
1878 // confidence window sizes identically at T>0.
1879 mut conf_emb: Option<&mut CudaSlice<f32>>,
1880 ) -> Result<(Vec<u32>, DsparkDraftSample), Box<dyn std::error::Error>> {
1881 let markov_on = std::env::var("MEMRA_DFLASH_MARKOV").as_deref() != Ok("0");
1882 let mut chain_d = e.stream().alloc_zeros::<u32>(nd + 1)?;
1883 e.set_u32_one(&mut chain_d, anchor)?;
1884 let mut th_all = e.zeros(nd)?;
1885 let mut z_all = e.zeros(nd)?;
1886 let mut mx_all = e.zeros(nd)?;
1887 let mut pb = e.zeros(n_vocab)?;
1888 for k in 0..nd {
1889 if let (Some(mk), true) = (&self.markov, markov_on) {
1890 let mut f = e.uninit(mk.rank)?;
1891 e.gather_row_bf16(&mk.w1_bf16, &chain_d, k, &mut f, mk.rank)?;
1892 if let Some(ce) = conf_emb.as_deref_mut() {
1893 let fv = e.view(&f, mk.rank);
1894 e.copy_view_into(ce, k * mk.rank, &fv, mk.rank)?;
1895 }
1896 let bias = e.matmul(&mk.w2, &f, 1)?;
1897 e.add_row_inplace(dl, &bias, n_vocab, k * n_vocab)?;
1898 } else if let (Some(ce), Some(mk)) = (conf_emb.as_deref_mut(), &self.markov) {
1899 // MARKOV=0 arm still stashes the embedding for the confidence head —
1900 // the greedy chain's exact behavior.
1901 let mut f = e.uninit(mk.rank)?;
1902 e.gather_row_bf16(&mk.w1_bf16, &chain_d, k, &mut f, mk.rank)?;
1903 let fv = e.view(&f, mk.rank);
1904 e.copy_view_into(ce, k * mk.rank, &fv, mk.rank)?;
1905 }
1906 let rows_k = e.htod_i32(&[k as i32])?;
1907 let (mut th1, mut z1, mut mx1) = (e.zeros(1)?, e.zeros(1)?, e.zeros(1)?);
1908 e.filter_stats(
1909 dl, n_vocab, &rows_k, &mut th1, &mut z1, &mut mx1, n_vocab, 1, sp.temp, sp.top_k,
1910 sp.top_p, sp.min_p,
1911 )?;
1912 e.gumbel_perturb_filtered_col(
1913 dl, k, &mut pb, n_vocab, sp.seed, *sctr, sp.temp, &mx1, &th1, 0,
1914 )?;
1915 *sctr = sctr.wrapping_add(1);
1916 e.argmax_token_device_col(&pb, 0, n_vocab, &mut chain_d, k + 1)?;
1917 e.copy_into(&mut th_all, k, &th1, 1)?;
1918 e.copy_into(&mut z_all, k, &z1, 1)?;
1919 e.copy_into(&mut mx_all, k, &mx1, 1)?;
1920 }
1921 let chain = e.dtoh_u32(&chain_d)?;
1922 let (thv, zv, mxv) = (e.dtoh(&th_all)?, e.dtoh(&z_all)?, e.dtoh(&mx_all)?);
1923 let stats = (0..nd).map(|i| (mxv[i], thv[i], zv[i])).collect();
1924 Ok((
1925 chain[1..].to_vec(),
1926 DsparkDraftSample::Rows {
1927 th: th_all,
1928 z: z_all,
1929 stats,
1930 },
1931 ))
1932 }
1933
1934 /// Family dispatch for the sampled proposal: Selector for DFlash2, Rows otherwise.
1935 /// Returns the drafted tokens (the round's `cand` tail) + the proposal record.
1936 #[allow(clippy::too_many_arguments)]
1937 pub(crate) fn dspark_propose_sampled(
1938 &self,
1939 e: &Engine,
1940 dl: &mut CudaSlice<f32>,
1941 rows: &CudaSlice<f32>,
1942 nd: usize,
1943 n_vocab: usize,
1944 anchor: u32,
1945 sp: &crate::spec::SpecSampling,
1946 sctr: &mut u32,
1947 uctr: &mut u32,
1948 conf_emb: Option<&mut CudaSlice<f32>>,
1949 ) -> Result<(Vec<u32>, DsparkDraftSample), Box<dyn std::error::Error>> {
1950 if let Some(d2) = self.dflash2.as_ref() {
1951 // The confidence stash is a markov-family program; DFlash2 has no
1952 // accept-rate head (the policy resolver never arms it for this family).
1953 debug_assert!(
1954 conf_emb.is_none(),
1955 "conf_emb stash requested on a DFlash2 selector proposal"
1956 );
1957 let (path, q_chosen, cand, q_rows) = self.dflash2_propose_sampled(
1958 e, dl, rows, nd, n_vocab, anchor, sp.temp, sp.seed, uctr,
1959 )?;
1960 Ok((
1961 path,
1962 DsparkDraftSample::Selector {
1963 cand,
1964 q_rows,
1965 q_chosen,
1966 top_k: d2.top_k,
1967 },
1968 ))
1969 } else {
1970 self.dspark_chain_sampled(e, dl, nd, n_vocab, anchor, sp, sctr, conf_emb)
1971 }
1972 }
1973
1974 /// FIRST-LIGHT forward (oracle contract): full non-causal attention over
1975 /// [ctx_features ; block], NO draft KV cache, NO sliding window (the oracle bypasses
1976 /// the reference mask machinery the same way — window/caching land in the round arm).
1977 ///
1978 /// `target_hidden`: [ctx, n_taps*hidden] (f32, device) — raw tapped states.
1979 /// `noise_emb`: [block, hidden] — target embed rows for [accepted, MASK x b-1].
1980 /// `pos`: absolute positions for ctx rows THEN block rows (ctx+block i32).
1981 /// Returns final normed hidden [block, hidden] (feed target lm_head for draft logits).
1982 /// ctx features for `t` tapped rows: hidden_norm(fc(taps)) — the drafter's context
1983 /// representation, cacheable across rounds (append-only in committed-token order).
1984 pub fn ctx_features(
1985 &self,
1986 e: &Engine,
1987 taps: &CudaSlice<f32>,
1988 t: usize,
1989 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
1990 let c = &self.cfg;
1991 let n_taps = c.target_layer_ids.len();
1992 let fc_out = self.mm(e, &self.fc, taps, t, n_taps * c.hidden, c.hidden)?;
1993 let mut out = e.uninit(t * c.hidden)?;
1994 e.rms_norm(&fc_out, &self.hidden_norm, &mut out, c.hidden, t, c.eps)?;
1995 Ok(out)
1996 }
1997
1998 pub fn forward(
1999 &self,
2000 e: &Engine,
2001 target_hidden: &CudaSlice<f32>,
2002 noise_emb: &CudaSlice<f32>,
2003 pos: &[i32],
2004 ctx: usize,
2005 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2006 let ctx_f = self.ctx_features(e, target_hidden, ctx)?;
2007 if let Ok(dir) = std::env::var("MEMRA_DFLASH_DUMP") {
2008 let v = e.dtoh(&ctx_f)?;
2009 let bytes: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
2010 std::fs::write(format!("{dir}/memra-ctx_features.f32"), bytes)?;
2011 }
2012 self.forward_block(e, &ctx_f, noise_emb, pos, ctx)
2013 }
2014
2015 /// Block forward over PRECOMPUTED ctx features (the round arm's entry: features are
2016 /// cached across rounds; only the block work repeats).
2017 pub fn forward_block(
2018 &self,
2019 e: &Engine,
2020 ctx_f: &CudaSlice<f32>,
2021 noise_emb: &CudaSlice<f32>,
2022 pos: &[i32],
2023 ctx: usize,
2024 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2025 let c = &self.cfg;
2026 let (h, nh, nkv, hd) = (c.hidden, c.n_head, c.n_kv, c.head_dim);
2027 let b = c.block_size;
2028 assert_eq!(pos.len(), ctx + b, "pos covers ctx rows then block rows");
2029
2030 let pos_blk = e.htod_i32(&pos[ctx..])?;
2031
2032 let mut x = e.clone_dtod(noise_emb)?; // [b, hidden] residual stream
2033 for (li, l) in self.layers.iter().enumerate() {
2034 // input_layernorm on the block rows only (ctx features are norm-free per ref:
2035 // k/v project the SAME ctx_f every layer, un-layernormed).
2036 let mut xn = e.uninit(b * h)?;
2037 e.rms_norm(&x, &l.ln_in, &mut xn, h, b, c.eps)?;
2038 // DFlash2: dynamic conv WRAPS attention — q/k_noise/v_noise all project the
2039 // CONVOLVED block rows (reference decoder layer: prepare -> self_attn ->
2040 // finish, all inside the residual branch). ctx_f is never convolved.
2041 let mut attn_dyn: Option<CudaSlice<f32>> = None;
2042 if let Some(d2) = &self.dflash2 {
2043 let (xc, dyn_) = self.d2_conv_prepare(e, &d2.attn_conv[li], &xn, b)?;
2044 xn = xc;
2045 attn_dyn = Some(dyn_);
2046 }
2047
2048 // q from block; k/v from [ctx_f ; block-normed]
2049 let q0 = self.mm(e, &l.wq, &xn, b, h, nh * hd)?;
2050 let k0c = self.mm(e, &l.wk, ctx_f, ctx, h, nkv * hd)?;
2051 let v0c = self.mm(e, &l.wv, ctx_f, ctx, h, nkv * hd)?;
2052 let k0b = self.mm(e, &l.wk, &xn, b, h, nkv * hd)?;
2053 let v0b = self.mm(e, &l.wv, &xn, b, h, nkv * hd)?;
2054
2055 // per-head q/k rms norm (v passes through: ones weight trick not needed — the
2056 // qkv kernel norms rq+rk rows; concatenate k first).
2057 let mut k0 = e.uninit((ctx + b) * nkv * hd)?;
2058 e.copy_into(&mut k0, 0, &k0c, ctx * nkv * hd)?;
2059 e.copy_into(&mut k0, ctx * nkv * hd, &k0b, b * nkv * hd)?;
2060 let mut v = e.uninit((ctx + b) * nkv * hd)?;
2061 e.copy_into(&mut v, 0, &v0c, ctx * nkv * hd)?;
2062 e.copy_into(&mut v, ctx * nkv * hd, &v0b, b * nkv * hd)?;
2063
2064 if li == 0 {
2065 if let Ok(dir) = std::env::var("MEMRA_DFLASH_DUMP") {
2066 let v = e.dtoh(&q0)?;
2067 let bytes: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
2068 std::fs::write(format!("{dir}/memra-l0_q0.f32"), bytes)?;
2069 }
2070 }
2071 let mut q = e.uninit(b * nh * hd)?;
2072 let mut k = e.uninit((ctx + b) * nkv * hd)?;
2073 // rms over head_dim rows: q has b*nh rows, k has (ctx+b)*nkv rows.
2074 e.rms_norm(&q0, &l.q_norm, &mut q, hd, b * nh, c.eps)?;
2075 if li == 0 {
2076 if let Ok(dir) = std::env::var("MEMRA_DFLASH_DUMP") {
2077 let v = e.dtoh(&q)?;
2078 let bytes: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
2079 std::fs::write(format!("{dir}/memra-l0_qn.f32"), bytes)?;
2080 }
2081 }
2082 e.rms_norm(&k0, &l.k_norm, &mut k, hd, (ctx + b) * nkv, c.eps)?;
2083
2084 // rope: q at block positions, k at ctx-then-block positions (absolute).
2085 let norope = std::env::var("MEMRA_DFLASH_NOROPE").is_ok();
2086 if !norope {
2087 self.rope_rows(e, &mut q, &pos_blk, nh, b)?;
2088 }
2089 if li == 0 {
2090 if let Ok(dir) = std::env::var("MEMRA_DFLASH_DUMP") {
2091 let dump = |name: &str,
2092 t: &cudarc::driver::CudaSlice<f32>|
2093 -> Result<(), Box<dyn std::error::Error>> {
2094 let v = e.dtoh(t)?;
2095 let bytes: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
2096 std::fs::write(format!("{dir}/memra-l0_{name}.f32"), bytes)?;
2097 Ok(())
2098 };
2099 dump("xn", &xn)?;
2100 dump("q_prerope", &q)?;
2101 }
2102 }
2103 // k rows are laid out [row, nkv, hd] with row-major tokens — rope_neox expects
2104 // (n_heads, n_tokens); ctx and block ropes run as one call over ctx+b tokens.
2105 let pos_all = e.htod_i32(pos)?;
2106 if !norope {
2107 self.rope_rows(e, &mut k, &pos_all, nkv, ctx + b)?;
2108 }
2109
2110 // full non-causal attention: every block query sees all ctx+b keys.
2111 let mut attn = e.uninit(b * nh * hd)?;
2112 let scale = 1.0f32 / (hd as f32).sqrt();
2113 // NAIVE SDPA for first light: fa_prefill's NON-CAUSAL arm with T != T_kv is
2114 // BROKEN (attn maxdiff 0.34 vs the torch oracle; q/k inputs bit-close — no
2115 // existing caller exercises that shape class, jsonl 2026-07-13). The 16 x
2116 // (ctx+16) block attention is tiny; the fa arm returns behind this seam once
2117 // its kernel is fixed + parity-gated.
2118 if self.dflash2.is_some() && c.layer_sliding[li] {
2119 // DFlash2 non-causal symmetric window (config is_causal=false, all
2120 // layers sliding). The kernel masks only keys OLDER than
2121 // q_pos-(window-1); the future side (k - q < window) never binds
2122 // because keys reach at most q_pos + block <= q_pos + window
2123 // (asserted at load). Positions must be contiguous — q_pos is derived
2124 // in-kernel as (T_kv - T) + qt.
2125 debug_assert!(pos.windows(2).all(|w| w[1] == w[0] + 1));
2126 d2_windowed_attn(e, &q, &k, &v, &mut attn, hd, nh, nkv, b, ctx + b, scale, c)?;
2127 } else if std::env::var("MEMRA_DFLASH_FA").is_ok() {
2128 e.fa_prefill(&q, &k, &v, &mut attn, hd, nh, nkv, b, ctx + b, scale, false)?;
2129 } else {
2130 e.sdpa_naive(&q, &k, &v, &mut attn, hd, nh, nkv, b, ctx + b, scale, false)?;
2131 }
2132
2133 let mut o = self.mm(e, &l.wo, &attn, b, nh * hd, h)?;
2134 if let (Some(d2), Some(dyn_)) = (&self.dflash2, &attn_dyn) {
2135 o = self.d2_conv_finish(e, &d2.attn_conv[li], &o, dyn_, b)?;
2136 }
2137 let mut x1 = e.uninit(b * h)?;
2138 e.add(&o, &x, &mut x1, b * h)?;
2139 if li == 0 {
2140 if let Ok(dir) = std::env::var("MEMRA_DFLASH_DUMP") {
2141 let dump = |name: &str,
2142 t: &cudarc::driver::CudaSlice<f32>|
2143 -> Result<(), Box<dyn std::error::Error>> {
2144 let v = e.dtoh(t)?;
2145 let bytes: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
2146 std::fs::write(format!("{dir}/memra-l0_{name}.f32"), bytes)?;
2147 Ok(())
2148 };
2149 dump("q", &q)?;
2150 dump("k", &k)?;
2151 dump("attn", &attn)?;
2152 dump("x1", &x1)?;
2153 }
2154 }
2155
2156 // mlp (DFlash2: the same conv wrap — prepare on the post-ln rows, mlp on
2157 // the convolved rows, finish on the mlp output, then the residual add)
2158 let mut x1n = e.uninit(b * h)?;
2159 e.rms_norm(&x1, &l.ln_post, &mut x1n, h, b, c.eps)?;
2160 let mut mlp_dyn: Option<CudaSlice<f32>> = None;
2161 if let Some(d2) = &self.dflash2 {
2162 let (xc, dyn_) = self.d2_conv_prepare(e, &d2.mlp_conv[li], &x1n, b)?;
2163 x1n = xc;
2164 mlp_dyn = Some(dyn_);
2165 }
2166 let gate = self.mm(e, &l.w_gate, &x1n, b, h, c.n_ff)?;
2167 let up_ = self.mm(e, &l.w_up, &x1n, b, h, c.n_ff)?;
2168 let mut act = e.uninit(b * c.n_ff)?;
2169 e.silu_mul(&gate, &up_, &mut act, b * c.n_ff)?;
2170 let mut down = self.mm(e, &l.w_down, &act, b, c.n_ff, h)?;
2171 if let (Some(d2), Some(dyn_)) = (&self.dflash2, &mlp_dyn) {
2172 down = self.d2_conv_finish(e, &d2.mlp_conv[li], &down, dyn_, b)?;
2173 }
2174 let mut x2 = e.uninit(b * h)?;
2175 e.add(&down, &x1, &mut x2, b * h)?;
2176 x = x2;
2177 if let Ok(dir) = std::env::var("MEMRA_DFLASH_DUMP") {
2178 let v = e.dtoh(&x)?;
2179 let bytes: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
2180 std::fs::write(format!("{dir}/memra-layer{li}_out.f32"), bytes)?;
2181 }
2182 }
2183 let mut out = e.uninit(b * h)?;
2184 e.rms_norm(&x, &self.norm, &mut out, h, b, c.eps)?;
2185 Ok(out)
2186 }
2187}
2188
2189/// Draft KV cache (round-cost fix, 2026-07-13): per-layer normed+roped ctx K and raw ctx V,
2190/// append-only in committed order. Block K/V land TRANSIENTLY at [len..len+b] each round
2191/// (never committed — the reference crops them identically). Kills the per-round full-ctx
2192/// projection recompute (first light was O(ctx)/round -> 7 tok/s).
2193pub struct DflashKv {
2194 pub k: Vec<CudaSlice<f32>>, // per layer [cap + block, nkv*hd]
2195 pub v: Vec<CudaSlice<f32>>,
2196 pub len: usize,
2197 pub cap: usize,
2198}
2199
2200impl DflashKv {
2201 pub fn new(
2202 e: &Engine,
2203 cfg: &DflashCfg,
2204 cap: usize,
2205 ) -> Result<Self, Box<dyn std::error::Error>> {
2206 let rowsz = cfg.n_kv * cfg.head_dim;
2207 let mut k = Vec::with_capacity(cfg.n_layer);
2208 let mut v = Vec::with_capacity(cfg.n_layer);
2209 for _ in 0..cfg.n_layer {
2210 k.push(e.uninit((cap + cfg.block_size) * rowsz)?);
2211 v.push(e.uninit((cap + cfg.block_size) * rowsz)?);
2212 }
2213 Ok(Self { k, v, len: 0, cap })
2214 }
2215}
2216
2217impl DflashDraft {
2218 /// Ingest `t` NEW ctx-feature rows (committed order, absolute positions `pos_new`) into
2219 /// the draft KV: per layer k/v projections + k head-norm + rope, appended at kv.len.
2220 pub fn ingest_ctx(
2221 &self,
2222 e: &Engine,
2223 kv: &mut DflashKv,
2224 feats: &CudaSlice<f32>,
2225 pos_new: &[i32],
2226 t: usize,
2227 ) -> Result<(), Box<dyn std::error::Error>> {
2228 let c = &self.cfg;
2229 let (h, nkv, hd) = (c.hidden, c.n_kv, c.head_dim);
2230 assert!(kv.len + t <= kv.cap, "draft kv overflow");
2231 let pos_d = e.htod_i32(pos_new)?;
2232 for (li, l) in self.layers.iter().enumerate() {
2233 let k0 = self.mm(e, &l.wk, feats, t, h, nkv * hd)?;
2234 let v0 = self.mm(e, &l.wv, feats, t, h, nkv * hd)?;
2235 let mut kn = e.uninit(t * nkv * hd)?;
2236 e.rms_norm(&k0, &l.k_norm, &mut kn, hd, t * nkv, c.eps)?;
2237 self.rope_rows(e, &mut kn, &pos_d, nkv, t)?;
2238 e.copy_into(&mut kv.k[li], kv.len * nkv * hd, &kn, t * nkv * hd)?;
2239 e.copy_into(&mut kv.v[li], kv.len * nkv * hd, &v0, t * nkv * hd)?;
2240 }
2241 kv.len += t;
2242 Ok(())
2243 }
2244
2245 /// Block forward over the CACHED ctx KV: only the 16 block rows are projected per layer;
2246 /// block K/V land transiently at kv[len..len+b]. Bit-class-identical to forward_block
2247 /// (same kernels, same per-row programs; ONLY the ctx K/V recompute is cached).
2248 pub fn forward_round(
2249 &self,
2250 e: &Engine,
2251 kv: &mut DflashKv,
2252 noise_emb: &CudaSlice<f32>,
2253 pos_block: &[i32],
2254 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>> {
2255 let c = &self.cfg;
2256 let (h, nh, nkv, hd) = (c.hidden, c.n_head, c.n_kv, c.head_dim);
2257 let b = c.block_size;
2258 assert_eq!(pos_block.len(), b);
2259 let ctx = kv.len;
2260 let pos_blk = e.htod_i32(pos_block)?;
2261 let mut x = e.clone_dtod(noise_emb)?;
2262 for (li, l) in self.layers.iter().enumerate() {
2263 let mut xn = e.uninit(b * h)?;
2264 e.rms_norm(&x, &l.ln_in, &mut xn, h, b, c.eps)?;
2265 // DFlash2: dynamic conv wraps attention (see forward_block).
2266 let mut attn_dyn: Option<CudaSlice<f32>> = None;
2267 if let Some(d2) = &self.dflash2 {
2268 let (xc, dyn_) = self.d2_conv_prepare(e, &d2.attn_conv[li], &xn, b)?;
2269 xn = xc;
2270 attn_dyn = Some(dyn_);
2271 }
2272 let q0 = self.mm(e, &l.wq, &xn, b, h, nh * hd)?;
2273 let k0b = self.mm(e, &l.wk, &xn, b, h, nkv * hd)?;
2274 let v0b = self.mm(e, &l.wv, &xn, b, h, nkv * hd)?;
2275 let mut q = e.uninit(b * nh * hd)?;
2276 let mut kb = e.uninit(b * nkv * hd)?;
2277 e.rms_norm(&q0, &l.q_norm, &mut q, hd, b * nh, c.eps)?;
2278 e.rms_norm(&k0b, &l.k_norm, &mut kb, hd, b * nkv, c.eps)?;
2279 self.rope_rows(e, &mut q, &pos_blk, nh, b)?;
2280 self.rope_rows(e, &mut kb, &pos_blk, nkv, b)?;
2281 e.copy_into(&mut kv.k[li], ctx * nkv * hd, &kb, b * nkv * hd)?;
2282 e.copy_into(&mut kv.v[li], ctx * nkv * hd, &v0b, b * nkv * hd)?;
2283 let mut attn = e.uninit(b * nh * hd)?;
2284 let scale = 1.0f32 / (hd as f32).sqrt();
2285 if self.dflash2.is_some() && c.layer_sliding[li] {
2286 // Non-causal symmetric window (config is_causal=false): kv row index
2287 // == absolute position for BOTH ctx rows (committed order) and the
2288 // transient block rows, so the kernel's q_pos = (T_kv - T) + qt is the
2289 // absolute position and the old-side mask is exact. The future side
2290 // never binds (block <= window, asserted at load).
2291 d2_windowed_attn(
2292 e,
2293 &q,
2294 &kv.k[li],
2295 &kv.v[li],
2296 &mut attn,
2297 hd,
2298 nh,
2299 nkv,
2300 b,
2301 ctx + b,
2302 scale,
2303 c,
2304 )?;
2305 } else if std::env::var("MEMRA_DFLASH_FA").is_ok() {
2306 e.fa_prefill(
2307 &q,
2308 &kv.k[li],
2309 &kv.v[li],
2310 &mut attn,
2311 hd,
2312 nh,
2313 nkv,
2314 b,
2315 ctx + b,
2316 scale,
2317 false,
2318 )?;
2319 } else {
2320 e.sdpa_naive(
2321 &q,
2322 &kv.k[li],
2323 &kv.v[li],
2324 &mut attn,
2325 hd,
2326 nh,
2327 nkv,
2328 b,
2329 ctx + b,
2330 scale,
2331 false,
2332 )?;
2333 }
2334 let mut o = self.mm(e, &l.wo, &attn, b, nh * hd, h)?;
2335 if let (Some(d2), Some(dyn_)) = (&self.dflash2, &attn_dyn) {
2336 o = self.d2_conv_finish(e, &d2.attn_conv[li], &o, dyn_, b)?;
2337 }
2338 let mut x1 = e.uninit(b * h)?;
2339 e.add(&o, &x, &mut x1, b * h)?;
2340 let mut x1n = e.uninit(b * h)?;
2341 e.rms_norm(&x1, &l.ln_post, &mut x1n, h, b, c.eps)?;
2342 let mut mlp_dyn: Option<CudaSlice<f32>> = None;
2343 if let Some(d2) = &self.dflash2 {
2344 let (xc, dyn_) = self.d2_conv_prepare(e, &d2.mlp_conv[li], &x1n, b)?;
2345 x1n = xc;
2346 mlp_dyn = Some(dyn_);
2347 }
2348 let gate = self.mm(e, &l.w_gate, &x1n, b, h, c.n_ff)?;
2349 let up_ = self.mm(e, &l.w_up, &x1n, b, h, c.n_ff)?;
2350 let mut act = e.uninit(b * c.n_ff)?;
2351 e.silu_mul(&gate, &up_, &mut act, b * c.n_ff)?;
2352 let mut down = self.mm(e, &l.w_down, &act, b, c.n_ff, h)?;
2353 if let (Some(d2), Some(dyn_)) = (&self.dflash2, &mlp_dyn) {
2354 down = self.d2_conv_finish(e, &d2.mlp_conv[li], &down, dyn_, b)?;
2355 }
2356 let mut x2 = e.uninit(b * h)?;
2357 e.add(&down, &x1, &mut x2, b * h)?;
2358 x = x2;
2359 }
2360 let mut out = e.uninit(b * h)?;
2361 e.rms_norm(&x, &self.norm, &mut out, h, b, c.eps)?;
2362 Ok(out)
2363 }
2364}
2365
2366// ================= DFlash spec round (greedy, first light) =================
2367// Exact contract: identical output stream to plain greedy decode BY CONSTRUCTION — the
2368// target's batched verify argmax decides every committed token; the drafter only proposes.
2369// (Same verify+rewind pattern as generate_spec_gemma's eager round; t=16 verify rides the
2370// straddle-split-safe fa_decode_rows.)
2371impl crate::hybrid::HybridModel {
2372 pub fn generate_spec_dflash(
2373 &self,
2374 e: &Engine,
2375 draft: &DflashDraft,
2376 prompt: &[u32],
2377 max_new: usize,
2378 eos: &[u32],
2379 ) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2380 use crate::cache::{Cache, DflashTapSink};
2381 let n_embd = self.cfg.n_embd as usize;
2382 let c = &draft.cfg;
2383 assert!(
2384 draft.dflash2.is_none(),
2385 "DFlash2 drafters ride the qwen-hybrid dspark round (selector + windowed \
2386 attention); the gemma arm has no consumer for the family's ops"
2387 );
2388 assert_eq!(n_embd, c.hidden, "draft hidden must match target n_embd");
2389 let b = c.block_size;
2390 let n_taps = c.target_layer_ids.len();
2391 let max_ctx = prompt.len() + max_new + b + 8;
2392 // First light holds ctx <= sliding_window: the draft was trained with 4 sliding
2393 // layers (window 2048) and the first-light attention is windowless full — inside
2394 // the window the two are identical. The depth cell (1736 + 128) fits.
2395 assert!(
2396 max_ctx <= c.sliding_window,
2397 "first-light dflash round is windowless — ctx cap {} exceeds the draft window {}",
2398 max_ctx,
2399 c.sliding_window
2400 );
2401 let mut cache = Cache::new(e, &self.cfg, max_ctx)?;
2402
2403 // ---- prime with taps armed ----
2404 let tp = prompt.len();
2405 cache.dflash_taps = Some(DflashTapSink {
2406 layer_ids: c.target_layer_ids.clone(),
2407 buf: e.uninit(tp * n_taps * n_embd)?,
2408 hidden: n_embd,
2409 t: tp,
2410 base: 0,
2411 });
2412 let t_prime = std::time::Instant::now();
2413 let (logits, _h_seed, _hiddens) = self.prime_cache(e, prompt, &mut cache, 0)?;
2414 let mut last = crate::forward::argmax(&logits) as u32;
2415 // draft KV cache: ingest the prompt's ctx features once; per round only the kept
2416 // rows ingest + the block projects (round cost O(block), not O(ctx)).
2417 let mut dkv = DflashKv::new(e, &draft.cfg, max_ctx)?;
2418 {
2419 // CHUNKED ingest (depth OOM fix): the 1736-row prompt tap buffer is ~224MB f32;
2420 // running fc + 5-layer k/v projection over it in one shot stacks another
2421 // ~300MB of transients on the ~21.3GB trunk peak. 256-row windows bound the
2422 // transient set; identical values (row-independent ops).
2423 let taps = cache.dflash_taps.take().unwrap();
2424 let n_taps_h = n_taps * n_embd;
2425 let mut r0 = 0usize;
2426 while r0 < tp {
2427 let t_c = (tp - r0).min(256);
2428 let tv = e.view(&taps.buf, tp * n_taps_h);
2429 let win = tv.slice(r0 * n_taps_h..(r0 + t_c) * n_taps_h);
2430 let mut chunk = e.uninit(t_c * n_taps_h)?;
2431 e.copy_view_into(&mut chunk, 0, &win, t_c * n_taps_h)?;
2432 let f = draft.ctx_features(e, &chunk, t_c)?;
2433 let pos_c: Vec<i32> = ((r0 as i32)..(r0 + t_c) as i32).collect();
2434 draft.ingest_ctx(e, &mut dkv, &f, &pos_c, t_c)?;
2435 r0 += t_c;
2436 }
2437 }
2438 let mut ctx_len = tp;
2439 e.stream().synchronize()?;
2440 // published prime wall (the run-spec/gemma-gate timing contract subtracts it)
2441 crate::PRIME_NANOS.store(
2442 t_prime.elapsed().as_nanos() as u64,
2443 std::sync::atomic::Ordering::Relaxed,
2444 );
2445
2446 // embed-scale seam (MEMRA_DFLASH_EMB_SCALE): gemma trunks scale embeddings by
2447 // sqrt(n_embd) INSIDE the forward; whether the z-lab gemma4 training fed the
2448 // drafter scaled or raw embed rows is not visible from the reference (qwen path
2449 // uses raw embed_tokens). Acceptance arbitrates; default raw.
2450 let emb_scale = if std::env::var("MEMRA_DFLASH_EMB_SCALE").as_deref() == Ok("1") {
2451 (n_embd as f32).sqrt()
2452 } else {
2453 1.0
2454 };
2455
2456 let mut out = Vec::with_capacity(max_new);
2457 let n_vocab = self.output.out_features();
2458 // VERIFY WIDTH (MEMRA_DFLASH_VERIFY_T, default 8): the drafter always drafts a full
2459 // block (its trained mask pattern) but only the first vt rows go through the target
2460 // verify — the t=16 verify rides the untuned b16 tier at ~32% of the byte wall
2461 // (65ms/verify) while b8 rides the tuned r2 tier; with ~2.7 committed/round the
2462 // deep block positions almost never survive anyway. Exactness unaffected (verify
2463 // still decides every committed token).
2464 let vt_cap: usize = std::env::var("MEMRA_DFLASH_VERIFY_T")
2465 .ok()
2466 .and_then(|v| v.parse().ok())
2467 .unwrap_or(8)
2468 .clamp(2, b);
2469 // adaptive verify width (MEMRA_DFLASH_ADAPT!=0, MTP accepted+1 recipe): next round
2470 // verifies one past this round's accepted run, clamped [3, cap].
2471 let adapt = std::env::var("MEMRA_DFLASH_ADAPT").as_deref() != Ok("0");
2472 let mut vt = vt_cap;
2473 let mut attempted = 0usize;
2474 let mut accepted = 0usize;
2475 // The whole round runs in the decode-exact matmul scope: the m=16 draft mms were
2476 // otherwise falling into the prefill-GEMM class (770us/matmul, 17% of the depth
2477 // round). Prime (before this loop) keeps the prefill GEMM path.
2478 e.set_verify_exact(true);
2479 'outer: while out.len() < max_new {
2480 let start = cache.pos; // committed length
2481 // ---- draft: block = [last, MASK x b-1] ----
2482 let mut block: Vec<u32> = vec![c.mask_token_id; b];
2483 block[0] = last;
2484 let mut noise = e.htod(&self.embd.gather(n_embd, &block))?;
2485 if emb_scale != 1.0 {
2486 e.scale_inplace(&mut noise, emb_scale, b * n_embd)?;
2487 }
2488 if std::env::var("MEMRA_DFLASH_DEBUG").as_deref() == Ok("1") && start == cache.pos {
2489 let nv = e.dtoh(&noise)?;
2490 let r0: f32 = nv[..n_embd].iter().map(|x| x * x).sum::<f32>().sqrt();
2491 let r1: f32 = nv[n_embd..2 * n_embd]
2492 .iter()
2493 .map(|x| x * x)
2494 .sum::<f32>()
2495 .sqrt();
2496 eprintln!(
2497 "[dflash noise] |row0(last)|={r0:.3} |row1(MASK id {})|={r1:.3}",
2498 c.mask_token_id
2499 );
2500 }
2501 let pos_block: Vec<i32> = ((start as i32)..(start + b) as i32).collect();
2502 let dh = draft.forward_round(e, &mut dkv, &noise, &pos_block)?;
2503 // draft tokens = argmax(lm_head(h rows 1..b))
2504 let mut rows = e.uninit((b - 1) * n_embd)?;
2505 {
2506 let dv = e.view(&dh, b * n_embd);
2507 let tail = dv.slice(n_embd..b * n_embd);
2508 e.copy_view_into(&mut rows, 0, &tail, (b - 1) * n_embd)?;
2509 }
2510 let mut dl = e.matmul(&self.output, &rows, b - 1)?;
2511 // SEMI-AR MARKOV CHAIN (DSpark head, when present + MEMRA_DFLASH_MARKOV!=0):
2512 // left-to-right, logits_k += W2(W1[prev realized token]) — the whole chain
2513 // stays on-device (chain_d[0] = the pending token; argmax k writes
2514 // chain_d[k+1], the k+1 bias gathers from it). Greedy mirror of the patch's
2515 // _markov_semiar_sample_block.
2516 let markov_on = std::env::var("MEMRA_DFLASH_MARKOV").as_deref() != Ok("0");
2517 let mut chain_d = e.stream().alloc_zeros::<u32>(b)?;
2518 if let (Some(mk), true) = (&draft.markov, markov_on) {
2519 e.set_u32_one(&mut chain_d, last)?;
2520 for k in 0..(b - 1) {
2521 let mut f = e.uninit(mk.rank)?;
2522 e.gather_row_bf16(&mk.w1_bf16, &chain_d, k, &mut f, mk.rank)?;
2523 let bias = e.matmul(&mk.w2, &f, 1)?;
2524 e.add_row_inplace(&mut dl, &bias, n_vocab, k * n_vocab)?;
2525 e.argmax_token_device_col(&dl, k, n_vocab, &mut chain_d, k + 1)?;
2526 }
2527 } else {
2528 for i in 0..(b - 1) {
2529 e.argmax_token_device_col(&dl, i, n_vocab, &mut chain_d, i + 1)?;
2530 }
2531 }
2532 let chain = e.dtoh_u32(&chain_d)?;
2533 let dtoks = &chain[1..];
2534 for (i, &dt) in dtoks.iter().enumerate() {
2535 block[i + 1] = dt;
2536 }
2537 let dbg = std::env::var("MEMRA_DFLASH_DEBUG").as_deref() == Ok("1");
2538
2539 // ---- verify: one t=vt target forward with taps armed ----
2540 let vblock = &block[..vt];
2541 cache.dflash_taps = Some(DflashTapSink {
2542 layer_ids: c.target_layer_ids.clone(),
2543 buf: e.uninit(vt * n_taps * n_embd)?,
2544 hidden: n_embd,
2545 t: vt,
2546 base: 0,
2547 });
2548 let (vam, _vh) = self.gemma4_decode_step_t_am(e, vblock, start, &mut cache)?;
2549 let taps = cache.dflash_taps.take().unwrap();
2550 if dbg {
2551 eprintln!(
2552 "[dflash r] start={start} last={last}\n draft={:?}\n vam ={:?}",
2553 &block[1..],
2554 &vam
2555 );
2556 }
2557
2558 // ---- accept ----
2559 let mut m = 0usize;
2560 while m < vt - 1 && block[m + 1] as usize == vam[m] as usize {
2561 m += 1;
2562 }
2563 attempted += vt - 1;
2564 accepted += m;
2565 out.push(last);
2566 if eos.contains(&last) {
2567 break 'outer;
2568 }
2569 for &dt in &block[1..=m] {
2570 out.push(dt);
2571 if eos.contains(&dt) {
2572 break 'outer;
2573 }
2574 if out.len() >= max_new {
2575 break 'outer;
2576 }
2577 }
2578 let next = vam[m] as u32;
2579
2580 // ---- commit/rollback: keep m+1 of the b appended rows ----
2581 let keep = m + 1;
2582 for kvl in cache.kv.iter_mut().flatten() {
2583 kvl.len -= vt - keep;
2584 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
2585 }
2586 cache.pos -= vt - keep;
2587
2588 // ---- ingest the kept rows' ctx features into the draft KV ----
2589 {
2590 let tv = e.view(&taps.buf, vt * n_taps * n_embd);
2591 let keep_view = tv.slice(0..keep * n_taps * n_embd);
2592 let mut kept = e.uninit(keep * n_taps * n_embd)?;
2593 e.copy_view_into(&mut kept, 0, &keep_view, keep * n_taps * n_embd)?;
2594 let f = draft.ctx_features(e, &kept, keep)?;
2595 let pos_k: Vec<i32> = ((ctx_len as i32)..(ctx_len + keep) as i32).collect();
2596 draft.ingest_ctx(e, &mut dkv, &f, &pos_k, keep)?;
2597 ctx_len += keep;
2598 }
2599 last = next;
2600 if adapt {
2601 vt = (m + 2).clamp(3, vt_cap);
2602 }
2603 }
2604 e.set_verify_exact(false);
2605 if std::env::var("MEMRA_SPEC_STATS").as_deref() == Ok("1") {
2606 eprintln!(
2607 "[dflash] acceptance {accepted}/{attempted} = {:.3}",
2608 accepted as f64 / attempted.max(1) as f64
2609 );
2610 }
2611 Ok(out)
2612 }
2613}
2614
2615// ================= Engine-bundle slice 1: batched GDN state snapshot ====================
2616// DSF-ROUNDCOST-20260820 §1.1 measured the dspark round's `cache.snapshot(e)` at 0.67 ms
2617// native wall — 48 linear layers x {conv, ssm} x (alloc_zeros + memcpy_dtod) of pure
2618// dispatch serialization, zero kernels. This batcher holds ONE persistent CacheSnapshot
2619// (buffers allocated on round 1, reused every round — kills the per-round alloc/memset
2620// churn) plus device pointer tables, so a round's snap is one small H2D table refresh
2621// (the ssm handles ping-pong per verify row, so live pointers are re-read each round;
2622// conv handles are rolled in place and never move) + TWO `copy_batch_uniform_f32`
2623// launches. Bytes, buffers and stream order are identical to `Cache::snapshot`; only the
2624// dispatch count changes, so acceptance and streams stay bit-identical (E2E-gated).
2625// `MEMRA_STATE_COPY_BATCH=0` reverts to the legacy per-layer snapshot.
2626
2627pub(crate) struct DsparkSnapBatch {
2628 pub(crate) snap: crate::cache::CacheSnapshot,
2629 /// Linear-attention layer indices, in `conv_table`/`ssm_table` order.
2630 lin: Vec<usize>,
2631 /// [src_0..src_{n-1}, dst_0..dst_{n-1}] — live conv states -> snapshot conv buffers.
2632 conv_table: CudaSlice<u64>,
2633 ssm_table: CudaSlice<u64>,
2634 host_ssm: Vec<u64>,
2635 conv_words: usize,
2636 ssm_words: usize,
2637}
2638
2639impl DsparkSnapBatch {
2640 /// Build from a fresh full snapshot (this IS round 1's snap — the caller uses
2641 /// `self.snap` directly after `new`). Returns None when the cache has no linear
2642 /// layers or their state sizes are non-uniform (a future hybrid shape) — the caller
2643 /// then stays on the legacy per-layer snapshot rather than copying wrong byte counts.
2644 pub(crate) fn new(
2645 e: &Engine,
2646 cache: &crate::cache::Cache,
2647 ) -> Result<Option<Self>, Box<dyn std::error::Error>> {
2648 use cudarc::driver::DevicePtr;
2649 let snap = cache.snapshot(e)?;
2650 let lin: Vec<usize> = (0..cache.recur.len())
2651 .filter(|&il| cache.recur[il].is_some())
2652 .collect();
2653 if lin.is_empty() {
2654 return Ok(None);
2655 }
2656 let first = cache.recur[lin[0]].as_ref().unwrap();
2657 let (conv_words, ssm_words) = (first.conv_state.len(), first.ssm_state.len());
2658 for &il in &lin {
2659 let rl = cache.recur[il].as_ref().unwrap();
2660 if rl.conv_state.len() != conv_words || rl.ssm_state.len() != ssm_words {
2661 return Ok(None);
2662 }
2663 }
2664 let n = lin.len();
2665 let mut host_conv = vec![0u64; 2 * n];
2666 let mut host_ssm = vec![0u64; 2 * n];
2667 {
2668 let s = &e.gpu.stream();
2669 for (k, &il) in lin.iter().enumerate() {
2670 let rl = cache.recur[il].as_ref().unwrap();
2671 let (pc, _g0) = rl.conv_state.device_ptr(s);
2672 let (ps, _g1) = rl.ssm_state.device_ptr(s);
2673 let (dc, _g2) = snap.conv[il].as_ref().unwrap().device_ptr(s);
2674 let (ds, _g3) = snap.ssm[il].as_ref().unwrap().device_ptr(s);
2675 host_conv[k] = pc as u64;
2676 host_conv[n + k] = dc as u64;
2677 host_ssm[k] = ps as u64;
2678 host_ssm[n + k] = ds as u64;
2679 }
2680 }
2681 let conv_table = e.htod_u64(&host_conv)?;
2682 let ssm_table = e.htod_u64(&host_ssm)?;
2683 Ok(Some(Self {
2684 snap,
2685 lin,
2686 conv_table,
2687 ssm_table,
2688 host_ssm,
2689 conv_words,
2690 ssm_words,
2691 }))
2692 }
2693
2694 /// The per-round snap: refresh kv lens/pos host-side (as `snapshot_into` does),
2695 /// re-read the live ssm handles into the table (gdn ping-pong moves them; the conv
2696 /// handles and every snapshot dst are stable), then two batched-copy launches.
2697 pub(crate) fn refresh(
2698 &mut self,
2699 e: &Engine,
2700 cache: &crate::cache::Cache,
2701 ) -> Result<(), Box<dyn std::error::Error>> {
2702 use cudarc::driver::DevicePtr;
2703 for il in 0..cache.kv.len() {
2704 self.snap.kv_len[il] = cache.kv[il].as_ref().map(|kvl| kvl.len);
2705 }
2706 self.snap.pos = cache.pos;
2707 let n = self.lin.len();
2708 {
2709 let s = &e.gpu.stream();
2710 for (k, &il) in self.lin.iter().enumerate() {
2711 let rl = cache.recur[il].as_ref().unwrap();
2712 let (ps, _g) = rl.ssm_state.device_ptr(s);
2713 self.host_ssm[k] = ps as u64;
2714 }
2715 }
2716 e.htod_u64_into(&self.host_ssm, &mut self.ssm_table)?;
2717 e.copy_batch_uniform_f32(&self.conv_table, n, self.conv_words)?;
2718 e.copy_batch_uniform_f32(&self.ssm_table, n, self.ssm_words)?;
2719 Ok(())
2720 }
2721}
2722
2723// ================= DSpark spec round, QWEN-HYBRID target (lane/dspark-q38-recover) =====
2724// The q38 twin of generate_spec_dflash. Same drafter machinery (rounds, markov chain,
2725// draft KV, adaptive verify width); the TARGET side swaps gemma4's dense verify for the
2726// qwen serving-class verify funnel (dspark_verify_t_am) + snapshot/rollback, because the
2727// hybrid GDN conv/ssm state mutates in place — dense KV truncation cannot roll it back.
2728// Exactness contract unchanged: identical stream to plain greedy BY CONSTRUCTION (the
2729// target's verify argmax decides every committed token).
2730impl crate::hybrid::HybridModel {
2731 pub fn generate_spec_dspark(
2732 &self,
2733 e: &Engine,
2734 draft: &DflashDraft,
2735 prompt: &[u32],
2736 max_new: usize,
2737 eos: &[u32],
2738 sampling: Option<&crate::spec::SpecSampling>,
2739 ) -> Result<Vec<u32>, Box<dyn std::error::Error>> {
2740 use crate::cache::{Cache, DflashTapSink};
2741 assert!(
2742 self.cfg.gemma4.is_none(),
2743 "gemma4 targets use generate_spec_dflash; this is the qwen-hybrid arm"
2744 );
2745 // SAMPLED ADMISSION (T>0, lane/dspark-sampled-admission-20260820): Some+temp>0
2746 // routes the round's proposal/accept through the rejection-sampling arms; None or
2747 // temp==0 keeps every greedy path byte-identical (the exactness instrument).
2748 let sp_on: Option<&crate::spec::SpecSampling> = sampling.filter(|s| s.temp > 0.0);
2749 // PENALTIES AT T==0 ARE A LOUD REFUSAL (lane/dspark-penalized-sampled-20260821):
2750 // the greedy walk argmaxes RAW verify columns, so a temp==0 config carrying
2751 // non-identity penalties would silently serve the UNPENALIZED greedy stream —
2752 // exactly the H-class silent-program-switch this route refuses everywhere else.
2753 // Penalized greedy stays on the plain path (worker admission owns the exclusion).
2754 if let Some(s) = sampling {
2755 if s.temp <= 0.0 && s.pen_on() {
2756 return Err(
2757 "dspark spec at temp==0 is the greedy route and would silently drop \
2758 the request's penalties; penalized greedy is served on the plain path"
2759 .into(),
2760 );
2761 }
2762 }
2763 // Penalized-sampled state: the session window (pen_window_seed — one definition
2764 // across both spec routes), extended with every committed token; each round's
2765 // accept receives the trimmed tail (min(penalty_last_n, PEN_WINDOW_MAX)).
2766 let pen_on = sp_on.is_some_and(|s| s.pen_on());
2767 let mut pen_hist: Vec<u32> = if pen_on {
2768 crate::spec::pen_window_seed(&[], prompt, sp_on.unwrap().penalty_last_n)
2769 } else {
2770 Vec::new()
2771 };
2772 let (mut sctr, mut uctr) = (0u32, 0u32);
2773 let n_embd = self.cfg.n_embd as usize;
2774 let c = &draft.cfg;
2775 assert_eq!(n_embd, c.hidden, "draft hidden must match target n_embd");
2776 let b = c.block_size;
2777 let n_taps = c.target_layer_ids.len();
2778 let max_ctx = prompt.len() + max_new + b + 8;
2779 // DFlash2 implements the reference's non-causal symmetric sliding window in
2780 // the round attention (sdpa_naive_w), so depth past the window is admitted;
2781 // other families keep the historical windowless contract.
2782 assert!(
2783 draft.dflash2.is_some() || max_ctx <= c.sliding_window,
2784 "dspark round is windowless — ctx cap {} exceeds the draft window {}",
2785 max_ctx,
2786 c.sliding_window
2787 );
2788 let mut cache = Cache::new(e, &self.cfg, max_ctx)?;
2789
2790 // ---- prime with taps armed (chunked prime writes at chunk offsets via sink.base) ----
2791 let tp = prompt.len();
2792 cache.dflash_taps = Some(DflashTapSink {
2793 layer_ids: c.target_layer_ids.clone(),
2794 buf: e.uninit(tp * n_taps * n_embd)?,
2795 hidden: n_embd,
2796 t: tp,
2797 base: 0,
2798 });
2799 let t_prime = std::time::Instant::now();
2800 let (logits, _h_seed, _hiddens) = self.prime_cache(e, prompt, &mut cache, 0)?;
2801 // Boundary token: greedy takes the argmax (byte contract); sampled draws it from
2802 // the request's own filtered target through the session Philox stream — the same
2803 // shipped composition the frspec route uses (sample_check arm 9 oracles it).
2804 let mut last = match sp_on {
2805 Some(sp) => crate::spec::sample_boundary_token(
2806 e,
2807 &logits,
2808 sp,
2809 &pen_hist,
2810 &mut sctr,
2811 "dspark-prime",
2812 )?,
2813 None => crate::forward::argmax(&logits) as u32,
2814 };
2815 let mut dkv = DflashKv::new(e, &draft.cfg, max_ctx)?;
2816 {
2817 let taps = cache.dflash_taps.take().unwrap();
2818 let n_taps_h = n_taps * n_embd;
2819 let mut r0 = 0usize;
2820 while r0 < tp {
2821 let t_c = (tp - r0).min(256);
2822 let tv = e.view(&taps.buf, tp * n_taps_h);
2823 let win = tv.slice(r0 * n_taps_h..(r0 + t_c) * n_taps_h);
2824 let mut chunk = e.uninit(t_c * n_taps_h)?;
2825 e.copy_view_into(&mut chunk, 0, &win, t_c * n_taps_h)?;
2826 let f = draft.ctx_features(e, &chunk, t_c)?;
2827 let pos_c: Vec<i32> = ((r0 as i32)..(r0 + t_c) as i32).collect();
2828 draft.ingest_ctx(e, &mut dkv, &f, &pos_c, t_c)?;
2829 r0 += t_c;
2830 }
2831 }
2832 let mut ctx_len = tp;
2833 e.stream().synchronize()?;
2834 crate::PRIME_NANOS.store(
2835 t_prime.elapsed().as_nanos() as u64,
2836 std::sync::atomic::Ordering::Relaxed,
2837 );
2838
2839 let mut out = Vec::with_capacity(max_new);
2840 let n_vocab = self.output.out_features();
2841 // Harvest convention (DSPARK-POSTMORTEM-20260820.md): which drafter output rows
2842 // become draft candidates. nd = drafts/round; verify carries [anchor, drafts]
2843 // = up to nd+1 rows. FAMILY-keyed for DFlash2 (mask-fill by construction),
2844 // else default = the CHECKPOINT's own strategy census (owner-ratified flip,
2845 // 2026-08-20); explicit env still wins (contradiction refuses).
2846 let harvest = DsparkHarvest::for_draft(draft);
2847 let nd = harvest.n_drafts(b);
2848 let r0 = harvest.first_row();
2849 let vt_cap: usize = std::env::var("MEMRA_DFLASH_VERIFY_T")
2850 .ok()
2851 .and_then(|v| v.parse().ok())
2852 .unwrap_or(nd + 1)
2853 .clamp(2, nd + 1);
2854 let adapt = std::env::var("MEMRA_DFLASH_ADAPT").as_deref() != Ok("0");
2855 // Verify-window policy (H4, DSPARK-POSTMORTEM-20260820.md): default =
2856 // confidence-slot tau=.5 when the checkpoint carries an accept-rate head
2857 // (owner-ratified flip 2026-08-20; cell-3 tau ladder knee) — each round's
2858 // window is sized from the head's own slot scores, post-draft pre-verify.
2859 // Head-less checkpoints and MEMRA_DFLASH_ADAPT=0 keep the reactive ladder.
2860 let vt_policy = DsparkVtPolicy::resolve(draft.confidence.is_some());
2861 if vt_policy.is_confidence() {
2862 assert!(
2863 draft.confidence.is_some(),
2864 "MEMRA_DSPARK_VT={vt_policy:?} needs a checkpoint with an accept-rate \
2865 head (confidence_head.* absent in this export)"
2866 );
2867 }
2868 let mut vt = vt_cap;
2869 let mut attempted = 0usize;
2870 let mut accepted = 0usize;
2871 // Engine-bundle slice 1: persistent batched snapshot (None until round 1; stays
2872 // None — legacy per-layer snapshot — under MEMRA_STATE_COPY_BATCH=0 or when the
2873 // batcher declines the cache shape).
2874 let mut snapb: Option<DsparkSnapBatch> = None;
2875 let mut snapb_off = !crate::spec::state_copy_batch_on();
2876 // Engine-bundle slice 2: deferred chain readback needs the resident embed table
2877 // (verify then embeds chain_d directly). Ladder/stash arms only — the confidence
2878 // policies size vt from a pre-verify head readback and keep the legacy order.
2879 let defer_rb = crate::spec::dspark_defer_readback_on() && !vt_policy.is_confidence();
2880 let (embd_qt, embd_rb) = self.embd.qt_and_row_bytes(n_embd);
2881 let embd_gpu = if !defer_rb || crate::spec::spec_host_embd() {
2882 None
2883 } else {
2884 Some(
2885 self.embd_gpu
2886 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload")),
2887 )
2888 };
2889 // Engine-bundle slice 3: per-(segment, vt) verify graphs for the linear-layer runs
2890 // (rides the slice-2 deferred path only — device tokens keep the whole verify off
2891 // the host). PERSISTENT across generations on the model (rebuilding per call
2892 // re-captured ~80 graphs per prompt — measured 97.8 -> 79.1 tok/s e2e); the
2893 // captured bodies are cache-independent: all state reads go through per-round
2894 // refreshed pointer tables and ctx-owned slabs. None = eager walk, byte-identical.
2895 let mut vg_guard = self.dspark_vgraphs.lock().unwrap();
2896 if vg_guard.is_none() && embd_gpu.is_some() && crate::spec::dspark_verify_graph_on() {
2897 *vg_guard = crate::spec::DsparkVerifyGraphs::new(e, &cache, vt_cap, n_embd)?;
2898 }
2899 let vgraphs: &mut Option<crate::spec::DsparkVerifyGraphs> = &mut vg_guard;
2900 // per-phase economics counters (ns) — the verify-toll dataset
2901 let (mut ns_draft, mut ns_snap, mut ns_verify, mut ns_roll, mut ns_ingest) =
2902 (0u64, 0u64, 0u64, 0u64, 0u64);
2903 let mut rounds = 0usize;
2904 let stats = std::env::var("MEMRA_SPEC_STATS").as_deref() == Ok("1");
2905 let clock = |on: bool, e: &Engine| -> std::time::Instant {
2906 if on {
2907 let _ = e.stream().synchronize();
2908 }
2909 std::time::Instant::now()
2910 };
2911 'outer: while out.len() < max_new {
2912 rounds += 1;
2913 let start = cache.pos; // committed length
2914 // ---- draft: block = [last, MASK x b-1] (decode-exact class for the m=b mms) ----
2915 let t0 = clock(stats, e);
2916 e.set_verify_exact(true);
2917 let mut block: Vec<u32> = vec![c.mask_token_id; b];
2918 block[0] = last;
2919 let noise = e.htod(&self.embd.gather(n_embd, &block))?;
2920 let pos_block: Vec<i32> = ((start as i32)..(start + b) as i32).collect();
2921 let dh = draft.forward_round(e, &mut dkv, &noise, &pos_block)?;
2922 // Harvest: logits over rows r0..r0+nd (Dflash: mask rows 1..b-1, fill
2923 // semantics; Dspark: ALL b rows, shifted semantics — row k predicts
2924 // anchor+k+1, so col k of `dl` is the draft for position start+k+1).
2925 let mut rows = e.uninit(nd * n_embd)?;
2926 {
2927 let dv = e.view(&dh, b * n_embd);
2928 let src = dv.slice(r0 * n_embd..(r0 + nd) * n_embd);
2929 e.copy_view_into(&mut rows, 0, &src, nd * n_embd)?;
2930 }
2931 let mut dl = e.matmul(&self.output, &rows, nd)?;
2932 // Family/sampling-keyed proposal (v0.100 train merge of the port and H4/
2933 // engine-bundle stacks — BOTH programs preserved):
2934 // - SAMPLED (sp_on): rejection-sampling proposal, records the true per-slot
2935 // q (family-keyed inside: selector for DFlash2, markov-corrected rows
2936 // otherwise). Host CDF/readback syncs inside — slice-2 deferral N/A.
2937 // - DFlash2 greedy: the candidate path selector REPLACES the markov chain
2938 // (reference DFlash2DraftModel.propose — greedy arm).
2939 // - markov/plain greedy chain: the engine-bundle arm; slice-2 readback
2940 // deferral decided below (needs the ckpt arm reads).
2941 // Confidence policy: stash each slot's markov prev-token embedding (the
2942 // exact `w1` row the chain gathers) into a [nd, rank] buffer — d2d async,
2943 // read back beside `rows` in one host sync after the chain.
2944 let want_conf_emb = vt_policy.is_confidence()
2945 && draft.confidence.as_ref().is_some_and(|ch| ch.with_markov);
2946 let mut conf_emb: Option<CudaSlice<f32>> = match (&draft.markov, want_conf_emb) {
2947 (Some(mk), true) => Some(e.uninit(nd * mk.rank)?),
2948 (None, true) => unreachable!(
2949 "with_markov confidence head without a markov table — the loader forbids it"
2950 ),
2951 _ => None,
2952 };
2953 let mut cand: Vec<u32> = Vec::with_capacity(nd + 1);
2954 let mut prop: Option<DsparkDraftSample> = None;
2955 let mut chain_dev: Option<CudaSlice<u32>> = None;
2956 if let Some(sp) = sp_on {
2957 let (tail, ds) = draft.dspark_propose_sampled(
2958 e,
2959 &mut dl,
2960 &rows,
2961 nd,
2962 n_vocab,
2963 last,
2964 sp,
2965 &mut sctr,
2966 &mut uctr,
2967 conf_emb.as_mut(),
2968 )?;
2969 e.set_verify_exact(false);
2970 cand.push(last);
2971 cand.extend_from_slice(&tail);
2972 prop = Some(ds);
2973 } else if draft.dflash2.is_some() {
2974 let path = draft.dflash2_propose_greedy(e, &dl, &rows, nd, n_vocab, last)?;
2975 e.set_verify_exact(false);
2976 cand.push(last);
2977 cand.extend_from_slice(&path);
2978 } else {
2979 let markov_on = std::env::var("MEMRA_DFLASH_MARKOV").as_deref() != Ok("0");
2980 let mut chain_d = e.stream().alloc_zeros::<u32>(nd + 1)?;
2981 if let (Some(mk), true) = (&draft.markov, markov_on) {
2982 e.set_u32_one(&mut chain_d, last)?;
2983 for k in 0..nd {
2984 let mut f = e.uninit(mk.rank)?;
2985 e.gather_row_bf16(&mk.w1_bf16, &chain_d, k, &mut f, mk.rank)?;
2986 if let Some(ce) = conf_emb.as_mut() {
2987 let fv = e.view(&f, mk.rank);
2988 e.copy_view_into(ce, k * mk.rank, &fv, mk.rank)?;
2989 }
2990 let bias = e.matmul(&mk.w2, &f, 1)?;
2991 e.add_row_inplace(&mut dl, &bias, n_vocab, k * n_vocab)?;
2992 e.argmax_token_device_col(&dl, k, n_vocab, &mut chain_d, k + 1)?;
2993 }
2994 } else {
2995 if want_conf_emb {
2996 // chain_d[0] must carry the anchor — slot 0's prev token.
2997 e.set_u32_one(&mut chain_d, last)?;
2998 }
2999 for i in 0..nd {
3000 if let (Some(ce), Some(mk)) = (conf_emb.as_mut(), &draft.markov) {
3001 let mut f = e.uninit(mk.rank)?;
3002 e.gather_row_bf16(&mk.w1_bf16, &chain_d, i, &mut f, mk.rank)?;
3003 let fv = e.view(&f, mk.rank);
3004 e.copy_view_into(ce, i * mk.rank, &fv, mk.rank)?;
3005 }
3006 e.argmax_token_device_col(&dl, i, n_vocab, &mut chain_d, i + 1)?;
3007 }
3008 }
3009 e.set_verify_exact(false);
3010 chain_dev = Some(chain_d);
3011 }
3012 // MEMRA_DSPARK_CKPT (default 1): verify with the MTP column-stash armed so a
3013 // partial accept restores state directly. =0 keeps the snapshot+replay arm
3014 // (the oracle the stash arm is gated against — MEMRA_DSPARK_CKPT_GATE=1 runs
3015 // BOTH per partial round and byte-compares the resulting cache state).
3016 // Read here (was at the verify site) — slice 2's deferral needs the arm
3017 // choice before deciding whether the chain readback can move past verify.
3018 let ckpt_on = std::env::var("MEMRA_DSPARK_CKPT").as_deref() != Ok("0");
3019 let ckpt_gate = std::env::var("MEMRA_DSPARK_CKPT_GATE").as_deref() == Ok("1");
3020 // SAMPLED x ckpt-gate refusal: the gate compares verify argmaxes across a
3021 // replay — a greedy-exactness instrument (port lane). Refuse loudly.
3022 if sp_on.is_some() && ckpt_gate {
3023 return Err(
3024 "MEMRA_DSPARK_CKPT_GATE compares verify argmaxes across a replay \
3025 — a greedy-exactness instrument; unset it for T>0 dspark rounds"
3026 .into(),
3027 );
3028 }
3029 // Slice 2: under the stash/gate arms with a resident embed table, the GREEDY
3030 // chain readback is DEFERRED past verify dispatch and merged with the argmax
3031 // readback into one sync. The replay arm (CKPT=0) verifies host tokens and
3032 // keeps the legacy order; the sampled and DFlash2 proposals already synced
3033 // at the walk (chain_dev is None there).
3034 let deferred = chain_dev.is_some() && embd_gpu.is_some() && (ckpt_on || ckpt_gate);
3035 // ---- H4 confidence window: size THIS round's verify from the head ----
3036 if vt_policy.is_confidence() {
3037 let ch = draft.confidence.as_ref().expect("asserted at loop entry");
3038 let (rows_h, emb_h) = match conf_emb.as_ref() {
3039 Some(ce) => {
3040 let (a, b2) = e.dtoh_pair(&rows, ce)?;
3041 (a, Some(b2))
3042 }
3043 None => (e.dtoh(&rows)?, None),
3044 };
3045 let rank = draft.markov.as_ref().map(|m| m.rank).unwrap_or(0);
3046 let mut raws = Vec::with_capacity(nd);
3047 for k in 0..nd {
3048 let hrow = &rows_h[k * n_embd..(k + 1) * n_embd];
3049 let emb = emb_h.as_ref().map(|eh| &eh[k * rank..(k + 1) * rank]);
3050 raws.push(ch.raw_score(hrow, emb));
3051 }
3052 vt = vt_policy
3053 .size_window(&raws, vt_cap)
3054 .expect("confidence policies always size the window");
3055 }
3056 // Verify candidates: [anchor, draft 1..nd]. Under Dflash this is the
3057 // historical `block` content; under Dspark it is one longer than the
3058 // drafter's input block (nd = b drafts + the anchor). The sampled/DFlash2
3059 // proposals built `cand` at the walk; deferred greedy rounds build it after
3060 // the merged readback — the bytes are identical (chain_d is written before
3061 // either sync).
3062 if let Some(chain_d) = chain_dev.as_ref() {
3063 if !deferred {
3064 let chain = e.dtoh_u32(chain_d)?;
3065 cand.push(last);
3066 cand.extend_from_slice(&chain[1..]);
3067 }
3068 }
3069 ns_draft += clock(stats, e).duration_since(t0).as_nanos() as u64;
3070
3071 // ---- snapshot (GDN conv/ssm state + KV lens), then verify t=vt ----
3072 let t1 = std::time::Instant::now();
3073 // Slice 1: batched snap (one table refresh + two copy launches) with the
3074 // legacy per-layer snapshot as the kill-switch / non-uniform fallback.
3075 let mut snap_legacy: Option<crate::cache::CacheSnapshot> = None;
3076 if !snapb_off && snapb.is_none() {
3077 snapb = DsparkSnapBatch::new(e, &cache)?;
3078 snapb_off = snapb.is_none();
3079 } else if let Some(sb) = snapb.as_mut() {
3080 sb.refresh(e, &cache)?;
3081 }
3082 let snap: &crate::cache::CacheSnapshot = match snapb.as_ref() {
3083 Some(sb) => &sb.snap,
3084 None => {
3085 snap_legacy = Some(cache.snapshot(e)?);
3086 snap_legacy.as_ref().unwrap()
3087 }
3088 };
3089 let _ = &snap_legacy;
3090 ns_snap += clock(stats, e).duration_since(t1).as_nanos() as u64;
3091 let t2 = std::time::Instant::now();
3092 // Slice 3: the tap-sink buffer is persistent per vt in the graphs ctx
3093 // (captured segments bake its address); fully rewritten by every verify.
3094 let tap_buf = match vgraphs.as_mut().and_then(|g| g.tap_bufs.remove(&vt)) {
3095 Some(buf) => buf,
3096 None => e.uninit(vt * n_taps * n_embd)?,
3097 };
3098 cache.dflash_taps = Some(DflashTapSink {
3099 layer_ids: c.target_layer_ids.clone(),
3100 buf: tap_buf,
3101 hidden: n_embd,
3102 t: vt,
3103 base: 0,
3104 });
3105 // The whole fallible verify window runs inside a closure so the Err path can
3106 // return the sink buffer to the ctx pool before propagating (v0.98 review
3107 // carry-over): five `?`s span the window, and an early return would drop
3108 // `cache.dflash_taps` — freeing the buffer whose ADDRESS the model-persistent
3109 // captured graphs bake, so the next generation's replayed tap copies would
3110 // write freed memory. The never-orphan invariant below now holds on EVERY
3111 // exit, not just the EOS/budget break.
3112 let verify_res = (|cache: &mut crate::cache::Cache,
3113 cand: &mut Vec<u32>,
3114 vgraphs: &mut Option<crate::spec::DsparkVerifyGraphs>|
3115 -> Result<
3116 (
3117 Vec<u32>,
3118 Option<CudaSlice<f32>>,
3119 Option<crate::spec::DsparkVerifyCkpt>,
3120 ),
3121 Box<dyn std::error::Error>,
3122 > {
3123 if sp_on.is_some() {
3124 // SAMPLED: keep the raw verify logits — the accept walk gathers
3125 // filtered p from them (argmaxes are the greedy arm's instrument,
3126 // not this one's).
3127 if ckpt_on {
3128 let (tl, vck) =
3129 self.dspark_verify_t_logits_ckpt(e, &cand[..vt], start, cache)?;
3130 Ok((Vec::new(), Some(tl), Some(vck)))
3131 } else {
3132 Ok((
3133 Vec::new(),
3134 Some(self.dspark_verify_t_logits(e, &cand[..vt], start, cache)?),
3135 None,
3136 ))
3137 }
3138 } else if deferred {
3139 // Slice 2: verify embeds the DEVICE chain (cand layout by construction:
3140 // chain_d[0] = anchor, chain_d[1..] = drafts), then ONE host sync reads
3141 // chain + verify argmaxes together — the host dispatched snap + all of
3142 // verify while the draft was still executing.
3143 let chain_d = chain_dev.as_ref().expect("deferred implies greedy chain");
3144 let g = embd_gpu.expect("deferred implies resident embed");
3145 let (am_d, vck) = self.dspark_verify_t_am_ckpt_dev(
3146 e,
3147 chain_d,
3148 vt,
3149 start,
3150 cache,
3151 (g, embd_qt, embd_rb),
3152 vgraphs.as_mut(),
3153 )?;
3154 let ch = e.stream().clone_dtoh(chain_d)?;
3155 let am = e.stream().clone_dtoh(&am_d)?;
3156 e.stream().synchronize()?;
3157 cand.push(last);
3158 cand.extend_from_slice(&ch[1..]);
3159 Ok((am, None, Some(vck)))
3160 } else if ckpt_on || ckpt_gate {
3161 let (vam, vck) = self.dspark_verify_t_am_ckpt(e, &cand[..vt], start, cache)?;
3162 Ok((vam, None, Some(vck)))
3163 } else {
3164 Ok((
3165 self.dspark_verify_t_am(e, &cand[..vt], start, cache)?,
3166 None,
3167 None,
3168 ))
3169 }
3170 })(&mut cache, &mut cand, vgraphs);
3171 let (vam, tl, vck) = match verify_res {
3172 Ok(v) => v,
3173 Err(err) => {
3174 if let (Some(g), Some(taps)) = (vgraphs.as_mut(), cache.dflash_taps.take()) {
3175 g.tap_bufs.insert(vt, taps.buf);
3176 }
3177 return Err(err);
3178 }
3179 };
3180 let taps = cache.dflash_taps.take().unwrap();
3181 // Return the tap buffer to the ctx pool IMMEDIATELY — an EOS/budget break
3182 // between accept and ingest must never orphan an address the captured
3183 // graphs bake (the next generation would alloc a fresh buffer and the
3184 // replayed tap copies would write freed memory). Ingest reads it borrowed.
3185 let tap_local: Option<CudaSlice<f32>> = match vgraphs.as_mut() {
3186 Some(g) => {
3187 g.tap_bufs.insert(vt, taps.buf);
3188 None
3189 }
3190 None => Some(taps.buf),
3191 };
3192 let tap_ref: &CudaSlice<f32> = match &tap_local {
3193 Some(b) => b,
3194 None => &vgraphs.as_ref().expect("ctx present above").tap_bufs[&vt],
3195 };
3196 ns_verify += clock(stats, e).duration_since(t2).as_nanos() as u64;
3197
3198 // ---- accept ----
3199 // Penalized-sampled: the anchor `last` is committed THIS round unconditionally
3200 // (the out.push below), so it joins the window before the accept walk — verify
3201 // row 0's state includes it. Accepted drafts extend the window after the walk;
3202 // `next` joins as the anchor of ITS round.
3203 if pen_on {
3204 pen_hist.push(last);
3205 }
3206 let (m, next) = match (sp_on, tl.as_ref()) {
3207 (Some(sp), Some(tl)) => {
3208 let w0 = pen_hist
3209 .len()
3210 .saturating_sub(sp.penalty_last_n.min(crate::spec::PEN_WINDOW_MAX));
3211 dspark_accept_sampled(
3212 e,
3213 tl,
3214 &cand,
3215 vt,
3216 n_vocab,
3217 &dl,
3218 prop.as_ref()
3219 .expect("sampled round without a proposal record"),
3220 sp,
3221 &pen_hist[w0..],
3222 &mut sctr,
3223 &mut uctr,
3224 )?
3225 }
3226 _ => {
3227 let m = dspark_accept_prefix(&cand, &vam, vt);
3228 (m, vam[m])
3229 }
3230 };
3231 if pen_on {
3232 pen_hist.extend_from_slice(&cand[1..=m]);
3233 }
3234 attempted += vt - 1;
3235 accepted += m;
3236 out.push(last);
3237 if eos.contains(&last) {
3238 break 'outer;
3239 }
3240 for &dt in &cand[1..=m] {
3241 // budget check BEFORE the push: at real acceptance the final round often
3242 // accepts a draft at the boundary, and push-then-check emitted max_new+1
3243 // tokens (plain emits exactly max_new — the E2E gate read it as a length
3244 // divergence at index max_new with the shared prefix byte-identical).
3245 if out.len() >= max_new {
3246 break 'outer;
3247 }
3248 out.push(dt);
3249 if eos.contains(&dt) {
3250 break 'outer;
3251 }
3252 }
3253
3254 // ---- commit/rollback: hybrid state cannot truncate — restore + replay kept ----
3255 let keep = m + 1;
3256 let t3 = std::time::Instant::now();
3257 // Slice 3: rounds whose linear column stash lives in the graphs ctx's slabs
3258 // commit through the slab twin (same semantics, slab-addressed sources).
3259 let slab_commit = vgraphs.as_ref().map(|g| g.round_slab).unwrap_or(false);
3260 if keep < vt {
3261 if ckpt_gate {
3262 // GATE ARM: stash-restore, snapshot S1; then the replay oracle, snapshot
3263 // S2; the two cache states must match BIT-FOR-BIT (kv lens, pos, every
3264 // conv/ssm buffer). Continue from the replay state (proven identical).
3265 if slab_commit {
3266 self.dspark_commit_prefix_slab(
3267 e,
3268 &mut cache,
3269 snap,
3270 vgraphs.as_ref().expect("slab_commit implies ctx"),
3271 keep,
3272 )?;
3273 } else {
3274 let vck = vck.as_ref().expect("gate arm always fills the ckpt");
3275 self.dspark_commit_prefix(e, &mut cache, snap, vck, keep)?;
3276 }
3277 // host-side state capture (NO device snapshot copies — two extra
3278 // device snapshots per round OOM'd beside the 15GB trunk)
3279 let capture = |cache: &Cache| -> Result<
3280 (usize, Vec<Option<usize>>, Vec<(Vec<f32>, Vec<f32>)>),
3281 Box<dyn std::error::Error>,
3282 > {
3283 let mut lens = Vec::new();
3284 let mut states = Vec::new();
3285 for il in 0..cache.kv.len() {
3286 lens.push(cache.kv[il].as_ref().map(|k| k.len));
3287 if let Some(rl) = &cache.recur[il] {
3288 states.push((e.dtoh(&rl.conv_state)?, e.dtoh(&rl.ssm_state)?));
3289 }
3290 }
3291 Ok((cache.pos, lens, states))
3292 };
3293 let (p1, l1, st1) = capture(&cache)?;
3294 crate::pp::restore_cache_checkpoint(e, &self.cfg, None, &mut cache, snap)?;
3295 let ram = self.dspark_verify_t_am(e, &cand[..keep], start, &mut cache)?;
3296 assert_eq!(
3297 &ram[..],
3298 &vam[..keep],
3299 "prefix replay must reproduce the verify argmaxes"
3300 );
3301 let (p2, l2, st2) = capture(&cache)?;
3302 assert_eq!(p1, p2, "ckpt-gate: pos mismatch");
3303 assert_eq!(l1, l2, "ckpt-gate: kv_len mismatch");
3304 for (il, ((c1, s1v), (c2, s2v))) in st1.iter().zip(&st2).enumerate() {
3305 let bits = |a: &[f32], b: &[f32]| {
3306 a.iter().zip(b).all(|(x, y)| x.to_bits() == y.to_bits())
3307 };
3308 assert!(
3309 bits(c1, c2),
3310 "ckpt-gate: linear layer {il} conv state differs"
3311 );
3312 assert!(
3313 bits(s1v, s2v),
3314 "ckpt-gate: linear layer {il} ssm state differs"
3315 );
3316 }
3317 } else if slab_commit {
3318 // STASH ARM, slab twin (slice 3): same restore, slab-addressed.
3319 self.dspark_commit_prefix_slab(
3320 e,
3321 &mut cache,
3322 snap,
3323 vgraphs.as_ref().expect("slab_commit implies ctx"),
3324 keep,
3325 )?;
3326 } else if let Some(vck) = vck.as_ref() {
3327 // STASH ARM (default): column-state restore, no replay forward.
3328 self.dspark_commit_prefix(e, &mut cache, snap, vck, keep)?;
3329 } else {
3330 // REPLAY ARM (MEMRA_DSPARK_CKPT=0): the original snapshot+replay oracle.
3331 crate::pp::restore_cache_checkpoint(e, &self.cfg, None, &mut cache, snap)?;
3332 debug_assert_eq!(cache.pos, start, "rollback landed off the round start");
3333 let ram = self.dspark_verify_t_am(e, &cand[..keep], start, &mut cache)?;
3334 if sp_on.is_none() {
3335 // the argmax-reproduction oracle is greedy-only; the sampled arm
3336 // replays purely to rebuild the cache state.
3337 debug_assert_eq!(
3338 &ram[..],
3339 &vam[..keep],
3340 "prefix replay must reproduce the verify argmaxes"
3341 );
3342 }
3343 }
3344 }
3345 ns_roll += clock(stats, e).duration_since(t3).as_nanos() as u64;
3346
3347 // ---- ingest the kept rows' ctx features into the draft KV ----
3348 let t4 = std::time::Instant::now();
3349 {
3350 let tv = e.view(tap_ref, vt * n_taps * n_embd);
3351 let keep_view = tv.slice(0..keep * n_taps * n_embd);
3352 let mut kept = e.uninit(keep * n_taps * n_embd)?;
3353 e.copy_view_into(&mut kept, 0, &keep_view, keep * n_taps * n_embd)?;
3354 let f = draft.ctx_features(e, &kept, keep)?;
3355 let pos_k: Vec<i32> = ((ctx_len as i32)..(ctx_len + keep) as i32).collect();
3356 draft.ingest_ctx(e, &mut dkv, &f, &pos_k, keep)?;
3357 ctx_len += keep;
3358 }
3359 ns_ingest += clock(stats, e).duration_since(t4).as_nanos() as u64;
3360 last = next;
3361 // Ladder update only — under the confidence policies vt is recomputed
3362 // from the head every round, post-draft pre-verify.
3363 if !vt_policy.is_confidence() && adapt {
3364 vt = (m + 2).clamp(3, vt_cap);
3365 }
3366 }
3367 if stats {
3368 let ms = |n: u64| n as f64 / 1e6;
3369 eprintln!(
3370 "[dspark-q38] acceptance {accepted}/{attempted} = {:.3} rounds={rounds} \
3371 draft={:.1}ms snap={:.1}ms verify={:.1}ms rollback+replay={:.1}ms ingest={:.1}ms",
3372 accepted as f64 / attempted.max(1) as f64,
3373 ms(ns_draft),
3374 ms(ns_snap),
3375 ms(ns_verify),
3376 ms(ns_roll),
3377 ms(ns_ingest)
3378 );
3379 }
3380 Ok(out)
3381 }
3382}
3383
3384// ================= DSpark SERVING session (lane/dspark-q38-recover serve route) =========
3385// Burst-scoped state for the worker's dspark spec arm — the qwen-hybrid twin of
3386// GemmaSpecSession. Holds the trunk cache + draft KV + the round loop's carry state
3387// (`last`, ctx_len, adaptive vt) so the scheduler round-robins other sessions between
3388// bursts. The round body is generate_spec_dspark's loop, hoisted; that bin arm stays the
3389// banked oracle (E2E gate), and the serve-route smoke gates this twin byte-identical to
3390// a spec-off boot over the real HTTP surface. Exactness contract unchanged: the target's
3391// verify argmax decides every committed token, so the stream equals plain greedy BY
3392// CONSTRUCTION on every accept path (ckpt stash, gate, replay).
3393pub struct DsparkSpecSession {
3394 pub cache: crate::cache::Cache,
3395 dkv: DflashKv,
3396 last: u32,
3397 ctx_len: usize,
3398 vt: usize,
3399 pub rounds: usize,
3400 max_ctx: usize,
3401 done: bool,
3402 /// Engine-bundle slice 1: persistent batched snapshot (buffers + pointer tables live
3403 /// with the session so bursts reuse them). None until the first round; stays None —
3404 /// legacy per-layer snapshot — when `snapb_off`.
3405 snapb: Option<DsparkSnapBatch>,
3406 snapb_off: bool,
3407 /// SAMPLED ADMISSION (T>0, lane/dspark-sampled-admission-20260820): the request's
3408 /// sampling config (None/temp==0 = the greedy route, byte-identical). Fixed for the
3409 /// session — the worker's admission owns the sampler identity.
3410 sampling: Option<crate::spec::SpecSampling>,
3411 /// Philox event counters, session-owned so randomness never repeats across bursts
3412 /// (the frspec session-continuity law): `sctr` = device sampling events (boundary,
3413 /// draft chain, bonus, residual), `uctr` = host uniforms (selector walk, accept tests).
3414 sctr: u32,
3415 uctr: u32,
3416 /// Penalized-sampled window (lane/dspark-penalized-sampled-20260821): seeded from
3417 /// the prompt tail (`pen_window_seed`), extended with every committed token, carried
3418 /// across bursts so a burst boundary never resets the stream the client asked us to
3419 /// penalize. Empty (and never touched) when the request carries no penalties.
3420 pen_hist: Vec<u32>,
3421}
3422
3423impl DsparkSpecSession {
3424 pub fn cache_max_ctx(&self) -> usize {
3425 self.max_ctx
3426 }
3427 pub fn finished(&self) -> bool {
3428 self.done
3429 }
3430 pub fn pos(&self) -> usize {
3431 self.cache.pos
3432 }
3433}
3434
3435impl crate::hybrid::HybridModel {
3436 /// Turn-1 prime: trunk prefill with taps armed + chunked ctx ingest into the draft KV.
3437 /// Mirrors generate_spec_dspark's prime block exactly (chunk offsets via sink.base are
3438 /// handled inside prime_cache's tick loop; the 256-row ingest chunks match the bin arm).
3439 pub fn dspark_spec_session_new(
3440 &self,
3441 e: &Engine,
3442 draft: &DflashDraft,
3443 prompt: &[u32],
3444 ctx_cap: usize,
3445 sampling: Option<crate::spec::SpecSampling>,
3446 ) -> Result<DsparkSpecSession, Box<dyn std::error::Error>> {
3447 use crate::cache::{Cache, DflashTapSink};
3448 assert!(
3449 self.cfg.gemma4.is_none(),
3450 "gemma4 targets use the assistant-drafter route; dspark is the qwen-hybrid arm"
3451 );
3452 // Penalized SAMPLED requests are IN scope (lane/dspark-penalized-sampled-20260821:
3453 // p-side penalties over the true per-state window, q the recorded proposal — the
3454 // accept walk's penalty arm). Penalties at temp==0 stay a LOUD refusal: the greedy
3455 // walk argmaxes RAW columns and would silently drop them — penalized greedy is
3456 // served exactly on the plain path (worker admission owns that exclusion).
3457 if let Some(sp) = sampling.as_ref() {
3458 if sp.temp <= 0.0 && sp.pen_on() {
3459 return Err(
3460 "dspark spec at temp==0 is the greedy route and would silently drop \
3461 the request's penalties; penalized greedy is served on the plain path"
3462 .into(),
3463 );
3464 }
3465 }
3466 let n_embd = self.cfg.n_embd as usize;
3467 let c = &draft.cfg;
3468 assert_eq!(n_embd, c.hidden, "draft hidden must match target n_embd");
3469 let b = c.block_size;
3470 let n_taps = c.target_layer_ids.len();
3471 // The dspark round is windowless: every position the session will ever hold must
3472 // fit the draft window. Clamp the session ctx to it and refuse prompts that
3473 // cannot take even one round — admission falls back to the plain path.
3474 // DFlash2 rounds implement the reference's symmetric sliding window
3475 // (sdpa_naive_w), so its sessions take the full ctx cap.
3476 let max_ctx = if draft.dflash2.is_some() {
3477 ctx_cap
3478 } else {
3479 ctx_cap.min(c.sliding_window)
3480 };
3481 if prompt.len() + b + 8 > max_ctx {
3482 return Err(format!(
3483 "dspark session needs {} ctx (prompt {} + block {b} + 8), cap {max_ctx}",
3484 prompt.len() + b + 8,
3485 prompt.len()
3486 )
3487 .into());
3488 }
3489 let mut cache = Cache::new(e, &self.cfg, max_ctx)?;
3490 let tp = prompt.len();
3491 cache.dflash_taps = Some(DflashTapSink {
3492 layer_ids: c.target_layer_ids.clone(),
3493 buf: e.uninit(tp * n_taps * n_embd)?,
3494 hidden: n_embd,
3495 t: tp,
3496 base: 0,
3497 });
3498 let (logits, _h_seed, _hiddens) = self.prime_cache(e, prompt, &mut cache, 0)?;
3499 // Boundary token: greedy argmax (byte contract) or the request's own filtered
3500 // draw through the session Philox stream (the frspec boundary composition) —
3501 // penalized over the prompt window when the request carries penalties.
3502 let mut sctr0 = 0u32;
3503 let pen_hist: Vec<u32> = match sampling.as_ref().filter(|s| s.temp > 0.0 && s.pen_on()) {
3504 Some(sp) => crate::spec::pen_window_seed(&[], prompt, sp.penalty_last_n),
3505 None => Vec::new(),
3506 };
3507 let last = match sampling.as_ref().filter(|s| s.temp > 0.0) {
3508 Some(sp) => crate::spec::sample_boundary_token(
3509 e,
3510 &logits,
3511 sp,
3512 &pen_hist,
3513 &mut sctr0,
3514 "dspark-prime",
3515 )?,
3516 None => crate::forward::argmax(&logits) as u32,
3517 };
3518 let mut dkv = DflashKv::new(e, &draft.cfg, max_ctx)?;
3519 {
3520 let taps = cache.dflash_taps.take().unwrap();
3521 let n_taps_h = n_taps * n_embd;
3522 let mut r0 = 0usize;
3523 while r0 < tp {
3524 let t_c = (tp - r0).min(256);
3525 let tv = e.view(&taps.buf, tp * n_taps_h);
3526 let win = tv.slice(r0 * n_taps_h..(r0 + t_c) * n_taps_h);
3527 let mut chunk = e.uninit(t_c * n_taps_h)?;
3528 e.copy_view_into(&mut chunk, 0, &win, t_c * n_taps_h)?;
3529 let f = draft.ctx_features(e, &chunk, t_c)?;
3530 let pos_c: Vec<i32> = ((r0 as i32)..(r0 + t_c) as i32).collect();
3531 draft.ingest_ctx(e, &mut dkv, &f, &pos_c, t_c)?;
3532 r0 += t_c;
3533 }
3534 }
3535 e.stream().synchronize()?;
3536 // Verify carries [anchor, drafts] = up to n_drafts+1 rows (harvest-dependent;
3537 // DSPARK-POSTMORTEM-20260820.md; family-keyed for DFlash2, else checkpoint
3538 // strategy census).
3539 let nd = DsparkHarvest::for_draft(draft).n_drafts(b);
3540 let vt_cap: usize = std::env::var("MEMRA_DFLASH_VERIFY_T")
3541 .ok()
3542 .and_then(|v| v.parse().ok())
3543 .unwrap_or(nd + 1)
3544 .clamp(2, nd + 1);
3545 Ok(DsparkSpecSession {
3546 cache,
3547 dkv,
3548 last,
3549 ctx_len: tp,
3550 vt: vt_cap,
3551 rounds: 0,
3552 max_ctx,
3553 done: false,
3554 snapb: None,
3555 snapb_off: !crate::spec::state_copy_batch_on(),
3556 sampling,
3557 sctr: sctr0,
3558 uctr: 0,
3559 pen_hist,
3560 })
3561 }
3562
3563 /// One scheduler burst: dspark rounds until >= `burst_target` tokens are committed,
3564 /// EOS lands, or the ctx cap is reached. Returns (tokens, drafted, accepted) for this
3565 /// burst — the worker clamps the public slice (engine overshoot within a round stays
3566 /// in the session cache, exactly the gemma-burst contract).
3567 pub fn dspark_spec_session_burst(
3568 &self,
3569 e: &Engine,
3570 draft: &DflashDraft,
3571 sess: &mut DsparkSpecSession,
3572 burst_target: usize,
3573 eos: &[u32],
3574 ) -> Result<(Vec<u32>, usize, usize), Box<dyn std::error::Error>> {
3575 use crate::cache::DflashTapSink;
3576 let n_embd = self.cfg.n_embd as usize;
3577 let c = &draft.cfg;
3578 let b = c.block_size;
3579 let n_taps = c.target_layer_ids.len();
3580 let n_vocab = self.output.out_features();
3581 // Harvest convention (DSPARK-POSTMORTEM-20260820.md) — identical to the bin arm
3582 // (family-keyed for DFlash2, else checkpoint strategy census; owner-ratified
3583 // flip 2026-08-20).
3584 let harvest = DsparkHarvest::for_draft(draft);
3585 let nd = harvest.n_drafts(b);
3586 let r0 = harvest.first_row();
3587 let vt_cap: usize = std::env::var("MEMRA_DFLASH_VERIFY_T")
3588 .ok()
3589 .and_then(|v| v.parse().ok())
3590 .unwrap_or(nd + 1)
3591 .clamp(2, nd + 1);
3592 let adapt = std::env::var("MEMRA_DFLASH_ADAPT").as_deref() != Ok("0");
3593 // Verify-window policy (H4, DSPARK-POSTMORTEM-20260820.md) — identical to the
3594 // bin arm: default = confidence-slot tau=.5 on a head-carrying checkpoint
3595 // (owner-ratified flip 2026-08-20); head-less (incl. the DFlash2 family) and
3596 // ADAPT=0 keep the ladder.
3597 let vt_policy = DsparkVtPolicy::resolve(draft.confidence.is_some());
3598 if vt_policy.is_confidence() {
3599 assert!(
3600 draft.confidence.is_some(),
3601 "MEMRA_DSPARK_VT={vt_policy:?} needs a checkpoint with an accept-rate \
3602 head (confidence_head.* absent in this export)"
3603 );
3604 }
3605 // SAMPLED ADMISSION (T>0): session-fixed config; counters live on the session so
3606 // randomness never repeats across bursts. None/temp==0 = the greedy route.
3607 let sp_on: Option<crate::spec::SpecSampling> = sess.sampling.filter(|s| s.temp > 0.0);
3608 let pen_on = sp_on.as_ref().is_some_and(|s| s.pen_on());
3609 let mut out: Vec<u32> = Vec::with_capacity(burst_target + b);
3610 let mut drafted = 0usize;
3611 let mut accepted_n = 0usize;
3612 // Engine-bundle slice 2 — identical to the bin arm: deferred chain readback under
3613 // the stash arm with a resident embed table (ladder policy only).
3614 let defer_rb = crate::spec::dspark_defer_readback_on() && !vt_policy.is_confidence();
3615 let (embd_qt, embd_rb) = self.embd.qt_and_row_bytes(n_embd);
3616 let embd_gpu = if !defer_rb || crate::spec::spec_host_embd() {
3617 None
3618 } else {
3619 Some(
3620 self.embd_gpu
3621 .get_or_init(|| e.upload_u8(&self.embd.raw).expect("embed table upload")),
3622 )
3623 };
3624 // Slice 3/4c SERVE ENGAGEMENT (graphs-serve lane; DSF-ROUNDCOST §9.3 -> §10): the
3625 // verify-graph pool lives on the MODEL (`dspark_vgraphs`, one per process) and its
3626 // keys — (segment, vt) and (vt, rung, hi) — carry NOTHING session-scoped, so ANY
3627 // session whose round matches a key replays the same capture (this is the
3628 // cache-reuse-pool the old bin-arm-only note asked for). Sharing is sound because
3629 // every per-session-varying address the captured bodies touch is indirect:
3630 // conv/ssm state and the ckpt stash resolve through the per-verify refreshed
3631 // pointer table (refresh_tables + copy_indirect_src_f32 — the slice-3
3632 // parity/lifetime law; a baked address is the known 12/12-divergence class), kv
3633 // bases through fa_table, residual/pos/tap through ctx-owned staging rewritten
3634 // every round; per-row t_kv derives in-kernel from pos_seq, and the per-round
3635 // host bookkeeping (parity swap, len bump) runs on THIS session's cache. The
3636 // guard spans the burst: the slab stash is live verify -> commit inside each
3637 // round, and the worker drives bursts from one scheduler thread
3638 // (step_dspark_spec), so sessions interleave at burst boundaries only.
3639 // DEFAULT ON on the serve route since the v0.103 train (owner-ratified
3640 // 2026-08-22, §10 re-gate at flip): MEMRA_DSPARK_VERIFY_GRAPH=0 is the
3641 // kill-switch that keeps this None — the eager walk, byte-identical (the
3642 // kill-switch arm of the serve battery). The bin arm keeps its own opt-in.
3643 let mut vg_guard = self.dspark_vgraphs.lock().unwrap();
3644 if vg_guard.is_none() && embd_gpu.is_some() && crate::spec::dspark_verify_graph_serve_on() {
3645 *vg_guard = crate::spec::DsparkVerifyGraphs::new(e, &sess.cache, vt_cap, n_embd)?;
3646 if vg_guard.is_some() {
3647 // Engagement receipt (the §8 dead-arm lesson): prove the door is LIVE on
3648 // the serve surface — S6b banked the tip server carrying zero door strings.
3649 eprintln!("[dspark-vg] serve pool ENGAGED (vt_cap={vt_cap})");
3650 }
3651 }
3652 let vgraphs: &mut Option<crate::spec::DsparkVerifyGraphs> = &mut vg_guard;
3653 'outer: while out.len() < burst_target && !sess.done {
3654 let start = sess.cache.pos;
3655 if start + nd + 1 > sess.max_ctx {
3656 sess.done = true;
3657 break;
3658 }
3659 sess.rounds += 1;
3660 let mut vt = sess.vt;
3661 // ---- draft: block = [last, MASK x b-1] (identical to the bin arm) ----
3662 e.set_verify_exact(true);
3663 let mut block: Vec<u32> = vec![c.mask_token_id; b];
3664 block[0] = sess.last;
3665 let noise = e.htod(&self.embd.gather(n_embd, &block))?;
3666 let pos_block: Vec<i32> = ((start as i32)..(start + b) as i32).collect();
3667 let dh = draft.forward_round(e, &mut sess.dkv, &noise, &pos_block)?;
3668 // Harvest: logits over rows r0..r0+nd (see the bin arm / the postmortem).
3669 let mut rows = e.uninit(nd * n_embd)?;
3670 {
3671 let dv = e.view(&dh, b * n_embd);
3672 let src = dv.slice(r0 * n_embd..(r0 + nd) * n_embd);
3673 e.copy_view_into(&mut rows, 0, &src, nd * n_embd)?;
3674 }
3675 let mut dl = e.matmul(&self.output, &rows, nd)?;
3676 // Family/sampling-keyed proposal — identical to the bin arm (see there for
3677 // the program law: sampled records the true q, DFlash2 rides the selector,
3678 // the markov/plain greedy chain keeps the slice-2 deferral). Confidence
3679 // policy: stash markov prev-token embeddings d2d during the chain, one host
3680 // readback after — identical to the bin arm.
3681 let want_conf_emb = vt_policy.is_confidence()
3682 && draft.confidence.as_ref().is_some_and(|ch| ch.with_markov);
3683 let mut conf_emb: Option<CudaSlice<f32>> = match (&draft.markov, want_conf_emb) {
3684 (Some(mk), true) => Some(e.uninit(nd * mk.rank)?),
3685 (None, true) => unreachable!(
3686 "with_markov confidence head without a markov table — the loader forbids it"
3687 ),
3688 _ => None,
3689 };
3690 let mut cand: Vec<u32> = Vec::with_capacity(nd + 1);
3691 let mut prop: Option<DsparkDraftSample> = None;
3692 let mut chain_dev: Option<CudaSlice<u32>> = None;
3693 // Slice 2: arm choice read before the chain readback (see the bin arm; the
3694 // serve arm has no CKPT_GATE oracle — the bin arm carries it).
3695 let ckpt_on = std::env::var("MEMRA_DSPARK_CKPT").as_deref() != Ok("0");
3696 let mut deferred = false;
3697 if let Some(sp) = sp_on.as_ref() {
3698 // SAMPLED proposal (family-keyed; identical to the bin arm).
3699 let (tail, ds) = draft.dspark_propose_sampled(
3700 e,
3701 &mut dl,
3702 &rows,
3703 nd,
3704 n_vocab,
3705 sess.last,
3706 sp,
3707 &mut sess.sctr,
3708 &mut sess.uctr,
3709 conf_emb.as_mut(),
3710 )?;
3711 e.set_verify_exact(false);
3712 cand.push(sess.last);
3713 cand.extend_from_slice(&tail);
3714 prop = Some(ds);
3715 } else if draft.dflash2.is_some() {
3716 // DFlash2: candidate path selector replaces the markov chain
3717 // (identical to the bin arm).
3718 let path = draft.dflash2_propose_greedy(e, &dl, &rows, nd, n_vocab, sess.last)?;
3719 e.set_verify_exact(false);
3720 cand.push(sess.last);
3721 cand.extend_from_slice(&path);
3722 } else {
3723 let markov_on = std::env::var("MEMRA_DFLASH_MARKOV").as_deref() != Ok("0");
3724 let mut chain_d = e.stream().alloc_zeros::<u32>(nd + 1)?;
3725 if let (Some(mk), true) = (&draft.markov, markov_on) {
3726 e.set_u32_one(&mut chain_d, sess.last)?;
3727 for k in 0..nd {
3728 let mut f = e.uninit(mk.rank)?;
3729 e.gather_row_bf16(&mk.w1_bf16, &chain_d, k, &mut f, mk.rank)?;
3730 if let Some(ce) = conf_emb.as_mut() {
3731 let fv = e.view(&f, mk.rank);
3732 e.copy_view_into(ce, k * mk.rank, &fv, mk.rank)?;
3733 }
3734 let bias = e.matmul(&mk.w2, &f, 1)?;
3735 e.add_row_inplace(&mut dl, &bias, n_vocab, k * n_vocab)?;
3736 e.argmax_token_device_col(&dl, k, n_vocab, &mut chain_d, k + 1)?;
3737 }
3738 } else {
3739 if want_conf_emb {
3740 // chain_d[0] must carry the anchor — slot 0's prev token.
3741 e.set_u32_one(&mut chain_d, sess.last)?;
3742 }
3743 for i in 0..nd {
3744 if let (Some(ce), Some(mk)) = (conf_emb.as_mut(), &draft.markov) {
3745 let mut f = e.uninit(mk.rank)?;
3746 e.gather_row_bf16(&mk.w1_bf16, &chain_d, i, &mut f, mk.rank)?;
3747 let fv = e.view(&f, mk.rank);
3748 e.copy_view_into(ce, i * mk.rank, &fv, mk.rank)?;
3749 }
3750 e.argmax_token_device_col(&dl, i, n_vocab, &mut chain_d, i + 1)?;
3751 }
3752 }
3753 e.set_verify_exact(false);
3754 deferred = embd_gpu.is_some() && ckpt_on;
3755 chain_dev = Some(chain_d);
3756 }
3757 // ---- H4 confidence window: size THIS round's verify from the head ----
3758 if vt_policy.is_confidence() {
3759 let ch = draft.confidence.as_ref().expect("asserted at burst entry");
3760 let (rows_h, emb_h) = match conf_emb.as_ref() {
3761 Some(ce) => {
3762 let (a, b2) = e.dtoh_pair(&rows, ce)?;
3763 (a, Some(b2))
3764 }
3765 None => (e.dtoh(&rows)?, None),
3766 };
3767 let rank = draft.markov.as_ref().map(|m| m.rank).unwrap_or(0);
3768 let mut raws = Vec::with_capacity(nd);
3769 for k in 0..nd {
3770 let hrow = &rows_h[k * n_embd..(k + 1) * n_embd];
3771 let emb = emb_h.as_ref().map(|eh| &eh[k * rank..(k + 1) * rank]);
3772 raws.push(ch.raw_score(hrow, emb));
3773 }
3774 vt = vt_policy
3775 .size_window(&raws, vt_cap)
3776 .expect("confidence policies always size the window");
3777 }
3778 // Non-deferred greedy chain readback (the sampled and DFlash2 proposals
3779 // built `cand` at the walk; deferred rounds build it after the merged
3780 // readback — bytes identical, chain_d written before either sync).
3781 if let Some(chain_d) = chain_dev.as_ref() {
3782 if !deferred {
3783 let chain = e.dtoh_u32(chain_d)?;
3784 cand.push(sess.last);
3785 cand.extend_from_slice(&chain[1..]);
3786 }
3787 }
3788
3789 // ---- snapshot, then verify t=vt (ckpt stash default; oracle arms kept) ----
3790 // Slice 1: batched snap (see DsparkSnapBatch) with the legacy per-layer
3791 // snapshot as the kill-switch / non-uniform fallback.
3792 let mut snap_legacy: Option<crate::cache::CacheSnapshot> = None;
3793 if !sess.snapb_off && sess.snapb.is_none() {
3794 sess.snapb = DsparkSnapBatch::new(e, &sess.cache)?;
3795 sess.snapb_off = sess.snapb.is_none();
3796 } else if let Some(sb) = sess.snapb.as_mut() {
3797 sb.refresh(e, &sess.cache)?;
3798 }
3799 let snap: &crate::cache::CacheSnapshot = match sess.snapb.as_ref() {
3800 Some(sb) => &sb.snap,
3801 None => {
3802 snap_legacy = Some(sess.cache.snapshot(e)?);
3803 snap_legacy.as_ref().unwrap()
3804 }
3805 };
3806 let _ = &snap_legacy;
3807 // Slice 3: the tap-sink buffer is persistent per vt in the graphs ctx
3808 // (captured segments bake its address — a per-round alloc here would make
3809 // every session's replayed tap copies write freed memory); fully rewritten
3810 // by every verify, so pool ownership changes no bytes.
3811 let tap_buf = match vgraphs.as_mut().and_then(|g| g.tap_bufs.remove(&vt)) {
3812 Some(buf) => buf,
3813 None => e.uninit(vt * n_taps * n_embd)?,
3814 };
3815 sess.cache.dflash_taps = Some(DflashTapSink {
3816 layer_ids: c.target_layer_ids.clone(),
3817 buf: tap_buf,
3818 hidden: n_embd,
3819 t: vt,
3820 base: 0,
3821 });
3822 // Composition guard (sampled admission × model-owned pool, this train's
3823 // cross-product): the slab flag is a per-round statement, but only the
3824 // graphs-aware verify (`_am_ckpt_dev`) clears it. Serve sessions MIX arms
3825 // within one process-lifetime pool — a SAMPLED round rides the raw-logits
3826 // twins (no graphs param) and must not inherit `round_slab=true` from a
3827 // previous greedy session's captured round, or its commit is steered at
3828 // slabs the round never wrote. Clear at the round boundary; the deferred
3829 // arm re-derives it inside the verify. (The bin arm has the same shape but
3830 // fixes its sampling mode per process, so no mixed rounds exist there.)
3831 if let Some(g) = vgraphs.as_mut() {
3832 g.round_slab = false;
3833 }
3834 // The whole fallible verify window runs inside a closure so the Err path
3835 // can return the sink buffer to the ctx pool before propagating — the
3836 // serve-surface twin of the EOS-orphan lesson: a mid-verify error
3837 // propagates OUT of the burst, the request dies, the session's cache is
3838 // dropped — but the PROCESS (and the pool, with the tap-buffer address
3839 // baked into its captures) lives on. Recover the ctx-owned buffer before
3840 // the error escapes, or the next session's replayed tap copies write
3841 // freed memory. The bin arm has no such path (a gate-binary error ends
3842 // the process).
3843 let verify_out = (|| -> Result<
3844 (
3845 Vec<u32>,
3846 Option<CudaSlice<f32>>,
3847 Option<crate::spec::DsparkVerifyCkpt>,
3848 ),
3849 Box<dyn std::error::Error>,
3850 > {
3851 if sp_on.is_some() {
3852 // SAMPLED: raw verify logits for the rejection walk (bin-arm twin).
3853 if ckpt_on {
3854 let (tl, vck) = self.dspark_verify_t_logits_ckpt(
3855 e,
3856 &cand[..vt],
3857 start,
3858 &mut sess.cache,
3859 )?;
3860 Ok((Vec::new(), Some(tl), Some(vck)))
3861 } else {
3862 Ok((
3863 Vec::new(),
3864 Some(self.dspark_verify_t_logits(
3865 e,
3866 &cand[..vt],
3867 start,
3868 &mut sess.cache,
3869 )?),
3870 None,
3871 ))
3872 }
3873 } else if deferred {
3874 // Slice 2: device-token verify + ONE merged readback (see the bin arm).
3875 let chain_d = chain_dev.as_ref().expect("deferred implies greedy chain");
3876 let g = embd_gpu.expect("deferred implies resident embed");
3877 let (am_d, vck) = self.dspark_verify_t_am_ckpt_dev(
3878 e,
3879 chain_d,
3880 vt,
3881 start,
3882 &mut sess.cache,
3883 (g, embd_qt, embd_rb),
3884 vgraphs.as_mut(),
3885 )?;
3886 let ch = e.stream().clone_dtoh(chain_d)?;
3887 let am = e.stream().clone_dtoh(&am_d)?;
3888 e.stream().synchronize()?;
3889 cand.push(sess.last);
3890 cand.extend_from_slice(&ch[1..]);
3891 Ok((am, None, Some(vck)))
3892 } else if ckpt_on {
3893 let (vam, vck) =
3894 self.dspark_verify_t_am_ckpt(e, &cand[..vt], start, &mut sess.cache)?;
3895 Ok((vam, None, Some(vck)))
3896 } else {
3897 Ok((
3898 self.dspark_verify_t_am(e, &cand[..vt], start, &mut sess.cache)?,
3899 None,
3900 None,
3901 ))
3902 }
3903 })();
3904 let (vam, tl, vck) = match verify_out {
3905 Ok(v) => v,
3906 Err(err) => {
3907 if let Some(taps) = sess.cache.dflash_taps.take() {
3908 if let Some(g) = vgraphs.as_mut() {
3909 g.tap_bufs.insert(vt, taps.buf);
3910 }
3911 }
3912 return Err(err);
3913 }
3914 };
3915 let taps = sess.cache.dflash_taps.take().unwrap();
3916 // Return the tap buffer to the ctx pool IMMEDIATELY — an EOS/budget break
3917 // between accept and ingest must never orphan an address the captured graphs
3918 // bake (the bin arm's lesson, and it holds doubly here: the pool outlives
3919 // the SESSION, not just the round). Ingest reads it borrowed.
3920 let tap_local: Option<CudaSlice<f32>> = match vgraphs.as_mut() {
3921 Some(g) => {
3922 g.tap_bufs.insert(vt, taps.buf);
3923 None
3924 }
3925 None => Some(taps.buf),
3926 };
3927 let tap_ref: &CudaSlice<f32> = match &tap_local {
3928 Some(b) => b,
3929 None => &vgraphs.as_ref().expect("ctx present above").tap_bufs[&vt],
3930 };
3931
3932 // ---- accept ----
3933 // Penalized-sampled: anchor joins the window before the walk (committed this
3934 // round via the out.push below); accepted drafts extend it after — identical
3935 // to the bin arm.
3936 if pen_on {
3937 sess.pen_hist.push(sess.last);
3938 }
3939 let (m, next) = match (sp_on.as_ref(), tl.as_ref()) {
3940 (Some(sp), Some(tl)) => {
3941 let w0 = sess
3942 .pen_hist
3943 .len()
3944 .saturating_sub(sp.penalty_last_n.min(crate::spec::PEN_WINDOW_MAX));
3945 dspark_accept_sampled(
3946 e,
3947 tl,
3948 &cand,
3949 vt,
3950 n_vocab,
3951 &dl,
3952 prop.as_ref()
3953 .expect("sampled round without a proposal record"),
3954 sp,
3955 &sess.pen_hist[w0..],
3956 &mut sess.sctr,
3957 &mut sess.uctr,
3958 )?
3959 }
3960 _ => {
3961 let m = dspark_accept_prefix(&cand, &vam, vt);
3962 (m, vam[m])
3963 }
3964 };
3965 if pen_on {
3966 sess.pen_hist.extend_from_slice(&cand[1..=m]);
3967 }
3968 drafted += vt - 1;
3969 accepted_n += m;
3970 out.push(sess.last);
3971 if eos.contains(&sess.last) {
3972 sess.done = true;
3973 break 'outer;
3974 }
3975 for &dt in &cand[1..=m] {
3976 out.push(dt);
3977 if eos.contains(&dt) {
3978 sess.done = true;
3979 break 'outer;
3980 }
3981 }
3982
3983 // ---- commit/rollback (stash arm default; replay oracle kept) ----
3984 let keep = m + 1;
3985 // Slice 3: rounds whose linear column stash lives in the graphs ctx's slabs
3986 // commit through the slab twin (same semantics, slab-addressed sources) —
3987 // identical to the bin arm's dispatch.
3988 let slab_commit = vgraphs.as_ref().map(|g| g.round_slab).unwrap_or(false);
3989 if keep < vt {
3990 if slab_commit {
3991 self.dspark_commit_prefix_slab(
3992 e,
3993 &mut sess.cache,
3994 snap,
3995 vgraphs.as_ref().expect("slab_commit implies ctx"),
3996 keep,
3997 )?;
3998 } else if let Some(vck) = vck.as_ref() {
3999 self.dspark_commit_prefix(e, &mut sess.cache, snap, vck, keep)?;
4000 } else {
4001 crate::pp::restore_cache_checkpoint(e, &self.cfg, None, &mut sess.cache, snap)?;
4002 debug_assert_eq!(sess.cache.pos, start, "rollback landed off the round start");
4003 let ram = self.dspark_verify_t_am(e, &cand[..keep], start, &mut sess.cache)?;
4004 if sp_on.is_none() {
4005 // greedy-only oracle; the sampled arm replays to rebuild state.
4006 debug_assert_eq!(
4007 &ram[..],
4008 &vam[..keep],
4009 "prefix replay must reproduce the verify argmaxes"
4010 );
4011 }
4012 }
4013 }
4014
4015 // ---- ingest the kept rows' ctx features into the draft KV ----
4016 {
4017 let tv = e.view(tap_ref, vt * n_taps * n_embd);
4018 let keep_view = tv.slice(0..keep * n_taps * n_embd);
4019 let mut kept = e.uninit(keep * n_taps * n_embd)?;
4020 e.copy_view_into(&mut kept, 0, &keep_view, keep * n_taps * n_embd)?;
4021 let f = draft.ctx_features(e, &kept, keep)?;
4022 let pos_k: Vec<i32> =
4023 ((sess.ctx_len as i32)..(sess.ctx_len + keep) as i32).collect();
4024 draft.ingest_ctx(e, &mut sess.dkv, &f, &pos_k, keep)?;
4025 sess.ctx_len += keep;
4026 }
4027 sess.last = next;
4028 // Ladder update only — the confidence policies recompute vt from the
4029 // head every round, post-draft pre-verify; their carry just keeps
4030 // observability (sess.vt = the last confidence-sized window).
4031 if vt_policy.is_confidence() {
4032 sess.vt = vt;
4033 } else if adapt {
4034 sess.vt = (m + 2).clamp(3, vt_cap);
4035 }
4036 }
4037 Ok((out, drafted, accepted_n))
4038 }
4039}
4040
4041// ================= Harvest-convention gate (CPU; DSPARK-POSTMORTEM-20260820.md) =========
4042// The parity oracle is row-count-agnostic (it reproduces the markov MODULE on whatever
4043// rows it is fed) and the E2E gate is harvest-independent (verify-side truth), so
4044// NEITHER can catch a wrong row->position mapping — that blindness is how the q38
4045// misalignment shipped. These tests pin the convention itself as logic the round
4046// consumes, so a mutation back to the mask-fill harvest under the Dspark variant fails
4047// HERE, naming the convention.
4048#[cfg(test)]
4049mod dflash2_tests {
4050 use super::{DsparkHarvest, dflash2_walk_greedy, dflash2_walk_sampled, rejection_accept_len};
4051
4052 /// f32 -> bf16 bytes (truncation; test values are bf16-exact small integers).
4053 fn bf16(vals: &[f32]) -> Vec<u8> {
4054 vals.iter()
4055 .flat_map(|v| ((v.to_bits() >> 16) as u16).to_le_bytes())
4056 .collect()
4057 }
4058
4059 const V: usize = 8; // test vocab
4060 const R: usize = 2; // selector rank
4061 const K: usize = 2; // top_k
4062
4063 /// Codebooks for the chain tests: pred rows are one-hot-ish, succ rows chosen so
4064 /// the slot-1 winner FLIPS with the slot-0 choice.
4065 fn books() -> (Vec<u8>, Vec<u8>) {
4066 let mut pred = vec![0f32; V * R];
4067 pred[0] = 1.0; // tok 0: [1, 0] (the anchor)
4068 pred[1 * R + 1] = 1.0; // tok 1: [0, 1]
4069 pred[2 * R] = 1.0; // tok 2: [1, 0]
4070 let mut succ = vec![0f32; V * R];
4071 succ[1 * R] = 2.0; // tok 1: [2, 0]
4072 succ[2 * R + 1] = 5.0; // tok 2: [0, 5]
4073 succ[3 * R + 1] = 3.0; // tok 3: [0, 3]
4074 succ[4 * R] = 10.0; // tok 4: [10, 0]
4075 (bf16(&pred), bf16(&succ))
4076 }
4077
4078 #[test]
4079 fn selector_walk_is_a_chain_not_per_slot_argmax() {
4080 let (pred, succ) = books();
4081 // slot 0 candidates {1, 2}, slot 1 candidates {3, 4}; hproj all-ones.
4082 let cand: Vec<u32> = vec![1, 2, 3, 4];
4083 let hproj = vec![1.0f32; 2 * R];
4084 // Anchor 0 (pred [1,0]): slot 0 scores = <[1,0],succ> -> tok1: 2, tok2: 0
4085 // -> picks 1. Slot 1 must then walk from pred[1]=[0,1]: tok3 scores 3,
4086 // tok4 scores 0 -> picks 3. A mutation that seeds every slot from the ANCHOR
4087 // (pred[0]=[1,0]) scores tok3: 0 / tok4: 10 and picks 4 instead — the chain
4088 // IS the semantics (reference CandidateSelector.select: `predecessor` is the
4089 // previously CHOSEN candidate, seeded by anchor_ids).
4090 let path = dflash2_walk_greedy(&pred, &succ, V, R, K, &[0.0; 4], &cand, &hproj, 0, 2);
4091 assert_eq!(
4092 path,
4093 vec![1, 3],
4094 "walk must seed slot p from slot p-1's CHOSEN candidate \
4095 (z-lab model.py CandidateSelector.select)"
4096 );
4097 }
4098
4099 #[test]
4100 fn selector_walk_unary_term_participates() {
4101 let (pred, succ) = books();
4102 let cand: Vec<u32> = vec![1, 2, 3, 4];
4103 let hproj = vec![1.0f32; 2 * R];
4104 // unary +10 on slot-0 candidate 2 overrides the bilinear 2-vs-0 margin;
4105 // the chain then walks from pred[2]=[1,0] and slot 1 flips to tok 4.
4106 let path = dflash2_walk_greedy(
4107 &pred,
4108 &succ,
4109 V,
4110 R,
4111 K,
4112 &[0.0, 10.0, 0.0, 0.0],
4113 &cand,
4114 &hproj,
4115 0,
4116 2,
4117 );
4118 assert_eq!(
4119 path,
4120 vec![2, 4],
4121 "score = unary + bilinear (reference: `unary[:, position] + einsum(...)`); \
4122 dropping the unary term picks tok 1 here"
4123 );
4124 }
4125
4126 #[test]
4127 fn selector_walk_hidden_gate_participates() {
4128 let (pred, succ) = books();
4129 let cand: Vec<u32> = vec![1, 2, 3, 4];
4130 // hproj [0, .] zeroes the pred[0]=[1,0] gate for slot 0: tok1's bilinear 2
4131 // vanishes, and the unary tiebreak (+1 on tok2) decides. The chain from tok2
4132 // (pred [1,0]) with slot-1 hproj [1,1] then picks tok4 (10 vs 0).
4133 let hproj = vec![0.0f32, 1.0, 1.0, 1.0];
4134 let path = dflash2_walk_greedy(
4135 &pred,
4136 &succ,
4137 V,
4138 R,
4139 K,
4140 &[0.0, 1.0, 0.0, 0.0],
4141 &cand,
4142 &hproj,
4143 0,
4144 2,
4145 );
4146 assert_eq!(
4147 path,
4148 vec![2, 4],
4149 "the bilinear gate is pred_row .* HIDDEN_PROJECTION (reference: \
4150 `predecessor_codebook(predecessor) * hidden[:, position]`); ignoring \
4151 hproj leaves tok1's margin standing"
4152 );
4153 }
4154
4155 #[test]
4156 fn dflash2_harvest_is_census_keyed() {
4157 // DFlash2 is mask-fill BY CONSTRUCTION (reference dflash_generate harvests
4158 // rows 1-verify_size:; card: "7 draft tokens per verification step").
4159 assert_eq!(
4160 DsparkHarvest::for_family_value(true, None, false),
4161 DsparkHarvest::Dflash
4162 );
4163 assert_eq!(
4164 DsparkHarvest::for_family_value(true, Some("dflash"), false),
4165 DsparkHarvest::Dflash
4166 );
4167 // The family key BEATS the strategy census: a (hypothetical) DFlash2 export
4168 // whose config also strategy-censuses dspark still harvests mask-fill.
4169 assert_eq!(
4170 DsparkHarvest::for_family_value(true, None, true),
4171 DsparkHarvest::Dflash
4172 );
4173 // An env override to the SHIFTED harvest contradicts the census — REFUSE,
4174 // never re-key (the postmortem's misalignment class in reverse).
4175 assert!(
4176 std::panic::catch_unwind(|| DsparkHarvest::for_family_value(
4177 true,
4178 Some("dspark"),
4179 false
4180 ))
4181 .is_err(),
4182 "MEMRA_DSPARK_HARVEST=dspark on a DFlash2 checkpoint must refuse"
4183 );
4184 // Non-DFlash2 checkpoints ride the strategy-keyed resolution (env wins).
4185 assert_eq!(
4186 DsparkHarvest::for_family_value(false, Some("dspark"), false),
4187 DsparkHarvest::Dspark
4188 );
4189 assert_eq!(
4190 DsparkHarvest::for_family_value(false, None, false),
4191 DsparkHarvest::Dflash
4192 );
4193 assert_eq!(
4194 DsparkHarvest::for_family_value(false, None, true),
4195 DsparkHarvest::Dspark,
4196 "unset env on a DSPARK-strategy export must keep the ratified census flip"
4197 );
4198 }
4199
4200 // ============ SAMPLED ADMISSION (T>0) gates — lane/dspark-sampled-admission-20260820 =
4201 // The device kernels are oracled by sample_check (filter_stats/gumbel/residual arms);
4202 // these pin the HOST math the route ships — the selector's sampled walk, the accept
4203 // rule, and the round COMPOSITION (accept + residual + bonus must reproduce the target
4204 // distribution p exactly; a mis-composition leaves every kernel individually correct,
4205 // which is why the composition arm exists — sample_check arm 6's lesson).
4206
4207 #[test]
4208 fn sampled_walk_tiny_temp_matches_greedy() {
4209 // T->0 continuity: at tiny temperature the candidate softmax concentrates on the
4210 // argmax and the sampled walk must reproduce the greedy chain token-for-token
4211 // (the frspec gate-(1) shape). Same fixture as the chain test.
4212 let (pred, succ) = books();
4213 let cand: Vec<u32> = vec![1, 2, 3, 4];
4214 let hproj = vec![1.0f32; 2 * R];
4215 let greedy = dflash2_walk_greedy(&pred, &succ, V, R, K, &[0.0; 4], &cand, &hproj, 0, 2);
4216 let mut u = || 0.5f32;
4217 let (path, q_chosen, q_rows) = dflash2_walk_sampled(
4218 &pred, &succ, V, R, K, &[0.0; 4], &cand, &hproj, 0, 2, 1e-6, &mut u,
4219 );
4220 assert_eq!(
4221 path, greedy,
4222 "tiny-T sampled walk must equal the greedy chain"
4223 );
4224 assert_eq!(q_rows.len(), 2 * K);
4225 for (p, &q) in path.iter().zip(&q_chosen) {
4226 let _ = p;
4227 assert!(
4228 q > 0.999,
4229 "tiny-T chosen-candidate prob must be ~1, got {q}"
4230 );
4231 }
4232 }
4233
4234 #[test]
4235 fn sampled_walk_records_the_distribution_it_samples() {
4236 // The recorded q IS the proposal: per slot the q_rows sum to ~1, q_chosen is the
4237 // row value at the drawn candidate, and the CDF walk picks the candidate whose
4238 // cumulative bracket contains the uniform.
4239 let (pred, succ) = books();
4240 let cand: Vec<u32> = vec![1, 2, 3, 4];
4241 let hproj = vec![1.0f32; 2 * R];
4242 // slot-0 scores at anchor 0: tok1 = 2.0, tok2 = 0.0; at T=2.0 the softmax is
4243 // e^1/(e^1+e^0) ~= 0.731 for tok1.
4244 let q1 = (1f64.exp() / (1f64.exp() + 1.0)) as f32;
4245 for (u0, want0) in [(q1 - 0.01, 1u32), (q1 + 0.01, 2u32)] {
4246 let mut seq = vec![u0, 0.0f32].into_iter();
4247 let mut u = move || seq.next().unwrap();
4248 let (path, q_chosen, q_rows) = dflash2_walk_sampled(
4249 &pred, &succ, V, R, K, &[0.0; 4], &cand, &hproj, 0, 2, 2.0, &mut u,
4250 );
4251 assert_eq!(
4252 path[0], want0,
4253 "CDF walk must place u={u0} in the right candidate bracket"
4254 );
4255 let row0: f32 = q_rows[..K].iter().sum();
4256 assert!(
4257 (row0 - 1.0).abs() < 1e-5,
4258 "slot-0 q must sum to 1, got {row0}"
4259 );
4260 let ci = cand[..K].iter().position(|&c| c == path[0]).unwrap();
4261 assert_eq!(
4262 q_chosen[0], q_rows[ci],
4263 "q_chosen must be the recorded row prob of the drawn candidate"
4264 );
4265 assert!(
4266 (q_rows[0] - q1).abs() < 1e-4,
4267 "slot-0 tok1 prob must be softmax(scores/T), got {} want {q1}",
4268 q_rows[0]
4269 );
4270 }
4271 }
4272
4273 #[test]
4274 fn sampled_walk_chains_the_drawn_candidate() {
4275 // The chain conditions on the DRAWN candidate, not the argmax: forcing the
4276 // low-prob slot-0 candidate (tok 2) flips slot 1's winner (tok 4 over tok 3),
4277 // exactly like the greedy chain test — a walk that seeds every slot from the
4278 // anchor (or the argmax) fails here.
4279 let (pred, succ) = books();
4280 let cand: Vec<u32> = vec![1, 2, 3, 4];
4281 let hproj = vec![1.0f32; 2 * R];
4282 let mut seq = vec![0.99f32, 0.01].into_iter();
4283 let mut u = move || seq.next().unwrap();
4284 let (path, _, _) = dflash2_walk_sampled(
4285 &pred, &succ, V, R, K, &[0.0; 4], &cand, &hproj, 0, 2, 2.0, &mut u,
4286 );
4287 assert_eq!(path[0], 2, "u=0.99 must draw the low-prob candidate");
4288 assert_eq!(
4289 path[1], 4,
4290 "slot 1 must walk from pred[2] (the DRAWN token), which scores tok4 at 10 \
4291 — chaining from the anchor or the argmax picks tok3"
4292 );
4293 }
4294
4295 #[test]
4296 fn rejection_accept_walk_is_the_leviathan_rule() {
4297 // accept while u*q < p, strict, prefix-stop at the first reject.
4298 assert_eq!(
4299 rejection_accept_len(&[0.5, 0.5], &[0.5, 0.5], &[0.9, 0.9]),
4300 2
4301 );
4302 assert_eq!(
4303 rejection_accept_len(&[0.5, 0.5], &[0.5, 0.5], &[1.0, 0.0]),
4304 0
4305 );
4306 // u*q == p is a REJECT (strict <) — the frspec test byte-for-byte.
4307 assert_eq!(rejection_accept_len(&[0.25], &[0.5], &[0.5]), 0);
4308 // q == 0 with p > 0 accepts unconditionally (the skey exactness signature).
4309 assert_eq!(rejection_accept_len(&[1e-6], &[0.0], &[0.999]), 1);
4310 // prefix stop: slot 1 rejects, slot 2 never tested.
4311 assert_eq!(
4312 rejection_accept_len(&[0.9, 0.0, 0.9], &[0.1, 0.9, 0.1], &[0.5, 0.5, 0.5]),
4313 1
4314 );
4315 }
4316
4317 // ---- round composition: the committed-token distribution must equal the target p ----
4318 // CPU mirror of the shipped rule for the FIRST post-anchor slot: draft x ~ q, accept
4319 // iff u*q(x) < p(x) (rejection_accept_len — the shipped fn), else commit a residual
4320 // sample ~ norm(max(0, p - q)). The marginal of the committed token is exactly p —
4321 // for ANY q — which is the whole correctness claim of the route's sampled admission.
4322
4323 fn tv(a: &[f64], b: &[f64]) -> f64 {
4324 a.iter().zip(b).map(|(x, y)| (x - y).abs()).sum::<f64>() / 2.0
4325 }
4326
4327 /// One composed trial with an injectable accept rule; returns the committed token.
4328 fn compose_once(
4329 p: &[f32],
4330 q: &[f32],
4331 u_draw: f32,
4332 u_accept: f32,
4333 u_resid: f32,
4334 invert_accept: bool,
4335 skip_q_in_residual: bool,
4336 ) -> usize {
4337 let n = p.len();
4338 // draft ~ q (CDF walk, the walk_sampled convention)
4339 let mut acc = 0f64;
4340 let mut x = n - 1;
4341 for (i, &qi) in q.iter().enumerate() {
4342 acc += qi as f64;
4343 if (u_draw as f64) < acc {
4344 x = i;
4345 break;
4346 }
4347 }
4348 let accepted = if invert_accept {
4349 !((u_accept as f64) * (q[x] as f64) < p[x] as f64)
4350 } else {
4351 rejection_accept_len(&p[x..=x], &q[x..=x], &[u_accept]) == 1
4352 };
4353 if accepted {
4354 return x;
4355 }
4356 // residual ~ norm(max(0, p - q)) (the device kernel's fixed-order CDF walk)
4357 let r: Vec<f64> = p
4358 .iter()
4359 .zip(q)
4360 .map(|(&pi, &qi)| {
4361 let qq = if skip_q_in_residual { 0.0 } else { qi as f64 };
4362 (pi as f64 - qq).max(0.0)
4363 })
4364 .collect();
4365 let total: f64 = r.iter().sum();
4366 let mut acc = 0f64;
4367 let target = u_resid as f64 * total;
4368 for (i, &ri) in r.iter().enumerate() {
4369 acc += ri;
4370 if acc >= target && ri > 0.0 {
4371 return i;
4372 }
4373 }
4374 n - 1
4375 }
4376
4377 fn compose_tv(q: &[f32], invert_accept: bool, skip_q_in_residual: bool) -> f64 {
4378 // target p: a spread-out 8-token distribution
4379 let p: Vec<f32> = vec![0.30, 0.22, 0.15, 0.12, 0.09, 0.06, 0.04, 0.02];
4380 let trials = 200_000usize;
4381 let mut counts = vec![0f64; V];
4382 for t in 0..trials {
4383 // three independent uniforms per trial off the host Philox stream
4384 let u_draw = crate::spec::host_u01(7, (t * 3) as u32);
4385 let u_accept = crate::spec::host_u01(7, (t * 3 + 1) as u32);
4386 let u_resid = crate::spec::host_u01(7, (t * 3 + 2) as u32);
4387 counts[compose_once(
4388 &p,
4389 q,
4390 u_draw,
4391 u_accept,
4392 u_resid,
4393 invert_accept,
4394 skip_q_in_residual,
4395 )] += 1.0;
4396 }
4397 let emp: Vec<f64> = counts.iter().map(|c| c / trials as f64).collect();
4398 let pf: Vec<f64> = p.iter().map(|&v| v as f64).collect();
4399 tv(&emp, &pf)
4400 }
4401
4402 #[test]
4403 fn sampled_round_composition_matches_the_target() {
4404 // Monte-Carlo floor at 200k draws over 8 tokens ~ 0.004 TV; bound 0.01.
4405 // (a) full-vocab q (the Rows families' shape), far from p;
4406 let q_rows: Vec<f32> = vec![0.02, 0.04, 0.06, 0.09, 0.12, 0.15, 0.22, 0.30];
4407 // (b) SPARSE candidate-set q (the DFlash2 selector shape: support on 2 of 8).
4408 let q_sparse: Vec<f32> = vec![0.0, 0.7, 0.0, 0.3, 0.0, 0.0, 0.0, 0.0];
4409 for (name, q) in [("rows", &q_rows), ("sparse", &q_sparse)] {
4410 let d = compose_tv(q, false, false);
4411 assert!(
4412 d < 0.01,
4413 "composition[{name}]: committed-token distribution must equal p \
4414 (TV {d:.4} >= 0.01)"
4415 );
4416 }
4417 }
4418
4419 #[test]
4420 fn composition_teeth_inverted_accept_fails() {
4421 // DECISIVE teeth: the same harness with the accept inequality inverted must
4422 // MISS the target — otherwise the composition gate is vacuous.
4423 let q: Vec<f32> = vec![0.02, 0.04, 0.06, 0.09, 0.12, 0.15, 0.22, 0.30];
4424 let d = compose_tv(&q, true, false);
4425 assert!(
4426 d > 0.05,
4427 "inverted accept rule must fail the composition bound (TV {d:.4})"
4428 );
4429 }
4430
4431 #[test]
4432 fn composition_teeth_residual_without_q_fails() {
4433 // Sampling the reject slot from p instead of norm(max(0, p-q)) double-counts
4434 // the overlap mass min(p,q) — the committed distribution leaves p.
4435 let q: Vec<f32> = vec![0.0, 0.7, 0.0, 0.3, 0.0, 0.0, 0.0, 0.0];
4436 let d = compose_tv(&q, false, true);
4437 assert!(
4438 d > 0.05,
4439 "residual that skips the q subtraction must fail the bound (TV {d:.4})"
4440 );
4441 }
4442
4443 // ---- PENALIZED round composition (lane/dspark-penalized-sampled-20260821) ----
4444 // Multi-slot rounds where the penalty state EVOLVES within the round. The base trunk
4445 // logits are state-independent, so ALL context dependence flows through penalties —
4446 // the sharpest fixture for "verify row j's target is penalized by the tokens accepted
4447 // before j in the same round", and the proposal concentrates on ONE token, so the
4448 // dominant drafted block is a self-hit (a drafted token penalizing its own successor
4449 // — the within-round case a frozen round-start window cannot see). The reference is
4450 // EXACT (analytic chain p1(a)·p2(b|a), the plain sampler's semantics); the spec arm
4451 // is the shipped round rule — chain draw from q, `rejection_accept_len`, residual
4452 // norm(max(0, p−q)) at the reject slot, bonus from the one-past row on full accept —
4453 // with per-slot penalized p (mirroring penalize_logits_rows_inc_f32's window rule).
4454
4455 const PV: usize = 6;
4456 const PEN_REP: f32 = 1.6;
4457 const PEN_FREQ: f32 = 0.8;
4458 const PEN_PRESENT: f32 = 1.2;
4459
4460 fn pen_base() -> Vec<f32> {
4461 vec![1.5, 0.8, 0.3, -0.2, -0.7, -1.2]
4462 }
4463
4464 /// The proposal: heavy on token 0 so drafted blocks repeat it (the self-hit case).
4465 fn pen_q() -> Vec<f32> {
4466 vec![0.85, 0.06, 0.04, 0.03, 0.01, 0.01]
4467 }
4468
4469 /// CPU mirror of penalize_logits_f32 / the plain sampler's apply_penalties: first
4470 /// occurrence does the whole adjustment, cnt = occurrences in the window, rep
4471 /// divides positive logits and multiplies negative ones.
4472 fn pen_apply(logits: &mut [f32], window: &[u32]) {
4473 let mut seen: Vec<u32> = Vec::new();
4474 for &id in window {
4475 if seen.contains(&id) {
4476 continue;
4477 }
4478 seen.push(id);
4479 let cnt = window.iter().filter(|&&h| h == id).count() as f32;
4480 let v = &mut logits[id as usize];
4481 if *v > 0.0 {
4482 *v /= PEN_REP;
4483 } else {
4484 *v *= PEN_REP;
4485 }
4486 *v -= PEN_FREQ * cnt + PEN_PRESENT;
4487 }
4488 }
4489
4490 /// Penalized target at history `window` (temp 1.0, no truncation filters — those are
4491 /// orthogonal and covered by the unpenalized composition tests + kernel oracles).
4492 fn pen_target(window: &[u32]) -> Vec<f64> {
4493 let mut l = pen_base();
4494 pen_apply(&mut l, window);
4495 let mx = l.iter().cloned().fold(f32::NEG_INFINITY, f32::max) as f64;
4496 let ex: Vec<f64> = l.iter().map(|&v| ((v as f64) - mx).exp()).collect();
4497 let z: f64 = ex.iter().sum();
4498 ex.iter().map(|v| v / z).collect()
4499 }
4500
4501 /// Which penalty-arm mutation the harness runs — `None` is the shipped rule.
4502 #[derive(Clone, Copy, PartialEq)]
4503 enum PenMutation {
4504 None,
4505 /// All rows (walk + bonus) penalized with the ROUND-START window only — the
4506 /// within-round update dropped (the frspec per-round posture; what a flat
4507 /// `penalize_logits_rows` launch would ship).
4508 FrozenWindow,
4509 /// Reject-slot residual computed from the UNPENALIZED p (a raw-tlogits column
4510 /// copy instead of the penalized buffer).
4511 UnpenalizedResidual,
4512 /// Full-accept bonus drawn from the UNPENALIZED one-past row.
4513 UnpenalizedBonus,
4514 }
4515
4516 /// Emit `want` committed tokens through spec rounds of `k` drafts (the shipped round
4517 /// rule, penalty-aware) and return them. `hist0` = the pre-stream window (the prompt
4518 /// seed); uniforms come off the injected stream.
4519 fn pen_round_stream(
4520 hist0: &[u32],
4521 k: usize,
4522 want: usize,
4523 mutation: PenMutation,
4524 next_u: &mut dyn FnMut() -> f32,
4525 ) -> Vec<u32> {
4526 let q = pen_q();
4527 let mut hist: Vec<u32> = hist0.to_vec();
4528 let mut committed: Vec<u32> = Vec::new();
4529 while committed.len() < want {
4530 // draft k tokens ~ q (fixed-order CDF walk, the walk_sampled convention)
4531 let drafted: Vec<u32> = (0..k)
4532 .map(|_| {
4533 let u = next_u() as f64;
4534 let mut acc = 0f64;
4535 let mut bi = 0usize;
4536 for (i, &qi) in q.iter().enumerate() {
4537 acc += qi as f64;
4538 if u < acc {
4539 bi = i;
4540 break;
4541 }
4542 }
4543 bi as u32
4544 })
4545 .collect();
4546 // per-slot penalized p at the drafted ids (row j's window = hist ++ drafted[..j])
4547 let pj: Vec<f32> = (0..k)
4548 .map(|j| {
4549 let win: Vec<u32> = if mutation == PenMutation::FrozenWindow {
4550 hist.clone()
4551 } else {
4552 hist.iter()
4553 .copied()
4554 .chain(drafted[..j].iter().copied())
4555 .collect()
4556 };
4557 pen_target(&win)[drafted[j] as usize] as f32
4558 })
4559 .collect();
4560 let qj: Vec<f32> = drafted.iter().map(|&d| q[d as usize]).collect();
4561 let us: Vec<f32> = (0..k).map(|_| next_u()).collect();
4562 let m = rejection_accept_len(&pj, &qj, &us);
4563 committed.extend_from_slice(&drafted[..m]);
4564 hist.extend_from_slice(&drafted[..m]);
4565 let next: u32 = if m == k {
4566 // bonus ~ p at the one-past row (window carries the WHOLE drafted block)
4567 let win: Vec<u32> = if matches!(
4568 mutation,
4569 PenMutation::FrozenWindow | PenMutation::UnpenalizedBonus
4570 ) {
4571 if mutation == PenMutation::UnpenalizedBonus {
4572 Vec::new() // raw row: no penalties at all
4573 } else {
4574 hist[..hist.len() - m].to_vec() // round-start window
4575 }
4576 } else {
4577 hist.clone()
4578 };
4579 let p = pen_target(&win);
4580 let u = next_u() as f64;
4581 let mut acc = 0f64;
4582 let mut bi = PV - 1;
4583 for (i, &pi) in p.iter().enumerate() {
4584 acc += pi;
4585 if u < acc {
4586 bi = i;
4587 break;
4588 }
4589 }
4590 bi as u32
4591 } else {
4592 // residual ~ norm(max(0, p_m − q)) at the reject slot's state
4593 let win: Vec<u32> = match mutation {
4594 PenMutation::UnpenalizedResidual => Vec::new(),
4595 PenMutation::FrozenWindow => hist[..hist.len() - m].to_vec(),
4596 _ => hist.clone(),
4597 };
4598 let p = pen_target(&win);
4599 let r: Vec<f64> = p
4600 .iter()
4601 .zip(&q)
4602 .map(|(&pi, &qi)| (pi - qi as f64).max(0.0))
4603 .collect();
4604 let total: f64 = r.iter().sum();
4605 let target = next_u() as f64 * total;
4606 let mut acc = 0f64;
4607 let mut bi = PV - 1;
4608 for (i, &ri) in r.iter().enumerate() {
4609 acc += ri;
4610 if acc >= target && ri > 0.0 {
4611 bi = i;
4612 break;
4613 }
4614 }
4615 bi as u32
4616 };
4617 committed.push(next);
4618 hist.push(next);
4619 }
4620 committed.truncate(want);
4621 committed
4622 }
4623
4624 /// Joint TV of the spec arm's first two committed tokens vs the EXACT penalized
4625 /// chain p1(a)·p2(b|a) — the plain sampler's distribution over the same two steps.
4626 fn pen_compose_tv(hist0: &[u32], k: usize, mutation: PenMutation) -> f64 {
4627 let trials = 300_000usize;
4628 let mut counts = vec![0f64; PV * PV];
4629 for t in 0..trials {
4630 // stride 64: a k<=2 round consumes <=2k+1 uniforms, <=2 rounds per trial
4631 let mut ctr = (t as u32) * 64;
4632 let mut next_u = move || {
4633 let u = crate::spec::host_u01(11, ctr);
4634 ctr = ctr.wrapping_add(1);
4635 u
4636 };
4637 let s = pen_round_stream(hist0, k, 2, mutation, &mut next_u);
4638 counts[s[0] as usize * PV + s[1] as usize] += 1.0;
4639 }
4640 let p1 = pen_target(hist0);
4641 let mut tv = 0f64;
4642 for a in 0..PV {
4643 let mut w: Vec<u32> = hist0.to_vec();
4644 w.push(a as u32);
4645 let p2 = pen_target(&w);
4646 for b in 0..PV {
4647 let refp = p1[a] * p2[b];
4648 tv += (counts[a * PV + b] / trials as f64 - refp).abs();
4649 }
4650 }
4651 tv / 2.0
4652 }
4653
4654 #[test]
4655 fn penalized_round_composition_matches_the_penalized_chain() {
4656 // MC floor at 300k trials over 36 cells ~ 0.004 TV; bound 0.01. Fixture (a):
4657 // k=2, empty prompt window — the drafted pair (0,0) dominates, so slot 2's
4658 // accept is the SELF-HIT case (its own predecessor was drafted this round).
4659 // Fixture (b): k=1, prompt window [1,1] — the bonus is the successor of a
4660 // same-round accepted draft, and cnt>1 exercises the freq×count path.
4661 for (name, hist0, k) in [
4662 ("k2-selfhit", vec![], 2usize),
4663 ("k1-bonus-successor", vec![1u32, 1u32], 1usize),
4664 ] {
4665 let d = pen_compose_tv(&hist0, k, PenMutation::None);
4666 eprintln!("penalized composition[{name}]: TV {d:.4} (bound 0.01)");
4667 assert!(
4668 d < 0.01,
4669 "penalized composition[{name}]: committed-token distribution must equal \
4670 the penalized chain (TV {d:.4} >= 0.01)"
4671 );
4672 }
4673 }
4674
4675 #[test]
4676 fn penalized_composition_teeth_frozen_window_fails() {
4677 // DECISIVE teeth: penalizing every verify row with the ROUND-START window —
4678 // dropping the within-round penalty update, i.e. a flat penalize_logits_rows
4679 // launch where the route ships penalize_logits_rows_inc — must MISS the
4680 // penalized chain, or the composition gate cannot see the one thing this lane
4681 // adds over the frozen-window prior art.
4682 let d = pen_compose_tv(&[], 2, PenMutation::FrozenWindow);
4683 eprintln!("penalized teeth[frozen-window]: TV {d:.4} (must exceed 0.05)");
4684 assert!(
4685 d > 0.05,
4686 "within-round penalty update dropped (frozen round-start window) must FAIL \
4687 the composition bound (TV {d:.4})"
4688 );
4689 }
4690
4691 #[test]
4692 fn penalized_composition_teeth_unpenalized_residual_fails() {
4693 // The reject-slot residual must read the PENALIZED column: a raw-tlogits column
4694 // copy (p_raw − q) commits from the wrong measure. Non-empty prompt window so
4695 // even round-start reject slots hit the mutation (an empty-window fixture only
4696 // sees it on within-round rejects and the margin thins to ~0.055).
4697 let d = pen_compose_tv(&[1, 1], 2, PenMutation::UnpenalizedResidual);
4698 eprintln!("penalized teeth[unpenalized-residual]: TV {d:.4} (must exceed 0.05)");
4699 assert!(
4700 d > 0.05,
4701 "residual computed from the unpenalized p must FAIL the composition bound \
4702 (TV {d:.4})"
4703 );
4704 }
4705
4706 #[test]
4707 fn penalized_composition_teeth_unpenalized_bonus_fails() {
4708 // The full-accept bonus row must carry the whole drafted block in its window:
4709 // a raw one-past row draw commits the unpenalized measure right after a
4710 // same-round accept.
4711 let d = pen_compose_tv(&[1, 1], 1, PenMutation::UnpenalizedBonus);
4712 eprintln!("penalized teeth[unpenalized-bonus]: TV {d:.4} (must exceed 0.05)");
4713 assert!(
4714 d > 0.05,
4715 "bonus drawn from the unpenalized one-past row must FAIL the composition \
4716 bound (TV {d:.4})"
4717 );
4718 }
4719}
4720
4721#[cfg(test)]
4722mod dspark_harvest_tests {
4723 use super::{DsparkHarvest, DsparkVtPolicy, dspark_accept_prefix, dspark_strategy_census};
4724
4725 const B: usize = 7; // q38 arm-a block_size
4726
4727 #[test]
4728 fn dspark_strategy_requires_shifted_harvest() {
4729 let h = DsparkHarvest::Dspark;
4730 assert_eq!(
4731 h.first_row(),
4732 0,
4733 "DSPARK-strategy checkpoints (SpecForge OnlineDSparkModel, \
4734 training.strategy=dspark — the q38 arm-a export) supervise ALL rows with \
4735 SHIFTED labels: label_offsets = arange(1, block_size+1), i.e. the ANCHOR \
4736 row's output is draft 1 (specforge/algorithms/common/\
4737 dflash_family_model.py:816; sglang v0.5.17 dspark_draft.py:248,260). \
4738 Harvesting from row 1 re-opens the DSPARK-POSTMORTEM-20260820 slot \
4739 misalignment (accept 2.9 -> 1.43)."
4740 );
4741 assert_eq!(
4742 h.n_drafts(B),
4743 B,
4744 "DSpark harvests gamma = block_size drafts per round (sglang \
4745 dspark_config.py:269, verify_num_draft_tokens = gamma+1); b-1 is the \
4746 DFlash mask-fill count and drops the best-trained slot \
4747 (DSPARK-POSTMORTEM-20260820.md §3-H1)."
4748 );
4749 for row in 0..B {
4750 assert_eq!(
4751 h.trained_offset_of_row(row),
4752 row + 1,
4753 "OnlineDSparkModel trains row k to predict anchor+k+1 \
4754 (dflash_family_model.py:816); a same-position (mask-fill) mapping \
4755 here verifies every slot one position early — the postmortem's \
4756 collapse."
4757 );
4758 }
4759 }
4760
4761 #[test]
4762 fn dflash_strategy_keeps_mask_fill_harvest() {
4763 // Guards the reverse mutation: z-lab dflash checkpoints (the gemma arm) are
4764 // mask-fill — row k FILLS anchor+k, the anchor row is loss-excluded
4765 // (dflash_family_model.py:453-472). Shifting THEM would break the gemma arm.
4766 let h = DsparkHarvest::Dflash;
4767 assert_eq!(h.first_row(), 1, "DFlash drafts start at mask row 1");
4768 assert_eq!(h.n_drafts(B), B - 1, "DFlash harvests block_size-1 drafts");
4769 for row in 1..B {
4770 assert_eq!(h.trained_offset_of_row(row), row);
4771 }
4772 }
4773
4774 #[test]
4775 fn every_candidate_verifies_the_position_its_row_was_trained_for() {
4776 // The round's invariant: draft candidate i (1-based; verified against the
4777 // trunk's prediction for anchor+i) is filled from drafter output row
4778 // first_row + i - 1. Alignment == that row was TRAINED for offset i.
4779 for h in [DsparkHarvest::Dflash, DsparkHarvest::Dspark] {
4780 for i in 1..=h.n_drafts(B) {
4781 let row = h.first_row() + i - 1;
4782 assert_eq!(
4783 h.trained_offset_of_row(row),
4784 i,
4785 "{h:?}: candidate {i} rides row {row}, which is trained for \
4786 offset {} — harvest misaligned",
4787 h.trained_offset_of_row(row)
4788 );
4789 }
4790 }
4791 }
4792
4793 #[test]
4794 fn env_seam_parses_and_refuses() {
4795 assert_eq!(
4796 DsparkHarvest::from_env_value(None),
4797 DsparkHarvest::Dflash,
4798 "the ENV-ONLY parser keeps the historical arm; the ratified strategy-keyed \
4799 default lives in resolve_value (checkpoint census), not here"
4800 );
4801 assert_eq!(
4802 DsparkHarvest::from_env_value(Some("dspark")),
4803 DsparkHarvest::Dspark
4804 );
4805 assert_eq!(
4806 DsparkHarvest::from_env_value(Some("dflash")),
4807 DsparkHarvest::Dflash
4808 );
4809 assert!(
4810 std::panic::catch_unwind(|| DsparkHarvest::from_env_value(Some("shifted"))).is_err(),
4811 "unknown harvest values must REFUSE, not default"
4812 );
4813 assert_eq!(
4814 DsparkHarvest::from_name("dspark"),
4815 Some(DsparkHarvest::Dspark)
4816 );
4817 assert_eq!(
4818 DsparkHarvest::from_name("dflash"),
4819 Some(DsparkHarvest::Dflash)
4820 );
4821 assert_eq!(DsparkHarvest::from_name("mask-fill"), None);
4822 }
4823
4824 /// The owner-ratified default flips (2026-08-20). Each assertion names its
4825 /// evidence; mutating either resolve back to the old default fails these.
4826 #[test]
4827 fn ratified_default_harvest_is_strategy_keyed() {
4828 // DSPARK-strategy checkpoint + unset env = the shifted harvest (B1: accept
4829 // 1.38->2.41 agentic / 1.53->3.66 math, E2E ALL EXACT x5, interleaved x5).
4830 assert_eq!(
4831 DsparkHarvest::resolve_value(None, true),
4832 DsparkHarvest::Dspark,
4833 "owner-ratified 2026-08-20: unset env defaults a DSPARK-strategy \
4834 checkpoint to the shifted harvest (DSPARK-POSTMORTEM-20260820.md B1)"
4835 );
4836 // mask-fill checkpoint + unset env = the historical arm, byte-identical.
4837 assert_eq!(
4838 DsparkHarvest::resolve_value(None, false),
4839 DsparkHarvest::Dflash
4840 );
4841 assert_eq!(
4842 DsparkHarvest::resolve_value(Some(""), false),
4843 DsparkHarvest::Dflash
4844 );
4845 // Explicit env overrides the census in BOTH directions (the A/B seam).
4846 assert_eq!(
4847 DsparkHarvest::resolve_value(Some("dflash"), true),
4848 DsparkHarvest::Dflash
4849 );
4850 assert_eq!(
4851 DsparkHarvest::resolve_value(Some("dspark"), false),
4852 DsparkHarvest::Dspark
4853 );
4854 // Unknown values still REFUSE through the resolve path.
4855 assert!(
4856 std::panic::catch_unwind(|| DsparkHarvest::resolve_value(Some("shifted"), true))
4857 .is_err()
4858 );
4859 }
4860
4861 #[test]
4862 fn strategy_census_reads_the_checkpoint_not_the_env() {
4863 // The q38 arm-a export shape: both signals present.
4864 let q38 = r#"{"architectures": ["Qwen3DSparkModel"], "block_size": 7,
4865 "dflash_config": {"projector_type": "dspark", "markov_rank": 256}}"#;
4866 assert!(dspark_strategy_census(q38));
4867 // Either signal alone suffices.
4868 assert!(dspark_strategy_census(
4869 r#"{"architectures": ["Qwen3DSparkModel"]}"#
4870 ));
4871 assert!(dspark_strategy_census(
4872 r#"{"dflash_config": {"projector_type": "dspark"}}"#
4873 ));
4874 // A mask-fill DFlash export carries neither -> historical default.
4875 let dflash = r#"{"architectures": ["Qwen3DFlashModel"],
4876 "dflash_config": {"attention_mode": "gqa"}}"#;
4877 assert!(!dspark_strategy_census(dflash));
4878 assert!(!dspark_strategy_census("{}"));
4879 }
4880
4881 #[test]
4882 fn ratified_default_vt_is_confidence_slot_tau_half() {
4883 // Head-carrying checkpoint + unset env = confidence-slot tau=.5 (H4 cell 3:
4884 // the tau ladder's knee; cell 2: 93.9%/97.7% of fixed-8 accept at wall >=
4885 // the reactive ladder, exactness 11/11 ALL EXACT).
4886 assert_eq!(
4887 DsparkVtPolicy::resolve_value(None, None, None, true),
4888 DsparkVtPolicy::ConfidenceSlot { tau: 0.5 },
4889 "owner-ratified 2026-08-20: unset MEMRA_DSPARK_VT defaults to \
4890 confidence-slot tau=.5 on a head-carrying checkpoint (H4 cells 2-3)"
4891 );
4892 // tau env still steers the default arm (and a bad tau still refuses).
4893 assert_eq!(
4894 DsparkVtPolicy::resolve_value(None, Some("0.35"), None, true),
4895 DsparkVtPolicy::ConfidenceSlot { tau: 0.35 }
4896 );
4897 assert!(
4898 std::panic::catch_unwind(|| DsparkVtPolicy::resolve_value(
4899 None,
4900 Some("nan-ish"),
4901 None,
4902 true
4903 ))
4904 .is_err()
4905 );
4906 // Census: no accept-rate head -> nothing to schedule with -> ladder.
4907 assert_eq!(
4908 DsparkVtPolicy::resolve_value(None, None, None, false),
4909 DsparkVtPolicy::Ladder
4910 );
4911 // MEMRA_DFLASH_ADAPT=0 is an explicit fixed-window request: honored.
4912 assert_eq!(
4913 DsparkVtPolicy::resolve_value(None, None, Some("0"), true),
4914 DsparkVtPolicy::Ladder
4915 );
4916 // Explicit values keep their exact prior semantics through resolve.
4917 assert_eq!(
4918 DsparkVtPolicy::resolve_value(Some("ladder"), None, None, true),
4919 DsparkVtPolicy::Ladder
4920 );
4921 assert_eq!(
4922 DsparkVtPolicy::resolve_value(Some("confidence"), Some("0.35"), None, true),
4923 DsparkVtPolicy::Confidence { tau: 0.35 }
4924 );
4925 // Explicit confidence mode with ADAPT=0 stays a refusal.
4926 assert!(
4927 std::panic::catch_unwind(|| DsparkVtPolicy::resolve_value(
4928 Some("confidence-slot"),
4929 None,
4930 Some("0"),
4931 true
4932 ))
4933 .is_err()
4934 );
4935 }
4936
4937 /// End-to-end alignment fixture in miniature: a mock drafter whose row r argmaxes
4938 /// to token BASE + (its trained offset under the DSPARK strategy), and a mock trunk
4939 /// whose prediction for anchor+j is BASE + j. The DSpark harvest accepts the whole
4940 /// block; feeding the same drafter through the mask-fill harvest accepts ZERO —
4941 /// the postmortem's collapse reproduced as pure logic.
4942 #[test]
4943 fn dspark_trained_rows_through_mask_fill_harvest_accept_nothing() {
4944 const BASE: u32 = 1000;
4945 let anchor: u32 = BASE; // token at the round anchor position (offset 0)
4946 // trunk verify argmaxes: vam[j] = prediction for anchor offset j+1
4947 let vam: Vec<u32> = (1..=B as u32 + 1).map(|j| BASE + j).collect();
4948 // drafter rows trained under the DSPARK strategy: row r predicts offset r+1
4949 let dspark_trained_row_argmax =
4950 |r: usize| BASE + DsparkHarvest::Dspark.trained_offset_of_row(r) as u32;
4951
4952 // Correct (shifted) harvest: candidate i <- row i-1.
4953 let h = DsparkHarvest::Dspark;
4954 let mut cand = vec![anchor];
4955 for i in 1..=h.n_drafts(B) {
4956 cand.push(dspark_trained_row_argmax(h.first_row() + i - 1));
4957 }
4958 let vt = h.n_drafts(B) + 1;
4959 assert_eq!(
4960 dspark_accept_prefix(&cand, &vam, vt),
4961 vt - 1,
4962 "aligned harvest must accept the full block"
4963 );
4964
4965 // Mask-fill harvest of the SAME dspark-trained drafter: candidate i <- row i,
4966 // which was trained for offset i+1 — every slot one position late.
4967 let wrong = DsparkHarvest::Dflash;
4968 let mut cand_wrong = vec![anchor];
4969 for i in 1..=wrong.n_drafts(B) {
4970 cand_wrong.push(dspark_trained_row_argmax(wrong.first_row() + i - 1));
4971 }
4972 let vt_wrong = wrong.n_drafts(B) + 1;
4973 assert_eq!(
4974 dspark_accept_prefix(&cand_wrong, &vam, vt_wrong),
4975 0,
4976 "mask-fill harvest of a dspark-trained drafter verifies every slot against \
4977 a position the row was not trained for (DSPARK-POSTMORTEM-20260820.md)"
4978 );
4979 }
4980}
4981
4982// ================= Verify-window policy gate (CPU; H4, DSPARK-POSTMORTEM-20260820.md) ===
4983// Pins the confidence-vt semantics as logic the round consumes: cumprod survival over
4984// sigmoid scores, thresholded, anchor + kept drafts, floor 2 / cap vt_cap — and the env
4985// seam's refuse-on-ambiguity. Mutating the policy (per-slot threshold instead of
4986// survival, off-by-one on the anchor, silent unknown-value fallback) fails HERE.
4987#[cfg(test)]
4988mod dspark_vt_tests {
4989 use super::{ConfidenceHead, DsparkVtPolicy, dspark_confidence_vt, dspark_slot_confidence_vt};
4990
4991 /// Pre-sigmoid logit for a target probability: sigmoid(logit(p)) == p.
4992 fn logit(p: f32) -> f32 {
4993 (p / (1.0 - p)).ln()
4994 }
4995
4996 #[test]
4997 fn confidence_vt_is_cumprod_survival_not_per_slot_threshold() {
4998 // sigmoids = [0.9, 0.8, 0.9, ...]: every PER-SLOT score clears tau=0.5, but
4999 // cumulative survival sinks below it at slot 6 (0.9, 0.72, 0.648, 0.583,
5000 // 0.525, then 0.472 < 0.5) — the window must stop where the EXPECTED
5001 // accepted-prefix stops paying, not where a slot looks locally fine.
5002 let raws: Vec<f32> = [0.9, 0.8, 0.9, 0.9, 0.9, 0.9, 0.9]
5003 .iter()
5004 .map(|&p| logit(p))
5005 .collect();
5006 assert_eq!(
5007 dspark_confidence_vt(&raws, 0.5, 8),
5008 6,
5009 "keeps 5 drafts + anchor"
5010 );
5011 // Tighter threshold closes the window sooner; looser opens it to the cap.
5012 assert_eq!(
5013 dspark_confidence_vt(&raws, 0.7, 8),
5014 3,
5015 "tau=0.7 keeps 2 drafts"
5016 );
5017 assert_eq!(
5018 dspark_confidence_vt(&raws, 0.05, 8),
5019 8,
5020 "tau→0 = full block"
5021 );
5022 }
5023
5024 #[test]
5025 fn slot_arm_truncates_at_first_low_confidence_slot() {
5026 // Owner directive (2026-08-20): submit only the longest prefix whose EVERY
5027 // slot clears tau on its own sigmoid. On the survival test's raws
5028 // ([0.9, 0.8, 0.9 x5], tau=0.5) every slot clears per-slot, so the slot arm
5029 // opens the full block where survival stopped at 6 — the two stopping
5030 // statistics must stay distinct arms.
5031 let raws: Vec<f32> = [0.9, 0.8, 0.9, 0.9, 0.9, 0.9, 0.9]
5032 .iter()
5033 .map(|&p| logit(p))
5034 .collect();
5035 assert_eq!(dspark_slot_confidence_vt(&raws, 0.5, 8), 8);
5036 assert_eq!(dspark_confidence_vt(&raws, 0.5, 8), 6);
5037 // A low-confidence tail never enters verify: [0.9, 0.9, 0.3, 0.9, ...]
5038 // truncates at slot 3 REGARDLESS of the confident slots behind it — a kept
5039 // slot after a dropped one could never commit (prefix accept rule).
5040 let tail: Vec<f32> = [0.9, 0.9, 0.3, 0.9, 0.9, 0.9, 0.9]
5041 .iter()
5042 .map(|&p| logit(p))
5043 .collect();
5044 assert_eq!(
5045 dspark_slot_confidence_vt(&tail, 0.5, 8),
5046 3,
5047 "2 drafts + anchor"
5048 );
5049 // Tighter tau keeps less.
5050 assert_eq!(
5051 dspark_slot_confidence_vt(&tail, 0.95, 8),
5052 2,
5053 "floor at tau=0.95"
5054 );
5055 }
5056
5057 #[test]
5058 fn confidence_vt_floor_and_cap() {
5059 // A hopeless round still verifies ONE draft (the draft forward is paid;
5060 // vt=1 would guarantee an empty round at the same cost class).
5061 let cold: Vec<f32> = [0.1f32, 0.1, 0.1].iter().map(|&p| logit(p)).collect();
5062 assert_eq!(
5063 dspark_confidence_vt(&cold, 0.5, 8),
5064 2,
5065 "floor = anchor + 1 draft"
5066 );
5067 assert_eq!(
5068 dspark_slot_confidence_vt(&cold, 0.5, 8),
5069 2,
5070 "slot arm same floor"
5071 );
5072 // The MEMRA_DFLASH_VERIFY_T cap still binds a confident round.
5073 let hot: Vec<f32> = vec![logit(0.99); 7];
5074 assert_eq!(dspark_confidence_vt(&hot, 0.5, 5), 5, "vt_cap binds");
5075 assert_eq!(
5076 dspark_confidence_vt(&hot, 0.5, 8),
5077 8,
5078 "full block when confident"
5079 );
5080 assert_eq!(
5081 dspark_slot_confidence_vt(&hot, 0.5, 5),
5082 5,
5083 "slot arm same cap"
5084 );
5085 // No scores (defensive): floor.
5086 assert_eq!(dspark_confidence_vt(&[], 0.5, 8), 2);
5087 assert_eq!(dspark_slot_confidence_vt(&[], 0.5, 8), 2);
5088 }
5089
5090 #[test]
5091 fn vt_policy_env_seam_parses_and_refuses() {
5092 assert_eq!(
5093 DsparkVtPolicy::from_env_value(None, None, None),
5094 DsparkVtPolicy::Ladder,
5095 "default stays the shipped ladder — the H4 arm is opt-in"
5096 );
5097 assert_eq!(
5098 DsparkVtPolicy::from_env_value(Some(""), None, None),
5099 DsparkVtPolicy::Ladder
5100 );
5101 assert_eq!(
5102 DsparkVtPolicy::from_env_value(Some("ladder"), None, Some("0")),
5103 DsparkVtPolicy::Ladder,
5104 "ladder + ADAPT=0 = the fixed-window arm, untouched"
5105 );
5106 assert_eq!(
5107 DsparkVtPolicy::from_env_value(Some("confidence"), None, None),
5108 DsparkVtPolicy::Confidence { tau: 0.5 },
5109 "tau defaults to 0.5 (raw sigmoid, no STS sidecar — postmortem §3-H4)"
5110 );
5111 assert_eq!(
5112 DsparkVtPolicy::from_env_value(Some("confidence"), Some("0.35"), Some("1")),
5113 DsparkVtPolicy::Confidence { tau: 0.35 }
5114 );
5115 assert_eq!(
5116 DsparkVtPolicy::from_env_value(Some("confidence-slot"), Some("0.6"), None),
5117 DsparkVtPolicy::ConfidenceSlot { tau: 0.6 },
5118 "the owner-directive per-slot arm parses with the same tau env"
5119 );
5120 assert!(
5121 std::panic::catch_unwind(|| DsparkVtPolicy::from_env_value(
5122 Some("confidence-slot"),
5123 None,
5124 Some("0")
5125 ))
5126 .is_err(),
5127 "confidence-slot + MEMRA_DFLASH_ADAPT=0 must REFUSE like confidence"
5128 );
5129 assert!(
5130 std::panic::catch_unwind(|| DsparkVtPolicy::from_env_value(Some("static"), None, None))
5131 .is_err(),
5132 "unknown policy values must REFUSE, not default — a typo silently \
5133 reverting the window policy invalidates an A/B"
5134 );
5135 assert!(
5136 std::panic::catch_unwind(|| DsparkVtPolicy::from_env_value(
5137 Some("confidence"),
5138 None,
5139 Some("0")
5140 ))
5141 .is_err(),
5142 "confidence + MEMRA_DFLASH_ADAPT=0 is contradictory and must REFUSE"
5143 );
5144 for bad in ["0", "1", "1.5", "-0.1", "nan"] {
5145 assert!(
5146 std::panic::catch_unwind(|| DsparkVtPolicy::from_env_value(
5147 Some("confidence"),
5148 Some(bad),
5149 None
5150 ))
5151 .is_err(),
5152 "tau={bad} must REFUSE (survival threshold lives in (0,1))"
5153 );
5154 }
5155 }
5156
5157 #[test]
5158 fn raw_score_matches_the_parity_gate_dot() {
5159 // The head is a raw linear proj over [hidden ; markov_prev_embedding] + b —
5160 // the exact stage-5 contract in dspark_q38_parity.rs.
5161 let ch = ConfidenceHead {
5162 w: vec![0.5, -1.0, 2.0, 0.25, -0.5],
5163 b: 0.125,
5164 in_dim: 5,
5165 with_markov: true,
5166 };
5167 let hidden = [1.0f32, 2.0, 3.0];
5168 let emb = [4.0f32, 8.0];
5169 let want = 0.125 + 0.5 * 1.0 - 1.0 * 2.0 + 2.0 * 3.0 + 0.25 * 4.0 - 0.5 * 8.0;
5170 assert_eq!(ch.raw_score(&hidden, Some(&emb)), want);
5171 let ch_plain = ConfidenceHead {
5172 w: vec![0.5, -1.0, 2.0],
5173 b: -0.25,
5174 in_dim: 3,
5175 with_markov: false,
5176 };
5177 let want_plain = -0.25 + 0.5 * 1.0 - 1.0 * 2.0 + 2.0 * 3.0;
5178 assert_eq!(ch_plain.raw_score(&hidden, None), want_plain);
5179 }
5180}