Spaces:
Sleeping
Sleeping
| """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"] | |