"""Phase 0 compiled-execution tests: keys, manifest, on-disk cache, native probe.""" import json from cism import _native as native from cism.compile_cache import ( MANIFEST_FILENAME, artifact_dir, cache_dir_for, canonical_config, read_artifact, write_artifact, ) # --- Minimal Python mirror of the FNV-1a key for determinism/sensitivity --- # Used only by the key-shape tests below (never by the native-probe test), # and asserted only for determinism/sensitivity, never for exact values. _FNV_OFFSET = 14695981039346656037 _FNV_PRIME = 1099511628211 _MASK64 = (1 << 64) - 1 def _py_compute_key(canonical, precision, act, threads, cpu, code): joined = ( f"canonical={canonical};precision={precision};act={act};" f"threads={threads};cpu={cpu};code={code}" ) h = _FNV_OFFSET for byte in joined.encode("utf-8"): h ^= byte h = (h * _FNV_PRIME) & _MASK64 return f"{h:016x}" BASE_INPUTS = dict( canonical='{"hidden_size":32,"layers":2}', precision="fp32", act="fp32", threads=1, cpu="avx2", code="cism-plan-v1", ) def test_key_determinism(): first = _py_compute_key(**BASE_INPUTS) second = _py_compute_key(**BASE_INPUTS) assert first == second assert len(first) == 16 assert all(c in "0123456789abcdef" for c in first) def test_key_sensitivity(): base = _py_compute_key(**BASE_INPUTS) variants = [ {**BASE_INPUTS, "precision": "int8"}, {**BASE_INPUTS, "threads": 2}, {**BASE_INPUTS, "act": "int8"}, {**BASE_INPUTS, "cpu": "scalar"}, {**BASE_INPUTS, "code": "cism-plan-v2"}, {**BASE_INPUTS, "canonical": '{"hidden_size":33,"layers":2}'}, ] for variant in variants: assert _py_compute_key(**variant) != base def test_manifest_parses_with_required_fields(): canonical = canonical_config({"hidden_size": 32, "num_hidden_layers": 2}) result = native.compile_key(canonical, "fp32", "fp32", 1) manifest = json.loads(result["manifest"]) for field in ("magic", "version", "key", "precision", "act_precision", "threads", "cpu_features", "shapes"): assert field in manifest assert manifest["magic"] == "cism-plan" assert manifest["version"] == "cism-plan-v1" assert manifest["key"] == result["key"] assert manifest["precision"] == "fp32" assert manifest["act_precision"] == "fp32" assert manifest["threads"] == 1 assert manifest["cpu_features"] == result["cpu_features"] assert isinstance(manifest["shapes"], list) def test_cache_roundtrip(tmp_path): canonical = canonical_config({"hidden_size": 32}) result = native.compile_key(canonical, "fp32", "fp32", 1) manifest = result["manifest"] cache = cache_dir_for(tmp_path, "org/model", "rev1") directory = artifact_dir(cache, result["key"]) write_artifact(directory, manifest) assert read_artifact(directory) == manifest def test_corrupt_file_reads_as_none(tmp_path): directory = artifact_dir(cache_dir_for(tmp_path, "org/model", "rev1"), "somekey") directory.mkdir(parents=True, exist_ok=True) (directory / MANIFEST_FILENAME).write_bytes(b"\xff\xfe\x00\x01not-json\x00\xff") assert read_artifact(directory) is None def test_missing_dir_reads_as_none(tmp_path): assert read_artifact(tmp_path / "does-not-exist-12345") is None def test_canonical_model_config_matches_native_pipe(): """Python 12-field pipe equals the native Model canonical (fingerprint). Pinned against the verified probe: FP32 hexfloat via FP32 round-trip, tied renders as 0/1. Guards the format without executing anything. """ from cism.compile_cache import canonical_model_config config = dict( model_type="llama", hidden_size=16, intermediate_size=32, num_hidden_layers=1, num_attention_heads=4, num_key_value_heads=2, head_dim=4, vocab_size=32, max_position_embeddings=32, rms_norm_eps=1e-5, rope_theta=10000.0, tie_word_embeddings=False, ) assert (canonical_model_config(config) == "llama|16|32|1|4|2|4|32|32|0x1.4f8b580000000p-17|0x1.3880000000000p+13|0") def test_native_compile_key_probe(): canonical = canonical_config({"hidden_size": 32, "num_hidden_layers": 2}) first = native.compile_key(canonical, "fp32", "fp32", 1) second = native.compile_key(canonical, "fp32", "fp32", 1) assert isinstance(first, dict) assert set(first) == {"key", "manifest", "cpu_features"} # Determinism across two calls; no hash recomputation here. assert first["key"] == second["key"] assert first["manifest"] == second["manifest"] assert first["cpu_features"] == second["cpu_features"] # Difference across precisions. other = native.compile_key(canonical, "int8", "fp32", 1) assert other["key"] != first["key"] # Manifest agrees with the returned key/cpu_features. parsed = json.loads(first["manifest"]) assert parsed["key"] == first["key"] assert parsed["cpu_features"] == first["cpu_features"] def test_model_info_compiled_key_matches_probe(): """Model.info compiled key equals the native probe over info canonical. Documented limitation: Python compile_cache.canonical_config uses a different field set (sorted-JSON config rendering) than the native Model.compile_canonical pipe-delimited string ("model_type|hidden|...|tied"), so cross-language key EQUALITY is explicitly deferred and not asserted here. This test only cross-checks the native Model key against the native compile_key probe using the Model-reported canonical string. No execution-path or numeric changes. """ import numpy as np rng = np.random.default_rng(20260912) hidden, intermediate, layers, heads, kv_heads, head_dim, vocab = 16, 32, 1, 4, 2, 4, 32 config = dict( model_type="llama", hidden_size=hidden, intermediate_size=intermediate, num_hidden_layers=layers, num_attention_heads=heads, num_key_value_heads=kv_heads, head_dim=head_dim, vocab_size=vocab, max_position_embeddings=32, rms_norm_eps=1e-5, rope_theta=10000.0, tie_word_embeddings=False, hidden_act="silu", ) weights = {} def matrix(name, shape): weights[name] = rng.normal(0, 0.14, shape).astype(np.float32) def norm(name, width): weights[name] = rng.uniform(0.75, 1.25, width).astype(np.float32) matrix("model.embed_tokens.weight", (vocab, hidden)) matrix("lm_head.weight", (vocab, hidden)) norm("model.norm.weight", hidden) prefix = "model.layers.0." norm(prefix + "input_layernorm.weight", hidden) norm(prefix + "post_attention_layernorm.weight", hidden) matrix(prefix + "self_attn.q_proj.weight", (heads * head_dim, hidden)) matrix(prefix + "self_attn.k_proj.weight", (kv_heads * head_dim, hidden)) matrix(prefix + "self_attn.v_proj.weight", (kv_heads * head_dim, hidden)) matrix(prefix + "self_attn.o_proj.weight", (hidden, heads * head_dim)) matrix(prefix + "mlp.gate_proj.weight", (intermediate, hidden)) matrix(prefix + "mlp.up_proj.weight", (intermediate, hidden)) matrix(prefix + "mlp.down_proj.weight", (hidden, intermediate)) model = native.Model(config, weights, "int8", 1) info = model.info compiled = info["compiled"] assert compiled["hit"] is False assert compiled["mode"] == "generic" assert isinstance(compiled["canonical"], str) and compiled["canonical"] probe = native.compile_key( compiled["canonical"], info["precision"], info["act_precision"], info["threads"] ) assert compiled["key"] == probe["key"] old_key = compiled["key"] model.set_act_precision("int8") new_info = model.info assert new_info["compiled"]["key"] != old_key recheck = native.compile_key( new_info["compiled"]["canonical"], new_info["precision"], new_info["act_precision"], new_info["threads"], ) assert new_info["compiled"]["key"] == recheck["key"]