1use 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;
44pub const DATA_ALIGNMENT: u64 = 4096;
46pub const TENSOR_ALIGNMENT: u64 = 64;
48pub const LARGE_TENSOR_MIN: u64 = 16 * 1024;
56pub const LARGE_TENSOR_ALIGN: u64 = 4096;
61pub const DIR_RECORD_LEN: usize = 56;
63pub const DIR_MAX_NDIM: usize = 6;
64
65pub 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 pub const LOOP_MASKS: u32 = 1 << 5;
79 pub const SKILL_FILE: u32 = 1 << 6;
85 pub const PRISM_HADAMARD: u32 = 1 << 7;
87 pub const PRISM_AFFINE: u32 = 1 << 8;
91 pub const SKILLS_V2: u32 = 1 << 9;
95 pub const ROUTER_V2: u32 = 1 << 10;
99 pub const DECISION: u32 = 1 << 11;
107 pub const BOUNDED_STATE: u32 = 1 << 12;
120 pub const GENOME: u32 = 1 << 13;
128
129 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
144pub const DECISION_ARCH_PREFIX: &str = "cortiq-decision-";
149
150pub fn is_decision_profile(arch_name: &str) -> bool {
152 arch_name.starts_with(DECISION_ARCH_PREFIX)
153}
154
155fn 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
168fn 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
184pub 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
193pub 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
228pub 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#[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 pub quant_type: QuantType,
289 #[serde(default, skip_serializing_if = "Option::is_none")]
290 pub provenance: Option<serde_json::Value>,
291 #[serde(default, skip_serializing_if = "Option::is_none")]
294 pub tokenizer_config: Option<TokenizerBundle>,
295 #[serde(default, skip_serializing_if = "Option::is_none")]
299 pub section_hashes: Option<SectionHashes>,
300 #[serde(default, skip_serializing_if = "Vec::is_empty")]
304 pub skills: Vec<SkillRecord>,
305 #[serde(default, skip_serializing_if = "Option::is_none")]
309 pub shard: Option<ShardInfo>,
310 #[serde(default, skip_serializing_if = "Option::is_none")]
316 pub calibration: Option<Calibration>,
317 #[serde(default, skip_serializing_if = "Option::is_none")]
322 pub routing: Option<RoutingCalibration>,
323 #[serde(default, skip_serializing_if = "Option::is_none")]
326 pub genome: Option<GenomeInfo>,
327 #[serde(default, skip_serializing_if = "Vec::is_empty")]
329 pub lineage: Vec<LineageEvent>,
330 #[serde(default, skip_serializing_if = "Option::is_none")]
333 pub router: Option<RouterPolicy>,
334 #[serde(default, skip_serializing_if = "Vec::is_empty")]
337 pub segments: Vec<Segment>,
338}
339
340#[derive(Debug, Clone, Serialize, Deserialize)]
346pub struct RoutingCalibration {
347 pub temperature: f32,
348 pub novelty_theta: f32,
349 pub samples: usize,
351 pub target_fpr: f32,
353}
354
355#[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#[derive(Debug, Clone, Serialize, Deserialize)]
369pub struct ShardInfo {
370 pub no: usize,
371 pub count: usize,
372}
373
374#[derive(Debug, Clone, Serialize, Deserialize)]
377pub struct SelectionDescriptor {
378 pub metric: String,
381 pub phi_layer: usize,
383 pub mean: String,
385 pub basis: String,
387 pub rank: usize,
388 #[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 #[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#[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 #[serde(default)]
411 pub layers: Vec<usize>,
412 #[serde(default, skip_serializing_if = "Option::is_none")]
415 pub selection: Option<SelectionDescriptor>,
416 #[serde(default, skip_serializing_if = "Option::is_none")]
418 pub input_mask_task: Option<String>,
419 #[serde(default, skip_serializing_if = "Option::is_none")]
421 pub quality: Option<serde_json::Value>,
422
423 #[serde(default, skip_serializing_if = "Option::is_none")]
428 pub base_dir_hash: Option<String>,
429 #[serde(default, skip_serializing_if = "Option::is_none")]
432 pub base_arch: Option<String>,
433 #[serde(default, skip_serializing_if = "Option::is_none")]
435 pub task: Option<String>,
436 #[serde(default, skip_serializing_if = "Option::is_none")]
438 pub provenance: Option<serde_json::Value>,
439
440 #[serde(default, skip_serializing_if = "Option::is_none")]
445 pub kind: Option<String>,
446 #[serde(default, skip_serializing_if = "Option::is_none")]
450 pub experts: Option<ExpertAppend>,
451 #[serde(default, skip_serializing_if = "Option::is_none")]
455 pub lookup: Option<LookupInfo>,
456 #[serde(default, skip_serializing_if = "Vec::is_empty")]
459 pub overrides: Vec<SkillOverride>,
460 #[serde(default, skip_serializing_if = "Option::is_none")]
463 pub bound: Option<SkillBound>,
464 #[serde(default, skip_serializing_if = "Option::is_none")]
467 pub state_effect: Option<StateEffect>,
468 #[serde(default, skip_serializing_if = "Option::is_none")]
470 pub status: Option<String>,
471 #[serde(default, skip_serializing_if = "Option::is_none")]
474 pub gate: Option<serde_json::Value>,
475 #[serde(default, skip_serializing_if = "Option::is_none")]
477 pub prompt_contract: Option<String>,
478 #[serde(default, skip_serializing_if = "Option::is_none")]
480 pub origin: Option<serde_json::Value>,
481}
482
483#[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#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
497pub struct TokenizerBundle {
498 #[serde(default, skip_serializing_if = "Option::is_none")]
500 pub chat_template: Option<String>,
501 #[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#[derive(Debug, Clone, PartialEq, Eq)]
516pub struct TensorEntry {
517 pub name: String,
518 pub dtype: TensorDtype,
519 pub shape: Vec<usize>,
520 pub off: u64,
522 pub nbytes: u64,
523 pub shard: usize,
526 pub hash: u64,
528}
529
530impl TensorEntry {
531 pub fn n_elems(&self) -> usize {
532 self.shape.iter().product()
533 }
534}
535
536#[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
545pub struct TensorSpecRef<'a> {
547 pub name: String,
548 pub dtype: TensorDtype,
549 pub shape: Vec<usize>,
550 pub data: &'a [u8],
551}
552
553#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
555pub struct SparseIndexEntry {
556 pub task_id: u32,
557 pub layer_idx: usize,
558 pub active_ffn_groups: Vec<u16>,
560 pub active_heads: Vec<u8>,
562}
563
564#[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 header_hash: u64,
576 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
594pub struct CmfModel {
596 pub path: PathBuf,
597 uid: u64,
603 pub header: CmfHeader,
604 pub required_features: u32,
605 pub tensors: Vec<TensorEntry>,
606 by_name: HashMap<u64, u32>,
613 name_overflow: Vec<u32>,
614 pub masks: MaskCatalog,
615 pub sparse_index: Vec<SparseIndexEntry>,
616 pub vocab: Option<Vec<u8>>,
618 backing: Backing,
619 data_off: u64,
620 envelope: Envelope,
621 extra_shards: Vec<(Backing, u64)>,
624 trunk: Option<u64>,
626 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
643static MODEL_UID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
645
646impl CmfModel {
647 pub fn uid(&self) -> u64 {
651 self.uid
652 }
653
654 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 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 #[cfg(unix)]
685 {
686 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 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 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 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 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 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 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 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), None => {
870 by_name.insert(h, i as u32);
871 }
872 }
873 }
874
875 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 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 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 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 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 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 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 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); };
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); 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 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 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 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 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 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 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 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 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 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 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 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 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 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 let _ = unsafe {
1441 m.unchecked_advise_range(memmap2::UncheckedAdvice::DontNeed, start, alen)
1442 };
1443 if let Some(f) = &file {
1444 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 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 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 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 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 pub fn verify(&self) -> Vec<String> {
1552 let mut problems = Vec::new();
1553
1554 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 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 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 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 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 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 header.segments.clear();
1748 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 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 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 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 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()); 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 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 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
1881pub 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
1892pub 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 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 if header.skills.iter().any(|s| s.base_dir_hash.is_some()) {
1922 f |= features::SKILL_FILE;
1923 }
1924 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
1936fn 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
1955pub 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 data_off: u64,
1983 cursor: u64,
1985 entries: Vec<TensorEntry>,
1986 manifest: Option<BufWriter<File>>,
1990}
1991
1992impl CmfStreamWriter {
1993 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 align_to(3 * head as u64, DATA_ALIGNMENT).max(1 << 20)
2005 }
2006
2007 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 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 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 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 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 f.get_ref().sync_data()?;
2127 sync_parent_dir(&path)?;
2128 self.manifest = Some(f);
2129 Ok(self)
2130 }
2131
2132 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 file.sync_data()?;
2140 sync_parent_dir(&path)?;
2141 self.manifest = Some(BufWriter::new(file));
2142 Ok(self)
2143 }
2144
2145 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 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 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 pub fn mark(&mut self, note: &str) -> Result<(), CmfError> {
2248 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 sync_parent_dir(&self.path)?;
2266 Ok(())
2267 }
2268
2269 pub fn tensor_count(&self) -> usize {
2270 self.entries.len()
2271 }
2272
2273 pub fn data_len(&self) -> u64 {
2275 self.cursor
2276 }
2277
2278 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 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 {
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 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
2427pub 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
2447pub 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
2470pub 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 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
2522pub 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#[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}