use anyhow::{Context, Result, ensure};
use cortiq_core::quant::{
Q2TP_CHUNK, Q4TP_NIB, bf16_to_f32, dequant_tensor, expected_nbytes, f16_to_f32,
q2tp_ladder, q2tp_sections, q4tp_code, q4tp_ladder, q4tp_sections,
};
use cortiq_core::{CmfModel, TensorDtype, mtp_sidecar_path};
use cortiq_engine::{MtpFfnProbe, Pipeline, SamplerConfig, tokenizer::Tokenizer};
use serde_json::{Map, Value, json};
use std::cmp::Ordering;
use std::fs::{self, OpenOptions};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::time::Instant;
use crate::tdata::{TdataReader, TdataSplitSelector};
use crate::tube::paths_collide;
const MTP_COMPACT_ALIGNMENT: usize = 32;
const MTP_ADMISSION_TARGET_TOKENS: usize = 256;
const MTP_PLAIN_ADMISSION_PROMPT_TOKENS: usize = 16;
const MTP_PLAIN_ADMISSION_OUTPUT_TOKENS: usize = 2;
static STAGE_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[derive(Debug)]
struct CompactDestination {
main: PathBuf,
draft: PathBuf,
external_draft: bool,
publish_main: bool,
}
fn has_cmf_suffix(path: &Path) -> bool {
path.extension().and_then(|extension| extension.to_str()) == Some("cmf")
}
fn is_mtp_sidecar(path: &Path) -> bool {
mtp_sidecar_path(path) == path
}
fn main_path_for_mtp_sidecar(sidecar: &Path) -> Result<PathBuf> {
let name = sidecar
.file_name()
.and_then(|name| name.to_str())
.context("MTP compact: legacy sidecar output has no UTF-8 filename")?;
let stem = name.strip_suffix(".mtp.cmf").ok_or_else(|| {
anyhow::anyhow!(
"MTP compact: legacy sidecar output {} must end in .mtp.cmf",
sidecar.display()
)
})?;
ensure!(
!stem.is_empty(),
"MTP compact: legacy sidecar output {} has an empty main-model stem",
sidecar.display()
);
Ok(sidecar.with_file_name(format!("{stem}.cmf")))
}
fn parent_dir(path: &Path) -> &Path {
path.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| if path.is_absolute() { path } else { Path::new(".") })
}
fn prospective_path_key(path: &Path) -> Result<PathBuf> {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()
.context("MTP compact: reading current directory for output path")?
.join(path)
};
let mut ancestor = absolute.as_path();
let canonical_ancestor = loop {
match fs::canonicalize(ancestor) {
Ok(canonical) => break canonical,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
let next = ancestor.parent().ok_or_else(|| {
anyhow::anyhow!(
"MTP compact: no existing ancestor while normalizing {}",
path.display()
)
})?;
if next == ancestor {
anyhow::bail!(
"MTP compact: no existing ancestor while normalizing {}",
path.display()
);
}
ancestor = next;
}
Err(error) => {
return Err(error).with_context(|| {
format!(
"MTP compact: resolving existing ancestor {} for {}",
ancestor.display(),
path.display()
)
});
}
}
};
let suffix = absolute.strip_prefix(ancestor).with_context(|| {
format!(
"MTP compact: deriving prospective path suffix {} from {}",
absolute.display(),
ancestor.display()
)
})?;
let mut key = canonical_ancestor;
for component in suffix.components() {
use std::path::Component;
match component {
Component::CurDir => {}
Component::Normal(name) => key.push(name),
Component::ParentDir => {
if !key.pop() {
anyhow::bail!(
"MTP compact: path {} escapes filesystem root while normalizing",
path.display()
);
}
}
Component::RootDir | Component::Prefix(_) => anyhow::bail!(
"MTP compact: unexpected absolute component while normalizing {}",
path.display()
),
}
}
Ok(key)
}
fn paths_overlap_or_nested(a: &Path, b: &Path) -> Result<bool> {
if paths_collide(a, b) {
return Ok(true);
}
let a_key = prospective_path_key(a)?;
let b_key = prospective_path_key(b)?;
Ok(a_key == b_key || a_key.starts_with(&b_key) || b_key.starts_with(&a_key))
}
fn compact_destination(
main_path: &Path,
draft_path: &Path,
output: &Path,
) -> Result<CompactDestination> {
ensure!(
has_cmf_suffix(output),
"MTP compact: --output must name a new .cmf candidate"
);
let external_draft = !paths_collide(main_path, draft_path);
if is_mtp_sidecar(output) {
ensure!(
external_draft,
"MTP compact: --output ending in .mtp.cmf is a legacy sidecar-only form and is valid only when MODEL has an external MTP sidecar"
);
return Ok(CompactDestination {
main: main_path_for_mtp_sidecar(output)?,
draft: output.to_path_buf(),
external_draft: true,
publish_main: false,
});
}
if !external_draft {
return Ok(CompactDestination {
main: output.to_path_buf(),
draft: output.to_path_buf(),
external_draft: false,
publish_main: true,
});
}
let draft = mtp_sidecar_path(output);
Ok(CompactDestination {
main: output.to_path_buf(),
draft,
external_draft: true,
publish_main: true,
})
}
fn require_main_model_input(path: &Path) -> Result<()> {
ensure!(
mtp_sidecar_path(path) != path,
"MTP compact: MODEL must name the main .cmf, not an .mtp.cmf sidecar; pass its companion main CMF"
);
Ok(())
}
fn check_new_file_leaf(path: &Path, label: &str) -> Result<()> {
path.file_name()
.filter(|name| !name.is_empty())
.ok_or_else(|| {
anyhow::anyhow!("MTP compact: {label} {} names no file", path.display())
})?;
match fs::symlink_metadata(path) {
Ok(_) => anyhow::bail!(
"MTP compact: refusing to overwrite existing {label} {}",
path.display()
),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error)
.with_context(|| format!("MTP compact: checking {label} {}", path.display())),
}
}
fn preflight_explicit_paths(
source_main: &Path,
output: &Path,
plan_path: &Path,
) -> Result<()> {
ensure!(
has_cmf_suffix(output),
"MTP compact: --output must name a new .cmf candidate"
);
ensure!(
!paths_overlap_or_nested(output, source_main)?
&& !paths_overlap_or_nested(plan_path, source_main)?
&& !paths_overlap_or_nested(plan_path, output)?,
"MTP compact: candidate output and --plan-out must be distinct ordinary-file paths, not aliases or nested below source/candidate paths"
);
check_new_file_leaf(output, "candidate output")?;
check_new_file_leaf(plan_path, "MTP plan")?;
Ok(())
}
fn prepare_new_file(path: &Path, label: &str) -> Result<()> {
check_new_file_leaf(path, label)?;
let parent = parent_dir(path);
let mut created_parents = Vec::new();
let mut ancestor = parent;
loop {
match fs::symlink_metadata(ancestor) {
Ok(_) => break,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
created_parents.push(ancestor.to_path_buf());
let next = parent_dir(ancestor);
ensure!(
next != ancestor,
"MTP compact: no existing ancestor while creating parent {} for {label}",
parent.display()
);
ancestor = next;
}
Err(error) => {
return Err(error).with_context(|| {
format!(
"MTP compact: checking parent ancestor {} for {label}",
ancestor.display()
)
});
}
}
}
fs::create_dir_all(parent).with_context(|| {
format!(
"MTP compact: creating parent for {label} {}",
path.display()
)
})?;
for created in created_parents.iter().rev() {
sync_parent_dir(created, "creating output parent directory")?;
}
ensure!(
fs::metadata(parent)
.with_context(|| format!("MTP compact: stat parent {}", parent.display()))?
.is_dir(),
"MTP compact: parent {} for {label} is not a directory",
parent.display()
);
check_new_file_leaf(path, label)
}
fn sync_parent_dir(path: &Path, what: &str) -> Result<()> {
#[cfg(unix)]
{
let parent = parent_dir(path);
fs::File::open(parent)
.with_context(|| {
format!(
"MTP compact: opening parent directory {} to sync {what}",
parent.display()
)
})?
.sync_all()
.with_context(|| {
format!(
"MTP compact: syncing parent directory {} after {what}",
parent.display()
)
})?;
}
#[cfg(not(unix))]
let _ = (path, what);
Ok(())
}
fn create_stage_dir(output: &Path, kind: &str) -> Result<PathBuf> {
let name = output
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| {
anyhow::anyhow!(
"MTP compact: staging output {} has no UTF-8 name",
output.display()
)
})?;
let parent = parent_dir(output);
for _ in 0..64 {
let nonce = STAGE_SEQUENCE
.fetch_add(1, AtomicOrdering::Relaxed)
.wrapping_add(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_nanos() as u64)
.unwrap_or(0),
);
let staged = parent.join(format!(
".{name}.mtp-compact-{kind}-{}-{nonce:016x}.stage",
std::process::id()
));
match fs::create_dir(&staged) {
Ok(()) => {
if let Err(error) = sync_parent_dir(&staged, "creating private stage") {
let _ = fs::remove_dir(&staged);
return Err(error);
}
return Ok(staged);
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
Err(error) => {
return Err(error).with_context(|| {
format!("MTP compact: creating private stage {}", staged.display())
});
}
}
}
anyhow::bail!(
"MTP compact: could not create a private stage beside {}",
output.display()
)
}
fn staged_child(stage_dir: &Path, output: &Path) -> Result<PathBuf> {
let name = output
.file_name()
.filter(|name| !name.is_empty())
.ok_or_else(|| anyhow::anyhow!("MTP compact: {} names no file", output.display()))?;
Ok(stage_dir.join(name))
}
fn publish_new_file(staged: &Path, output: &Path, label: &str) -> Result<()> {
match fs::hard_link(staged, output) {
Ok(()) => {
sync_parent_dir(output, &format!("publishing {label}"))?;
if let Err(error) = fs::remove_file(staged) {
eprintln!(
"warning: MTP compact published {label} {}, but could not remove private stage {}: {error}",
output.display(),
staged.display()
);
}
Ok(())
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => anyhow::bail!(
"MTP compact: refusing to overwrite {label} {}; staged validated artifact remains at {}",
output.display(),
staged.display()
),
Err(error) => Err(error).with_context(|| {
format!(
"MTP compact: atomically publishing {label} {}; staged validated artifact remains at {}",
output.display(),
staged.display()
)
}),
}
}
fn stage_backbone(source: &Path, staged: &Path) -> Result<&'static str> {
let source = fs::canonicalize(source)
.with_context(|| format!("MTP compact: resolving source backbone {}", source.display()))?;
match fs::hard_link(&source, staged) {
Ok(()) => {
sync_parent_dir(staged, "staging unchanged backbone")?;
Ok("hard-link")
}
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => anyhow::bail!(
"MTP compact: private backbone stage already exists: {}",
staged.display()
),
Err(link_error) => {
let mut input = fs::File::open(&source).with_context(|| {
format!("MTP compact: opening source backbone {}", source.display())
})?;
let expected = input
.metadata()
.with_context(|| format!("MTP compact: stat source backbone {}", source.display()))?
.len();
let mut output = OpenOptions::new()
.write(true)
.create_new(true)
.open(staged)
.with_context(|| {
format!(
"MTP compact: creating private backbone stage {} after hard-link failure ({link_error})",
staged.display()
)
})?;
let copied = io::copy(&mut input, &mut output).with_context(|| {
format!(
"MTP compact: copying {} to private stage {}",
source.display(),
staged.display()
)
})?;
output.sync_data().with_context(|| {
format!("MTP compact: syncing private backbone stage {}", staged.display())
})?;
ensure!(
copied == expected,
"MTP compact: incomplete backbone copy {} -> {} ({copied} != {expected} bytes)",
source.display(),
staged.display()
);
sync_parent_dir(staged, "staging copied backbone")?;
Ok("copy")
}
}
}
fn write_new_file(path: &Path, data: &[u8], label: &str) -> Result<()> {
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.with_context(|| format!("MTP compact: creating staged {label} {}", path.display()))?;
file.write_all(data)
.with_context(|| format!("MTP compact: writing staged {label} {}", path.display()))?;
file.sync_data()
.with_context(|| format!("MTP compact: syncing staged {label} {}", path.display()))?;
sync_parent_dir(path, &format!("writing staged {label}"))?;
Ok(())
}
fn verify_open_cmf(model: &CmfModel, path: &Path, label: &str) -> Result<()> {
let started = Instant::now();
eprintln!("mtp compact: verifying {label} {}", path.display());
let integrity = model.verify();
ensure!(
integrity.is_empty(),
"MTP compact: {label} integrity failed: {}",
integrity
.iter()
.take(3)
.cloned()
.collect::<Vec<_>>()
.join("; ")
);
eprintln!(
"mtp compact: verified {label} in {:.1}s",
started.elapsed().as_secs_f64()
);
Ok(())
}
fn verify_cmf(path: &Path, label: &str) -> Result<Arc<CmfModel>> {
let model = Arc::new(
CmfModel::open_sharded(path)
.with_context(|| format!("MTP compact: reopening {label} {}", path.display()))?,
);
verify_open_cmf(&model, path, label)?;
Ok(model)
}
fn validate_tdata_tokenizer_contract(
manifest: &Value,
vocab: Option<&[u8]>,
tokenizer_config_bytes: &[u8],
source_vocab_size: usize,
) -> Result<&'static str> {
let Some(tokenizer) = manifest.get("tokenizer") else {
eprintln!(
"mtp compact: warning: calibration .tdata has no tokenizer contract; token-id compatibility is not cryptographically bound to the source model"
);
return Ok("absent-legacy");
};
let tokenizer = tokenizer
.as_object()
.context("MTP compact: .tdata manifest.tokenizer must be an object")?;
let expected_vocab_size = tokenizer
.get("vocab_size")
.and_then(Value::as_u64)
.context("MTP compact: .tdata manifest.tokenizer.vocab_size is required")?;
let expected_vocab_size = usize::try_from(expected_vocab_size)
.context("MTP compact: .tdata tokenizer vocab_size does not fit this platform")?;
ensure!(
expected_vocab_size == source_vocab_size,
"MTP compact: .tdata tokenizer vocab_size {expected_vocab_size} differs from source tokenizer {source_vocab_size}"
);
let expected_vocab_hash = tokenizer
.get("vocab_sha256")
.and_then(Value::as_str)
.context("MTP compact: .tdata manifest.tokenizer.vocab_sha256 is required")?;
let source_vocab = vocab.context(
"MTP compact: .tdata has a tokenizer contract but source main CMF has no embedded tokenizer bytes",
)?;
let source_vocab_hash = crate::knowledge::sha256_hex(source_vocab);
ensure!(
expected_vocab_hash == source_vocab_hash,
"MTP compact: .tdata tokenizer vocab_sha256 differs from source main tokenizer"
);
let expected_config_hash = tokenizer
.get("config_sha256")
.and_then(Value::as_str)
.context("MTP compact: .tdata manifest.tokenizer.config_sha256 is required")?;
let source_config_hash = crate::knowledge::sha256_hex(tokenizer_config_bytes);
ensure!(
expected_config_hash == source_config_hash,
"MTP compact: .tdata tokenizer config_sha256 differs from source main tokenizer metadata"
);
Ok("verified-v1")
}
fn revalidate_source_identity(
phase: &str,
main_path: &Path,
draft_path: &Path,
expected_main_hash: u64,
expected_draft_hash: u64,
expected_main_trunk_hash: u64,
expected_draft_trunk_hash: u64,
) -> Result<Arc<CmfModel>> {
let main_label = format!("source main {phase}");
let main = verify_cmf(main_path, &main_label)?;
ensure!(
main.dir_hash() == expected_main_hash,
"MTP compact: source main changed {phase} (expected directory hash {expected_main_hash:016x}, got {:016x})",
main.dir_hash()
);
ensure!(
main.trunk_hash() == expected_main_trunk_hash,
"MTP compact: source main tensors changed {phase} (expected trunk hash {expected_main_trunk_hash:016x}, got {:016x})",
main.trunk_hash()
);
if paths_collide(main_path, draft_path) {
ensure!(
expected_main_hash == expected_draft_hash
&& expected_main_trunk_hash == expected_draft_trunk_hash,
"MTP compact: embedded source main/draft identity changed during calibration"
);
} else {
let draft_label = format!("source MTP draft {phase}");
let draft = verify_cmf(draft_path, &draft_label)?;
ensure!(
draft.dir_hash() == expected_draft_hash,
"MTP compact: source MTP draft changed {phase} (expected directory hash {expected_draft_hash:016x}, got {:016x})",
draft.dir_hash()
);
ensure!(
draft.trunk_hash() == expected_draft_trunk_hash,
"MTP compact: source MTP draft tensors changed {phase} (expected trunk hash {expected_draft_trunk_hash:016x}, got {:016x})",
draft.trunk_hash()
);
}
Ok(main)
}
fn plain_cpu_greedy_admission(
model: &Arc<CmfModel>,
probe_tokens: &[u32],
label: &str,
) -> Result<Vec<u32>> {
let prompt_len = probe_tokens.len().min(MTP_PLAIN_ADMISSION_PROMPT_TOKENS);
ensure!(
prompt_len >= 2,
"MTP compact: {label} CPU greedy admission prefix is too short"
);
let mut sampler = SamplerConfig::default();
sampler.temperature = 0.0;
sampler.top_p = 1.0;
sampler.top_k = 0;
sampler.min_p = 0.0;
sampler.repetition_penalty = 1.0;
sampler.presence_penalty = 0.0;
sampler.suppress_tokens.clear();
let result = cortiq_engine::gpu::cpu_scope(|| -> std::result::Result<_, String> {
let mut pipeline =
Pipeline::from_model(model, sampler).map_err(|error| error.to_string())?;
pipeline.mtp = None;
pipeline.mimo_mtp = None;
pipeline.speculative = false;
pipeline.ignore_eos = true;
pipeline.generate_from_ids(
&probe_tokens[..prompt_len],
MTP_PLAIN_ADMISSION_OUTPUT_TOKENS,
None,
None,
)
})
.map_err(|error| anyhow::anyhow!("MTP compact: CPU plain greedy admission for {label}: {error}"))?;
ensure!(
result.token_ids.len() == MTP_PLAIN_ADMISSION_OUTPUT_TOKENS,
"MTP compact: CPU plain greedy admission for {label} emitted {} instead of {} token(s)",
result.token_ids.len(),
MTP_PLAIN_ADMISSION_OUTPUT_TOKENS
);
Ok(result.token_ids)
}
fn admit_candidate(
source_main: &Arc<CmfModel>,
destination: &CompactDestination,
staged_main: &Path,
staged_draft: &Path,
expected_mimo: bool,
expected_unchanged_main_trunk_hash: Option<u64>,
probe_tokens: &[u32],
kept_widths: &[usize],
) -> Result<()> {
let main = verify_cmf(staged_main, "staged candidate main")?;
if let Some(expected_main_trunk_hash) = expected_unchanged_main_trunk_hash {
ensure!(
main.trunk_hash() == expected_main_trunk_hash,
"MTP compact: staged candidate main differs from calibrated source (expected trunk hash {expected_main_trunk_hash:016x}, got {:016x})",
main.trunk_hash()
);
}
if destination.external_draft {
let _draft = verify_cmf(staged_draft, "staged candidate MTP sidecar")?;
}
let mut pipeline = Pipeline::from_model(&main, SamplerConfig::default()).map_err(|error| {
anyhow::anyhow!("MTP compact: loading staged candidate runtime: {error}")
})?;
if expected_mimo {
ensure!(
pipeline.mimo_mtp.is_some(),
"MTP compact: staged candidate did not load its MiMo MTP draft"
);
} else {
ensure!(
pipeline.mtp.is_some(),
"MTP compact: staged candidate did not load its MTP draft"
);
}
let probe = pipeline.probe_mtp_ffn(probe_tokens).map_err(|error| {
anyhow::anyhow!("MTP compact: staged candidate draft runtime probe failed: {error}")
})?;
validate_probe(&probe).context("MTP compact: staged candidate returned invalid draft probe")?;
ensure!(
probe.len() == kept_widths.len(),
"MTP compact: staged candidate has {} draft layers, plan kept {}",
probe.len(),
kept_widths.len()
);
for (layer, (row, &kept_width)) in probe.iter().zip(kept_widths).enumerate() {
ensure!(
row.observed_positions > 0
&& row.mass.len() == kept_width
&& row.squared_mass.len() == kept_width,
"MTP compact: staged candidate layer {layer} did not run at its compact width {kept_width}"
);
}
drop(pipeline);
let source_ids = plain_cpu_greedy_admission(source_main, probe_tokens, "source main")?;
let candidate_ids = plain_cpu_greedy_admission(&main, probe_tokens, "staged candidate")?;
ensure!(
source_ids == candidate_ids,
"MTP compact: CPU plain greedy admission differs between source and candidate (source {:?}, candidate {:?})",
source_ids,
candidate_ids
);
Ok(())
}
fn source_hash_fields(main_hash: u64, draft_hash: u64) -> Map<String, Value> {
let mut fields = Map::with_capacity(2);
fields.insert(
"source_directory_hash".into(),
json!(format!("{main_hash:016x}")),
);
fields.insert(
"source_draft_directory_hash".into(),
json!(format!("{draft_hash:016x}")),
);
fields
}
fn source_trunk_hash_fields(main_hash: u64, draft_hash: u64) -> Map<String, Value> {
let mut fields = Map::with_capacity(2);
fields.insert(
"source_trunk_hash".into(),
json!(format!("{main_hash:016x}")),
);
fields.insert(
"source_draft_trunk_hash".into(),
json!(format!("{draft_hash:016x}")),
);
fields
}
fn tensor_f32(model: &CmfModel, name: &str) -> Result<Vec<f32>> {
let entry = model
.tensor(name)
.with_context(|| format!("MTP compact: missing tensor {name}"))?;
let mut values = vec![0.0f32; entry.n_elems()];
dequant_tensor(entry, model.tensor_bytes(name)?, &mut values)
.map_err(|error| anyhow::anyhow!("MTP compact: dequantizing {name}: {error}"))?;
Ok(values)
}
fn direct_down_column_norm_sq(
dtype: TensorDtype,
source: &[u8],
rows: usize,
cols: usize,
) -> Result<Option<Vec<f64>>> {
ensure!(
cols > 0,
"MTP compact: direct down-projection scoring needs a positive column width"
);
let expected = || {
expected_nbytes(dtype, &[rows, cols])
.context("MTP compact: direct down-projection score shape overflows")
};
let require_payload = |expected: usize| -> Result<()> {
ensure!(
source.len() == expected,
"MTP compact: direct down-projection score payload has {} bytes, expected {expected}",
source.len()
);
Ok(())
};
match dtype {
TensorDtype::F32 => {
require_payload(expected()?)?;
let mut column_norm_sq = vec![0.0f64; cols];
for (index, bytes) in source.chunks_exact(4).enumerate() {
let value = f32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as f64;
let column = index % cols;
column_norm_sq[column] += value * value;
}
return Ok(Some(column_norm_sq));
}
TensorDtype::F16 | TensorDtype::Bf16 => {
require_payload(expected()?)?;
let mut column_norm_sq = vec![0.0f64; cols];
for (index, bytes) in source.chunks_exact(2).enumerate() {
let bits = u16::from_le_bytes([bytes[0], bytes[1]]);
let value = match dtype {
TensorDtype::F16 => f16_to_f32(bits),
TensorDtype::Bf16 => bf16_to_f32(bits),
_ => unreachable!("matched f16/bf16 dtype"),
} as f64;
let column = index % cols;
column_norm_sq[column] += value * value;
}
return Ok(Some(column_norm_sq));
}
TensorDtype::Q8Row => {
require_payload(expected()?)?;
let q_len = rows
.checked_mul(cols)
.context("MTP compact: direct q8 score weight plane overflows")?;
let (quant, scales) = source.split_at(q_len);
let mut column_norm_sq = vec![0.0f64; cols];
for row in 0..rows {
let scale = f16_to_f32(u16::from_le_bytes([
scales[row * 2],
scales[row * 2 + 1],
]));
for column in 0..cols {
let value = quant[row * cols + column] as i8 as f32 * scale;
let value = value as f64;
column_norm_sq[column] += value * value;
}
}
return Ok(Some(column_norm_sq));
}
TensorDtype::Q8_2f => {
require_payload(expected()?)?;
let q_len = rows
.checked_mul(cols)
.context("MTP compact: direct q8_2f score weight plane overflows")?;
let row_scale_len = rows
.checked_mul(2)
.context("MTP compact: direct q8_2f score row-scale plane overflows")?;
let (quant, rest) = source.split_at(q_len);
let (row_scales, column_scales) = rest.split_at(row_scale_len);
let column_scales = (0..cols)
.map(|column| {
f16_to_f32(u16::from_le_bytes([
column_scales[column * 2],
column_scales[column * 2 + 1],
]))
})
.collect::<Vec<_>>();
let mut column_norm_sq = vec![0.0f64; cols];
for row in 0..rows {
let row_scale = f16_to_f32(u16::from_le_bytes([
row_scales[row * 2],
row_scales[row * 2 + 1],
]));
for column in 0..cols {
let value = quant[row * cols + column] as i8 as f32
* row_scale
* column_scales[column];
let value = value as f64;
column_norm_sq[column] += value * value;
}
}
return Ok(Some(column_norm_sq));
}
_ => {}
}
let tile_bytes = match dtype {
TensorDtype::Q4TiledP => Q4TP_NIB,
TensorDtype::Q2TiledP => Q2TP_CHUNK,
_ => return Ok(None),
};
ensure!(
cols != 0 && cols % 32 == 0,
"MTP compact: {dtype:?} direct down-projection scoring needs a positive 32-aligned width, got {cols}"
);
require_payload(expected()?)?;
let groups_per_row = cols / 32;
let row_weight_bytes = groups_per_row
.checked_mul(tile_bytes)
.context("MTP compact: direct down-projection score row width overflows")?;
let (params_offset, codes_offset, code_stride) = match dtype {
TensorDtype::Q4TiledP => q4tp_sections(rows, cols),
TensorDtype::Q2TiledP => q2tp_sections(rows, cols),
_ => unreachable!("non-tiled dtype returned above"),
};
ensure!(
params_offset
== rows
.checked_mul(row_weight_bytes)
.context("MTP compact: direct down-projection score plane offset overflows")?,
"MTP compact: direct down-projection score plane layout is inconsistent"
);
let params_len = rows
.checked_mul(4)
.context("MTP compact: direct down-projection score parameter plane overflows")?;
let params_end = params_offset
.checked_add(params_len)
.context("MTP compact: direct down-projection score parameter plane end overflows")?;
let codes_len = rows
.checked_mul(code_stride)
.context("MTP compact: direct down-projection score code plane overflows")?;
ensure!(
params_end <= source.len()
&& codes_offset
.checked_add(codes_len)
.is_some_and(|end| end == source.len()),
"MTP compact: direct down-projection score plane bounds are inconsistent"
);
let params = &source[params_offset..params_end];
let mut column_norm_sq = vec![0.0f64; cols];
for row in 0..rows {
let codes = &source[codes_offset + row * code_stride..codes_offset + (row + 1) * code_stride];
match dtype {
TensorDtype::Q4TiledP => {
let ladder = q4tp_ladder(params, row);
for group in 0..groups_per_row {
let scale = ladder[q4tp_code(codes, group)];
let tile_offset = row * row_weight_bytes + group * Q4TP_NIB;
for (index, &byte) in source[tile_offset..tile_offset + Q4TP_NIB]
.iter()
.enumerate()
{
let column = group * 32 + index * 2;
let low = ((byte & 0x0f) as f32 - 8.0) * scale;
let high = ((byte >> 4) as f32 - 8.0) * scale;
let low = low as f64;
let high = high as f64;
column_norm_sq[column] += low * low;
column_norm_sq[column + 1] += high * high;
}
}
}
TensorDtype::Q2TiledP => {
let ladder = q2tp_ladder(params, row);
for group in 0..groups_per_row {
let scale = ladder[q4tp_code(codes, group)];
let tile_offset = row * row_weight_bytes + group * Q2TP_CHUNK;
for (index, &byte) in source[tile_offset..tile_offset + Q2TP_CHUNK]
.iter()
.enumerate()
{
let column = group * 32 + index * 4;
for shift in 0..4 {
let value = (((byte >> (2 * shift)) & 0x03) as f32 - 1.5) * scale;
let value = value as f64;
column_norm_sq[column + shift] += value * value;
}
}
}
}
_ => unreachable!("non-tiled dtype returned above"),
}
}
Ok(Some(column_norm_sq))
}
fn down_column_norm_sq(model: &CmfModel, name: &str, rows: usize, cols: usize) -> Result<Vec<f64>> {
let entry = model
.tensor(name)
.with_context(|| format!("MTP compact: missing tensor {name}"))?;
ensure!(
entry.shape.as_slice() == [rows, cols],
"MTP compact: {name} has shape {:?}, expected [{rows}, {cols}]",
entry.shape
);
let source = model
.tensor_bytes(name)
.with_context(|| format!("MTP compact: reading tensor {name}"))?;
if let Some(norms) = direct_down_column_norm_sq(entry.dtype, source, rows, cols)? {
return Ok(norms);
}
let down = tensor_f32(model, name)?;
let expected = rows
.checked_mul(cols)
.context("MTP compact: down-projection shape overflows usize")?;
ensure!(
down.len() == expected,
"MTP compact: {name} dequantized to {} values, expected {expected}",
down.len()
);
let mut norms = vec![0.0f64; cols];
for row in down.chunks_exact(cols) {
for (column, &value) in row.iter().enumerate() {
let value = value as f64;
norms[column] += value * value;
}
}
Ok(norms)
}
fn tensor_shape(model: &CmfModel, name: &str) -> Result<Vec<usize>> {
Ok(model
.tensor(name)
.with_context(|| format!("MTP compact: missing tensor {name}"))?
.shape
.clone())
}
fn aligned_keep(width: usize, keep_ratio: f64) -> Result<usize> {
ensure!(
keep_ratio.is_finite() && keep_ratio > 0.0 && keep_ratio <= 1.0,
"MTP compact: --keep-ratio must be finite and in (0, 1]"
);
ensure!(
width >= MTP_COMPACT_ALIGNMENT && width % MTP_COMPACT_ALIGNMENT == 0,
"MTP compact: draft FFN width {width} is not a positive {MTP_COMPACT_ALIGNMENT}-channel codec group"
);
let keep = ((width as f64 * keep_ratio).floor() as usize / MTP_COMPACT_ALIGNMENT)
* MTP_COMPACT_ALIGNMENT;
ensure!(
keep >= MTP_COMPACT_ALIGNMENT,
"MTP compact: --keep-ratio {keep_ratio} leaves fewer than one {MTP_COMPACT_ALIGNMENT}-channel group from width {width}"
);
Ok(keep)
}
fn admission_probe_len(record_tokens: usize, draft_layers: usize) -> Result<usize> {
let required = draft_layers
.checked_add(2)
.context("MTP compact: draft layer count overflows admission prefix")?;
ensure!(
record_tokens >= required,
"MTP compact: a calibration record with {record_tokens} token(s) is too short to admit a {draft_layers}-layer draft"
);
Ok(record_tokens.min(MTP_ADMISSION_TARGET_TOKENS.max(required)))
}
fn admission_probe_prefix(tokens: &[u32], draft_layers: usize) -> Result<Option<&[u32]>> {
let required = draft_layers
.checked_add(2)
.context("MTP compact: draft layer count overflows admission prefix")?;
if tokens.len() < required {
return Ok(None);
}
let len = admission_probe_len(tokens.len(), draft_layers)?;
Ok(Some(&tokens[..len]))
}
#[cfg(test)]
fn rank_channel_scores(
squared_mass: &[f64],
down: &[f32],
hidden: usize,
width: usize,
) -> Result<Vec<(usize, f64)>> {
let expected_down = hidden
.checked_mul(width)
.context("MTP compact: down-projection shape overflows usize")?;
ensure!(
squared_mass.len() == width && down.len() == expected_down,
"MTP compact: score inputs do not match a {hidden} × {width} down projection"
);
let mut column_norm_sq = vec![0.0f64; width];
for row in down.chunks_exact(width) {
for (channel, &value) in row.iter().enumerate() {
let value = value as f64;
column_norm_sq[channel] += value * value;
}
}
let mut ranked: Vec<(usize, f64)> = (0..width)
.map(|channel| {
let col_norm_sq = column_norm_sq[channel];
let activation = squared_mass[channel];
ensure!(
activation.is_finite() && activation >= 0.0 && col_norm_sq.is_finite(),
"MTP compact: draft channel {channel} has invalid score inputs"
);
let score = activation * col_norm_sq;
ensure!(
score.is_finite(),
"MTP compact: draft channel {channel} score overflow"
);
Ok((channel, score))
})
.collect::<Result<_>>()?;
ranked.sort_by(|(ia, a), (ib, b)| {
b.partial_cmp(a)
.unwrap_or(Ordering::Equal)
.then_with(|| ia.cmp(ib))
});
Ok(ranked)
}
fn rank_channel_norm_scores(
squared_mass: &[f64],
column_norm_sq: &[f64],
width: usize,
) -> Result<Vec<(usize, f64)>> {
ensure!(
squared_mass.len() == width && column_norm_sq.len() == width,
"MTP compact: score inputs do not match a {width}-channel down projection"
);
let mut ranked: Vec<(usize, f64)> = (0..width)
.map(|channel| {
let col_norm_sq = column_norm_sq[channel];
let activation = squared_mass[channel];
ensure!(
activation.is_finite() && activation >= 0.0 && col_norm_sq.is_finite(),
"MTP compact: draft channel {channel} has invalid score inputs"
);
let score = activation * col_norm_sq;
ensure!(
score.is_finite(),
"MTP compact: draft channel {channel} score overflow"
);
Ok((channel, score))
})
.collect::<Result<_>>()?;
ranked.sort_by(|(ia, a), (ib, b)| {
b.partial_cmp(a)
.unwrap_or(Ordering::Equal)
.then_with(|| ia.cmp(ib))
});
Ok(ranked)
}
fn validate_probe(probe: &[MtpFfnProbe]) -> Result<()> {
ensure!(
!probe.is_empty(),
"MTP compact: draft probe returned no layers"
);
for (layer, row) in probe.iter().enumerate() {
ensure!(
row.mass.len() == row.squared_mass.len()
&& row
.mass
.iter()
.all(|value| value.is_finite() && *value >= 0.0)
&& row
.squared_mass
.iter()
.all(|value| value.is_finite() && *value >= 0.0),
"MTP compact: draft layer {layer} returned invalid activation statistics"
);
}
Ok(())
}
fn validate_aggregate_probe(probe: &[MtpFfnProbe]) -> Result<()> {
validate_probe(probe)?;
for (layer, row) in probe.iter().enumerate() {
ensure!(
row.observed_positions > 0,
"MTP compact: draft layer {layer} saw no teacher-forced positions"
);
let energy = row
.squared_mass
.iter()
.try_fold(0.0f64, |total, &value| {
let next = total + value;
ensure!(
next.is_finite(),
"MTP compact: draft layer {layer} activation energy overflow"
);
Ok(next)
})?;
ensure!(
energy > 0.0,
"MTP compact: draft layer {layer} has zero activation energy; refuse arbitrary channel selection"
);
}
Ok(())
}
fn accumulate_probe(total: &mut [MtpFfnProbe], probe: &[MtpFfnProbe]) -> Result<()> {
ensure!(
total.len() == probe.len(),
"MTP compact: draft layer count changed from {} to {} during calibration",
total.len(),
probe.len()
);
for (layer, (sum, row)) in total.iter_mut().zip(probe.iter()).enumerate() {
ensure!(
sum.mass.len() == row.mass.len() && sum.squared_mass.len() == row.squared_mass.len(),
"MTP compact: draft layer {layer} width changed during calibration"
);
for (channel, (a, b)) in sum.mass.iter_mut().zip(&row.mass).enumerate() {
let next = *a + *b;
ensure!(
next.is_finite() && next >= 0.0,
"MTP compact: draft layer {layer} mass overflow at channel {channel}"
);
*a = next;
}
for (channel, (a, b)) in sum
.squared_mass
.iter_mut()
.zip(&row.squared_mass)
.enumerate()
{
let next = *a + *b;
ensure!(
next.is_finite() && next >= 0.0,
"MTP compact: draft layer {layer} squared mass overflow at channel {channel}"
);
*a = next;
}
sum.observed_positions = sum
.observed_positions
.checked_add(row.observed_positions)
.ok_or_else(|| anyhow::anyhow!("MTP compact: observed-position counter overflow"))?;
}
Ok(())
}
fn rank_draft_channels(
model: &CmfModel,
layer: usize,
squared_mass: &[f64],
keep_ratio: f64,
) -> Result<(usize, Vec<usize>, Value)> {
let prefix = format!("model.mtp.layers.{layer}.mlp.");
let gate_name = format!("{prefix}gate_proj.weight");
let up_name = format!("{prefix}up_proj.weight");
let down_name = format!("{prefix}down_proj.weight");
let gate_shape = tensor_shape(model, &gate_name)?;
let up_shape = tensor_shape(model, &up_name)?;
let down_shape = tensor_shape(model, &down_name)?;
let down_dtype = model
.tensor(&down_name)
.with_context(|| format!("MTP compact: missing tensor {down_name}"))?
.dtype;
ensure!(
gate_shape.len() == 2 && up_shape.len() == 2 && down_shape.len() == 2,
"MTP compact: draft layer {layer} FFN tensors must be matrices"
);
let (width, hidden) = (gate_shape[0], gate_shape[1]);
ensure!(
up_shape == gate_shape && down_shape == vec![hidden, width],
"MTP compact: draft layer {layer} has mismatched FFN shapes gate={gate_shape:?}, up={up_shape:?}, down={down_shape:?}"
);
ensure!(
squared_mass.len() == width,
"MTP compact: draft layer {layer} probe width {} != tensor width {width}",
squared_mass.len()
);
ensure!(
hidden == model.arch().hidden_size,
"MTP compact: draft layer {layer} hidden width {hidden} != header hidden_size {}",
model.arch().hidden_size
);
let keep = aligned_keep(width, keep_ratio)?;
let draft_layers = model.arch().mtp.as_ref().map_or(1, |mtp| mtp.num_layers);
let scoring_started = Instant::now();
eprintln!(
"mtp compact: scoring MTP draft layer {}/{} down_proj ({} × {}, {})",
layer.saturating_add(1),
draft_layers,
hidden,
width,
down_dtype.name()
);
let column_norm_sq = down_column_norm_sq(model, &down_name, hidden, width)?;
eprintln!(
"mtp compact: scored MTP draft layer {}/{} in {:.1}s",
layer.saturating_add(1),
draft_layers,
scoring_started.elapsed().as_secs_f64()
);
let ranked = rank_channel_norm_scores(squared_mass, &column_norm_sq, width)
.with_context(|| format!("MTP compact: draft layer {layer}"))?;
let total: f64 = ranked.iter().map(|(_, score)| *score).sum();
ensure!(
total.is_finite() && total >= 0.0,
"MTP compact: invalid score total"
);
let selected_score: f64 = ranked.iter().take(keep).map(|(_, score)| *score).sum();
let order = ranked
.iter()
.take(keep)
.map(|(channel, _)| *channel)
.collect();
Ok((
width,
order,
json!({
"layer": layer,
"source_width": width,
"kept_width": keep,
"down_proj_dtype": down_dtype.name(),
"score_total": total,
"score_kept": selected_score,
"score_kept_fraction": if total > 0.0 { Some(selected_score / total) } else { None },
}),
))
}
fn draft_source(main: &Arc<CmfModel>, main_path: &str) -> Result<(PathBuf, Arc<CmfModel>)> {
if main
.tensors
.iter()
.any(|tensor| tensor.name.starts_with("model.mtp."))
{
return Ok((PathBuf::from(main_path), main.clone()));
}
let sidecar = mtp_sidecar_path(Path::new(main_path));
ensure!(
sidecar.exists(),
"MTP compact: {} carries no model.mtp.* tensors and its expected sidecar {} is absent",
main_path,
sidecar.display()
);
let side = Arc::new(
CmfModel::open_sharded(&sidecar)
.with_context(|| format!("MTP compact: opening sidecar {}", sidecar.display()))?,
);
ensure!(
side.tensors
.iter()
.any(|tensor| tensor.name.starts_with("model.mtp.")),
"MTP compact: sidecar {} carries no model.mtp.* payload",
sidecar.display()
);
Ok((sidecar, side))
}
pub fn cmd_mtp_compact(
model_path: &str,
data_selector: &str,
output: &str,
keep_ratio: f64,
plan_out: Option<&str>,
max_records: usize,
max_record_tokens: usize,
) -> Result<()> {
let source_main = Path::new(model_path);
require_main_model_input(source_main)?;
let output_path = Path::new(output);
let default_plan = format!("{output}.mtp-plan.json");
let plan_path = PathBuf::from(plan_out.unwrap_or(&default_plan));
preflight_explicit_paths(source_main, output_path, &plan_path)?;
let main = verify_cmf(source_main, "source main")?;
let (draft_path, draft) = draft_source(&main, model_path)?;
if !paths_collide(source_main, &draft_path) {
verify_open_cmf(&draft, &draft_path, "source MTP draft sidecar")?;
}
let destination = compact_destination(source_main, &draft_path, output_path)?;
ensure!(
(!destination.publish_main
|| (!paths_overlap_or_nested(&destination.main, source_main)?
&& !paths_overlap_or_nested(&destination.main, &draft_path)?))
&& !paths_overlap_or_nested(&destination.draft, source_main)?
&& !paths_overlap_or_nested(&destination.draft, &draft_path)?
&& (!destination.external_draft
|| !destination.publish_main
|| !paths_overlap_or_nested(&destination.main, &destination.draft)?),
"MTP compact: candidate output must be distinct from source main/draft and paired main/sidecar paths must not alias or nest"
);
ensure!(
!destination.external_draft || draft.header.shard.is_none(),
"MTP compact: a sharded external .mtp.cmf sidecar is not supported; first materialize one canonical sidecar"
);
let expected_mimo = main.arch().arch_name == "mimo_v2";
ensure!(
!destination.external_draft || expected_mimo,
"MTP compact: an external .mtp.cmf draft is supported only for the MiMo runtime"
);
if destination.external_draft {
check_new_file_leaf(&destination.draft, "candidate MTP sidecar")?;
}
let draft_widths = crate::tube::preflight_mtp_compact_source(&draft)
.context("MTP compact: validating source draft for physical compaction")?;
let draft_layer_count = draft
.arch()
.mtp
.as_ref()
.map(|config| config.num_layers)
.context("MTP compact: loaded draft has no MTP layer declaration")?;
ensure!(
draft_widths.len() == draft_layer_count,
"MTP compact: source draft declared {draft_layer_count} layer(s), but physical preflight found {}",
draft_widths.len()
);
for (layer, width) in draft_widths.iter().copied().enumerate() {
aligned_keep(width, keep_ratio).with_context(|| {
format!("MTP compact: validating --keep-ratio for draft layer {layer}")
})?;
}
let minimum_chunk_tokens = draft_layer_count
.checked_add(2)
.context("MTP compact: draft layer count overflows calibration cap")?
.max(3);
ensure!(
max_record_tokens >= minimum_chunk_tokens,
"MTP compact: --max-record-tokens must be at least {minimum_chunk_tokens} for a {draft_layer_count}-layer draft"
);
ensure!(
!paths_overlap_or_nested(&plan_path, source_main)?
&& !paths_overlap_or_nested(&plan_path, &draft_path)?
&& (!destination.publish_main
|| !paths_overlap_or_nested(&plan_path, &destination.main)?)
&& !paths_overlap_or_nested(&plan_path, &destination.draft)?,
"MTP compact: --plan-out must be a distinct ordinary-file path, not an alias or child of a source/candidate CMF"
);
let (reader, selector) = TdataReader::open_selector(data_selector)
.with_context(|| format!("MTP compact: opening data selector {data_selector:?}"))?;
ensure!(
selector.split == TdataSplitSelector::Train,
"MTP compact: data selector must be the train split; use PATH.tdata or PATH.tdata:train so validation/test remain untouched"
);
let tokenizer_contract = match reader.manifest.raw.get("tokenizer") {
Some(_) => {
let source_vocab = main.vocab.as_deref().context(
"MTP compact: .tdata declares a tokenizer contract but source main CMF has no embedded tokenizer bytes",
)?;
let source_tokenizer = Tokenizer::from_bytes(source_vocab).map_err(|error| {
anyhow::anyhow!("MTP compact: parsing source tokenizer for .tdata contract: {error}")
})?;
let tokenizer_config_bytes = serde_json::to_vec(&main.header.tokenizer_config)
.map_err(|error| anyhow::anyhow!("MTP compact: serializing source tokenizer metadata: {error}"))?;
validate_tdata_tokenizer_contract(
&reader.manifest.raw,
Some(source_vocab),
&tokenizer_config_bytes,
source_tokenizer.vocab_size(),
)?
}
None => validate_tdata_tokenizer_contract(&reader.manifest.raw, None, &[], 0)?,
};
reader
.verify_for_calibration(&selector)
.with_context(|| format!("MTP compact: verifying {data_selector:?}"))?;
if destination.publish_main {
prepare_new_file(&destination.main, "candidate main")?;
}
if destination.external_draft {
prepare_new_file(&destination.draft, "candidate MTP sidecar")?;
}
prepare_new_file(&plan_path, "MTP plan")?;
let mut pipeline = Pipeline::from_model(&main, SamplerConfig::default())
.map_err(|error| anyhow::anyhow!("MTP compact: loading source runtime: {error}"))?;
if expected_mimo {
ensure!(
pipeline.mimo_mtp.is_some(),
"MTP compact: source did not load its MiMo MTP draft (check CMF_MIMO_MTP)"
);
} else {
ensure!(
pipeline.mtp.is_some(),
"MTP compact: source did not load its MTP draft (check CMF_MTP)"
);
}
let mut aggregate: Option<Vec<MtpFfnProbe>> = None;
let mut admission_tokens: Option<Vec<u32>> = None;
let mut records = 0usize;
let mut selected_records = 0usize;
let mut chunks = 0usize;
let mut tokens = 0usize;
let calibration_started = Instant::now();
for item in reader.stream(selector)? {
if max_records != 0 && selected_records >= max_records {
break;
}
let sample = item?;
selected_records = selected_records
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("MTP compact: selected-record counter overflow"))?;
let mut start = 0usize;
let mut used_record = false;
while sample.tokens.len().saturating_sub(start) >= 2 {
let remaining = sample.tokens.len() - start;
let mut take = remaining.min(max_record_tokens);
if remaining > max_record_tokens && remaining - take == 1 {
take -= 1;
}
debug_assert!(take >= 2 && take <= max_record_tokens);
let end = start
.checked_add(take)
.expect("calibration chunk end checked against source length");
let chunk = &sample.tokens[start..end];
let probe = pipeline.probe_mtp_ffn(chunk).map_err(|error| {
anyhow::anyhow!(
"MTP compact: probing record {:?}, chunk {}: {error}",
sample.record.rid,
chunks.saturating_add(1)
)
})?;
validate_probe(&probe).with_context(|| {
format!(
"MTP compact: invalid draft activation statistics for record {:?}, chunk {}",
sample.record.rid,
chunks.saturating_add(1)
)
})?;
if let Some(total) = aggregate.as_mut() {
accumulate_probe(total, &probe).with_context(|| {
format!(
"MTP compact: aggregating draft activation statistics for record {:?}, chunk {}",
sample.record.rid,
chunks.saturating_add(1)
)
})?;
} else {
aggregate = Some(probe);
}
if let Some(prefix) = admission_probe_prefix(chunk, draft_layer_count)? {
if admission_tokens
.as_ref()
.map_or(true, |current| prefix.len() > current.len())
{
admission_tokens = Some(prefix.to_vec());
}
}
chunks = chunks
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("MTP compact: chunk counter overflow"))?;
tokens = tokens
.checked_add(chunk.len())
.ok_or_else(|| anyhow::anyhow!("MTP compact: token counter overflow"))?;
used_record = true;
if chunks == 1 || chunks % 8 == 0 {
let record_progress = if used_record {
records.saturating_add(1)
} else {
records
};
eprintln!(
"mtp compact: calibrating {chunks} chunk(s), {tokens} token(s), {record_progress} source record(s), elapsed {:.1}s",
calibration_started.elapsed().as_secs_f64()
);
}
start = end;
}
if used_record {
records = records
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("MTP compact: record counter overflow"))?;
}
}
let aggregate = aggregate.ok_or_else(|| {
anyhow::anyhow!(
"MTP compact: no usable calibration record with at least two tokens was selected"
)
})?;
validate_aggregate_probe(&aggregate)?;
let admission_tokens = admission_tokens.ok_or_else(|| {
anyhow::anyhow!(
"MTP compact: no calibration record is long enough to admit all {draft_layer_count} draft layer(s); need at least {} token(s)",
draft_layer_count.saturating_add(2)
)
})?;
drop(pipeline);
let mut mtp_layers = Vec::with_capacity(aggregate.len());
let mut report_layers = Vec::with_capacity(aggregate.len());
let mut kept_widths = Vec::with_capacity(aggregate.len());
for (layer, probe) in aggregate.iter().enumerate() {
let (source_width, order, mut report) =
rank_draft_channels(&draft, layer, &probe.squared_mass, keep_ratio)?;
let obj = report
.as_object_mut()
.expect("locally constructed JSON object");
obj.insert("observed_positions".into(), json!(probe.observed_positions));
obj.insert(
"activation_mass".into(),
json!(probe.mass.iter().sum::<f64>()),
);
obj.insert(
"activation_squared_mass".into(),
json!(probe.squared_mass.iter().sum::<f64>()),
);
let kept_width = order.len();
mtp_layers.push(json!({"order": order, "widths": [kept_width]}));
kept_widths.push(kept_width);
report_layers.push(json!({"source_width": source_width, "report": report}));
}
let source_main_dir_hash = main.dir_hash();
let source_draft_dir_hash = draft.dir_hash();
let source_main_trunk_hash = main.trunk_hash();
let source_draft_trunk_hash = draft.trunk_hash();
let draft_activation_contract = draft.arch().hidden_act.clone();
let mut selection = json!({
"kind": "draft-activation-weighted-down-column-energy-v1",
"formula": "sum_teacher_forced((activation(gate, up))^2) * down_proj_column_l2_squared",
"activation_contract": draft_activation_contract,
"keep_ratio": keep_ratio,
"alignment": MTP_COMPACT_ALIGNMENT,
"calibration_records": records,
"calibration_selected_records": selected_records,
"calibration_chunks": chunks,
"calibration_tokens": tokens,
"max_record_tokens": max_record_tokens,
"calibration_chunking": "fresh-causal-sequence-per-contiguous-chunk-v1",
"admission_probe_tokens": admission_tokens.len(),
"tokenizer_contract": tokenizer_contract,
"data_selector": data_selector,
"source_main": model_path,
"source_draft": draft_path.display().to_string(),
"output_layout": if destination.external_draft {
if destination.publish_main {
"runnable-main-and-sidecar-v1"
} else {
"legacy-sidecar-only-v1"
}
} else {
"runnable-embedded-main-v1"
},
"layers": report_layers,
});
selection
.as_object_mut()
.expect("locally constructed JSON object")
.extend(source_hash_fields(source_main_dir_hash, source_draft_dir_hash));
selection
.as_object_mut()
.expect("locally constructed JSON object")
.extend(source_trunk_hash_fields(
source_main_trunk_hash,
source_draft_trunk_hash,
));
let plan = json!({
"format": "cmf-mtp-ffn-plan",
"version": 2,
"trunk": {"mode": "preserve"},
"mtp": {"mode": "compact", "layers": mtp_layers},
"selection": selection,
});
drop(draft);
drop(main);
let source_for_bake = revalidate_source_identity(
"after calibration",
source_main,
&draft_path,
source_main_dir_hash,
source_draft_dir_hash,
source_main_trunk_hash,
source_draft_trunk_hash,
)?;
let stage_anchor = if destination.publish_main {
&destination.main
} else {
&destination.draft
};
let main_stage_dir = create_stage_dir(stage_anchor, "model")?;
let plan_stage_dir = match create_stage_dir(&plan_path, "plan") {
Ok(path) => path,
Err(error) => {
let _ = fs::remove_dir_all(&main_stage_dir);
return Err(error);
}
};
let staged_main = match staged_child(&main_stage_dir, &destination.main) {
Ok(path) => path,
Err(error) => {
let _ = fs::remove_dir_all(&main_stage_dir);
let _ = fs::remove_dir_all(&plan_stage_dir);
return Err(error);
}
};
let staged_draft = if destination.external_draft {
match staged_child(&main_stage_dir, &destination.draft) {
Ok(path) => path,
Err(error) => {
let _ = fs::remove_dir_all(&main_stage_dir);
let _ = fs::remove_dir_all(&plan_stage_dir);
return Err(error);
}
}
} else {
staged_main.clone()
};
let staged_plan = match staged_child(&plan_stage_dir, &plan_path) {
Ok(path) => path,
Err(error) => {
let _ = fs::remove_dir_all(&main_stage_dir);
let _ = fs::remove_dir_all(&plan_stage_dir);
return Err(error);
}
};
let mut backbone_materialization: Option<String> = None;
let backbone_staged = (|| -> Result<()> {
if destination.external_draft {
if let Some(shard) = &source_for_bake.header.shard {
eprintln!(
"mtp compact: streaming/coalescing {} source main shard(s) into a private candidate stage",
shard.count
);
crate::knowledge::refuse_genome_rewrite(
&source_for_bake,
"mtp-compact sharded MiMo backbone",
)?;
crate::tube::stream_copy_as_unsharded_new(&source_for_bake, &staged_main)?;
backbone_materialization = Some(format!(
"streamed coalescing of {} source shard(s)",
shard.count
));
} else {
eprintln!(
"mtp compact: staging unchanged main backbone (hard-link when the filesystem permits)"
);
backbone_materialization = Some(stage_backbone(source_main, &staged_main)?.into());
}
}
Ok(())
})();
if let Err(error) = backbone_staged {
let _ = fs::remove_dir_all(&main_stage_dir);
let _ = fs::remove_dir_all(&plan_stage_dir);
return Err(error);
}
drop(source_for_bake);
let staged = (|| -> Result<()> {
write_new_file(&staged_plan, &serde_json::to_vec_pretty(&plan)?, "MTP plan")?;
eprintln!("mtp compact: streaming compact MTP draft into a private candidate stage");
crate::tube::cmd_tube_bake_new(
draft_path
.to_str()
.ok_or_else(|| anyhow::anyhow!("MTP compact: non-UTF-8 source draft path"))?,
staged_plan
.to_str()
.ok_or_else(|| anyhow::anyhow!("MTP compact: non-UTF-8 staged plan path"))?,
staged_draft
.to_str()
.ok_or_else(|| anyhow::anyhow!("MTP compact: non-UTF-8 staged draft path"))?,
)?;
let source_for_admission = revalidate_source_identity(
"after bake",
source_main,
&draft_path,
source_main_dir_hash,
source_draft_dir_hash,
source_main_trunk_hash,
source_draft_trunk_hash,
)?;
admit_candidate(
&source_for_admission,
&destination,
&staged_main,
&staged_draft,
expected_mimo,
destination.external_draft.then_some(source_main_trunk_hash),
&admission_tokens,
&kept_widths,
)
})();
if let Err(error) = staged {
let _ = fs::remove_dir_all(&main_stage_dir);
let _ = fs::remove_dir_all(&plan_stage_dir);
return Err(error);
}
let published = (|| -> Result<()> {
eprintln!("mtp compact: publishing validated candidate without overwriting existing files");
publish_new_file(&staged_plan, &plan_path, "MTP plan")?;
if destination.external_draft {
publish_new_file(&staged_draft, &destination.draft, "candidate MTP sidecar")?;
}
if destination.publish_main {
publish_new_file(&staged_main, &destination.main, "candidate main")?;
}
Ok(())
})();
if let Err(error) = published {
eprintln!(
"warning: MTP compact publication failed; any already-published plan or sidecar is intentionally left in place rather than risking deletion of a concurrently created file"
);
let _ = fs::remove_dir_all(&main_stage_dir);
let _ = fs::remove_dir_all(&plan_stage_dir);
return Err(error);
}
if let Err(error) = fs::remove_dir_all(&main_stage_dir) {
eprintln!(
"warning: MTP compact published candidate, but could not remove private stage {}: {error}",
main_stage_dir.display()
);
}
if let Err(error) = fs::remove_dir_all(&plan_stage_dir) {
eprintln!(
"warning: MTP compact published candidate, but could not remove private stage {}: {error}",
plan_stage_dir.display()
);
}
let published_main = if destination.publish_main {
format!("\n candidate main: {}", destination.main.display())
} else {
"\n candidate main: not published (legacy sidecar-only output; pair the draft with a compatible main CMF yourself)".to_owned()
};
println!(
"mtp compact: runtime admission passed after {} calibration record(s), {} bounded chunk(s), {} token(s), {} draft layer(s)\n calibration chunk cap: {} token(s){}\n candidate draft: {}\n plan: {}{}\n note: this proves CMF integrity, draft execution, and a short CPU plain greedy source/candidate check; measure held-out quality and Metal/Vulkan (including any speculative graph) separately",
records,
chunks,
tokens,
aggregate.len(),
max_record_tokens,
published_main,
destination.draft.display(),
plan_path.display(),
backbone_materialization
.map(|how| format!("\n backbone materialization: {how}"))
.unwrap_or_default(),
);
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
accumulate_probe, admission_probe_len, admission_probe_prefix, aligned_keep,
cmd_mtp_compact, compact_destination, main_path_for_mtp_sidecar, paths_collide,
paths_overlap_or_nested, preflight_explicit_paths, prepare_new_file, publish_new_file,
rank_channel_norm_scores, rank_channel_scores, require_main_model_input, source_hash_fields,
source_trunk_hash_fields, stage_backbone,
validate_aggregate_probe,
direct_down_column_norm_sq, validate_probe, validate_tdata_tokenizer_contract,
};
use crate::convert::{Quant, quantize_2d};
use cortiq_core::format::TokenizerBundle;
use cortiq_core::{
CMF_VERSION, CmfHeader, CmfModel, LayerType, ModelArch, MtpConfig, NormStyle, QuantType,
TensorDtype, TensorSpec,
};
use cortiq_engine::{MtpFfnProbe, Pipeline, SamplerConfig};
use serde_json::json;
use std::fs;
use std::sync::Arc;
const MIMO_TEST_HIDDEN: usize = 4;
const MIMO_TEST_VOCAB: usize = 16;
const MIMO_TEST_DRAFT_WIDTH: usize = 64;
fn test_f16_bytes(values: &[f32]) -> Vec<u8> {
values
.iter()
.flat_map(|value| cortiq_core::quant::f32_to_f16(*value).to_le_bytes())
.collect()
}
fn test_f16_tensor(name: impl Into<String>, shape: &[usize], salt: usize) -> TensorSpec {
let len = shape.iter().product();
let values = (0..len)
.map(|index| (((index * 17 + salt * 13) % 29) as f32 + 1.0) / 128.0)
.collect::<Vec<_>>();
TensorSpec {
name: name.into(),
dtype: TensorDtype::F16,
shape: shape.to_vec(),
data: test_f16_bytes(&values),
}
}
fn test_f16_vector(name: impl Into<String>, values: &[f32]) -> TensorSpec {
TensorSpec {
name: name.into(),
dtype: TensorDtype::F16,
shape: vec![values.len()],
data: test_f16_bytes(values),
}
}
fn mimo_test_header() -> CmfHeader {
CmfHeader {
format: "cmf".into(),
version: CMF_VERSION,
tokenizer_config: None,
section_hashes: None,
skills: Vec::new(),
shard: None,
calibration: None,
routing: None,
genome: None,
lineage: Vec::new(),
router: None,
segments: Vec::new(),
arch: ModelArch {
arch_name: "mimo_v2".into(),
hidden_size: MIMO_TEST_HIDDEN,
intermediate_size: MIMO_TEST_HIDDEN,
num_layers: 1,
num_attention_heads: 1,
num_kv_heads: 1,
head_dim: MIMO_TEST_HIDDEN,
vocab_size: MIMO_TEST_VOCAB,
layer_types: vec![LayerType::FullAttention],
rms_norm_eps: 1e-6,
norm_style: NormStyle::Qwen,
rope_theta: 10_000.0,
tie_word_embeddings: false,
partial_rotary_factor: 1.0,
yarn: None,
attention_heads_per_layer: None,
kv_heads_per_layer: None,
v_head_dim: Some(MIMO_TEST_HIDDEN),
hidden_act: "silu".into(),
embed_multiplier: 1.0,
query_pre_attn_scalar: None,
sliding_window: None,
sliding_window_pattern: None,
rope_local_base_freq: None,
local_partial_rotary_factor: None,
global_head_dim: None,
num_global_kv_heads: None,
global_partial_rotary_factor: None,
final_logit_softcapping: None,
mla: None,
no_rope_layers: None,
attention_sink: false,
activation_situ_beta: None,
activation_situ_linear_beta: None,
attn_logit_softcapping: None,
attn_v_norm: false,
qk_norm_after_rope: false,
mtp: Some(MtpConfig {
num_layers: 1,
ffn_intermediate_sizes: Some(vec![MIMO_TEST_DRAFT_WIDTH]),
share_lm_head: true,
share_embed: true,
}),
moe: None,
qwen4_exp: None,
block_diffusion: None,
deepseek_v41: None,
linear_core: None,
anchor_core: None,
head_clusters: None,
max_position_embeddings: 64,
linear_conv_kernel_dim: None,
linear_num_key_heads: None,
linear_num_value_heads: None,
linear_key_head_dim: None,
linear_value_head_dim: None,
rope_freq_factors: None,
logit_multiplier: None,
g3n: None,
kda_gate_lower_bound: None,
num_loops: 1,
loop_final_norm: false,
prism_hadamard: None,
},
quant_type: QuantType::F16,
provenance: None,
}
}
fn mimo_test_main_tensors() -> Vec<TensorSpec> {
let h = MIMO_TEST_HIDDEN;
vec![
test_f16_tensor("model.embed_tokens.weight", &[MIMO_TEST_VOCAB, h], 1),
test_f16_vector("model.norm.weight", &[1.0; MIMO_TEST_HIDDEN]),
test_f16_tensor("lm_head.weight", &[MIMO_TEST_VOCAB, h], 2),
test_f16_vector(
"model.layers.0.input_layernorm.weight",
&[1.0; MIMO_TEST_HIDDEN],
),
test_f16_vector(
"model.layers.0.post_attention_layernorm.weight",
&[1.0; MIMO_TEST_HIDDEN],
),
test_f16_tensor("model.layers.0.self_attn.q_proj.weight", &[h, h], 3),
test_f16_tensor("model.layers.0.self_attn.k_proj.weight", &[h, h], 4),
test_f16_tensor("model.layers.0.self_attn.v_proj.weight", &[h, h], 5),
test_f16_tensor("model.layers.0.self_attn.o_proj.weight", &[h, h], 6),
test_f16_tensor("model.layers.0.mlp.gate_proj.weight", &[h, h], 7),
test_f16_tensor("model.layers.0.mlp.up_proj.weight", &[h, h], 8),
test_f16_tensor("model.layers.0.mlp.down_proj.weight", &[h, h], 9),
]
}
fn mimo_test_sidecar_tensors() -> Vec<TensorSpec> {
let h = MIMO_TEST_HIDDEN;
let p = "model.mtp.layers.0.";
vec![
test_f16_vector(format!("{p}enorm.weight"), &[1.0; MIMO_TEST_HIDDEN]),
test_f16_vector(format!("{p}hnorm.weight"), &[1.0; MIMO_TEST_HIDDEN]),
test_f16_tensor(format!("{p}eh_proj.weight"), &[h, 2 * h], 10),
test_f16_vector(
format!("{p}input_layernorm.weight"),
&[1.0; MIMO_TEST_HIDDEN],
),
test_f16_tensor(format!("{p}self_attn.q_proj.weight"), &[h, h], 11),
test_f16_tensor(format!("{p}self_attn.k_proj.weight"), &[h, h], 12),
test_f16_tensor(format!("{p}self_attn.v_proj.weight"), &[h, h], 13),
test_f16_tensor(format!("{p}self_attn.o_proj.weight"), &[h, h], 14),
test_f16_vector(format!("{p}self_attn.sinks"), &[0.0]),
test_f16_vector(
format!("{p}post_attention_layernorm.weight"),
&[1.0; MIMO_TEST_HIDDEN],
),
test_f16_tensor(
format!("{p}mlp.gate_proj.weight"),
&[MIMO_TEST_DRAFT_WIDTH, h],
15,
),
test_f16_tensor(
format!("{p}mlp.up_proj.weight"),
&[MIMO_TEST_DRAFT_WIDTH, h],
16,
),
test_f16_tensor(
format!("{p}mlp.down_proj.weight"),
&[h, MIMO_TEST_DRAFT_WIDTH],
17,
),
test_f16_vector(
format!("{p}final_layernorm.weight"),
&[1.0; MIMO_TEST_HIDDEN],
),
]
}
fn write_mimo_test_tdata_with_manifest(path: &std::path::Path, manifest: serde_json::Value) {
use crate::tdata::{TdataRecord, TdataWriter};
let tokens = [1_u32, 2, 3, 4, 5, 6, 7, 8, 9];
let mut writer = TdataWriter::create(path, manifest).unwrap();
let row = json!({
"rid": "mimo#0",
"pid": "mimo",
"sample": 0,
"category": "test",
"split": "train",
"usable": true,
"messages": [{"role": "user", "content": "test"}],
"response": {"think": "", "content": "test"},
"finish": "stop",
"tok": {"index": 0, "offset": 0, "prompt": 1, "think": 0, "answer": 8, "total": 9},
"quality": {"truncated": false, "loop_suspect": false, "check": null},
"meta": {}
});
let record = TdataRecord::from_json_line(&row.to_string(), 1).unwrap();
writer.append_record(record, &tokens).unwrap();
writer.finalize().unwrap();
}
fn write_mimo_test_tdata(path: &std::path::Path) {
write_mimo_test_tdata_with_manifest(path, json!({"teacher": "synthetic"}));
}
#[test]
fn compact_width_stays_codec_aligned_and_fail_closed() {
assert_eq!(aligned_keep(128, 1.0).unwrap(), 128);
assert_eq!(aligned_keep(128, 0.76).unwrap(), 96);
assert_eq!(aligned_keep(128, 0.25).unwrap(), 32);
assert!(aligned_keep(128, 0.24).is_err());
assert!(aligned_keep(127, 0.9).is_err());
assert!(aligned_keep(128, f64::NAN).is_err());
}
#[test]
fn admission_probe_prefix_is_bounded_but_covers_deep_drafts() {
assert_eq!(admission_probe_len(10_000, 5).unwrap(), 256);
assert_eq!(admission_probe_len(302, 300).unwrap(), 302);
assert!(admission_probe_len(5, 5).is_err());
assert!(admission_probe_prefix(&[1, 2], 1).unwrap().is_none());
assert_eq!(
admission_probe_prefix(&[1, 2, 3, 4], 1).unwrap(),
Some(&[1, 2, 3, 4][..])
);
}
#[test]
fn ranking_uses_draft_activation_and_down_column_energy() {
let down = vec![1.0, 1.0, 3.0, 10.0, 0.0, 0.0, 0.0, 0.0];
let ranked = rank_channel_scores(&[1.0, 16.0, 4.0, 0.1], &down, 2, 4).unwrap();
assert_eq!(
ranked.iter().map(|(index, _)| *index).collect::<Vec<_>>(),
vec![2, 1, 3, 0]
);
let tied = rank_channel_scores(&[1.0, 1.0], &[1.0, 1.0], 1, 2).unwrap();
assert_eq!(
tied.iter().map(|(index, _)| *index).collect::<Vec<_>>(),
vec![0, 1]
);
assert!(rank_channel_scores(&[f64::NAN], &[1.0], 1, 1).is_err());
assert!(rank_channel_scores(&[f64::MAX], &[f32::MAX], 1, 1).is_err());
}
#[test]
fn direct_down_scoring_matches_full_dequant_ranking() {
let (rows, cols) = (7usize, 64usize);
let values = (0..rows * cols)
.map(|index| ((index * 37 % 149) as f32 - 74.0) / 43.0)
.collect::<Vec<_>>();
let mass = (0..cols)
.map(|index| if index % 11 == 0 { 1.0 } else { (index % 7 + 1) as f64 })
.collect::<Vec<_>>();
let mut encoded = vec![
(
TensorDtype::F32,
values
.iter()
.flat_map(|value| value.to_le_bytes())
.collect::<Vec<_>>(),
),
(
TensorDtype::Bf16,
values
.iter()
.flat_map(|value| ((value.to_bits() >> 16) as u16).to_le_bytes())
.collect::<Vec<_>>(),
),
];
encoded.extend(
[
Quant::F16,
Quant::Q8Row,
Quant::Q8_2f,
Quant::Q4TiledP,
Quant::Q2TiledP,
]
.into_iter()
.map(|quant| quantize_2d(quant, &values, rows, cols)),
);
for (dtype, source) in encoded {
let direct_norms = direct_down_column_norm_sq(dtype, &source, rows, cols)
.unwrap()
.expect("tiled codec must use direct score path");
let mut full = vec![0.0f32; rows * cols];
match dtype {
TensorDtype::F32 => {
for (slot, bytes) in full.iter_mut().zip(source.chunks_exact(4)) {
*slot = f32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
}
}
TensorDtype::F16 => {
for (slot, bytes) in full.iter_mut().zip(source.chunks_exact(2)) {
*slot = cortiq_core::quant::f16_to_f32(u16::from_le_bytes([
bytes[0], bytes[1],
]));
}
}
TensorDtype::Bf16 => {
for (slot, bytes) in full.iter_mut().zip(source.chunks_exact(2)) {
*slot = cortiq_core::quant::bf16_to_f32(u16::from_le_bytes([
bytes[0], bytes[1],
]));
}
}
TensorDtype::Q8Row => {
cortiq_core::quant::dequant_q8_row(&source, rows, cols, &mut full)
}
TensorDtype::Q8_2f => {
cortiq_core::quant::dequant_q8_2f(&source, rows, cols, &mut full)
}
TensorDtype::Q4TiledP => {
cortiq_core::quant::dequant_q4tp(&source, rows, cols, &mut full)
}
TensorDtype::Q2TiledP => {
cortiq_core::quant::dequant_q2tp(&source, rows, cols, &mut full)
}
_ => panic!("unexpected direct-score dtype {dtype:?}"),
}
let reference = rank_channel_scores(&mass, &full, rows, cols).unwrap();
let direct = rank_channel_norm_scores(&mass, &direct_norms, cols).unwrap();
assert_eq!(
direct, reference,
"{dtype:?} direct score path changed ranking or score bits"
);
}
}
#[test]
fn selection_records_independent_main_and_draft_hashes() {
let fields = source_hash_fields(0xabc, 0xdef);
assert_eq!(
fields
.get("source_directory_hash")
.and_then(serde_json::Value::as_str),
Some("0000000000000abc")
);
assert_eq!(
fields
.get("source_draft_directory_hash")
.and_then(serde_json::Value::as_str),
Some("0000000000000def")
);
let trunks = source_trunk_hash_fields(0x123, 0x456);
assert_eq!(
trunks
.get("source_trunk_hash")
.and_then(serde_json::Value::as_str),
Some("0000000000000123")
);
assert_eq!(
trunks
.get("source_draft_trunk_hash")
.and_then(serde_json::Value::as_str),
Some("0000000000000456")
);
}
#[test]
fn calibration_tokenizer_contract_is_verified_or_explicitly_legacy() {
let vocab = b"synthetic-tokenizer-v1";
let config = br#"{"eos_token_id":2}"#;
let manifest = json!({
"tokenizer": {
"vocab_size": 17,
"vocab_sha256": crate::knowledge::sha256_hex(vocab),
"config_sha256": crate::knowledge::sha256_hex(config),
}
});
assert_eq!(
validate_tdata_tokenizer_contract(&manifest, Some(vocab), config, 17).unwrap(),
"verified-v1"
);
let mismatch = validate_tdata_tokenizer_contract(
&manifest,
Some(b"different-tokenizer"),
config,
17,
)
.unwrap_err()
.to_string();
assert!(mismatch.contains("vocab_sha256"), "{mismatch}");
let partial = json!({"tokenizer": {"vocab_size": 17}});
assert!(validate_tdata_tokenizer_contract(&partial, Some(vocab), config, 17).is_err());
assert_eq!(
validate_tdata_tokenizer_contract(&json!({"teacher": "legacy"}), None, &[], 0)
.unwrap(),
"absent-legacy"
);
}
#[test]
fn tokenizer_contract_is_checked_before_any_output_stage_is_created() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.cmf");
let source_sidecar = temp.path().join("source.mtp.cmf");
let matching_data = temp.path().join("matching.tdata");
let matching_output = temp.path().join("matching.cmf");
let vocab = br#"{"model":{"vocab":{"a":0,"b":1,"c":2,"d":3,"e":4,"f":5,"g":6,"h":7,"i":8,"j":9,"k":10,"l":11,"m":12,"n":13,"o":14,"p":15},"merges":[]},"added_tokens":[]}"#;
let mut header = mimo_test_header();
header.tokenizer_config = Some(TokenizerBundle {
chat_template: None,
eos_token_ids: vec![15],
bos_token_id: None,
pad_token_id: None,
});
let config = serde_json::to_vec(&header.tokenizer_config).unwrap();
let matching_manifest = json!({
"teacher": "synthetic",
"tokenizer": {
"vocab_size": MIMO_TEST_VOCAB,
"vocab_sha256": crate::knowledge::sha256_hex(vocab),
"config_sha256": crate::knowledge::sha256_hex(&config),
}
});
CmfModel::write(
&source,
&header,
&mimo_test_main_tensors(),
None,
Some(vocab),
)
.unwrap();
CmfModel::write(
&source_sidecar,
&header,
&mimo_test_sidecar_tensors(),
None,
None,
)
.unwrap();
write_mimo_test_tdata_with_manifest(&matching_data, matching_manifest.clone());
cmd_mtp_compact(
source.to_str().unwrap(),
matching_data.to_str().unwrap(),
matching_output.to_str().unwrap(),
0.5,
None,
1,
4,
)
.unwrap();
let plan: serde_json::Value = serde_json::from_slice(
&fs::read(temp.path().join("matching.cmf.mtp-plan.json")).unwrap(),
)
.unwrap();
assert_eq!(
plan["selection"]["tokenizer_contract"],
"verified-v1",
"matching embedded tokenizer must be auditable in the plan"
);
let mismatched_data = temp.path().join("mismatched.tdata");
let mut mismatched_manifest = matching_manifest;
mismatched_manifest["tokenizer"]["vocab_sha256"] = json!("00");
write_mimo_test_tdata_with_manifest(&mismatched_data, mismatched_manifest);
let nested_output = temp.path().join("must-not-exist/nested/candidate.cmf");
let error = cmd_mtp_compact(
source.to_str().unwrap(),
mismatched_data.to_str().unwrap(),
nested_output.to_str().unwrap(),
0.5,
None,
1,
4,
)
.unwrap_err()
.to_string();
assert!(error.contains("vocab_sha256"), "{error}");
assert!(
!nested_output.parent().unwrap().exists(),
"bad tokenizer data must fail before staging/output parents"
);
}
#[test]
fn probe_aggregation_fails_closed_on_invalid_or_overflowed_statistics() {
let invalid = vec![MtpFfnProbe {
mass: vec![1.0],
squared_mass: vec![f64::NAN],
observed_positions: 1,
}];
assert!(validate_probe(&invalid).is_err());
let mut total = vec![MtpFfnProbe {
mass: vec![f64::MAX],
squared_mass: vec![1.0],
observed_positions: 1,
}];
let next = vec![MtpFfnProbe {
mass: vec![f64::MAX],
squared_mass: vec![1.0],
observed_positions: 1,
}];
assert!(accumulate_probe(&mut total, &next).is_err());
let zero = vec![MtpFfnProbe {
mass: vec![0.0],
squared_mass: vec![0.0],
observed_positions: 1,
}];
assert!(validate_aggregate_probe(&zero).is_err());
}
#[test]
fn external_sidecar_output_means_a_ready_candidate_pair() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("base.cmf");
let source_draft = temp.path().join("base.mtp.cmf");
fs::write(&source, b"main").unwrap();
fs::write(&source_draft, b"draft").unwrap();
let candidate = temp.path().join("candidate/model.cmf");
let dest = compact_destination(&source, &source_draft, &candidate).unwrap();
assert!(dest.external_draft);
assert!(dest.publish_main);
assert_eq!(dest.main, candidate);
assert_eq!(dest.draft, temp.path().join("candidate/model.mtp.cmf"));
let legacy = temp.path().join("candidate/model.mtp.cmf");
let legacy_dest = compact_destination(&source, &source_draft, &legacy).unwrap();
assert!(legacy_dest.external_draft);
assert!(!legacy_dest.publish_main);
assert_eq!(legacy_dest.main, temp.path().join("candidate/model.cmf"));
assert_eq!(legacy_dest.draft, legacy);
assert_eq!(
main_path_for_mtp_sidecar(&temp.path().join("candidate/model.mtp.cmf")).unwrap(),
temp.path().join("candidate/model.cmf")
);
}
#[test]
fn source_model_must_be_main_not_mtp_sidecar() {
assert!(require_main_model_input(std::path::Path::new("base.cmf")).is_ok());
let error = require_main_model_input(std::path::Path::new("base.mtp.cmf"))
.unwrap_err()
.to_string();
assert!(error.contains("main .cmf"), "{error}");
}
#[test]
fn sidecar_output_requires_an_external_draft() {
let source = std::path::Path::new("embedded-source.cmf");
let error = compact_destination(
source,
source,
std::path::Path::new("candidate.mtp.cmf"),
)
.unwrap_err()
.to_string();
assert!(error.contains("legacy sidecar-only"), "{error}");
}
#[test]
fn malformed_plan_topology_is_rejected_before_source_verification() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("not-even-a-cmf.cmf");
fs::write(&source, b"not a CMF").unwrap();
let candidate = temp.path().join("candidate.cmf");
let nested_plan = candidate.join("plan.json");
let error = cmd_mtp_compact(
source.to_str().unwrap(),
temp.path().join("unread.tdata").to_str().unwrap(),
candidate.to_str().unwrap(),
0.5,
Some(nested_plan.to_str().unwrap()),
1,
4,
)
.unwrap_err()
.to_string();
assert!(error.contains("candidate output and --plan-out"), "{error}");
assert!(!candidate.exists(), "topology preflight must not create output");
}
#[test]
fn sidecar_shaped_output_preflight_is_nonclobber_and_defers_topology() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.cmf");
let output = temp.path().join("candidate.mtp.cmf");
let plan = temp.path().join("candidate.mtp.cmf.mtp-plan.json");
preflight_explicit_paths(&source, &output, &plan).unwrap();
assert!(!output.exists(), "preflight must not create a sidecar output");
assert!(!plan.exists(), "preflight must not create a plan output");
}
#[test]
fn nested_plan_path_is_rejected_before_data_or_output_creation() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.cmf");
let source_sidecar = temp.path().join("source.mtp.cmf");
let candidate = temp.path().join("candidate.cmf");
let nested_plan = candidate.join("plan.json");
CmfModel::write(&source, &mimo_test_header(), &mimo_test_main_tensors(), None, None)
.unwrap();
CmfModel::write(
&source_sidecar,
&mimo_test_header(),
&mimo_test_sidecar_tensors(),
None,
None,
)
.unwrap();
let error = cmd_mtp_compact(
source.to_str().unwrap(),
temp.path().join("unread.tdata").to_str().unwrap(),
candidate.to_str().unwrap(),
0.5,
Some(nested_plan.to_str().unwrap()),
1,
4,
)
.unwrap_err()
.to_string();
assert!(error.contains("--plan-out"), "{error}");
assert!(
!candidate.exists(),
"nested plan must not create a directory at the candidate CMF leaf"
);
assert!(paths_overlap_or_nested(
&temp.path().join("not-created/../candidate.cmf"),
&candidate,
)
.unwrap());
}
#[test]
fn invalid_keep_ratio_is_rejected_before_opening_calibration_data() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.cmf");
let source_sidecar = temp.path().join("source.mtp.cmf");
let candidate = temp.path().join("must-not-exist/candidate.cmf");
CmfModel::write(&source, &mimo_test_header(), &mimo_test_main_tensors(), None, None)
.unwrap();
CmfModel::write(
&source_sidecar,
&mimo_test_header(),
&mimo_test_sidecar_tensors(),
None,
None,
)
.unwrap();
let error = cmd_mtp_compact(
source.to_str().unwrap(),
temp.path().join("unread.tdata").to_str().unwrap(),
candidate.to_str().unwrap(),
0.01,
None,
1,
4,
)
.unwrap_err()
.to_string();
assert!(error.contains("--keep-ratio"), "{error}");
assert!(
!candidate.parent().unwrap().exists(),
"invalid keep ratio must fail before output parent creation"
);
}
#[test]
fn unsupported_mtp_codec_is_rejected_before_opening_calibration_data() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.cmf");
let source_sidecar = temp.path().join("source.mtp.cmf");
let candidate = temp.path().join("must-not-exist/candidate.cmf");
CmfModel::write(&source, &mimo_test_header(), &mimo_test_main_tensors(), None, None)
.unwrap();
let mut sidecar = mimo_test_sidecar_tensors();
let gate = sidecar
.iter_mut()
.find(|tensor| tensor.name == "model.mtp.layers.0.mlp.gate_proj.weight")
.expect("test sidecar contains MTP gate");
gate.dtype = TensorDtype::U8;
gate.data = vec![0; gate.shape.iter().product()];
CmfModel::write(&source_sidecar, &mimo_test_header(), &sidecar, None, None).unwrap();
let error = cmd_mtp_compact(
source.to_str().unwrap(),
temp.path().join("unread.tdata").to_str().unwrap(),
candidate.to_str().unwrap(),
0.5,
None,
1,
4,
)
.unwrap_err();
let error = format!("{error:#}");
assert!(error.contains("physical compact writer cannot materialize"), "{error}");
assert!(
!candidate.parent().unwrap().exists(),
"unsupported codec must fail before output parent creation"
);
}
#[test]
fn mimo_external_sidecar_compacts_to_a_runnable_candidate_pair() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.cmf");
let source_sidecar = temp.path().join("source.mtp.cmf");
let data = temp.path().join("calibration.tdata");
let candidate = temp.path().join("candidate.cmf");
let candidate_sidecar = temp.path().join("candidate.mtp.cmf");
let plan = temp.path().join("candidate.cmf.mtp-plan.json");
let header = mimo_test_header();
CmfModel::write(&source, &header, &mimo_test_main_tensors(), None, None).unwrap();
CmfModel::write(
&source_sidecar,
&header,
&mimo_test_sidecar_tensors(),
None,
None,
)
.unwrap();
write_mimo_test_tdata(&data);
let too_small = temp.path().join("too-small.cmf");
let error = cmd_mtp_compact(
source.to_str().unwrap(),
data.to_str().unwrap(),
too_small.to_str().unwrap(),
0.5,
None,
1,
2,
)
.unwrap_err()
.to_string();
assert!(error.contains("--max-record-tokens"), "{error}");
assert!(!too_small.exists());
cmd_mtp_compact(
source.to_str().unwrap(),
data.to_str().unwrap(),
candidate.to_str().unwrap(),
0.5,
None,
1,
4,
)
.unwrap();
assert!(candidate.exists());
assert!(candidate_sidecar.exists());
assert!(plan.exists());
let compact_draft = CmfModel::open(&candidate_sidecar).unwrap();
assert!(
compact_draft.verify().is_empty(),
"{:?}",
compact_draft.verify()
);
assert_eq!(
compact_draft
.arch()
.mtp
.as_ref()
.and_then(|config| config.ffn_intermediate_sizes.as_deref()),
Some(&[32][..])
);
assert_eq!(
compact_draft
.tensor("model.mtp.layers.0.mlp.gate_proj.weight")
.unwrap()
.shape,
vec![32, MIMO_TEST_HIDDEN]
);
assert_eq!(
compact_draft
.tensor("model.mtp.layers.0.mlp.down_proj.weight")
.unwrap()
.shape,
vec![MIMO_TEST_HIDDEN, 32]
);
let candidate_main = Arc::new(CmfModel::open(&candidate).unwrap());
assert!(
candidate_main.verify().is_empty(),
"{:?}",
candidate_main.verify()
);
let mut runtime = Pipeline::from_model(&candidate_main, SamplerConfig::default()).unwrap();
assert!(
runtime.mimo_mtp.is_some(),
"candidate must find its sidecar"
);
let probe = runtime.probe_mtp_ffn(&[1, 2, 3, 4, 5, 6]).unwrap();
assert_eq!(probe.len(), 1);
assert_eq!(probe[0].mass.len(), 32);
assert_eq!(probe[0].squared_mass.len(), 32);
assert!(probe[0].observed_positions > 0);
let plan: serde_json::Value = serde_json::from_slice(&fs::read(&plan).unwrap()).unwrap();
assert_eq!(plan["selection"]["calibration_records"], 1);
assert_eq!(plan["selection"]["calibration_selected_records"], 1);
assert_eq!(plan["selection"]["calibration_chunks"], 3);
assert_eq!(plan["selection"]["calibration_tokens"], 9);
assert_eq!(plan["selection"]["max_record_tokens"], 4);
assert_eq!(
plan["selection"]["calibration_chunking"],
"fresh-causal-sequence-per-contiguous-chunk-v1"
);
assert_eq!(plan["selection"]["admission_probe_tokens"], 4);
let legacy_draft = temp.path().join("legacy.mtp.cmf");
let legacy_main = temp.path().join("legacy.cmf");
let legacy_plan = temp.path().join("legacy.mtp.cmf.mtp-plan.json");
cmd_mtp_compact(
source.to_str().unwrap(),
data.to_str().unwrap(),
legacy_draft.to_str().unwrap(),
0.5,
None,
1,
4,
)
.unwrap();
assert!(legacy_draft.exists());
assert!(legacy_plan.exists());
assert!(
!legacy_main.exists(),
"legacy sidecar-only output must not create a new backbone"
);
let legacy = CmfModel::open(&legacy_draft).unwrap();
assert!(legacy.verify().is_empty(), "{:?}", legacy.verify());
let legacy_plan: serde_json::Value =
serde_json::from_slice(&fs::read(&legacy_plan).unwrap()).unwrap();
assert_eq!(
legacy_plan["selection"]["output_layout"],
"legacy-sidecar-only-v1"
);
assert!(CmfModel::open(&source).unwrap().verify().is_empty());
assert!(CmfModel::open(&source_sidecar).unwrap().verify().is_empty());
}
#[cfg(unix)]
#[test]
fn dangling_output_symlink_is_refused() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().unwrap();
let dangling = temp.path().join("candidate.cmf");
symlink("missing-target.cmf", &dangling).unwrap();
assert!(prepare_new_file(&dangling, "candidate main").is_err());
assert!(std::fs::symlink_metadata(&dangling).is_ok());
}
#[cfg(unix)]
#[test]
fn hard_linked_backbone_resolves_a_source_symlink_first() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().unwrap();
let real = temp.path().join("real.cmf");
let source_link = temp.path().join("source.cmf");
let staged = temp.path().join("staged.cmf");
fs::write(&real, b"immutable source").unwrap();
symlink(&real, &source_link).unwrap();
assert_eq!(stage_backbone(&source_link, &staged).unwrap(), "hard-link");
assert!(
!fs::symlink_metadata(&staged)
.unwrap()
.file_type()
.is_symlink()
);
assert_eq!(fs::read(&staged).unwrap(), b"immutable source");
}
#[test]
fn candidate_and_plan_paths_never_overwrite_existing_files() {
let temp = tempfile::tempdir().unwrap();
let existing = temp.path().join("existing.cmf");
fs::write(&existing, b"keep").unwrap();
assert!(prepare_new_file(&existing, "candidate main").is_err());
assert_eq!(fs::read(&existing).unwrap(), b"keep");
let fresh = temp.path().join("nested/fresh.cmf");
prepare_new_file(&fresh, "candidate main").unwrap();
assert!(!fresh.exists());
}
#[test]
fn validated_stage_publishes_without_clobbering() {
let temp = tempfile::tempdir().unwrap();
let staged = temp.path().join("stage.cmf");
let output = temp.path().join("candidate.cmf");
fs::write(&staged, b"validated").unwrap();
publish_new_file(&staged, &output, "candidate main").unwrap();
assert_eq!(fs::read(&output).unwrap(), b"validated");
assert!(!staged.exists());
let staged_again = temp.path().join("stage-again.cmf");
fs::write(&staged_again, b"new").unwrap();
assert!(publish_new_file(&staged_again, &output, "candidate main").is_err());
assert_eq!(fs::read(&output).unwrap(), b"validated");
assert!(staged_again.exists());
}
#[test]
fn source_and_alias_output_are_detected_as_the_same_path() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("draft.cmf");
fs::write(&source, b"cmf").unwrap();
let alias = temp.path().join(".").join("draft.cmf");
assert!(paths_collide(&source, &alias));
assert!(!paths_collide(&source, &temp.path().join("new-output.cmf")));
}
}