"""Full-depth real-weight component checks on small synthetic CPU fixtures.""" import gc import hashlib import importlib.metadata as metadata import importlib.util import json import math import os import sys import time from pathlib import Path import torch import torch.nn.functional as F from safetensors import safe_open ROOT = Path(__file__).resolve().parent def main(name): root = ROOT; started = time.monotonic(); checks = [] torch.set_num_threads(5); torch.manual_seed(1027) def record(label, passed, **details): row = {'name': label, 'passed': bool(passed), **details}; checks.append(row) print(json.dumps(row), flush=True) assert passed, label def close(label, actual, expected, atol=2e-4, rtol=2e-4): record(label, torch.allclose(actual, expected, atol=atol, rtol=rtol), max_abs=float((actual-expected).abs().max()), atol=atol, rtol=rtol) extraction = json.loads((root / 'extraction.json').read_text()) with (root / 'model.safetensors').open('rb') as stream: digest = hashlib.file_digest(stream, 'sha256').hexdigest() record('export_weight_sha256', digest == extraction['sha256']) shapes = {}; dtypes = {}; parameters = 0 with safe_open(root / 'model.safetensors', framework='pt') as handle: for key in handle.keys(): value = handle.get_slice(key); shapes[key] = value.get_shape() parameters += math.prod(shapes[key]); dtype = value.get_dtype() dtypes[dtype] = dtypes.get(dtype, 0) + math.prod(shapes[key]) spec = importlib.util.spec_from_file_location('priority_component_' + name, root / 'component.py') module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) print('LOAD_NATIVE', name, flush=True) component = module.load_component(root, dtype=torch.float32, device='cpu') native = component.native record('strict_native_load_all_keys_and_shapes', {k: list(v.shape) for k, v in native.state_dict().items()} == shapes, tensors=len(shapes), parameters=parameters, original_dtypes=dtypes) calls = [0]; original_sdpa = F.scaled_dot_product_attention def explicit(q, k, v, attn_mask=None, dropout_p=0.0, is_causal=False, scale=None, enable_gqa=False): calls[0] += 1 if enable_gqa: k = k.repeat_interleave(q.shape[-3] // k.shape[-3], -3) v = v.repeat_interleave(q.shape[-3] // v.shape[-3], -3) logits = (q @ k.transpose(-2, -1)) * (scale if scale is not None else q.shape[-1] ** -0.5) if is_causal: logits = logits.masked_fill(~torch.ones(q.shape[-2], k.shape[-2], dtype=torch.bool, device=q.device).tril(), -float('inf')) if attn_mask is not None: logits = logits.masked_fill(~attn_mask, -float('inf')) if attn_mask.dtype == torch.bool else logits + attn_mask return logits.softmax(-1) @ v with torch.inference_mode(): if name == 'internvlu': record('nonlearned_rope_cache_materialized', not native.decoder.pos_embed.all_inv_freqs.is_meta and torch.isfinite(native.decoder.pos_embed.all_inv_freqs).all()) record('native_complete_20_layers', len(native.decoder.transformer_blocks) == 20) x = torch.randn(1, 16, 4, 6) * .2; condition = torch.randn(1, 4, 4096) * .1 y = component.predict(x, 500., condition) record('rectangular_native_latent_forward', y.shape == x.shape and torch.isfinite(y).all(), rms=float(y.square().mean().sqrt())) reference = native.decoder(hidden_states=x, encoder_hidden_states=condition, timestep=torch.tensor([500.]), encoder_attention_mask=torch.ones(1,4,dtype=torch.bool), encoder_image_token_mask=torch.zeros(1,4,dtype=torch.bool), image_fhw_cond=[torch.empty(0,3,dtype=torch.long)], return_dict=False)[0] close('original_native_vs_exported_interface', y, reference, 0, 0) F.scaled_dot_product_attention = explicit try: alternative = component.predict(x, 500., condition) finally: F.scaled_dot_product_attention = original_sdpa close('all_layers_sdpa_vs_explicit_attention', y, alternative, 5e-4, 5e-4) record('conditioning_changes_prediction', not torch.allclose(y, component.predict(x, 500., condition * -.7))) record('time_changes_prediction', not torch.allclose(y, component.predict(x, 700., condition))) condition2 = condition.clone(); condition2[:, -1] = torch.randn(1,4096) * 100 mask = torch.tensor([[1,1,1,0]],dtype=torch.bool) close('masked_condition_isolation', component.predict(x,500.,condition,condition_mask=mask), component.predict(x,500.,condition2,condition_mask=mask), 3e-4, 3e-4) batch = component.predict(torch.cat([x,x*.7]), torch.tensor([500.,700.]), torch.cat([condition,condition*-.7])) close('batch_isolation_first', batch[:1], y, 5e-4, 5e-4) close('batch_isolation_second', batch[1:], component.predict(x*.7,700.,condition*-.7), 5e-4, 5e-4) visual = component.predict(x,500.,condition,condition_image_mask=torch.ones(1,4,dtype=torch.bool),condition_grid=[torch.tensor([[1,2,2]])]) record('semantic_visual_condition_grid', visual.shape == x.shape and torch.isfinite(visual).all()) edit_x = torch.randn(1,16,8,8)*.2 edited = component.predict(edit_x,500.,condition,reference_latents=[torch.randn(1,16,8,8)*.2]) record('reference_vae_editing_input', edited.shape == edit_x.shape and torch.isfinite(edited).all()) scheduler = component.make_scheduler(); scheduler.set_timesteps(3) record('native_dpm_flow_scheduler', type(scheduler).__name__ == 'DPMSolverMultistepScheduler' and scheduler.config.prediction_type == 'flow_prediction' and scheduler.config.use_flow_sigmas and scheduler.config.flow_shift == 3.) z = x.clone(); trajectory = [] for timestep in scheduler.timesteps: pred = component.predict(z,timestep,condition) z = scheduler.step(pred,timestep,z,return_dict=False)[0] trajectory.append(float(z.square().mean().sqrt())) record('three_step_native_dpm_trajectory', torch.isfinite(z).all() and not torch.equal(z,x), latent_rms=trajectory) else: record('native_complete_eight_layers', len(native.blocks) == 8) x = torch.randn(1,4,64)*.2; history = torch.randn(1,32,64)*.2; condition = torch.randn(1,1,896)*.1 y = component.predict(x,.4,condition,history) record('native_current_patch_prediction', y.shape == x.shape and torch.isfinite(y).all(), rms=float(y.square().mean().sqrt())) reference = native(x,torch.tensor([.4]),condition,history,None)[:, -4:] close('original_native_vs_exported_interface',y,reference,0,0) F.scaled_dot_product_attention = explicit try: alternative = component.predict(x,.4,condition,history) finally: F.scaled_dot_product_attention = original_sdpa close('all_layers_sdpa_vs_explicit_attention',y,alternative) record('talker_condition_changes_prediction',not torch.allclose(y,component.predict(x,.4,condition*-.7,history))) record('history_changes_prediction',not torch.allclose(y,component.predict(x,.4,condition,history*-.7))) record('time_changes_prediction',not torch.allclose(y,component.predict(x,.7,condition,history))) batch = component.predict(torch.cat([x,x*.7]),torch.tensor([.4,.7]),torch.cat([condition,condition*-.7]),torch.cat([history,history*-.7])) close('batch_isolation_first',batch[:1],y) close('batch_isolation_second',batch[1:],component.predict(x*.7,.7,condition*-.7,history*-.7)) null = component.predict(x,.4,torch.zeros_like(condition),history) packed = native.forward_with_cfg(x,torch.tensor(.4),condition,history,None) close('native_cfg_condition_branch',packed[:1],y) close('native_cfg_null_branch',packed[1:],null) epss = component.times_function(10,x.device,x.dtype) record('native_nonuniform_ten_step_times',torch.equal(epss,torch.tensor([0,2,4,6,8,12,16,20,24,28,32])/32)) generated = component.sample(condition,history,x) record('ten_step_native_cfm_trajectory',generated.shape==x.shape and torch.isfinite(generated).all() and not torch.equal(generated,x),rms=float(generated.square().mean().sqrt())) # Independent manual integration validates sway, CFG semantics and SDE. sway = epss - (torch.cos(torch.pi/2*epss)-1+epss) z = x.clone() for i in range(10): cond = component.predict(z,sway[i],condition,history) uncond = component.predict(z,sway[i],torch.zeros_like(condition),history) z = z + (cond + 2*(cond-uncond)) * (sway[i+1]-sway[i]) close('cfm_vs_independent_manual_cfg_integration',generated,z,4e-4,4e-4) explicit_noise = [torch.randn_like(x) for _ in range(10)] stochastic = component.sample(condition,history,x,sigma=.1,temperature=.2,sde_noise=explicit_noise) record('native_sde_noise_path',torch.isfinite(stochastic).all() and not torch.allclose(stochastic,generated)) record('independent_attention_reference_exercised',calls[0] >= (20 if name=='internvlu' else 8), calls=calls[0]) versions = {k: metadata.version(k) for k in ('torch','safetensors','diffusers','transformers','einops','huggingface_hub')} if name == 'ming_audio': versions['x-transformers'] = metadata.version('x-transformers') report = {'as_of':'2026-10-27','strict_load':True,'source_tensor_bytes_equal':True,'smoke':{'passed':True,'device':'cpu','runtime_dtype':'float32','synthetic_conditions':True,'real_text_or_audio_conditioning_tested':False,'codec_decoding_tested':False},'checks':checks,'tensors':len(shapes),'parameters':parameters,'original_dtype_elements':dtypes,'elapsed_seconds':round(time.monotonic()-started,3),'versions':versions,'limitations':['Full-depth real-weight CPU component fixtures using synthetic conditions and native latents.','No full VLM/Talker conditioning or image/audio codec decoding; no end-to-end generation-quality claim.','Checkpoint dtypes are preserved; float32 is a runtime validation cast only.']} (root/'validation.json').write_text(json.dumps(report,indent=2)) print('VALIDATED',name,len(checks),flush=True) if __name__ == '__main__': main(sys.argv[1])