import hashlib
import json
import math
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
CANONICAL_K21 = 1 << 41
U64_SPACE = 1 << 64
EXPECTED_HASH_NAMES = ["jamhash", "murmur3", "xxhash3", "wang64"]
EXPECTED_FRAC_SCALES = [10, 100, 1000, 10000]
EXPECTED_MUTATION_VALUES = {
"jamhash": (31.999994582691556, 218.80340756751377),
"murmur3": (32.00000005870091, 251.07203094638322),
"xxhash3": (32.000003531250215, 1887897.8494955385),
"wang64": (32.051766198057926, 3914408417.2748675),
}
def require(condition: bool, message: str) -> None:
if not condition:
raise SystemExit(f"evidence verification failed: {message}")
def close(actual: float, expected: float, *, tolerance: float = 1e-12) -> bool:
return math.isclose(actual, expected, rel_tol=tolerance, abs_tol=tolerance)
def verify_checksums() -> None:
manifest = ROOT / "evidence/checksums.sha256"
for line in manifest.read_text(encoding="utf-8").splitlines():
expected, relative_path = line.split(maxsplit=1)
path = ROOT / relative_path
actual = hashlib.sha256(path.read_bytes()).hexdigest()
require(actual == expected, f"checksum mismatch for {relative_path}")
def load_json(relative_path: str) -> dict:
with (ROOT / relative_path).open(encoding="utf-8") as source:
return json.load(source)
def verify_combined_metrics() -> list[dict]:
report = load_json("evidence/k21/combined_metrics.json")
hashes = report.get("hashes", [])
require(len(hashes) == 4, "combined report must contain four hash functions")
require(
[entry["hash_name"] for entry in hashes] == EXPECTED_HASH_NAMES,
"combined hash identifiers",
)
expected_pairs = CANONICAL_K21 * (CANONICAL_K21 - 1) / (2 * U64_SPACE)
for entry in hashes:
name = entry["hash_name"]
require(entry["total_hashes"] == CANONICAL_K21, f"{name}: total hash count")
require(entry["expected_hashes"] == CANONICAL_K21, f"{name}: expected hash count")
require(0 <= entry["min_hash"] <= entry["max_hash"] < U64_SPACE, f"{name}: range")
require(
close(entry["expected_collision_pairs"], expected_pairs, tolerance=1e-9),
f"{name}: random-map collision expectation",
)
scales = entry["frac_scales"]
counts = entry["frac_pass_counts"]
observed = entry["frac_observed"]
expected = entry["frac_expected"]
errors = entry["frac_abs_error"]
require(scales == EXPECTED_FRAC_SCALES, f"{name}: FracMinHash scales")
require(
len(scales) == len(counts) == len(observed) == len(expected) == len(errors),
f"{name}: FracMinHash array lengths",
)
for scale, count, obs, exp, error in zip(scales, counts, observed, expected, errors):
require(scale > 0 and 0 <= count <= CANONICAL_K21, f"{name}: scale {scale}")
require(close(obs, count / CANONICAL_K21), f"{name}: observed scale {scale}")
require(close(exp, 1 / scale), f"{name}: expected scale {scale}")
require(close(error, abs(obs - exp)), f"{name}: error scale {scale}")
require(
close(entry["frac_max_abs_error"], max(errors)),
f"{name}: maximum FracMinHash error",
)
pairs = entry["observed_collision_pairs"]
duplicates = entry["duplicate_hashes"]
elements = entry["collision_elements"]
multiplicity = entry["max_multiplicity"]
require(pairs >= 0 and duplicates >= 0 and elements >= 0, f"{name}: collision counts")
if multiplicity == 0:
require((pairs, duplicates, elements) == (0, 0, 0), f"{name}: empty collisions")
elif multiplicity == 2:
require(pairs == duplicates, f"{name}: all collision groups are doublets")
require(elements == 2 * duplicates, f"{name}: affected input count")
else:
require(False, f"{name}: unsupported stored multiplicity {multiplicity}")
require(
close(entry["collision_ratio_obs_exp"], pairs / expected_pairs, tolerance=1e-9),
f"{name}: collision ratio",
)
require(0 <= entry["bit_max_abs_error"] <= 0.5, f"{name}: bit balance")
require(entry["chi_squared_statistic"] >= 0, f"{name}: prefix chi-squared")
require(0 <= entry["ks_statistic"] <= 1, f"{name}: KS statistic")
jamhash = hashes[0]
require(jamhash["min_hash"] == 0, "jamhash minimum hash")
require(jamhash["observed_collision_pairs"] == 130928, "jamhash collision pairs")
require(jamhash["collision_elements"] == 261856, "jamhash affected inputs")
require(jamhash["max_multiplicity"] == 2, "jamhash maximum multiplicity")
return hashes
def verify_linear_metrics() -> list[dict]:
report = load_json("evidence/k21/linear_test.json")
require(report.get("total_canonical") == CANONICAL_K21, "linear canonical count")
require(report.get("tasks") == 32, "linear task count")
hashes = report.get("hashes", [])
require(
[entry["hash_name"] for entry in hashes] == EXPECTED_HASH_NAMES,
"linear hash identifiers",
)
for entry in hashes:
name = entry["hash_name"]
values = [value for key, value in entry.items() if key != "hash_name"]
require(all(math.isfinite(value) for value in values), f"{name}: finite linear metrics")
require(-1 <= entry["serial_correlation"] <= 1, f"{name}: serial correlation")
require(0 <= entry["order_preservation_frac"] <= 1, f"{name}: ordering fraction")
require(entry["gap_chi_squared"] >= 0, f"{name}: gap chi-squared")
require(
0
<= entry["mutation_min_hamming"]
<= entry["mutation_avg_hamming"]
<= entry["mutation_max_hamming"]
<= 64,
f"{name}: mutation Hamming range",
)
require(entry["mutation_hamming_std"] >= 0, f"{name}: mutation Hamming deviation")
require(entry["mutation_diff_chi_squared"] >= 0, f"{name}: mutation chi-squared")
expected_hamming, expected_chi = EXPECTED_MUTATION_VALUES[name]
require(
close(entry["mutation_avg_hamming"], expected_hamming),
f"{name}: documented Hamming mean",
)
require(
close(entry["mutation_diff_chi_squared"], expected_chi),
f"{name}: documented mutation chi-squared",
)
return hashes
def verify_zero_policy(jamhash: dict) -> dict:
report = load_json("evidence/k21/zero_policy.json")
dataset = report["dataset"]
require(dataset["id"] == "complete-canonical-packed-k21", "zero-policy dataset identifier")
require(dataset["canonical_count"] == CANONICAL_K21, "zero-policy canonical count")
require(report["hash"]["id"] == "jamhash_u64", "zero-policy hash identifier")
require(
report["hash"]["source_revision"] == "0015dcc31abeeae1a8b83fdfb9a09d2f16d39401",
"zero-policy hash source revision",
)
measurement = report["zero_measurement"]
zero_count = measurement["zero_output_count"]
require(zero_count == 1, "jamhash zero-output count")
counter_hash = hashlib.sha256((ROOT / "scripts/count_zero_outputs.py").read_bytes()).hexdigest()
require(counter_hash == measurement["counter_sha256"], "zero counter checksum")
require(
measurement["counter_source_revision"] == "b98348576b46c08641bd59498a8495326e3c95c1",
"zero counter source revision",
)
artifact = measurement["artifact"]
require(artifact["size_bytes"] == artifact["partition_values"] * 8, "sorted partition size")
require(artifact["partition_values"] == 34359721240, "sorted prefix-0 population")
require(artifact["values_examined"] == zero_count + 1, "zero measurement prefix length")
require(artifact["first_nonzero_output"] > 0, "first nonzero output")
combined_hash = hashlib.sha256(
(ROOT / "evidence/k21/combined_metrics.json").read_bytes()
).hexdigest()
require(
combined_hash == artifact["combined_metrics_sha256"],
"zero measurement aggregate checksum",
)
reporting = report["threshold_reporting"]
archived = reporting["archived"]
runtime = reporting["runtime_equivalent"]
require(archived["predicate"] == "hash < u64::MAX / scale", "archived threshold predicate")
require(
runtime["predicate"] == "hash != 0 && hash < u64::MAX / scale",
"runtime threshold predicate",
)
require(archived["zero_policy"] == "inclusive", "archived zero policy")
require(runtime["zero_policy"] == "excluded", "runtime zero policy")
require(archived["scales"] == EXPECTED_FRAC_SCALES, "zero-policy FracMinHash scales")
require(archived["pass_counts"] == jamhash["frac_pass_counts"], "archived threshold counts")
require(
runtime["pass_counts"]
== [count - zero_count for count in archived["pass_counts"]],
"zero-excluding threshold counts",
)
require(
report["pipeline_provenance"]["source_snapshot_revision"]
== "58e81cc179bf9a7f745c58bcf7d73aac20dc2d05",
"pipeline source snapshot",
)
return report
def verify_readme(jamhash: dict, linear_hashes: list[dict], zero_policy: dict) -> None:
readme = (ROOT / "README.md").read_text(encoding="utf-8")
require(f"{CANONICAL_K21:,}" in readme, "README canonical population")
require(
"not selected, sampled, or biologically filtered" in readme,
"README exhaustive dataset language",
)
require("A separate, smaller study" in readme, "README mutation-study separation")
require(
"sensitivity, specificity, false-positive" in readme,
"README screening evidence boundary",
)
collision_row = (
f"| `jamhash_u64` | {jamhash['observed_collision_pairs']:,} | "
f"{jamhash['collision_ratio_obs_exp']:.6f} | {jamhash['collision_elements']:,} | "
f"{jamhash['max_multiplicity']} |"
)
require(collision_row in readme, "README jamhash collision row")
labels = {
"jamhash": "`jamhash_u64`",
"murmur3": "fastmurmur3, seed 42",
"xxhash3": "XXH3",
"wang64": "Wang 64-bit mixer",
}
for entry in linear_hashes:
mutation_row = (
f"| {labels[entry['hash_name']]} | "
f"{entry['mutation_diff_chi_squared']:,.2f} | "
f"{entry['mutation_avg_hamming']:.5f} |"
)
require(mutation_row in readme, f"README {entry['hash_name']} mutation row")
measurement = zero_policy["zero_measurement"]
require(
f"exactly {measurement['zero_output_count']} zero-valued output" in readme,
"README zero-output count",
)
reporting = zero_policy["threshold_reporting"]
for scale, archived, runtime in zip(
reporting["archived"]["scales"],
reporting["archived"]["pass_counts"],
reporting["runtime_equivalent"]["pass_counts"],
):
row = f"| {scale:,} | {archived:,} | {runtime:,} |"
require(row in readme, f"README scale-{scale} zero-policy row")
def main() -> None:
verify_checksums()
combined = verify_combined_metrics()
linear = verify_linear_metrics()
jamhash = combined[0]
zero_policy = verify_zero_policy(jamhash)
verify_readme(jamhash, linear, zero_policy)
print(
"evidence verified: "
f"{CANONICAL_K21:,} canonical k=21 values, "
f"{jamhash['observed_collision_pairs']:,} jamhash collision pairs, "
f"{zero_policy['zero_measurement']['zero_output_count']} zero output, "
f"max multiplicity {jamhash['max_multiplicity']}"
)
if __name__ == "__main__":
main()