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