Skip to main content

cortiq_core/
format.rs

1//! CMF v2 binary container — envelope, tensor directory, mmap access.
2//!
3//! See `docs/CMF_V2_SPEC.md`. Layout summary:
4//!
5//! ```text
6//! [0x00]  magic "CMF\x01" | version u32 = 2 | flags u32 | required_features u32
7//! [0x10]  header_off/len | dir_off/len | data_off/len   (u64 LE each)
8//! [0x40]  masks_off/len  | vocab_off/len | index_off/len
9//! [0x70]  header_hash u64 | dir_hash u64   (hash64; 0 = absent, §8.1)
10//! [0x80]  header JSON → tensor directory → weight blob (4096-aligned,
11//!         tensors 64-aligned) → masks → vocab → sparse index
12//! ```
13//!
14//! That order is what a fresh write produces; readers address every
15//! section ONLY through the envelope. A tail append (§9.3,
16//! [`CmfModel::append_skill`]) extends the data section over the old
17//! masks/vocab/index and republishes header + directory after it.
18//!
19//! The tensor directory is the ONLY source of truth for the weight blob
20//! layout — there is no computable layout, by design (v1 bug class #1).
21//! Every validation failure is a hard error: no silent fallbacks.
22
23use crate::hash::hash64;
24pub use crate::knowledge::{
25    ExpertAppend, GenomeInfo, GenomeParent, LineageEvent, LookupInfo, PhiSpec, RouterPolicy,
26    Segment, SkillBound, SkillOverride, StateEffect, expert_append_state_effect,
27    ffn_replace_state_effect, trunk_hash,
28};
29use crate::mask::{MaskCatalog, TaskMask, decode_masks_section, encode_masks_section};
30use crate::quant::expected_nbytes;
31use crate::types::{ModelArch, QuantType, TensorDtype};
32use serde::{Deserialize, Serialize};
33use std::collections::HashMap;
34use std::fs::File;
35use std::io::{self, BufWriter, Seek, SeekFrom, Write};
36use std::path::{Path, PathBuf};
37
38mod append;
39pub use append::{AppendReport, LegacyBitsMigration, PendingAppend};
40
41pub const CMF_MAGIC: [u8; 4] = *b"CMF\x01";
42pub const CMF_VERSION: u32 = 2;
43pub const ENVELOPE_LEN: usize = 128;
44/// Weight blob is page-aligned for mmap.
45pub const DATA_ALIGNMENT: u64 = 4096;
46/// Every tensor inside the blob is 64-byte aligned (SIMD / cache line).
47pub const TENSOR_ALIGNMENT: u64 = 64;
48/// Tensors at least this large are additionally page-aligned to
49/// [`LARGE_TENSOR_ALIGN`], so a cold skill / MoE-expert / mask weight sits on
50/// its own page(s): the "unused weights cost 0 RSS" guarantee then holds at
51/// page granularity (a lazily-paged tensor pulls exactly its own bytes, not a
52/// neighbour's), and per-layer `madvise(WILLNEED)` covers clean ranges. Small
53/// tensors (norms, biases, 1-D f16) keep the 64-byte SIMD alignment so the
54/// padding stays negligible.
55pub const LARGE_TENSOR_MIN: u64 = 16 * 1024;
56/// Page alignment applied to large tensors. 4096 is the common page size
57/// (x86, most ARM/Android); it is a multiple of [`TENSOR_ALIGNMENT`], so
58/// existing readers — which only require `off % 64 == 0` — accept these files
59/// unchanged. Purely a writer-side, backward-compatible layout choice.
60pub const LARGE_TENSOR_ALIGN: u64 = 4096;
61/// One directory record is 56 bytes (see `.vmfc` v2).
62pub const DIR_RECORD_LEN: usize = 56;
63pub const DIR_MAX_NDIM: usize = 6;
64
65/// `required_features` bits. A reader MUST refuse a file with any bit
66/// it does not support.
67pub mod features {
68    pub const TENSOR_DIR: u32 = 1 << 0;
69    pub const BINARY_MASKS: u32 = 1 << 1;
70    pub const QUANT_2F: u32 = 1 << 2;
71    pub const DELTA_MASKS: u32 = 1 << 3;
72    pub const HOT_PACKS: u32 = 1 << 4;
73    /// FFN mask rows are stored PER VIRTUAL LAYER (physical × loops) so
74    /// a Looped Transformer can mask a neuron in one pass and keep it in
75    /// the other. A reader without this bit would slice the mask area by
76    /// the physical count and misread every row after the first pass —
77    /// silently — so the bit makes it refuse instead.
78    pub const LOOP_MASKS: u32 = 1 << 5;
79    /// The file is a STANDALONE SKILL — a partial tensor set cut against a
80    /// specific base (`SkillRecord.base_dir_hash`), not a runnable model.
81    /// Readers that predate the bit refuse the file loudly instead of
82    /// running half a network; runtimes that know it refuse to RUN it and
83    /// say `cortiq skill apply` instead.
84    pub const SKILL_FILE: u32 = 1 << 6;
85    /// Signed FWHT Prism/Bonsai matrices and activation-boundary contract.
86    pub const PRISM_HADAMARD: u32 = 1 << 7;
87    /// Explicit q2tp affine correction (`(c - 1.0) * s`) for the listed
88    /// Prism matrices.  This is deliberately separate from the transform bit
89    /// so readers cannot infer the correction from arch_name alone.
90    pub const PRISM_AFFINE: u32 = 1 << 8;
91    /// Some skill record is v2 (`kind` set): every v2 record is validated
92    /// against its genome binding and the directory; an unknown or
93    /// reserved `kind` is refused. Derived from content.
94    pub const SKILLS_V2: u32 = 1 << 9;
95    /// The file declares a router policy v2 (`header.router`): only that
96    /// `backbone_gated` policy may auto-route — a reader that would fall
97    /// back to argmin-over-skills must refuse. Derived from content.
98    pub const ROUTER_V2: u32 = 1 << 10;
99    /// Non-generative decision profile (`cortiq-decision`): an encoder,
100    /// a hashing contract and resonance skills, described by the
101    /// `decision.manifest` tensor. It is executed by the decision runtime
102    /// (`cortiq decide`, `cortiq serve`), never by the language-model
103    /// pipeline. Derived by every writer from the arch name prefix
104    /// [`super::DECISION_ARCH_PREFIX`]; a reader refuses a file whose bit
105    /// and prefix disagree.
106    pub const DECISION: u32 = 1 << 11;
107    /// The file carries a natively BOUNDED attention operator
108    /// (`LayerType::BoundedAttention` + `arch.anchor_core`): every layer's
109    /// state is a fixed-size record and the reader MUST execute the anchor
110    /// as ring + trained sinks — never as a masked full-attention layer over
111    /// a growing KV. Readers that predate the bit refuse the file loudly;
112    /// readers that know it must not let `CMF_O1_*` / `--o1` override the
113    /// operator the file declares.
114    ///
115    /// Numbered `1 << 12` since 0.8.1. Unreleased embryo-o1 builds wrote
116    /// it as `1 << 7`, which every released reader knows as
117    /// [`PRISM_HADAMARD`]; such a file is refused (bit ⇔ metadata) and must
118    /// be re-exported.
119    pub const BOUNDED_STATE: u32 = 1 << 12;
120    /// The file carries a FROZEN GENOME (`header.genome`): the reader
121    /// recomputes `trunk_hash` from the directory, arch, vocab and
122    /// tokenizer bundle and refuses the file on mismatch — the trunk cannot
123    /// change without a new genome. Derived from content by every writer.
124    ///
125    /// Numbered `1 << 13` since 0.8.1 (unreleased embryo-o1 builds: `1 << 8`,
126    /// which released readers know as [`PRISM_AFFINE`]).
127    pub const GENOME: u32 = 1 << 13;
128
129    /// Features this reader implements today.
130    pub const SUPPORTED: u32 = TENSOR_DIR
131        | BINARY_MASKS
132        | QUANT_2F
133        | LOOP_MASKS
134        | SKILL_FILE
135        | PRISM_HADAMARD
136        | PRISM_AFFINE
137        | SKILLS_V2
138        | ROUTER_V2
139        | DECISION
140        | BOUNDED_STATE
141        | GENOME;
142}
143
144/// Arch-name prefix of every decision profile (`cortiq-decision-ph-v1`,
145/// `cortiq-decision-overlay-v1`, …). The [`features::DECISION`] bit and
146/// this prefix travel together: writers derive the bit from the name,
147/// readers refuse a file where they disagree.
148pub const DECISION_ARCH_PREFIX: &str = "cortiq-decision-";
149
150/// Is `arch_name` a decision profile (see [`DECISION_ARCH_PREFIX`])?
151pub fn is_decision_profile(arch_name: &str) -> bool {
152    arch_name.starts_with(DECISION_ARCH_PREFIX)
153}
154
155/// The Prism bits a header demands: `PRISM_HADAMARD` with the
156/// `prism_hadamard` record, `PRISM_AFFINE` with its `affine` descriptor.
157fn prism_bits(arch: &ModelArch) -> u32 {
158    let mut f = 0;
159    if let Some(prism) = arch.prism_hadamard.as_ref() {
160        f |= features::PRISM_HADAMARD;
161        if prism.affine.is_some() {
162            f |= features::PRISM_AFFINE;
163        }
164    }
165    f
166}
167
168/// Extra words for a Prism bit/metadata mismatch on a file that carries an
169/// Embryo record: an unreleased embryo-o1 build wrote `BOUNDED_STATE` and
170/// `GENOME` on bits 7/8, which are the Prism bits of every released reader.
171fn legacy_embryo_bits_hint(header: &CmfHeader) -> &'static str {
172    if header.arch.anchor_core.is_some() || header.genome.is_some() {
173        " (this file carries anchor_core/genome: it was written by a pre-0.8.1 embryo-o1 \
174         build that numbered BOUNDED_STATE/GENOME as bits 7/8 — they are 12/13 now. \
175         Migrate it in place with `cortiq migrate-embryo-bits <file>`: it moves only those two \
176         envelope bits, which neither header_hash/dir_hash nor trunk_hash cover, and keeps \
177         every appended record. Do NOT re-export a file with appended records or a lineage — \
178         a re-export drops them. A detached .sig signs the whole file: re-sign after migrating)"
179    } else {
180        ""
181    }
182}
183
184/// The bits a pre-0.8.1 embryo-o1 build wrote for `BOUNDED_STATE` and
185/// `GENOME` (now [`features::BOUNDED_STATE`] / [`features::GENOME`]). They
186/// collide with the released `PRISM_HADAMARD` / `PRISM_AFFINE`.
187pub mod legacy_embryo_bits {
188    pub const BOUNDED_STATE: u32 = 1 << 7;
189    pub const GENOME: u32 = 1 << 8;
190    pub const BOTH: u32 = BOUNDED_STATE | GENOME;
191}
192
193/// The envelope `required_features` and the raw header JSON of a CMF
194/// file, read with no feature gate and no validation — for tooling that
195/// must recognise a file every reader refuses (overwrite guards, the
196/// legacy-bit migration). `Ok(None)`: not a CMF file (too short or wrong
197/// magic).
198pub fn peek_header(path: impl AsRef<Path>) -> Result<Option<(u32, serde_json::Value)>, CmfError> {
199    use std::io::Read;
200    let path = path.as_ref();
201    let mut f = File::open(path)?;
202    let len = f.metadata()?.len();
203    if len < ENVELOPE_LEN as u64 {
204        return Ok(None);
205    }
206    let mut head = [0u8; ENVELOPE_LEN];
207    f.read_exact(&mut head)?;
208    if head[0..4] != CMF_MAGIC {
209        return Ok(None);
210    }
211    let u64le = |o: usize| u64::from_le_bytes(head[o..o + 8].try_into().unwrap());
212    let required = u32::from_le_bytes(head[12..16].try_into().unwrap());
213    let (off, n) = (u64le(0x10), u64le(0x18));
214    if n == 0 || n > (1 << 30) || off.checked_add(n).is_none_or(|end| end > len) {
215        return Err(CmfError::Bounds(format!(
216            "{}: header section [{off}, +{n}) lies outside the file ({len} bytes)",
217            path.display()
218        )));
219    }
220    let mut buf = vec![0u8; n as usize];
221    f.seek(SeekFrom::Start(off))?;
222    f.read_exact(&mut buf)?;
223    let v = serde_json::from_slice(&buf)
224        .map_err(|e| CmfError::Parse(format!("{}: header JSON: {e}", path.display())))?;
225    Ok(Some((required, v)))
226}
227
228/// Does a raw header carry a frozen genome, a lineage journal or appended
229/// data segments? Keyed on CONTENT, not on the envelope bit: a pre-0.8.1
230/// embryo-o1 file carries `GENOME` on bit 8, and a file with appended
231/// segments loses them on any rewrite.
232pub fn header_json_is_genome(header: &serde_json::Value) -> bool {
233    let present = |k: &str| {
234        header
235            .get(k)
236            .is_some_and(|x| !x.is_null() && x.as_array().is_none_or(|a| !a.is_empty()))
237    };
238    present("genome") || present("lineage") || present("segments")
239}
240
241fn validate_prism_affine_targets(
242    arch: &ModelArch,
243    tensors: &[TensorEntry],
244) -> Result<(), CmfError> {
245    let Some(prism) = arch.prism_hadamard.as_ref() else {
246        return Ok(());
247    };
248    let Some(affine) = prism.affine.as_ref() else {
249        return Ok(());
250    };
251    affine.validate().map_err(CmfError::Parse)?;
252    for name in &affine.target_names {
253        let matches = tensors
254            .iter()
255            .filter(|t| t.name == *name)
256            .collect::<Vec<_>>();
257        if matches.len() != 1 {
258            return Err(CmfError::Parse(format!(
259                "Prism affine target '{}' appears {} times (expected exactly once)",
260                name,
261                matches.len()
262            )));
263        }
264        let t = matches[0];
265        if t.dtype != TensorDtype::Q2TiledP
266            || t.shape.len() != 2
267            || t.shape[1] == 0
268            || t.shape[1] % 32 != 0
269        {
270            return Err(CmfError::Parse(format!(
271                "Prism affine target '{}' must be dtype q2tp [rows, positive cols multiple of 32], got {:?}{:?}",
272                name, t.dtype, t.shape
273            )));
274        }
275    }
276    Ok(())
277}
278
279/// JSON header — architecture and provenance (human-readable part;
280/// machine-critical data lives in binary sections).
281#[derive(Debug, Clone, Serialize, Deserialize)]
282pub struct CmfHeader {
283    #[serde(default = "default_format")]
284    pub format: String,
285    pub version: u32,
286    pub arch: ModelArch,
287    /// Informational default; per-tensor truth is in the directory.
288    pub quant_type: QuantType,
289    #[serde(default, skip_serializing_if = "Option::is_none")]
290    pub provenance: Option<serde_json::Value>,
291    /// Chat/eos bundle (spec §6.1): the file — not the binary — defines
292    /// chat behavior. Additive: absent in older files.
293    #[serde(default, skip_serializing_if = "Option::is_none")]
294    pub tokenizer_config: Option<TokenizerBundle>,
295    /// Section-level integrity (spec §8.1): hex hash64 of the raw bytes
296    /// of the optional sections. header/dir hashes live in the envelope
297    /// reserved bytes — JSON cannot protect the JSON that carries it.
298    #[serde(default, skip_serializing_if = "Option::is_none")]
299    pub section_hashes: Option<SectionHashes>,
300    /// Per-skill records (spec §9): replacement tensors live in the
301    /// directory as `skill.{id}.{name}`; this registry carries the
302    /// selection descriptor and the honest quality contract.
303    #[serde(default, skip_serializing_if = "Vec::is_empty")]
304    pub skills: Vec<SkillRecord>,
305    /// Sharding (spec §10): this file is shard `no` of `count`; every
306    /// shard is a standalone valid .cmf carrying a tensor subset.
307    /// Naming convention: `…-{no:05}-of-{count:05}.cmf`.
308    #[serde(default, skip_serializing_if = "Option::is_none")]
309    pub shard: Option<ShardInfo>,
310    /// Measured confidence calibration (B1): a temperature fit on held-out
311    /// so the displayed softmax confidence is a true property of the
312    /// model (softmax(logits/T)), not a raw estimate. Additive; absent =
313    /// use raw (T=1). Written by `set_calibration.py` after `cortiq
314    /// calibrate` measures the reliability/ECE.
315    #[serde(default, skip_serializing_if = "Option::is_none")]
316    pub calibration: Option<Calibration>,
317    /// Skill-router calibration (spec §9 routing; the cortiq-router recipe):
318    /// temperature of the softmax over −error and the novelty threshold θ,
319    /// fitted on the skills' held-out φ samples carried in their descriptors.
320    /// Additive; absent = raw recon-argmin with a fixed E threshold.
321    #[serde(default, skip_serializing_if = "Option::is_none")]
322    pub routing: Option<RoutingCalibration>,
323    /// Frozen genome (bit `GENOME`, spec §9.2): identity + trunk hash of
324    /// the backbone every v2 skill binds to.
325    #[serde(default, skip_serializing_if = "Option::is_none")]
326    pub genome: Option<GenomeInfo>,
327    /// Append-only journal of the genome's life (spec §9.2).
328    #[serde(default, skip_serializing_if = "Vec::is_empty")]
329    pub lineage: Vec<LineageEvent>,
330    /// Router policy v2 (bit `ROUTER_V2`, spec §9.4): per-request
331    /// backbone-gated routing with the backbone as the default.
332    #[serde(default, skip_serializing_if = "Option::is_none")]
333    pub router: Option<RouterPolicy>,
334    /// Data-range ownership after tail appends (spec §9.3). Layout
335    /// metadata: a full rewrite clears it.
336    #[serde(default, skip_serializing_if = "Vec::is_empty")]
337    pub segments: Vec<Segment>,
338}
339
340/// Router calibration (see `SelectionDescriptor`): the recipe of the
341/// cortiq-router service. Confidence is `softmax(−err/T)`; novelty is
342/// `0.5·σ(z_top) + 0.25·1/(1+8·margin) + 0.25·(1−confidence)`; an input is
343/// novel iff its novelty exceeds θ, the `1−fpr` quantile of the in-scope
344/// held-out novelty scores.
345#[derive(Debug, Clone, Serialize, Deserialize)]
346pub struct RoutingCalibration {
347    pub temperature: f32,
348    pub novelty_theta: f32,
349    /// held-out samples the calibration used
350    pub samples: usize,
351    /// target in-scope false-positive rate of the novelty flag
352    pub target_fpr: f32,
353}
354
355/// Confidence-calibration record (spec §6.2). `temperature` scales the
356/// logits before softmax when reporting confidence; `ece_before`/`after`
357/// are the measured Expected Calibration Error (honest provenance).
358#[derive(Debug, Clone, Serialize, Deserialize)]
359pub struct Calibration {
360    pub temperature: f32,
361    #[serde(default, skip_serializing_if = "Option::is_none")]
362    pub ece_before: Option<f32>,
363    #[serde(default, skip_serializing_if = "Option::is_none")]
364    pub ece_after: Option<f32>,
365}
366
367/// Shard coordinates (1-based, gguf-split style).
368#[derive(Debug, Clone, Serialize, Deserialize)]
369pub struct ShardInfo {
370    pub no: usize,
371    pub count: usize,
372}
373
374/// Recon-argmin routing parameters (spec §9; P1 signal-consistency):
375/// E = ‖(φ−mean) − B·Bᵀ(φ−mean)‖² / ‖φ−mean‖²; pick argmin over skills.
376#[derive(Debug, Clone, Serialize, Deserialize)]
377pub struct SelectionDescriptor {
378    /// "mse" (normalized reconstruction error over raw φ) or "mse_unit"
379    /// (φ unit-normalized first — the only metric router v2 accepts).
380    pub metric: String,
381    /// Backbone layer whose mean-pooled hidden is φ(x).
382    pub phi_layer: usize,
383    /// Subspace mean, f16 LE base64, len = hidden.
384    pub mean: String,
385    /// Orthonormal basis rows, f16 LE base64, len = rank·hidden.
386    pub basis: String,
387    pub rank: usize,
388    /// Training reconstruction-error statistics (mean, std) — the z-score
389    /// "energy" of the novelty ensemble. Additive (cortiq-router recipe).
390    #[serde(default, skip_serializing_if = "Option::is_none")]
391    pub err_mean: Option<f32>,
392    #[serde(default, skip_serializing_if = "Option::is_none")]
393    pub err_std: Option<f32>,
394    /// Held-out in-scope φ samples, f16 LE base64, len = holdout_n·hidden —
395    /// what the file-level temperature/θ calibration is fitted on, so a
396    /// container recalibrates itself after every appended skill.
397    #[serde(default, skip_serializing_if = "Option::is_none")]
398    pub holdout: Option<String>,
399    #[serde(default, skip_serializing_if = "Option::is_none")]
400    pub holdout_n: Option<usize>,
401}
402
403/// One skill of the swarm (spec §9; Patent 15 per-skill record).
404#[derive(Debug, Clone, Default, Serialize, Deserialize)]
405pub struct SkillRecord {
406    pub id: String,
407    #[serde(default, skip_serializing_if = "Option::is_none")]
408    pub name: Option<String>,
409    /// Layers this skill specializes (a proper subset).
410    #[serde(default)]
411    pub layers: Vec<usize>,
412    /// Selection descriptor for recon-argmin routing (208c, P1):
413    /// per-skill affine subspace over φ(x) = mean-pooled hidden state.
414    #[serde(default, skip_serializing_if = "Option::is_none")]
415    pub selection: Option<SelectionDescriptor>,
416    /// Optional input-mask task name (208b), applied with the skill.
417    #[serde(default, skip_serializing_if = "Option::is_none")]
418    pub input_mask_task: Option<String>,
419    /// Measured quality (claim 16): overlaid vs backbone, held-out.
420    #[serde(default, skip_serializing_if = "Option::is_none")]
421    pub quality: Option<serde_json::Value>,
422
423    // ── Standalone-skill-file keys (features::SKILL_FILE) ──
424    /// hex hash64 of the BASE model's tensor directory this skill was cut
425    /// against. `skill apply` refuses a base whose directory hash differs:
426    /// a skill is a delta against exact bytes, not an architecture.
427    #[serde(default, skip_serializing_if = "Option::is_none")]
428    pub base_dir_hash: Option<String>,
429    /// The base's architecture name — the cheap human-readable identity
430    /// check next to the exact one above.
431    #[serde(default, skip_serializing_if = "Option::is_none")]
432    pub base_arch: Option<String>,
433    /// Mask-catalog task this skill activates once applied.
434    #[serde(default, skip_serializing_if = "Option::is_none")]
435    pub task: Option<String>,
436    /// Free-form provenance: corpus, steps, recipe, who baked it.
437    #[serde(default, skip_serializing_if = "Option::is_none")]
438    pub provenance: Option<serde_json::Value>,
439
440    // ── v2 record keys (features::SKILLS_V2, spec §9.5) ──
441    /// Record kind: "ffn_replace" and "expert_append" (implemented).
442    /// Reserved and refused until implemented: "anchor_sinks", "mask_only",
443    /// "fcd_vector". `None` = a v1 record (unchanged semantics).
444    #[serde(default, skip_serializing_if = "Option::is_none")]
445    pub kind: Option<String>,
446    /// `expert_append` parameters (spec §9.5.1): `count` new experts in
447    /// each of `layers`, their descriptor `rank` and the `shell_quantile`
448    /// the trainer used. Required for that kind, refused on any other.
449    #[serde(default, skip_serializing_if = "Option::is_none")]
450    pub experts: Option<ExpertAppend>,
451    /// `lookup` parameters (spec §9.5.2): the explicit key → card table
452    /// under `skill.{id}.lookup.*` — entries, keys, key normalisation,
453    /// languages, fields. Required for that kind, refused on any other.
454    #[serde(default, skip_serializing_if = "Option::is_none")]
455    pub lookup: Option<LookupInfo>,
456    /// Base tensors this record replaces, with the hash of the base entry
457    /// it was trained against. Exactly one per `skill.{id}.X` tensor.
458    #[serde(default, skip_serializing_if = "Vec::is_empty")]
459    pub overrides: Vec<SkillOverride>,
460    /// The genome (id, generation, f32-master trunk hash) the record is
461    /// bound to.
462    #[serde(default, skip_serializing_if = "Option::is_none")]
463    pub bound: Option<SkillBound>,
464    /// What switching this skill does to the sequence state (computed by
465    /// the writer from `arch.layer_types`).
466    #[serde(default, skip_serializing_if = "Option::is_none")]
467    pub state_effect: Option<StateEffect>,
468    /// "quarantine" | "active" | "stale_regate" | "retired".
469    #[serde(default, skip_serializing_if = "Option::is_none")]
470    pub status: Option<String>,
471    /// Measured gate; auto-routing requires `gate.status == "measured"`
472    /// and `status == "active"`.
473    #[serde(default, skip_serializing_if = "Option::is_none")]
474    pub gate: Option<serde_json::Value>,
475    /// Prompt template the skill was trained under (e.g. "cmf-im-v1").
476    #[serde(default, skip_serializing_if = "Option::is_none")]
477    pub prompt_contract: Option<String>,
478    /// `{trigger, dataset_sha256[], recipe, …}`.
479    #[serde(default, skip_serializing_if = "Option::is_none")]
480    pub origin: Option<serde_json::Value>,
481}
482
483/// Hex-encoded hash64 per optional section (u64 as JSON number would
484/// lose precision past 2^53).
485#[derive(Debug, Clone, Default, Serialize, Deserialize)]
486pub struct SectionHashes {
487    #[serde(default, skip_serializing_if = "Option::is_none")]
488    pub masks: Option<String>,
489    #[serde(default, skip_serializing_if = "Option::is_none")]
490    pub vocab: Option<String>,
491    #[serde(default, skip_serializing_if = "Option::is_none")]
492    pub index: Option<String>,
493}
494
495/// Chat template + generation stop tokens carried by the container.
496#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
497pub struct TokenizerBundle {
498    /// Jinja chat template (chat_template.jinja / tokenizer_config.json)
499    #[serde(default, skip_serializing_if = "Option::is_none")]
500    pub chat_template: Option<String>,
501    /// All ids that terminate generation (generation_config + im_end)
502    #[serde(default)]
503    pub eos_token_ids: Vec<u32>,
504    #[serde(default, skip_serializing_if = "Option::is_none")]
505    pub bos_token_id: Option<u32>,
506    #[serde(default, skip_serializing_if = "Option::is_none")]
507    pub pad_token_id: Option<u32>,
508}
509
510fn default_format() -> String {
511    "cmf".to_string()
512}
513
514/// One tensor directory entry.
515#[derive(Debug, Clone, PartialEq, Eq)]
516pub struct TensorEntry {
517    pub name: String,
518    pub dtype: TensorDtype,
519    pub shape: Vec<usize>,
520    /// Offset relative to the OWNING shard's `data_off`, multiple of 64.
521    pub off: u64,
522    pub nbytes: u64,
523    /// Runtime-only: which shard's mmap holds the bytes (0 for the
524    /// single-file case; not part of the 56-byte record).
525    pub shard: usize,
526    /// `hash64` of the tensor bytes.
527    pub hash: u64,
528}
529
530impl TensorEntry {
531    pub fn n_elems(&self) -> usize {
532        self.shape.iter().product()
533    }
534}
535
536/// Input for the Rust writer: one tensor with its encoded bytes.
537#[derive(Debug, Clone)]
538pub struct TensorSpec {
539    pub name: String,
540    pub dtype: TensorDtype,
541    pub shape: Vec<usize>,
542    pub data: Vec<u8>,
543}
544
545/// `TensorSpec` with a borrowed payload — see [`CmfModel::write_ref`].
546pub struct TensorSpecRef<'a> {
547    pub name: String,
548    pub dtype: TensorDtype,
549    pub shape: Vec<usize>,
550    pub data: &'a [u8],
551}
552
553/// Sparse index entry — precomputed per-task per-layer active group IDs.
554#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
555pub struct SparseIndexEntry {
556    pub task_id: u32,
557    pub layer_idx: usize,
558    /// Active quant-group indices for FFN (sorted, group = 32 neurons).
559    pub active_ffn_groups: Vec<u16>,
560    /// Active head indices for attention (sorted).
561    pub active_heads: Vec<u8>,
562}
563
564/// Section ranges parsed from the fixed envelope.
565#[derive(Debug, Clone, Copy, Default)]
566struct Envelope {
567    required_features: u32,
568    header: (u64, u64),
569    dir: (u64, u64),
570    data: (u64, u64),
571    masks: (u64, u64),
572    vocab: (u64, u64),
573    index: (u64, u64),
574    /// hash64 of the header JSON bytes (reserved [0x70]); 0 = absent.
575    header_hash: u64,
576    /// hash64 of the tensor-directory bytes (reserved [0x78]); 0 = absent.
577    dir_hash: u64,
578}
579
580enum Backing {
581    Mmap(memmap2::Mmap),
582    Owned(Vec<u8>),
583}
584
585impl Backing {
586    fn bytes(&self) -> &[u8] {
587        match self {
588            Backing::Mmap(m) => m,
589            Backing::Owned(v) => v,
590        }
591    }
592}
593
594/// A loaded CMF model: metadata owned, weights zero-copy via mmap.
595pub struct CmfModel {
596    pub path: PathBuf,
597    /// Identifies THIS open of the file, monotonically. GPU backends cache
598    /// device weights by (model, tensor); keying that on the mapping's
599    /// address made a reloaded model inherit the previous one's buffers
600    /// whenever the new mmap landed where the old had been — silent wrong
601    /// weights in any process that unloads and reloads, which a server does.
602    uid: u64,
603    pub header: CmfHeader,
604    pub required_features: u32,
605    pub tensors: Vec<TensorEntry>,
606    /// name-hash → tensor index. Keying on the hash (not the name) avoids
607    /// cloning every tensor name into the map at `open()` — that halves the
608    /// open-time allocations and the map's footprint, which matters for large
609    /// MoE / skills files with tens of thousands of tensors. A genuine 64-bit
610    /// hash collision between two *distinct* names — astronomically unlikely —
611    /// lands in `name_overflow`, so lookups stay exact.
612    by_name: HashMap<u64, u32>,
613    name_overflow: Vec<u32>,
614    pub masks: MaskCatalog,
615    pub sparse_index: Vec<SparseIndexEntry>,
616    /// Embedded tokenizer.json bytes, if present.
617    pub vocab: Option<Vec<u8>>,
618    backing: Backing,
619    data_off: u64,
620    envelope: Envelope,
621    /// Shards 2..N (spec §10): (backing, data_off) per extra file;
622    /// `TensorEntry.shard` 0 = this file, i>0 = extra_shards[i-1].
623    extra_shards: Vec<(Backing, u64)>,
624    /// Trunk hash verified at open (GENOME files), from the raw header.
625    trunk: Option<u64>,
626    /// hash64 of the raw mask / sparse-index sections (execution-relevant
627    /// parts of the trunk outside the directory; see
628    /// [`crate::knowledge::trunk_hash_exec`]).
629    exec: crate::knowledge::ExecHashes,
630}
631
632impl std::fmt::Debug for CmfModel {
633    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
634        f.debug_struct("CmfModel")
635            .field("path", &self.path)
636            .field("arch", &self.header.arch.arch_name)
637            .field("tensors", &self.tensors.len())
638            .field("masks", &self.masks.masks.len())
639            .finish()
640    }
641}
642
643/// Hands out a fresh id per open. Wraps only after 2^64 opens.
644static MODEL_UID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
645
646impl CmfModel {
647    /// A key that is unique to this open of the file and never recycled.
648    /// Backends that cache anything derived from the weights must key on
649    /// this, not on the mapping's address, which the allocator reuses.
650    pub fn uid(&self) -> u64 {
651        self.uid
652    }
653
654    /// Open and strictly validate a CMF v2 file. Any inconsistency is an
655    /// error — this function never substitutes defaults.
656    pub fn open(path: impl AsRef<Path>) -> Result<Self, CmfError> {
657        let path = path.as_ref().to_path_buf();
658        if !path.exists() {
659            return Err(CmfError::FileNotFound(path.display().to_string()));
660        }
661        let file = File::open(&path)?;
662        let file_len = file.metadata()?.len();
663
664        // `CMF_MMAP_POPULATE=1`: fault the whole file in at open (MAP_POPULATE)
665        // — for hosts whose RAM holds the file, so a device path that
666        // re-reads cold experts from the mapping every token never pays a
667        // page fault inside the token. Costs the read time once, up front.
668        let populate = std::env::var("CMF_MMAP_POPULATE").as_deref() == Ok("1");
669        let mut opts = memmap2::MmapOptions::new();
670        if populate {
671            opts.populate();
672        }
673        let backing = match unsafe { opts.map(&file) } {
674            Ok(m) => {
675                // Decode touches every weight page each token, so tell
676                // the kernel up front: WillNeed front-loads readahead
677                // (first-token page-fault storm becomes streaming I/O —
678                // this is TTFT on phones, where the file is a large
679                // share of RAM). Advisory only: a memory-pressured
680                // kernel is free to ignore it. CMF_MMAP_ADVISE=0 turns
681                // it off; CMF_MLOCK=1 additionally tries to pin the
682                // mapping (needs RLIMIT_MEMLOCK headroom — refusal is
683                // logged, not fatal).
684                #[cfg(unix)]
685                {
686                    // …unless these pages are headed for a device that will
687                    // then drop them. Reading the file ahead only to throw it
688                    // out behind the uploader has the kernel fetching the same
689                    // bytes twice. The two advices contradict each other, so
690                    // the one that asked for eviction wins.
691                    //
692                    // The conditions are the uploader's, approximated from what
693                    // is knowable at open time (the GPU is not up yet): Linux,
694                    // because `evict_ranges` is a no-op elsewhere; a backend
695                    // actually requested, because a CPU run wants the readahead;
696                    // and eviction not turned off. On UMA the mapping IS the
697                    // working copy and nothing is evicted — that is why this
698                    // cannot key on `CMF_GPU` alone.
699                    let evicting = cfg!(target_os = "linux")
700                        && std::env::var("CMF_GPU").is_ok_and(|v| v != "0" && v != "off")
701                        && std::env::var("CMF_UPLOAD_EVICT")
702                            .map(|v| v != "0")
703                            .unwrap_or(true);
704                    // WILLNEED is a whole-file readahead and macOS runs it
705                    // SYNCHRONOUSLY — on a 12.9 GB MoE file it held open()
706                    // for ~6.5 s while polluting RAM with experts that a
707                    // routed decode never touches. Small dense files keep
708                    // the readahead (it pays there); big files rely on
709                    // demand paging. CMF_MMAP_ADVISE=1 forces the old
710                    // blanket advise, =0 disables it entirely.
711                    const ADVISE_CAP: usize = 4 << 30;
712                    let advise = match std::env::var("CMF_MMAP_ADVISE").as_deref() {
713                        Ok("0") => false,
714                        Ok("1") => true,
715                        _ => m.len() <= ADVISE_CAP,
716                    };
717                    if !evicting && advise {
718                        let _ = m.advise(memmap2::Advice::WillNeed);
719                    }
720                    if std::env::var("CMF_MLOCK")
721                        .map(|v| v == "1")
722                        .unwrap_or(false)
723                    {
724                        if let Err(e) = m.lock() {
725                            tracing::warn!(
726                                "CMF_MLOCK=1: mlock refused ({e}) — continuing unpinned"
727                            );
728                        }
729                    }
730                }
731                Backing::Mmap(m)
732            }
733            Err(e) => {
734                tracing::warn!("mmap failed ({e}), reading file into memory");
735                Backing::Owned(std::fs::read(&path)?)
736            }
737        };
738
739        let env = Self::parse_envelope(backing.bytes(), file_len)?;
740
741        let bytes = backing.bytes();
742        let section = |off: u64, len: u64| -> &[u8] { &bytes[off as usize..(off + len) as usize] };
743
744        // Header JSON
745        let header: CmfHeader = serde_json::from_slice(section(env.header.0, env.header.1))
746            .map_err(|e| CmfError::Parse(format!("header JSON: {e}")))?;
747        // A decision profile and its feature bit travel together: the bit
748        // keeps generative readers away from the file, the prefix is what
749        // the decision runtime keys on. Either one alone is a file that
750        // some reader would misread.
751        let decision_bit = env.required_features & features::DECISION != 0;
752        if decision_bit != is_decision_profile(&header.arch.arch_name) {
753            return Err(CmfError::Parse(
754                "decision profile and DECISION feature bit disagree".into(),
755            ));
756        }
757        let prism_bit = env.required_features & features::PRISM_HADAMARD != 0;
758        let affine_bit = env.required_features & features::PRISM_AFFINE != 0;
759        let has_prism = header.arch.prism_hadamard.is_some();
760        let has_affine = header
761            .arch
762            .prism_hadamard
763            .as_ref()
764            .and_then(|p| p.affine.as_ref())
765            .is_some();
766        if prism_bit != has_prism {
767            return Err(CmfError::Parse(format!(
768                "required_features PRISM_HADAMARD={} disagrees with prism_hadamard metadata={has_prism}{}",
769                prism_bit,
770                legacy_embryo_bits_hint(&header)
771            )));
772        }
773        if affine_bit != has_affine {
774            return Err(CmfError::Parse(format!(
775                "required_features PRISM_AFFINE={} disagrees with affine metadata={has_affine}{}",
776                affine_bit,
777                legacy_embryo_bits_hint(&header)
778            )));
779        }
780        if let Some(prism) = header.arch.prism_hadamard.as_ref() {
781            prism.validate().map_err(CmfError::Parse)?;
782            if header.arch.arch_name != "prism_hadamard_qwen35" {
783                return Err(CmfError::Parse(
784                    "prism_hadamard metadata requires arch_name prism_hadamard_qwen35".into(),
785                ));
786            }
787        }
788        header
789            .arch
790            .validate_operator_metadata()
791            .map_err(CmfError::Parse)?;
792        // The bounded-anchor operator and its feature bit travel together:
793        // a file that declares the operator without the bit would run as
794        // full-KV on a reader that ignores unknown header fields, and a bit
795        // without the record has nothing to execute.
796        let bounded_declared = header.arch.anchor_core.is_some();
797        let bounded_bit = env.required_features & features::BOUNDED_STATE != 0;
798        if bounded_declared != bounded_bit {
799            return Err(CmfError::Parse(format!(
800                "anchor_core record ({}) and BOUNDED_STATE feature bit ({}) disagree",
801                if bounded_declared {
802                    "present"
803                } else {
804                    "absent"
805                },
806                if bounded_bit { "set" } else { "clear" }
807            )));
808        }
809
810        // Tensor directory
811        let tensors = Self::decode_directory(section(env.dir.0, env.dir.1))?;
812        validate_prism_affine_targets(&header.arch, &tensors)?;
813        for t in &tensors {
814            if t.off % TENSOR_ALIGNMENT != 0 {
815                return Err(CmfError::Bounds(format!(
816                    "tensor '{}': offset {} not 64-aligned",
817                    t.name, t.off
818                )));
819            }
820            let tensor_end = t.off.checked_add(t.nbytes).ok_or_else(|| {
821                CmfError::Bounds(format!("tensor '{}': offset + length overflows", t.name))
822            })?;
823            if tensor_end > env.data.1 {
824                return Err(CmfError::Bounds(format!(
825                    "tensor '{}': [{}, {}) exceeds data section ({} bytes)",
826                    t.name, t.off, tensor_end, env.data.1
827                )));
828            }
829            t.shape
830                .iter()
831                .try_fold(1usize, |n, &dim| n.checked_mul(dim))
832                .ok_or_else(|| {
833                    CmfError::Bounds(format!(
834                        "tensor '{}': shape product overflows usize",
835                        t.name
836                    ))
837                })?;
838            if let Some(expect) = expected_nbytes(t.dtype, &t.shape) {
839                if expect as u64 != t.nbytes {
840                    return Err(CmfError::Bounds(format!(
841                        "tensor '{}': nbytes {} != expected {} for {:?}{:?}",
842                        t.name, t.nbytes, expect, t.dtype, t.shape
843                    )));
844                }
845            }
846            // Payload-dependent lengths (vbit): exact check against the
847            // width header, bounds-before-slice (roadmap §4.9).
848            if matches!(t.dtype, TensorDtype::Vbit | TensorDtype::VbitRo) {
849                let payload = section(env.data.0 + t.off, t.nbytes);
850                crate::quant::validate_payload(t.dtype, &t.shape, payload)
851                    .map_err(|e| CmfError::Bounds(format!("tensor '{}': {e}", t.name)))?;
852            }
853        }
854        // Duplicate names would silently shadow each other in the
855        // HashMap (directory scan and by_name would disagree) — refuse
856        // the file instead (roadmap §4.9).
857        let mut by_name: HashMap<u64, u32> = HashMap::with_capacity(tensors.len());
858        let mut name_overflow: Vec<u32> = Vec::new();
859        for i in 0..tensors.len() {
860            let h = hash64(tensors[i].name.as_bytes());
861            match by_name.get(&h) {
862                Some(&j) if tensors[j as usize].name == tensors[i].name => {
863                    return Err(CmfError::Parse(format!(
864                        "duplicate tensor name '{}' in directory",
865                        tensors[i].name
866                    )));
867                }
868                Some(_) => name_overflow.push(i as u32), // hash collision of distinct names
869                None => {
870                    by_name.insert(h, i as u32);
871                }
872            }
873        }
874
875        // Masks
876        let masks = if env.masks.1 > 0 {
877            decode_masks_section(section(env.masks.0, env.masks.1), &header.arch)
878                .map_err(CmfError::Parse)?
879        } else {
880            MaskCatalog::empty()
881        };
882
883        // Vocab (tokenizer.json)
884        let vocab = if env.vocab.1 > 0 {
885            Some(section(env.vocab.0, env.vocab.1).to_vec())
886        } else {
887            None
888        };
889
890        // Sparse index
891        let sparse_index = if env.index.1 > 0 {
892            decode_sparse_index(section(env.index.0, env.index.1))?
893        } else {
894            vec![]
895        };
896
897        // Knowledge contract (§9.2–§9.5): GENOME / SKILLS_V2 / ROUTER_V2
898        // bits ⇔ content both ways, the trunk hash recomputed from the
899        // directory and the header JSON AS WRITTEN (an arch field this
900        // reader's struct does not know is still covered), every v2 skill
901        // record and the router policy. A file without the bits and without
902        // v2 content passes untouched.
903        // The mask catalog and sparse index are part of how the trunk
904        // executes (run/serve apply the catalog's fallback mask to every
905        // request): a genome's hash covers their raw bytes.
906        let exec = if header.genome.is_some() {
907            crate::knowledge::ExecHashes::of(
908                (env.masks.1 > 0).then(|| section(env.masks.0, env.masks.1)),
909                (env.index.1 > 0).then(|| section(env.index.0, env.index.1)),
910            )
911        } else {
912            crate::knowledge::ExecHashes::default()
913        };
914        let trunk = if header.genome.is_some() {
915            let raw: serde_json::Value =
916                serde_json::from_slice(section(env.header.0, env.header.1))
917                    .map_err(|e| CmfError::Parse(format!("header JSON: {e}")))?;
918            let null = serde_json::Value::Null;
919            Some(crate::knowledge::trunk_hash_json_exec(
920                raw.get("arch").unwrap_or(&null),
921                raw.get("tokenizer_config").unwrap_or(&null),
922                &tensors,
923                vocab.as_deref(),
924                exec,
925            ))
926        } else {
927            None
928        };
929        crate::knowledge::validate_knowledge(
930            &header,
931            &tensors,
932            vocab.as_deref(),
933            env.required_features,
934            Some(env.data.1),
935            trunk,
936            exec,
937        )?;
938        // A lookup policy this reader does not know (a newer writer's) is
939        // read as router_and_key by the runtime — the file stays readable.
940        for (id, p) in crate::knowledge::unknown_lookup_policies(&header) {
941            tracing::warn!(
942                "skill '{id}': lookup.policy '{p}' is unknown to this reader (knows {}) — read as \
943                 {}",
944                crate::knowledge::lookup_policy::ALL.join(" | "),
945                crate::knowledge::lookup_policy::DEFAULT
946            );
947        }
948        // expert_append descriptor values (bias 0, finite shell): a few
949        // 4-byte reads from the mapped payloads.
950        crate::knowledge::validate_expert_append_values(&header, &tensors, |t| {
951            let start = (env.data.0 + t.off) as usize;
952            bytes
953                .get(start..start + t.nbytes as usize)
954                .and_then(crate::knowledge::f32_le_head)
955        })?;
956        // lookup tables (sorted unique key hashes, entries in range,
957        // monotone offsets, UTF-8 blob, object-shaped slots): one pass
958        // over the mapped payloads of each lookup record — the slots are
959        // NOT parsed here (the runtime reads a card lazily; the writers
960        // parsed every slot before committing it).
961        crate::knowledge::validate_lookup_values(&header, &tensors, false, |t| {
962            let start = (env.data.0 + t.off) as usize;
963            bytes
964                .get(start..start + t.nbytes as usize)
965                .map(std::borrow::Cow::Borrowed)
966        })?;
967
968        tracing::info!(
969            "Opened CMF v2: {} | {} tensors | {} masks | vocab {} | {:.1} MB",
970            header.arch.arch_name,
971            tensors.len(),
972            masks.masks.len(),
973            if vocab.is_some() { "embedded" } else { "none" },
974            file_len as f64 / 1e6
975        );
976
977        Ok(Self {
978            uid: MODEL_UID.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
979            path,
980            header,
981            required_features: env.required_features,
982            tensors,
983            by_name,
984            name_overflow,
985            masks,
986            sparse_index,
987            vocab,
988            backing,
989            data_off: env.data.0,
990            envelope: env,
991            extra_shards: Vec::new(),
992            trunk,
993            exec,
994        })
995    }
996
997    /// Open a sharded model (spec §10): pass shard 1; siblings found by
998    /// the `-{no:05}-of-{count:05}.cmf` convention. Directories merge;
999    /// masks/vocab/index/skills come from shard 1.
1000    pub fn open_sharded(path: impl AsRef<Path>) -> Result<Self, CmfError> {
1001        let path = path.as_ref();
1002        let mut first = Self::open(path)?;
1003        let Some(info) = first.header.shard.clone() else {
1004            return Ok(first); // not sharded — plain open
1005        };
1006        if info.no != 1 {
1007            return Err(CmfError::Parse(format!(
1008                "open shard 1, not {} (of {})",
1009                info.no, info.count
1010            )));
1011        }
1012        let name = path
1013            .file_name()
1014            .and_then(|n| n.to_str())
1015            .ok_or_else(|| CmfError::Parse("bad shard path".into()))?;
1016        let tag1 = format!("-{:05}-of-{:05}.cmf", 1, info.count);
1017        if !name.ends_with(&tag1) {
1018            return Err(CmfError::Parse(format!(
1019                "shard file must end with '{tag1}' (got '{name}')"
1020            )));
1021        }
1022        let stem = &name[..name.len() - tag1.len()];
1023        for no in 2..=info.count {
1024            let sib = path.with_file_name(format!("{stem}-{:05}-of-{:05}.cmf", no, info.count));
1025            let sh = Self::open(&sib)?;
1026            if sh.header.arch.linear_core_identity() != first.header.arch.linear_core_identity() {
1027                return Err(CmfError::Parse(format!(
1028                    "{}: linear-core identity differs from shard 1",
1029                    sib.display()
1030                )));
1031            }
1032            match &sh.header.shard {
1033                Some(si) if si.no == no && si.count == info.count => {}
1034                other => {
1035                    return Err(CmfError::Parse(format!(
1036                        "{}: wrong shard coords {other:?}",
1037                        sib.display()
1038                    )));
1039                }
1040            }
1041            let shard_idx = first.extra_shards.len() + 1;
1042            first.extra_shards.push((sh.backing, sh.envelope.data.0));
1043            for mut t in sh.tensors {
1044                t.shard = shard_idx;
1045                let idx = first.tensors.len() as u32;
1046                let h = hash64(t.name.as_bytes());
1047                match first.by_name.get(&h) {
1048                    Some(&j) if first.tensors[j as usize].name == t.name => {
1049                        return Err(CmfError::Parse(format!(
1050                            "duplicate tensor name '{}' across shards",
1051                            t.name
1052                        )));
1053                    }
1054                    Some(_) => first.name_overflow.push(idx),
1055                    None => {
1056                        first.by_name.insert(h, idx);
1057                    }
1058                }
1059                first.tensors.push(t);
1060            }
1061        }
1062        tracing::info!(
1063            "sharded model: {} files, {} tensors total",
1064            info.count,
1065            first.tensors.len()
1066        );
1067        Ok(first)
1068    }
1069
1070    fn parse_envelope(bytes: &[u8], file_len: u64) -> Result<Envelope, CmfError> {
1071        if bytes.len() < ENVELOPE_LEN {
1072            return Err(CmfError::Bounds(format!(
1073                "file too small for CMF envelope: {} bytes",
1074                bytes.len()
1075            )));
1076        }
1077        if bytes[0..4] != CMF_MAGIC {
1078            return Err(CmfError::InvalidMagic);
1079        }
1080        let u32le = |o: usize| u32::from_le_bytes(bytes[o..o + 4].try_into().unwrap());
1081        let u64le = |o: usize| u64::from_le_bytes(bytes[o..o + 8].try_into().unwrap());
1082
1083        let version = u32le(4);
1084        if version != CMF_VERSION {
1085            return Err(CmfError::UnsupportedVersion(version));
1086        }
1087        let _flags = u32le(8); // reserved
1088        let required_features = u32le(12);
1089        check_required_features(required_features, features::SUPPORTED)?;
1090
1091        let env = Envelope {
1092            required_features,
1093            header: (u64le(0x10), u64le(0x18)),
1094            dir: (u64le(0x20), u64le(0x28)),
1095            data: (u64le(0x30), u64le(0x38)),
1096            masks: (u64le(0x40), u64le(0x48)),
1097            vocab: (u64le(0x50), u64le(0x58)),
1098            index: (u64le(0x60), u64le(0x68)),
1099            header_hash: u64le(0x70),
1100            dir_hash: u64le(0x78),
1101        };
1102
1103        for (name, (off, len), required) in [
1104            ("header", env.header, true),
1105            ("dir", env.dir, true),
1106            ("data", env.data, false),
1107            ("masks", env.masks, false),
1108            ("vocab", env.vocab, false),
1109            ("index", env.index, false),
1110        ] {
1111            if required && len == 0 {
1112                return Err(CmfError::Bounds(format!("section '{name}' is required")));
1113            }
1114            if len > 0
1115                && off
1116                    .checked_add(len)
1117                    .map(|end| end > file_len)
1118                    .unwrap_or(true)
1119            {
1120                return Err(CmfError::Bounds(format!(
1121                    "section '{name}' [{off}, {}) exceeds file ({file_len} bytes)",
1122                    off.saturating_add(len)
1123                )));
1124            }
1125            if len > 0
1126                && (usize::try_from(off).is_err()
1127                    || usize::try_from(len).is_err()
1128                    || usize::try_from(off + len).is_err())
1129            {
1130                return Err(CmfError::Bounds(format!(
1131                    "section '{name}' cannot be addressed on this platform"
1132                )));
1133            }
1134        }
1135        if env.data.1 > 0 && env.data.0 % DATA_ALIGNMENT != 0 {
1136            return Err(CmfError::Bounds(format!(
1137                "data section offset {} not {}-aligned",
1138                env.data.0, DATA_ALIGNMENT
1139            )));
1140        }
1141        Ok(env)
1142    }
1143
1144    fn decode_directory(bytes: &[u8]) -> Result<Vec<TensorEntry>, CmfError> {
1145        if bytes.len() < 16 {
1146            return Err(CmfError::Parse("tensor directory too short".into()));
1147        }
1148        let count = u64::from_le_bytes(bytes[0..8].try_into().unwrap()) as usize;
1149        let pool_off = u64::from_le_bytes(bytes[8..16].try_into().unwrap()) as usize;
1150        let records_len = count
1151            .checked_mul(DIR_RECORD_LEN)
1152            .ok_or_else(|| CmfError::Parse("tensor directory record count overflows".into()))?;
1153        let records_end = 16usize
1154            .checked_add(records_len)
1155            .ok_or_else(|| CmfError::Parse("tensor directory size overflows".into()))?;
1156        if records_end > bytes.len() || pool_off > bytes.len() || pool_off < records_end {
1157            return Err(CmfError::Parse(format!(
1158                "tensor directory malformed: count={count}, pool_off={pool_off}, len={}",
1159                bytes.len()
1160            )));
1161        }
1162        let pool = &bytes[pool_off..];
1163
1164        let mut out = Vec::with_capacity(count);
1165        for i in 0..count {
1166            let r = &bytes[16 + i * DIR_RECORD_LEN..16 + (i + 1) * DIR_RECORD_LEN];
1167            let name_off = u32::from_le_bytes(r[0..4].try_into().unwrap()) as usize;
1168            let name_len = u16::from_le_bytes(r[4..6].try_into().unwrap()) as usize;
1169            let dtype_id = r[6];
1170            let ndim = r[7] as usize;
1171            if ndim > DIR_MAX_NDIM {
1172                return Err(CmfError::Parse(format!("tensor #{i}: ndim {ndim} > 6")));
1173            }
1174            let mut shape = Vec::with_capacity(ndim);
1175            for d in 0..ndim {
1176                shape.push(
1177                    u32::from_le_bytes(r[8 + d * 4..12 + d * 4].try_into().unwrap()) as usize,
1178                );
1179            }
1180            let off = u64::from_le_bytes(r[32..40].try_into().unwrap());
1181            let nbytes = u64::from_le_bytes(r[40..48].try_into().unwrap());
1182            let hash = u64::from_le_bytes(r[48..56].try_into().unwrap());
1183
1184            let name_end = name_off
1185                .checked_add(name_len)
1186                .ok_or_else(|| CmfError::Parse(format!("tensor #{i}: name range overflows")))?;
1187            if name_end > pool.len() {
1188                return Err(CmfError::Parse(format!("tensor #{i}: name out of pool")));
1189            }
1190            let name = std::str::from_utf8(&pool[name_off..name_end])
1191                .map_err(|_| CmfError::Parse(format!("tensor #{i}: name is not UTF-8")))?
1192                .to_string();
1193            let dtype = TensorDtype::from_id(dtype_id).ok_or(CmfError::UnknownDtype(dtype_id))?;
1194
1195            out.push(TensorEntry {
1196                name,
1197                dtype,
1198                shape,
1199                off,
1200                nbytes,
1201                shard: 0,
1202                hash,
1203            });
1204        }
1205        Ok(out)
1206    }
1207
1208    // ───────────────────────── access ─────────────────────────
1209
1210    pub fn arch(&self) -> &ModelArch {
1211        &self.header.arch
1212    }
1213
1214    pub fn tensor(&self, name: &str) -> Option<&TensorEntry> {
1215        self.tensor_index(name).map(|i| &self.tensors[i])
1216    }
1217
1218    /// Directory index of a tensor by name (same resolution as
1219    /// [`Self::tensor`] — engines must not re-scan the directory). O(1) via the
1220    /// name-hash index; the name is verified against the entry so a hash
1221    /// collision can never return the wrong tensor, and the rare distinct-name
1222    /// collision falls back to the tiny overflow list.
1223    pub fn tensor_index(&self, name: &str) -> Option<usize> {
1224        let h = hash64(name.as_bytes());
1225        if let Some(&i) = self.by_name.get(&h) {
1226            if self.tensors[i as usize].name == name {
1227                return Some(i as usize);
1228            }
1229        }
1230        self.name_overflow
1231            .iter()
1232            .copied()
1233            .find(|&i| self.tensors[i as usize].name == name)
1234            .map(|i| i as usize)
1235    }
1236
1237    /// Tensor-source indirection (spec §9, Patent 15 fig3/302): the
1238    /// skill's replacement is read IN PLACE OF the backbone tensor —
1239    /// either/or, never combined. None skill → backbone directly.
1240    ///
1241    /// For a v2 skill record, a name listed in its `overrides` whose
1242    /// `skill.{id}.X` tensor is missing resolves to `None` (logged) — never
1243    /// silently to the backbone tensor. [`Self::try_resolve_tensor`] returns
1244    /// the reason.
1245    pub fn resolve_tensor(&self, name: &str, skill: Option<&str>) -> Option<&TensorEntry> {
1246        match self.try_resolve_tensor(name, skill) {
1247            Ok(t) => t,
1248            Err(e) => {
1249                tracing::error!("{e}");
1250                None
1251            }
1252        }
1253    }
1254
1255    /// [`Self::resolve_tensor`] with the no-silent-fallback case as an
1256    /// error: `Ok(None)` = no such tensor at all, `Err` = a v2 skill
1257    /// declares an override for `name` but carries no replacement.
1258    pub fn try_resolve_tensor(
1259        &self,
1260        name: &str,
1261        skill: Option<&str>,
1262    ) -> Result<Option<&TensorEntry>, CmfError> {
1263        if let Some(sid) = skill {
1264            if let Some(t) = self.tensor(&format!("skill.{sid}.{name}")) {
1265                return Ok(Some(t));
1266            }
1267            let declared = self
1268                .header
1269                .skills
1270                .iter()
1271                .find(|s| s.id == sid)
1272                .is_some_and(|s| s.kind.is_some() && s.overrides.iter().any(|o| o.name == name));
1273            if declared {
1274                return Err(CmfError::MissingTensor(format!(
1275                    "skill.{sid}.{name} (v2 skill '{sid}' overrides '{name}'; \
1276                     no fallback to the backbone tensor)"
1277                )));
1278            }
1279        }
1280        Ok(self.tensor(name))
1281    }
1282
1283    /// hash64 of the trunk as this file encodes it ([`trunk_hash`]): every
1284    /// directory entry outside `skill.`/`route.`, the arch, the vocab, the
1285    /// tokenizer bundle and — when present — the mask catalog and sparse
1286    /// index ([`crate::knowledge::trunk_hash_exec`]). Equals
1287    /// `header.genome.trunk_hash` for any GENOME file that opened.
1288    pub fn trunk_hash(&self) -> u64 {
1289        self.trunk.unwrap_or_else(|| {
1290            crate::knowledge::trunk_hash_exec(
1291                &self.header,
1292                &self.tensors,
1293                self.vocab.as_deref(),
1294                self.exec_hashes(),
1295            )
1296        })
1297    }
1298
1299    /// hash64 of this file's raw mask / sparse-index sections (computed at
1300    /// open for GENOME files, on demand otherwise).
1301    pub fn exec_hashes(&self) -> crate::knowledge::ExecHashes {
1302        if self.header.genome.is_some() {
1303            return self.exec;
1304        }
1305        let bytes = self.backing.bytes();
1306        let sec = |(off, len): (u64, u64)| -> Option<&[u8]> {
1307            (len > 0).then(|| bytes.get(off as usize..(off + len) as usize))?
1308        };
1309        crate::knowledge::ExecHashes::of(sec(self.envelope.masks), sec(self.envelope.index))
1310    }
1311
1312    /// The per-skill delta index view (claim 2): directory entries of
1313    /// one skill — exactly the byte ranges lazy loading pages in.
1314    pub fn skill_tensors(&self, skill_id: &str) -> impl Iterator<Item = &TensorEntry> {
1315        let prefix = format!("skill.{skill_id}.");
1316        self.tensors
1317            .iter()
1318            .filter(move |t| t.name.starts_with(&prefix))
1319    }
1320
1321    /// Zero-copy bytes of a tensor from the mmap'd data section.
1322    pub fn tensor_bytes(&self, name: &str) -> Result<&[u8], CmfError> {
1323        let entry = self
1324            .tensor(name)
1325            .ok_or_else(|| CmfError::MissingTensor(name.to_string()))?;
1326        Ok(self.entry_bytes(entry))
1327    }
1328
1329    /// All bytes of the primary mapping (GPU path: no-copy Metal buffer
1330    /// over the same mmap — unified memory, zero copying).
1331    /// hash64 of this file's tensor directory — the identity a standalone
1332    /// skill binds to (`SkillRecord.base_dir_hash`).
1333    pub fn dir_hash(&self) -> u64 {
1334        self.envelope.dir_hash
1335    }
1336
1337    pub fn primary_bytes(&self) -> &[u8] {
1338        self.backing.bytes()
1339    }
1340
1341    /// Absolute offset of the tensor within the primary mapping
1342    /// (None for tensors from sibling shards).
1343    /// Best-effort page-cache release for every tensor whose name passes
1344    /// `pred` (unix, primary shard only): the merged ranges are madvised
1345    /// DONTNEED so a one-shot stage's weights — a prompt encoder that
1346    /// runs once per generation — stop competing for RAM with the
1347    /// stages after it. A 25.7 GB fl2va file on a 24 GB Mac spent 40
1348    /// minutes paging the SSD during denoise for exactly this reason.
1349    /// Re-reading a dropped range later just refaults from disk.
1350    /// Returns the bytes released.
1351    pub fn advise_done(&self, pred: impl Fn(&str) -> bool) -> usize {
1352        #[cfg(unix)]
1353        {
1354            let base = self.primary_bytes().as_ptr() as usize;
1355            let map_len = self.primary_bytes().len();
1356            let page = 16384usize.max(unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as usize);
1357            let mut ranges: Vec<(usize, usize)> = self
1358                .tensors
1359                .iter()
1360                .filter(|e| e.shard == 0 && pred(&e.name))
1361                .filter_map(|e| {
1362                    let abs = self.entry_abs_offset(e)?;
1363                    Some((abs, abs + e.nbytes as usize))
1364                })
1365                .collect();
1366            ranges.sort_unstable();
1367            let mut dropped = 0usize;
1368            let mut merged: Vec<(usize, usize)> = Vec::new();
1369            for (s, e) in ranges {
1370                match merged.last_mut() {
1371                    Some(l) if s <= l.1 => l.1 = l.1.max(e),
1372                    _ => merged.push((s, e)),
1373                }
1374            }
1375            for (s, e) in merged {
1376                // Align INWARD: a page shared with a kept tensor stays.
1377                let s = s.div_ceil(page) * page;
1378                let e = (e.min(map_len)) / page * page;
1379                if e > s {
1380                    let r = unsafe {
1381                        libc::madvise((base + s) as *mut libc::c_void, e - s, libc::MADV_DONTNEED)
1382                    };
1383                    if r == 0 {
1384                        dropped += e - s;
1385                    }
1386                }
1387            }
1388            dropped
1389        }
1390        #[cfg(not(unix))]
1391        {
1392            let _ = pred;
1393            0
1394        }
1395    }
1396
1397    pub fn entry_abs_offset(&self, entry: &TensorEntry) -> Option<usize> {
1398        (entry.shard == 0).then(|| (self.data_off + entry.off) as usize)
1399    }
1400
1401    pub fn entry_bytes(&self, entry: &TensorEntry) -> &[u8] {
1402        let (bytes, data_off) = if entry.shard == 0 {
1403            (self.backing.bytes(), self.data_off)
1404        } else {
1405            let (b, o) = &self.extra_shards[entry.shard - 1];
1406            (b.bytes(), *o)
1407        };
1408        let start = (data_off + entry.off) as usize;
1409        &bytes[start..start + entry.nbytes as usize]
1410    }
1411
1412    /// The CPU is done with these byte ranges of the primary mapping
1413    /// (absolute offsets, as `entry_abs_offset` hands them out): drop them
1414    /// from the resident set and let the page cache release the file pages.
1415    /// Both calls are advisory and the mapping stays valid — a range that
1416    /// gets touched again re-faults from disk, so a caller can only cost
1417    /// time here, never correctness. Ranges are aligned OUTWARD to page
1418    /// boundaries; the neighbours those edges claw in re-fault the same way.
1419    /// Linux + mmap backing only; everywhere else a no-op.
1420    pub fn evict_ranges(&self, ranges: &[(usize, usize)]) {
1421        #[cfg(target_os = "linux")]
1422        {
1423            let Backing::Mmap(m) = &self.backing else {
1424                return;
1425            };
1426            let page = 4096usize;
1427            // One fd for the whole batch: fadvise targets the inode's page
1428            // cache, any fd on the same file will do.
1429            use std::os::unix::io::AsRawFd;
1430            let file = File::open(&self.path).ok();
1431            for &(off, len) in ranges {
1432                if len == 0 || off.saturating_add(len) > m.len() {
1433                    continue;
1434                }
1435                let start = off & !(page - 1);
1436                let end = off + len;
1437                let alen = end.next_multiple_of(page).min(m.len()) - start;
1438                // SAFETY: read-only MAP_SHARED file mapping — DONTNEED here
1439                // only drops clean pages; the next access re-faults them.
1440                let _ = unsafe {
1441                    m.unchecked_advise_range(memmap2::UncheckedAdvice::DontNeed, start, alen)
1442                };
1443                if let Some(f) = &file {
1444                    // SAFETY: plain fd + numeric range; advisory by contract.
1445                    unsafe {
1446                        libc::posix_fadvise(
1447                            f.as_raw_fd(),
1448                            start as libc::off_t,
1449                            alen as libc::off_t,
1450                            libc::POSIX_FADV_DONTNEED,
1451                        );
1452                    }
1453                }
1454            }
1455        }
1456        #[cfg(not(target_os = "linux"))]
1457        let _ = ranges;
1458    }
1459
1460    /// Tensors belonging to layer `i` (prefix `model.layers.{i}.`).
1461    pub fn layer_tensors(&self, layer_idx: usize) -> Vec<&TensorEntry> {
1462        let prefix = format!("model.layers.{layer_idx}.");
1463        self.tensors
1464            .iter()
1465            .filter(|t| t.name.starts_with(&prefix))
1466            .collect()
1467    }
1468
1469    /// Total parameter count estimated from matrix tensors (ndim ≥ 2).
1470    pub fn total_param_count(&self) -> u64 {
1471        self.tensors
1472            .iter()
1473            .filter(|t| t.shape.len() >= 2)
1474            .map(|t| t.n_elems() as u64)
1475            .sum()
1476    }
1477
1478    /// Recode selected tensors IN PLACE: each new payload must fit its old
1479    /// slot, the entry keeps its offset and the file keeps its length — the
1480    /// bytes between the new end and the old simply go dark (every reader
1481    /// walks the directory, nothing addresses the gap). This is what lets a
1482    /// published 100+ GB file change a tensor's layout on a disk too small
1483    /// to hold two copies of it. Patches are `(directory index, new dtype,
1484    /// new payload)`; entry hashes and the directory hash are recomputed so
1485    /// `verify` stays clean. Not atomic: a crash between the payload writes
1486    /// and the directory write leaves the old dtype over new bytes — verify
1487    /// (or re-fetch the source) after an interrupted run.
1488    pub fn recode_entries_in_place(
1489        path: &str,
1490        patches: &[(usize, TensorDtype, Vec<u8>)],
1491    ) -> Result<(), CmfError> {
1492        use std::io::{Read, Seek, SeekFrom, Write};
1493        let mut f = std::fs::OpenOptions::new()
1494            .read(true)
1495            .write(true)
1496            .open(path)?;
1497        let file_len = f.metadata()?.len();
1498        let mut head = vec![0u8; ENVELOPE_LEN];
1499        f.read_exact(&mut head)?;
1500        let env = Self::parse_envelope(&head, file_len)?;
1501        // By content too: a pre-0.8.1 embryo-o1 file carries GENOME on bit 8.
1502        let genome_by_content = peek_header(path)?
1503            .is_some_and(|(_, header)| header.get("genome").is_some_and(|g| !g.is_null()));
1504        if env.required_features & features::GENOME != 0 || genome_by_content {
1505            return Err(CmfError::Parse(format!(
1506                "recode: {path} carries a frozen genome (GENOME bit) — rewriting its \
1507                 payloads in place would change the trunk without a new genome. A requant \
1508                 must write a new file that rewrites genome.trunk_hash/encoding and records \
1509                 a `requant` lineage event"
1510            )));
1511        }
1512
1513        let mut dir = vec![0u8; env.dir.1 as usize];
1514        f.seek(SeekFrom::Start(env.dir.0))?;
1515        f.read_exact(&mut dir)?;
1516        let count = u64::from_le_bytes(dir[0..8].try_into().unwrap()) as usize;
1517
1518        for (i, dtype, data) in patches {
1519            if *i >= count {
1520                return Err(CmfError::Bounds(format!(
1521                    "recode: tensor #{i} out of directory ({count} entries)"
1522                )));
1523            }
1524            let rb = 16 + i * DIR_RECORD_LEN;
1525            let rec = &mut dir[rb..rb + DIR_RECORD_LEN];
1526            let off = u64::from_le_bytes(rec[32..40].try_into().unwrap());
1527            let old_n = u64::from_le_bytes(rec[40..48].try_into().unwrap());
1528            if data.len() as u64 > old_n {
1529                return Err(CmfError::Bounds(format!(
1530                    "recode: tensor #{i} payload {} > slot {old_n}",
1531                    data.len()
1532                )));
1533            }
1534            f.seek(SeekFrom::Start(env.data.0 + off))?;
1535            f.write_all(data)?;
1536            rec[6] = dtype.id();
1537            rec[40..48].copy_from_slice(&(data.len() as u64).to_le_bytes());
1538            rec[48..56].copy_from_slice(&hash64(data).to_le_bytes());
1539        }
1540
1541        f.seek(SeekFrom::Start(env.dir.0))?;
1542        f.write_all(&dir)?;
1543        f.seek(SeekFrom::Start(0x78))?;
1544        f.write_all(&hash64(&dir).to_le_bytes())?;
1545        f.sync_all()?;
1546        Ok(())
1547    }
1548
1549    /// Recompute all tensor hashes; returns human-readable problems
1550    /// (empty = file intact).
1551    pub fn verify(&self) -> Vec<String> {
1552        let mut problems = Vec::new();
1553
1554        // Section-level integrity (spec §8.1). Zero/absent = legacy file.
1555        let bytes = self.backing.bytes();
1556        let env = &self.envelope;
1557        let sect = |(off, len): (u64, u64)| &bytes[off as usize..(off + len) as usize];
1558        let check = |name: &str, stored: u64, span: (u64, u64)| -> Option<String> {
1559            if stored != 0 && span.1 > 0 {
1560                let actual = hash64(sect(span));
1561                if actual != stored {
1562                    return Some(format!(
1563                        "section '{name}': hash mismatch (stored {stored:016x}, \
1564                         actual {actual:016x})"
1565                    ));
1566                }
1567            }
1568            None
1569        };
1570        problems.extend(check("header", env.header_hash, env.header));
1571        problems.extend(check("dir", env.dir_hash, env.dir));
1572        if let Some(sh) = &self.header.section_hashes {
1573            for (name, hex, span) in [
1574                ("masks", &sh.masks, env.masks),
1575                ("vocab", &sh.vocab, env.vocab),
1576                ("index", &sh.index, env.index),
1577            ] {
1578                if let Some(hex) = hex {
1579                    match u64::from_str_radix(hex, 16) {
1580                        Ok(stored) => problems.extend(check(name, stored, span)),
1581                        Err(_) => {
1582                            problems.push(format!("section '{name}': malformed hash '{hex}'"))
1583                        }
1584                    }
1585                }
1586            }
1587        }
1588
1589        for t in &self.tensors {
1590            let actual = hash64(self.entry_bytes(t));
1591            if actual != t.hash {
1592                problems.push(format!(
1593                    "tensor '{}': hash mismatch (stored {:016x}, actual {:016x})",
1594                    t.name, t.hash, actual
1595                ));
1596            }
1597        }
1598        problems
1599    }
1600
1601    /// Approximate active weight bytes under a mask, from real tensor
1602    /// sizes in the directory (not from a formula).
1603    pub fn compute_active_size(&self, mask: &TaskMask) -> u64 {
1604        let arch = &self.header.arch;
1605        let mut total = 0u64;
1606        for li in 0..arch.num_layers {
1607            if !mask.layer_alive(li) {
1608                continue;
1609            }
1610            let ffn_frac = mask.ffn_active_count(li) as f64 / arch.intermediate_size.max(1) as f64;
1611            let head_frac =
1612                mask.active_head_count(li) as f64 / arch.num_attention_heads.max(1) as f64;
1613            for t in self.layer_tensors(li) {
1614                let frac = if t.name.contains(".mlp.") {
1615                    ffn_frac
1616                } else if t.name.contains(".self_attn.") {
1617                    head_frac
1618                } else {
1619                    1.0
1620                };
1621                total += (t.nbytes as f64 * frac) as u64;
1622            }
1623        }
1624        total
1625    }
1626
1627    // ───────────────────────── writer ─────────────────────────
1628
1629    /// Write a CMF v2 file. Offsets, alignment, hashes and the sparse
1630    /// index are computed here — the caller supplies content only.
1631    pub fn write(
1632        path: impl AsRef<Path>,
1633        header: &CmfHeader,
1634        tensors: &[TensorSpec],
1635        masks: Option<&MaskCatalog>,
1636        vocab: Option<&[u8]>,
1637    ) -> Result<(), CmfError> {
1638        let refs: Vec<TensorSpecRef> = tensors
1639            .iter()
1640            .map(|t| TensorSpecRef {
1641                name: t.name.clone(),
1642                dtype: t.dtype,
1643                shape: t.shape.clone(),
1644                data: &t.data,
1645            })
1646            .collect();
1647        Self::write_ref(path, header, &refs, masks, vocab)
1648    }
1649
1650    /// `write` with BORROWED tensor payloads — repack tools slice the
1651    /// source file's mmap directly, so a 19 GB container rewrites without
1652    /// materializing its tensors in RAM (the OS streams pages through).
1653    pub fn write_ref(
1654        path: impl AsRef<Path>,
1655        header: &CmfHeader,
1656        tensors: &[TensorSpecRef],
1657        masks: Option<&MaskCatalog>,
1658        vocab: Option<&[u8]>,
1659    ) -> Result<(), CmfError> {
1660        let path = path.as_ref();
1661        if let Some(prism) = header.arch.prism_hadamard.as_ref() {
1662            prism.validate().map_err(CmfError::Parse)?;
1663            if header.arch.arch_name != "prism_hadamard_qwen35" {
1664                return Err(CmfError::Parse(
1665                    "prism_hadamard metadata requires arch_name prism_hadamard_qwen35".into(),
1666                ));
1667            }
1668        }
1669        header
1670            .arch
1671            .validate_operator_metadata()
1672            .map_err(CmfError::Parse)?;
1673
1674        // Directory + data layout.
1675        let mut entries = Vec::with_capacity(tensors.len());
1676        let mut data_cursor = 0u64;
1677        for t in tensors {
1678            if t.shape.len() > DIR_MAX_NDIM {
1679                return Err(CmfError::Parse(format!(
1680                    "tensor '{}': ndim {} > 6",
1681                    t.name,
1682                    t.shape.len()
1683                )));
1684            }
1685            if let Some(expect) = expected_nbytes(t.dtype, &t.shape) {
1686                if expect != t.data.len() {
1687                    return Err(CmfError::Bounds(format!(
1688                        "tensor '{}': data {} bytes != expected {} for {:?}{:?}",
1689                        t.name,
1690                        t.data.len(),
1691                        expect,
1692                        t.dtype,
1693                        t.shape
1694                    )));
1695                }
1696            }
1697            let align = if t.data.len() as u64 >= LARGE_TENSOR_MIN {
1698                LARGE_TENSOR_ALIGN
1699            } else {
1700                TENSOR_ALIGNMENT
1701            };
1702            data_cursor = align_to(data_cursor, align);
1703            entries.push(TensorEntry {
1704                name: t.name.clone(),
1705                dtype: t.dtype,
1706                shape: t.shape.clone(),
1707                off: data_cursor,
1708                nbytes: t.data.len() as u64,
1709                shard: 0,
1710                hash: hash64(t.data),
1711            });
1712            data_cursor += t.data.len() as u64;
1713        }
1714        let data_len = data_cursor;
1715
1716        validate_prism_affine_targets(&header.arch, &entries)?;
1717
1718        let dir_bytes = Self::encode_directory(&entries);
1719
1720        let masks_bytes = match masks {
1721            Some(catalog) if !catalog.masks.is_empty() => {
1722                Some(encode_masks_section(catalog, &header.arch).map_err(CmfError::Parse)?)
1723            }
1724            _ => None,
1725        };
1726        let index_bytes = match masks {
1727            Some(catalog) if !catalog.masks.is_empty() => {
1728                let idx = build_sparse_index(catalog, &header.arch);
1729                Some(encode_sparse_index(&idx))
1730            }
1731            _ => None,
1732        };
1733
1734        // Section hashes go INTO the header (so the envelope's header
1735        // hash transitively covers them), then the header is serialized.
1736        let hex = |b: Option<&[u8]>| b.map(|b| format!("{:016x}", hash64(b)));
1737        let mut header = header.clone();
1738        if masks_bytes.is_some() || vocab.is_some() || index_bytes.is_some() {
1739            header.section_hashes = Some(SectionHashes {
1740                masks: hex(masks_bytes.as_deref()),
1741                vocab: hex(vocab),
1742                index: hex(index_bytes.as_deref()),
1743            });
1744        }
1745        // A full rewrite lays every tensor out afresh: it is a compaction,
1746        // and the tail-append segment table no longer describes the file.
1747        header.segments.clear();
1748        // Genome: fill an empty trunk hash from this content, refuse a
1749        // stale one (the trunk changed without a new genome). The mask
1750        // catalog / sparse index are part of the trunk's execution.
1751        let exec = crate::knowledge::ExecHashes::of(masks_bytes.as_deref(), index_bytes.as_deref());
1752        crate::knowledge::seal_genome(&mut header, &entries, vocab, exec)?;
1753
1754        let required_features = derive_required_features(&header, &entries, masks_bytes.is_some());
1755        // The writer refuses what open() would refuse — before a byte hits
1756        // the disk.
1757        crate::knowledge::validate_knowledge(
1758            &header,
1759            &entries,
1760            vocab,
1761            required_features,
1762            Some(data_len),
1763            None,
1764            exec,
1765        )?;
1766        crate::knowledge::check_lookup_policies(&header)?;
1767        // entries[i] is tensors[i]: the payload of an entry by position.
1768        crate::knowledge::validate_expert_append_values(&header, &entries, |e| {
1769            let i = entries.iter().position(|x| std::ptr::eq(x, e))?;
1770            crate::knowledge::f32_le_head(tensors[i].data)
1771        })?;
1772        crate::knowledge::validate_lookup_values(&header, &entries, true, |e| {
1773            let i = entries.iter().position(|x| std::ptr::eq(x, e))?;
1774            Some(std::borrow::Cow::Borrowed(tensors[i].data))
1775        })?;
1776        let header_json =
1777            serde_json::to_vec(&header).map_err(|e| CmfError::Parse(format!("header: {e}")))?;
1778
1779        // Section offsets.
1780        let header_off = ENVELOPE_LEN as u64;
1781        let dir_off = header_off + header_json.len() as u64;
1782        let data_off = align_to(dir_off + dir_bytes.len() as u64, DATA_ALIGNMENT);
1783        let masks_off = data_off + data_len;
1784        let masks_len = masks_bytes.as_ref().map(|b| b.len() as u64).unwrap_or(0);
1785        let vocab_off = masks_off + masks_len;
1786        let vocab_len = vocab.map(|b| b.len() as u64).unwrap_or(0);
1787        let index_off = vocab_off + vocab_len;
1788        let index_len = index_bytes.as_ref().map(|b| b.len() as u64).unwrap_or(0);
1789
1790        // Envelope.
1791        let mut env = Vec::with_capacity(ENVELOPE_LEN);
1792        env.extend_from_slice(&CMF_MAGIC);
1793        env.extend_from_slice(&CMF_VERSION.to_le_bytes());
1794        env.extend_from_slice(&0u32.to_le_bytes()); // flags
1795        env.extend_from_slice(&required_features.to_le_bytes());
1796        for (off, len) in [
1797            (header_off, header_json.len() as u64),
1798            (dir_off, dir_bytes.len() as u64),
1799            (data_off, data_len),
1800            (if masks_len > 0 { masks_off } else { 0 }, masks_len),
1801            (if vocab_len > 0 { vocab_off } else { 0 }, vocab_len),
1802            (if index_len > 0 { index_off } else { 0 }, index_len),
1803        ] {
1804            env.extend_from_slice(&off.to_le_bytes());
1805            env.extend_from_slice(&len.to_le_bytes());
1806        }
1807        // Reserved bytes carry header/dir integrity (spec §8.1).
1808        env.extend_from_slice(&hash64(&header_json).to_le_bytes());
1809        env.extend_from_slice(&hash64(&dir_bytes).to_le_bytes());
1810        env.resize(ENVELOPE_LEN, 0);
1811
1812        // Write out.
1813        let mut f = BufWriter::new(File::create(path)?);
1814        f.write_all(&env)?;
1815        f.write_all(&header_json)?;
1816        f.write_all(&dir_bytes)?;
1817        let mut pos = dir_off + dir_bytes.len() as u64;
1818        f.write_all(&zeros((data_off - pos) as usize))?;
1819        pos = data_off;
1820        for (spec, entry) in tensors.iter().zip(&entries) {
1821            let target = data_off + entry.off;
1822            f.write_all(&zeros((target - pos) as usize))?;
1823            f.write_all(spec.data)?;
1824            pos = target + spec.data.len() as u64;
1825        }
1826        debug_assert_eq!(pos, data_off + data_len);
1827        if let Some(mb) = &masks_bytes {
1828            f.write_all(mb)?;
1829        }
1830        if let Some(vb) = vocab {
1831            f.write_all(vb)?;
1832        }
1833        if let Some(ib) = &index_bytes {
1834            f.write_all(ib)?;
1835        }
1836        f.flush()?;
1837
1838        tracing::info!(
1839            "Wrote CMF v2: {} ({} tensors, {} masks, {:.1} MB)",
1840            path.display(),
1841            entries.len(),
1842            masks.map(|m| m.masks.len()).unwrap_or(0),
1843            std::fs::metadata(path)?.len() as f64 / 1e6
1844        );
1845        Ok(())
1846    }
1847
1848    pub(crate) fn encode_directory(entries: &[TensorEntry]) -> Vec<u8> {
1849        let mut pool = Vec::new();
1850        let mut name_offs = Vec::with_capacity(entries.len());
1851        for e in entries {
1852            name_offs.push((pool.len() as u32, e.name.len() as u16));
1853            pool.extend_from_slice(e.name.as_bytes());
1854        }
1855        let pool_off = 16 + entries.len() * DIR_RECORD_LEN;
1856
1857        let mut out = Vec::with_capacity(pool_off + pool.len());
1858        out.extend_from_slice(&(entries.len() as u64).to_le_bytes());
1859        out.extend_from_slice(&(pool_off as u64).to_le_bytes());
1860        for (e, (noff, nlen)) in entries.iter().zip(&name_offs) {
1861            out.extend_from_slice(&noff.to_le_bytes());
1862            out.extend_from_slice(&nlen.to_le_bytes());
1863            out.push(e.dtype.id());
1864            out.push(e.shape.len() as u8);
1865            for d in 0..DIR_MAX_NDIM {
1866                out.extend_from_slice(&(e.shape.get(d).copied().unwrap_or(0) as u32).to_le_bytes());
1867            }
1868            out.extend_from_slice(&e.off.to_le_bytes());
1869            out.extend_from_slice(&e.nbytes.to_le_bytes());
1870            out.extend_from_slice(&e.hash.to_le_bytes());
1871        }
1872        out.extend_from_slice(&pool);
1873        out
1874    }
1875}
1876
1877fn align_to(x: u64, a: u64) -> u64 {
1878    x.div_ceil(a) * a
1879}
1880
1881/// Refuse `required` bits outside `supported` — the envelope gate every
1882/// reader applies before touching a section (`UnsupportedFeature`). Public
1883/// so tooling can ask what a reader with an older `SUPPORTED` would do.
1884pub fn check_required_features(required: u32, supported: u32) -> Result<(), CmfError> {
1885    let unknown = required & !supported;
1886    if unknown != 0 {
1887        return Err(CmfError::UnsupportedFeature(unknown));
1888    }
1889    Ok(())
1890}
1891
1892/// The `required_features` a header + directory demand — the ONE place
1893/// the writers (`write_ref`, the streaming `finish`, the tail append)
1894/// derive the bits from content, so no bit is ever set by hand and none is
1895/// silently lost on a rewrite.
1896pub fn derive_required_features(
1897    header: &CmfHeader,
1898    entries: &[TensorEntry],
1899    masks_present: bool,
1900) -> u32 {
1901    let mut f = features::TENSOR_DIR | prism_bits(&header.arch);
1902    // A decision profile is not a language model: the bit makes every
1903    // generative reader refuse it (older ones by the unknown bit).
1904    if is_decision_profile(&header.arch.arch_name) {
1905        f |= features::DECISION;
1906    }
1907    if masks_present {
1908        f |= features::BINARY_MASKS;
1909        if header.arch.num_loops > 1 {
1910            f |= features::LOOP_MASKS;
1911        }
1912    }
1913    if entries
1914        .iter()
1915        .any(|t| matches!(t.dtype, TensorDtype::Q8_2f | TensorDtype::Vbit))
1916    {
1917        f |= features::QUANT_2F;
1918    }
1919    // A skill record bound to a base directory makes this file a
1920    // standalone skill, and the bit keeps every reader honest about it.
1921    if header.skills.iter().any(|s| s.base_dir_hash.is_some()) {
1922        f |= features::SKILL_FILE;
1923    }
1924    // A declared bounded anchor is an operator every reader must execute
1925    // exactly; the bit makes readers that cannot refuse.
1926    if header.arch.anchor_core.is_some() {
1927        f |= features::BOUNDED_STATE;
1928    }
1929    f | crate::knowledge::knowledge_bits(header)
1930}
1931
1932fn zeros(n: usize) -> Vec<u8> {
1933    vec![0u8; n]
1934}
1935
1936/// Persist a directory entry after a file it contains is created or updated.
1937///
1938/// A checkpoint mark is the boundary at which the converter may consume its
1939/// source shard.  Syncing the containing directory after the output payload
1940/// and manifest has been synced closes the small rename/create window where a
1941/// crash could otherwise leave durable bytes without a durable directory
1942/// entry.  Unix filesystems expose directory fsync; other platforms retain
1943/// the file-level ordering and use the no-op fallback below.
1944fn sync_parent_dir(path: &Path) -> Result<(), CmfError> {
1945    #[cfg(unix)]
1946    {
1947        let parent = path.parent().unwrap_or_else(|| Path::new("."));
1948        File::open(parent)?.sync_all()?;
1949    }
1950    #[cfg(not(unix))]
1951    let _ = path;
1952    Ok(())
1953}
1954
1955// ─────────────────── one-pass streaming writer (§8.4) ───────────────────
1956
1957/// Writes a CMF file in a single pass, payloads first.
1958///
1959/// [`CmfModel::write_ref`] needs every payload addressable at once, so a
1960/// converter has to hold the whole encoded model — RAM, or a spill file it
1961/// then copies into the output. For a 300B-class MoE that is ~120 GB written
1962/// twice. This writer instead reserves a gap at the head of the file, appends
1963/// each payload the moment it is encoded, and patches the envelope, header and
1964/// directory into that gap at the end. The bytes are written once and peak
1965/// disk cost is the finished file.
1966///
1967/// The gap is the one thing that can go wrong: the directory is not sized
1968/// until the last tensor arrives. [`CmfStreamWriter::finish`] therefore
1969/// refuses loudly if the head does not fit rather than truncating it, and
1970/// [`CmfStreamWriter::head_reserve_for`] gives callers a safe estimate.
1971/// What a resumed writer recovers from its manifest: the tensors already on
1972/// disk and any milestones the producer noted.
1973pub struct ResumeState {
1974    pub names: Vec<String>,
1975    pub marks: Vec<String>,
1976}
1977
1978pub struct CmfStreamWriter {
1979    file: BufWriter<File>,
1980    path: PathBuf,
1981    /// Absolute offset of the weight blob — also the size of the reserved gap.
1982    data_off: u64,
1983    /// Write cursor, relative to `data_off`.
1984    cursor: u64,
1985    entries: Vec<TensorEntry>,
1986    /// Append-only sidecar describing every payload already on disk. A Colab
1987    /// box can vanish mid-conversion; with this the finished payloads can be
1988    /// turned into a valid file instead of re-encoding for hours.
1989    manifest: Option<BufWriter<File>>,
1990}
1991
1992impl CmfStreamWriter {
1993    /// A gap that comfortably holds the head for `n_tensors` whose names run
1994    /// to `avg_name` bytes: the directory's fixed records, the name pool, the
1995    /// envelope, and a header JSON with room for arch metadata — then doubled,
1996    /// because being wrong here costs a whole re-run.
1997    pub fn head_reserve_for(n_tensors: usize, avg_name: usize) -> u64 {
1998        let dir = 16 + n_tensors * (DIR_RECORD_LEN + 6 + avg_name);
1999        let head = ENVELOPE_LEN + dir + (1 << 20);
2000        // Tripled, on top of a megabyte of slack that is already ~20x a
2001        // small model's directory. The asymmetry is deliberate: an
2002        // over-estimate costs zeros at the head of the file, an
2003        // under-estimate costs the entire conversion that produced it.
2004        align_to(3 * head as u64, DATA_ALIGNMENT).max(1 << 20)
2005    }
2006
2007    /// `gap` bytes are reserved for envelope + header + directory.
2008    pub fn new(path: impl AsRef<Path>, gap: u64) -> Result<Self, CmfError> {
2009        let path = path.as_ref().to_path_buf();
2010        let data_off = align_to(gap.max(ENVELOPE_LEN as u64 + 1), DATA_ALIGNMENT);
2011        // Read + write: `finish` reads a few descriptor values back from
2012        // the payloads already on disk (expert_append records).
2013        let file = std::fs::OpenOptions::new()
2014            .read(true)
2015            .write(true)
2016            .create(true)
2017            .truncate(true)
2018            .open(&path)?;
2019        let mut file = BufWriter::new(file);
2020        file.write_all(&zeros(data_off as usize))?;
2021        Ok(Self {
2022            file,
2023            path,
2024            data_off,
2025            cursor: 0,
2026            entries: Vec::new(),
2027            manifest: None,
2028        })
2029    }
2030
2031    /// Append one tensor. The payload is consumed here, so the caller can drop
2032    /// it immediately — that is the entire point of this writer.
2033    pub fn push(
2034        &mut self,
2035        name: &str,
2036        dtype: TensorDtype,
2037        shape: &[usize],
2038        data: &[u8],
2039    ) -> Result<(), CmfError> {
2040        self.push_bounded(name, dtype, shape, data, data.len().max(1))
2041    }
2042
2043    /// Append one tensor while limiting each write syscall to `chunk_bytes`.
2044    ///
2045    /// The source is normally an mmap, so this does not change the caller's
2046    /// memory residency. It does make the bounded-copy contract explicit for
2047    /// very large auxiliary tensors (DeepSeek-V4.1's native Engram tables are
2048    /// hundreds of gigabytes): the writer never asks an I/O layer to stage the
2049    /// whole payload as one buffer. Hashing remains one streaming pass over the
2050    /// borrowed bytes, and the directory entry is identical to [`Self::push`].
2051    pub fn push_bounded(
2052        &mut self,
2053        name: &str,
2054        dtype: TensorDtype,
2055        shape: &[usize],
2056        data: &[u8],
2057        chunk_bytes: usize,
2058    ) -> Result<(), CmfError> {
2059        if shape.len() > DIR_MAX_NDIM {
2060            return Err(CmfError::Parse(format!(
2061                "tensor '{}': ndim {} > {}",
2062                name,
2063                shape.len(),
2064                DIR_MAX_NDIM
2065            )));
2066        }
2067        if let Some(expect) = expected_nbytes(dtype, shape) {
2068            if expect != data.len() {
2069                return Err(CmfError::Bounds(format!(
2070                    "tensor '{}': data {} bytes != expected {} for {:?}{:?}",
2071                    name,
2072                    data.len(),
2073                    expect,
2074                    dtype,
2075                    shape
2076                )));
2077            }
2078        }
2079        let align = if data.len() as u64 >= LARGE_TENSOR_MIN {
2080            LARGE_TENSOR_ALIGN
2081        } else {
2082            TENSOR_ALIGNMENT
2083        };
2084        let off = align_to(self.cursor, align);
2085        self.file.write_all(&zeros((off - self.cursor) as usize))?;
2086        for chunk in data.chunks(chunk_bytes.max(1)) {
2087            self.file.write_all(chunk)?;
2088        }
2089        self.entries.push(TensorEntry {
2090            name: name.to_string(),
2091            dtype,
2092            shape: shape.to_vec(),
2093            off,
2094            nbytes: data.len() as u64,
2095            shard: 0,
2096            hash: hash64(data),
2097        });
2098        self.cursor = off + data.len() as u64;
2099        if let Some(m) = self.manifest.as_mut() {
2100            let e = self.entries.last().unwrap();
2101            writeln!(
2102                m,
2103                "{{\"name\":{},\"dtype\":{},\"shape\":{:?},\"off\":{},\"nbytes\":{},\"hash\":{}}}",
2104                serde_json::to_string(&e.name).unwrap_or_else(|_| "\"?\"".into()),
2105                dtype.id(),
2106                e.shape,
2107                e.off,
2108                e.nbytes,
2109                e.hash
2110            )?;
2111            m.flush()?;
2112        }
2113        Ok(())
2114    }
2115
2116    /// Start recording a sidecar manifest at `path`. One JSON line per
2117    /// tensor, flushed as it goes, plus a first line pinning the gap size.
2118    pub fn with_manifest(mut self, path: impl AsRef<Path>) -> Result<Self, CmfError> {
2119        let path = path.as_ref().to_path_buf();
2120        let mut f = BufWriter::new(File::create(&path)?);
2121        writeln!(f, "{{\"data_off\":{}}}", self.data_off)?;
2122        f.flush()?;
2123        // The first line establishes the resume geometry.  Make both its
2124        // contents and its directory entry durable before any source shard
2125        // can reach a later checkpoint.
2126        f.get_ref().sync_data()?;
2127        sync_parent_dir(&path)?;
2128        self.manifest = Some(f);
2129        Ok(self)
2130    }
2131
2132    /// Keep recording into an existing manifest — for a writer from
2133    /// [`CmfStreamWriter::resume`], whose earlier lines must survive.
2134    pub fn appending_manifest(mut self, path: impl AsRef<Path>) -> Result<Self, CmfError> {
2135        let path = path.as_ref().to_path_buf();
2136        let file = std::fs::OpenOptions::new().append(true).open(&path)?;
2137        // A resumed writer is allowed to consume the next shard immediately;
2138        // preserve the same directory ordering guarantee as a fresh writer.
2139        file.sync_data()?;
2140        sync_parent_dir(&path)?;
2141        self.manifest = Some(BufWriter::new(file));
2142        Ok(self)
2143    }
2144
2145    /// Rebuild a writer over an output file whose payloads are already on
2146    /// disk, from the manifest that recorded them. The file is reopened for
2147    /// writing without truncation and the cursor is placed after the last
2148    /// recorded tensor, so `finish` can complete a conversion that died.
2149    pub fn resume(
2150        path: impl AsRef<Path>,
2151        manifest: impl AsRef<Path>,
2152    ) -> Result<(Self, ResumeState), CmfError> {
2153        let path = path.as_ref().to_path_buf();
2154        let text = std::fs::read_to_string(manifest.as_ref())?;
2155        let mut lines = text.lines();
2156        let first = lines
2157            .next()
2158            .ok_or_else(|| CmfError::Parse("manifest is empty".into()))?;
2159        let head: serde_json::Value = serde_json::from_str(first)
2160            .map_err(|e| CmfError::Parse(format!("manifest head: {e}")))?;
2161        let data_off = head["data_off"]
2162            .as_u64()
2163            .ok_or_else(|| CmfError::Parse("manifest head has no data_off".into()))?;
2164
2165        let mut entries = Vec::new();
2166        let mut names = Vec::new();
2167        let mut marks = Vec::new();
2168        let (mut safe_upto, mut safe_entries) = (0u64, 0usize);
2169        for (i, line) in lines.enumerate() {
2170            if line.trim().is_empty() {
2171                continue;
2172            }
2173            // A truncated last line is expected if the process was killed
2174            // mid-write; it is dropped, not an error.
2175            let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
2176                tracing::warn!("manifest line {} is truncated — ignoring it", i + 2);
2177                break;
2178            };
2179            if let Some(mark) = v["mark"].as_str() {
2180                marks.push(mark.to_string());
2181                // Everything up to here is durable; anything the manifest
2182                // records after the LAST mark belongs to a shard that was
2183                // interrupted and will be redone, so it must not be kept —
2184                // otherwise the redo appends those tensors a second time.
2185                safe_upto = v["at"].as_u64().unwrap_or(0);
2186                safe_entries = entries.len();
2187                continue;
2188            }
2189            let dtype = TensorDtype::from_id(v["dtype"].as_u64().unwrap_or(0) as u8)
2190                .ok_or_else(|| CmfError::Parse(format!("manifest line {}: dtype", i + 2)))?;
2191            let name = v["name"].as_str().unwrap_or_default().to_string();
2192            names.push(name.clone());
2193            entries.push(TensorEntry {
2194                name,
2195                dtype,
2196                shape: v["shape"]
2197                    .as_array()
2198                    .map(|a| {
2199                        a.iter()
2200                            .filter_map(|x| x.as_u64())
2201                            .map(|x| x as usize)
2202                            .collect()
2203                    })
2204                    .unwrap_or_default(),
2205                off: v["off"].as_u64().unwrap_or(0),
2206                nbytes: v["nbytes"].as_u64().unwrap_or(0),
2207                shard: 0,
2208                hash: v["hash"].as_u64().unwrap_or(0),
2209            });
2210        }
2211        entries.truncate(safe_entries);
2212        names.truncate(safe_entries);
2213        let cursor = safe_upto;
2214        debug_assert_eq!(
2215            entries.last().map(|e| e.off + e.nbytes).unwrap_or(0),
2216            cursor,
2217            "the last mark disagrees with the entries before it"
2218        );
2219        let on_disk = std::fs::metadata(&path)?.len();
2220        if on_disk < data_off + cursor {
2221            return Err(CmfError::Bounds(format!(
2222                "{} is {on_disk} bytes but its last checkpoint claims {} — \
2223                 the file is shorter than its own record",
2224                path.display(),
2225                data_off + cursor
2226            )));
2227        }
2228        let mut file = std::fs::OpenOptions::new().write(true).open(&path)?;
2229        file.seek(SeekFrom::Start(data_off + cursor))?;
2230        Ok((
2231            Self {
2232                file: BufWriter::new(file),
2233                path,
2234                data_off,
2235                cursor,
2236                entries,
2237                manifest: None,
2238            },
2239            ResumeState { names, marks },
2240        ))
2241    }
2242
2243    /// Note a milestone in the manifest — a source shard fully consumed, say.
2244    /// Resume reads these back, which is what lets a restart skip work whose
2245    /// payloads are already in the file rather than only skipping tensors it
2246    /// happens to recognise by name.
2247    pub fn mark(&mut self, note: &str) -> Result<(), CmfError> {
2248        // The payloads must be on disk BEFORE the mark claims they are.
2249        // Without this the manifest runs ahead of a buffered writer, and a
2250        // kill in between leaves a record of bytes that were never written.
2251        self.file.flush()?;
2252        self.file.get_ref().sync_data()?;
2253        if let Some(m) = self.manifest.as_mut() {
2254            writeln!(
2255                m,
2256                "{{\"mark\":{},\"at\":{}}}",
2257                serde_json::to_string(note).unwrap_or_default(),
2258                self.cursor
2259            )?;
2260            m.flush()?;
2261            m.get_ref().sync_data()?;
2262        }
2263        // This is deliberately after both file syncs: the converter removes
2264        // a consumed source shard only after mark() returns.
2265        sync_parent_dir(&self.path)?;
2266        Ok(())
2267    }
2268
2269    pub fn tensor_count(&self) -> usize {
2270        self.entries.len()
2271    }
2272
2273    /// Bytes of weight blob written so far.
2274    pub fn data_len(&self) -> u64 {
2275        self.cursor
2276    }
2277
2278    /// Write the trailing sections, then patch the head into the reserved gap.
2279    pub fn finish(
2280        mut self,
2281        header: &CmfHeader,
2282        masks: Option<&MaskCatalog>,
2283        vocab: Option<&[u8]>,
2284    ) -> Result<(), CmfError> {
2285        let data_len = self.cursor;
2286
2287        let masks_bytes = match masks {
2288            Some(catalog) if !catalog.masks.is_empty() => {
2289                Some(encode_masks_section(catalog, &header.arch).map_err(CmfError::Parse)?)
2290            }
2291            _ => None,
2292        };
2293        let index_bytes = match masks {
2294            Some(catalog) if !catalog.masks.is_empty() => Some(encode_sparse_index(
2295                &build_sparse_index(catalog, &header.arch),
2296            )),
2297            _ => None,
2298        };
2299        if let Some(mb) = &masks_bytes {
2300            self.file.write_all(mb)?;
2301        }
2302        if let Some(vb) = vocab {
2303            self.file.write_all(vb)?;
2304        }
2305        if let Some(ib) = &index_bytes {
2306            self.file.write_all(ib)?;
2307        }
2308        self.file.flush()?;
2309
2310        let dir_bytes = CmfModel::encode_directory(&self.entries);
2311
2312        validate_prism_affine_targets(&header.arch, &self.entries)?;
2313
2314        let hex = |b: Option<&[u8]>| b.map(|b| format!("{:016x}", hash64(b)));
2315        let mut header = header.clone();
2316        if masks_bytes.is_some() || vocab.is_some() || index_bytes.is_some() {
2317            header.section_hashes = Some(SectionHashes {
2318                masks: hex(masks_bytes.as_deref()),
2319                vocab: hex(vocab),
2320                index: hex(index_bytes.as_deref()),
2321            });
2322        }
2323        // Same contract as the ordinary writer: the bounded anchor and its
2324        // bit travel together (open() refuses a file where they disagree),
2325        // and the knowledge bits are derived from content, never set by hand.
2326        header
2327            .arch
2328            .validate_operator_metadata()
2329            .map_err(CmfError::Parse)?;
2330        header.segments.clear();
2331        let exec = crate::knowledge::ExecHashes::of(masks_bytes.as_deref(), index_bytes.as_deref());
2332        crate::knowledge::seal_genome(&mut header, &self.entries, vocab, exec)?;
2333        let required_features =
2334            derive_required_features(&header, &self.entries, masks_bytes.is_some());
2335        crate::knowledge::validate_knowledge(
2336            &header,
2337            &self.entries,
2338            vocab,
2339            required_features,
2340            Some(data_len),
2341            None,
2342            exec,
2343        )?;
2344        crate::knowledge::check_lookup_policies(&header)?;
2345        // The payloads are already on disk: read the descriptor values back.
2346        {
2347            use std::io::Read;
2348            let (file, data_off) = (self.file.get_mut(), self.data_off);
2349            crate::knowledge::validate_expert_append_values(&header, &self.entries, |e| {
2350                let mut b = [0u8; 4];
2351                file.seek(SeekFrom::Start(data_off + e.off)).ok()?;
2352                file.read_exact(&mut b).ok()?;
2353                Some(f32::from_le_bytes(b))
2354            })?;
2355            // Lookup tables are small next to the weights: read them back.
2356            crate::knowledge::validate_lookup_values(&header, &self.entries, true, |e| {
2357                let mut b = vec![0u8; e.nbytes as usize];
2358                file.seek(SeekFrom::Start(data_off + e.off)).ok()?;
2359                file.read_exact(&mut b).ok()?;
2360                Some(std::borrow::Cow::Owned(b))
2361            })?;
2362        }
2363        let header_json =
2364            serde_json::to_vec(&header).map_err(|e| CmfError::Parse(format!("header: {e}")))?;
2365
2366        let header_off = ENVELOPE_LEN as u64;
2367        let dir_off = header_off + header_json.len() as u64;
2368        let head_len = dir_off + dir_bytes.len() as u64;
2369        if head_len > self.data_off {
2370            return Err(CmfError::Parse(format!(
2371                "streamed head is {head_len} bytes but only {} were reserved — \
2372                 the payloads are already on disk at a fixed offset, so this \
2373                 file cannot be salvaged; re-run with a larger reserve",
2374                self.data_off
2375            )));
2376        }
2377        let data_off = self.data_off;
2378        let masks_off = data_off + data_len;
2379        let masks_len = masks_bytes.as_ref().map(|b| b.len() as u64).unwrap_or(0);
2380        let vocab_off = masks_off + masks_len;
2381        let vocab_len = vocab.map(|b| b.len() as u64).unwrap_or(0);
2382        let index_off = vocab_off + vocab_len;
2383        let index_len = index_bytes.as_ref().map(|b| b.len() as u64).unwrap_or(0);
2384
2385        let mut env = Vec::with_capacity(ENVELOPE_LEN);
2386        env.extend_from_slice(&CMF_MAGIC);
2387        env.extend_from_slice(&CMF_VERSION.to_le_bytes());
2388        env.extend_from_slice(&0u32.to_le_bytes());
2389        env.extend_from_slice(&required_features.to_le_bytes());
2390        for (off, len) in [
2391            (header_off, header_json.len() as u64),
2392            (dir_off, dir_bytes.len() as u64),
2393            (data_off, data_len),
2394            (if masks_len > 0 { masks_off } else { 0 }, masks_len),
2395            (if vocab_len > 0 { vocab_off } else { 0 }, vocab_len),
2396            (if index_len > 0 { index_off } else { 0 }, index_len),
2397        ] {
2398            env.extend_from_slice(&off.to_le_bytes());
2399            env.extend_from_slice(&len.to_le_bytes());
2400        }
2401        env.extend_from_slice(&hash64(&header_json).to_le_bytes());
2402        env.extend_from_slice(&hash64(&dir_bytes).to_le_bytes());
2403        env.resize(ENVELOPE_LEN, 0);
2404
2405        let mut f = self
2406            .file
2407            .into_inner()
2408            .map_err(|e| CmfError::Io(e.into_error()))?;
2409        f.seek(SeekFrom::Start(0))?;
2410        f.write_all(&env)?;
2411        f.write_all(&header_json)?;
2412        f.write_all(&dir_bytes)?;
2413        f.flush()?;
2414        f.sync_all()?;
2415        sync_parent_dir(&self.path)?;
2416
2417        tracing::info!(
2418            "Wrote CMF v2 (streamed): {} ({} tensors, {:.1} MB)",
2419            self.path.display(),
2420            self.entries.len(),
2421            (data_off + data_len + masks_len + vocab_len + index_len) as f64 / 1e6
2422        );
2423        Ok(())
2424    }
2425}
2426
2427// ───────────────────── MTP sidecar naming ─────────────────────
2428
2429/// The multi-token-prediction sidecar that belongs to a main CMF file:
2430/// `<stem>.mtp.cmf` beside it (`mimo-v26-flash-q4tp.cmf` →
2431/// `mimo-v26-flash-q4tp.mtp.cmf`). A path that already names a sidecar is
2432/// returned unchanged. The converter writes the draft layers there so a
2433/// 164 GB main file never has to be rewritten to gain (or drop) them, and
2434/// the loader looks exactly there — one rule, shared by both.
2435pub fn mtp_sidecar_path(main: &Path) -> std::path::PathBuf {
2436    let name = main
2437        .file_name()
2438        .map(|n| n.to_string_lossy().into_owned())
2439        .unwrap_or_default();
2440    if name.ends_with(".mtp.cmf") {
2441        return main.to_path_buf();
2442    }
2443    let stem = name.strip_suffix(".cmf").unwrap_or(&name);
2444    main.with_file_name(format!("{stem}.mtp.cmf"))
2445}
2446
2447/// The Qwen3 DFlash/DSpark draft sidecar that belongs to a main CMF file:
2448/// `<stem>.dspark.cmf` beside it (`minicpm-v47-q4tp.cmf` →
2449/// `minicpm-v47-q4tp.dspark.cmf`).  DSpark is a separate draft architecture,
2450/// not an MTP head, so this deliberately never returns an `.mtp.cmf` path.
2451/// In particular, passing an MTP path produces a distinct
2452/// `…mtp.dspark.cmf` name rather than overwriting the MTP payload.
2453///
2454/// A path that already names a DSpark sidecar is returned unchanged.  The
2455/// runtime does not auto-load this file until it has a dedicated DFlash
2456/// executor; keeping the naming rule here makes converter output and that
2457/// future loader share one collision-free identity.
2458pub fn dspark_sidecar_path(main: &Path) -> std::path::PathBuf {
2459    let name = main
2460        .file_name()
2461        .map(|n| n.to_string_lossy().into_owned())
2462        .unwrap_or_default();
2463    if name.ends_with(".dspark.cmf") {
2464        return main.to_path_buf();
2465    }
2466    let stem = name.strip_suffix(".cmf").unwrap_or(&name);
2467    main.with_file_name(format!("{stem}.dspark.cmf"))
2468}
2469
2470// ───────────────────── sparse index (§7 of the spec) ─────────────────────
2471
2472/// Build the sparse index from mask bitfields: a 32-neuron FFN group is
2473/// active if it contains at least one active bit.
2474pub fn build_sparse_index(catalog: &MaskCatalog, arch: &ModelArch) -> Vec<SparseIndexEntry> {
2475    let mut out = Vec::new();
2476    for m in &catalog.masks {
2477        for li in 0..arch.num_layers {
2478            if !m.layer_alive(li) {
2479                continue;
2480            }
2481            let mut groups = Vec::new();
2482            if let Some(bits) = m.ffn_masks.get(li) {
2483                let n_groups = arch.intermediate_size.div_ceil(32);
2484                for g in 0..n_groups {
2485                    // Group g covers bits [g*32, g*32+32) = bytes [g*4, g*4+4).
2486                    // A per-layer FFN width may be SHORTER than the
2487                    // arch's (tube files size the arch to the widest
2488                    // layer), so the start needs clamping too — not just
2489                    // the end, or a narrow layer indexes past its row.
2490                    let lo = (g * 4).min(bits.len());
2491                    let active = bits[lo..(g * 4 + 4).min(bits.len())]
2492                        .iter()
2493                        .any(|&b| b != 0);
2494                    if active {
2495                        groups.push(g as u16);
2496                    }
2497                }
2498            }
2499            let mut heads = Vec::new();
2500            if let Some(bits) = m.head_masks.get(li) {
2501                for h in 0..arch.num_attention_heads {
2502                    if bits
2503                        .get(h / 8)
2504                        .map(|b| b & (1 << (h % 8)) != 0)
2505                        .unwrap_or(false)
2506                    {
2507                        heads.push(h as u8);
2508                    }
2509                }
2510            }
2511            out.push(SparseIndexEntry {
2512                task_id: m.task_id,
2513                layer_idx: li,
2514                active_ffn_groups: groups,
2515                active_heads: heads,
2516            });
2517        }
2518    }
2519    out
2520}
2521
2522/// `[u32 n_entries][u32 reserved]` then per entry:
2523/// `[u32 task][u32 layer][u32 n_groups][u32 n_heads][u16×g][u8×h][pad→4]`.
2524pub fn encode_sparse_index(entries: &[SparseIndexEntry]) -> Vec<u8> {
2525    let mut out = Vec::new();
2526    out.extend_from_slice(&(entries.len() as u32).to_le_bytes());
2527    out.extend_from_slice(&0u32.to_le_bytes());
2528    for e in entries {
2529        out.extend_from_slice(&e.task_id.to_le_bytes());
2530        out.extend_from_slice(&(e.layer_idx as u32).to_le_bytes());
2531        out.extend_from_slice(&(e.active_ffn_groups.len() as u32).to_le_bytes());
2532        out.extend_from_slice(&(e.active_heads.len() as u32).to_le_bytes());
2533        for g in &e.active_ffn_groups {
2534            out.extend_from_slice(&g.to_le_bytes());
2535        }
2536        out.extend_from_slice(&e.active_heads);
2537        while out.len() % 4 != 0 {
2538            out.push(0);
2539        }
2540    }
2541    out
2542}
2543
2544pub fn decode_sparse_index(bytes: &[u8]) -> Result<Vec<SparseIndexEntry>, CmfError> {
2545    let err = |msg: &str| CmfError::Parse(format!("sparse index: {msg}"));
2546    if bytes.len() < 8 {
2547        return Err(err("too short"));
2548    }
2549    let n = u32::from_le_bytes(bytes[0..4].try_into().unwrap()) as usize;
2550    let mut pos = 8usize;
2551    let mut out = Vec::with_capacity(n);
2552    for _ in 0..n {
2553        if pos + 16 > bytes.len() {
2554            return Err(err("entry header out of bounds"));
2555        }
2556        let task_id = u32::from_le_bytes(bytes[pos..pos + 4].try_into().unwrap());
2557        let layer_idx = u32::from_le_bytes(bytes[pos + 4..pos + 8].try_into().unwrap()) as usize;
2558        let n_groups = u32::from_le_bytes(bytes[pos + 8..pos + 12].try_into().unwrap()) as usize;
2559        let n_heads = u32::from_le_bytes(bytes[pos + 12..pos + 16].try_into().unwrap()) as usize;
2560        pos += 16;
2561        if pos + n_groups * 2 + n_heads > bytes.len() {
2562            return Err(err("entry data out of bounds"));
2563        }
2564        let mut groups = Vec::with_capacity(n_groups);
2565        for g in 0..n_groups {
2566            groups.push(u16::from_le_bytes(
2567                bytes[pos + g * 2..pos + g * 2 + 2].try_into().unwrap(),
2568            ));
2569        }
2570        pos += n_groups * 2;
2571        let heads = bytes[pos..pos + n_heads].to_vec();
2572        pos += n_heads;
2573        pos = pos.div_ceil(4) * 4;
2574        out.push(SparseIndexEntry {
2575            task_id,
2576            layer_idx,
2577            active_ffn_groups: groups,
2578            active_heads: heads,
2579        });
2580    }
2581    Ok(out)
2582}
2583
2584/// Errors from CMF operations. Every failure mode is loud.
2585#[derive(Debug, thiserror::Error)]
2586pub enum CmfError {
2587    #[error("File not found: {0}")]
2588    FileNotFound(String),
2589    #[error("Invalid CMF magic bytes")]
2590    InvalidMagic,
2591    #[error("Unsupported CMF version: {0}")]
2592    UnsupportedVersion(u32),
2593    #[error("File requires unsupported features (bits {0:#x})")]
2594    UnsupportedFeature(u32),
2595    #[error("Unknown tensor dtype id: {0}")]
2596    UnknownDtype(u8),
2597    #[error("Tensor not found: {0}")]
2598    MissingTensor(String),
2599    #[error("Bounds error: {0}")]
2600    Bounds(String),
2601    #[error("IO error: {0}")]
2602    Io(#[from] io::Error),
2603    #[error("Parse error: {0}")]
2604    Parse(String),
2605}