#!/usr/bin/env python3 """Strict C4 audit and deterministic table for native author-code baselines.""" import argparse import hashlib import json import math import os import re ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DEFAULT_PROTOCOLS = os.path.join( ROOT, "experiments", "native_baseline_protocols.json") PROTOCOL_IDS = ( "burstccn_cifar10_symplastic_paper_seed0", "dualprop_cifar10_vgg16_a0_b0p1_seed1988", ) SHA256_RE = re.compile(r"^[0-9a-f]{64}$") GIT_REVISION_RE = re.compile(r"^[0-9a-f]{40,64}$") def file_hash(path): digest = hashlib.sha256() with open(path, "rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def require(condition, message): if not condition: raise ValueError(message) def number(value, label, low=None, high=None): require(isinstance(value, (int, float)) and math.isfinite(value), f"{label} must be finite, found {value!r}") if low is not None: require(value >= low, f"{label} must be >= {low}, found {value}") if high is not None: require(value <= high, f"{label} must be <= {high}, found {value}") return float(value) def accuracy(metrics, key, label): return number(metrics.get(key), f"{label}.{key}", 0.0, 100.0) def verify_artifacts(records, label): require(isinstance(records, list) and records, f"{label} must be nonempty") for index, record in enumerate(records): prefix = f"{label}[{index}]" require(isinstance(record.get("path"), str) and record["path"], f"{prefix}.path is missing") require(isinstance(record.get("bytes"), int) and record["bytes"] > 0, f"{prefix}.bytes must be positive") require(bool(SHA256_RE.fullmatch(str(record.get("sha256", "")))), f"{prefix}.sha256 is invalid") def validate_common(row, protocol_id, protocol): label = protocol_id require(row.get("schema_version") == 1, f"{label}: unsupported schema") require(row.get("protocol_id") == protocol_id, f"{label}: protocol ID mismatch") require(row.get("protocol") == protocol, f"{label}: frozen protocol drift") require(row.get("method") == protocol["method"], f"{label}: method mismatch") require(row.get("dataset") == protocol["dataset"], f"{label}: dataset mismatch") require(row.get("seed") == protocol["seed"], f"{label}: seed mismatch") audit = row.get("audit", {}) for key in ("complete", "finite", "source_revision_match", "source_tracked_clean", "selection_semantics_frozen"): require(audit.get(key) is True, f"{label}: audit.{key} is not true") require(audit.get("expected_epochs") == protocol["expected_epochs"], f"{label}: expected epoch audit mismatch") source = row.get("source_provenance", {}) require(source.get("revision") == protocol["source_revision"], f"{label}: author source revision mismatch") require(source.get("tracked_dirty") is False, f"{label}: author source is not tracked-clean") importer = row.get("importer_provenance", {}) require(bool(GIT_REVISION_RE.fullmatch(str(importer.get("revision", "")))), f"{label}: importer revision is invalid") require(importer.get("tracked_dirty") is False, f"{label}: importer source was not tracked-clean") environment = row.get("runtime_environment", {}) freeze = environment.get("pip_freeze") require(isinstance(freeze, list) and freeze, f"{label}: package freeze missing") freeze_text = ("\n".join(freeze) + "\n").encode() require(hashlib.sha256(freeze_text).hexdigest() == environment.get("pip_freeze_sha256"), f"{label}: package freeze hash mismatch") gpu = row.get("gpu", {}) for key, expected in protocol["expected_gpu"].items(): require(gpu.get(key) == expected, f"{label}: GPU {key} expected {expected!r}, found {gpu.get(key)!r}") verify_artifacts(row.get("dataset_artifacts"), f"{label}.dataset_artifacts") dataset_hashes = {item["sha256"] for item in row["dataset_artifacts"]} require(set(protocol["required_dataset_sha256"]).issubset(dataset_hashes), f"{label}: required dataset hashes missing") verify_artifacts(row.get("run_artifacts"), f"{label}.run_artifacts") metrics = row.get("metrics", {}) require(metrics.get("epochs_observed") == protocol["expected_epochs"], f"{label}: incomplete metric history") number(metrics.get("wall_s"), f"{label}.wall_s", 0.0) require(metrics["wall_s"] > 0, f"{label}: wall time must be positive") require(isinstance(metrics.get("wall_definition"), str) and metrics["wall_definition"], f"{label}: wall definition missing") def validate_burst(row, protocol): label = row["protocol_id"] validate_common(row, label, protocol) metrics = row["metrics"] require(metrics.get("optimization_epochs") == 399, f"{label}: author-loop optimization epoch count mismatch") require(metrics.get("initial_evaluation_epoch") == 0, f"{label}: epoch-0 evaluation missing") require(metrics.get("primary_final_epoch") == 399, f"{label}: final epoch mismatch") require(metrics.get("test_evaluations") == 400, f"{label}: native per-epoch test count mismatch") for key in ( "primary_final_test_accuracy_percent", "final_validation_accuracy_percent", "validation_selected_validation_accuracy_percent", "validation_selected_test_accuracy_percent", "test_selected_test_accuracy_percent"): accuracy(metrics, key, label) require(0 <= metrics.get("validation_selected_epoch", -1) <= 399, f"{label}: validation-selected epoch invalid") require(0 <= metrics.get("test_selected_epoch", -1) <= 399, f"{label}: test-selected epoch invalid") require(metrics["test_selected_test_accuracy_percent"] >= metrics["primary_final_test_accuracy_percent"], f"{label}: test-selected result is below final result") def validate_dualprop(row, protocol): label = row["protocol_id"] validate_common(row, label, protocol) metrics = row["metrics"] require(metrics.get("final_epoch") == 130, f"{label}: final epoch mismatch") require(metrics.get("test_evaluations") == 1, f"{label}: expected a single test evaluation") require(metrics.get("test_selected_test_accuracy_percent") is None, f"{label}: test-selected metric must be unavailable") for key in ( "final_train_accuracy_percent", "final_validation_accuracy_percent", "validation_selected_validation_accuracy_percent", "validation_selected_test_accuracy_percent", "primary_test_accuracy_percent", "primary_test_top5_accuracy_percent"): accuracy(metrics, key, label) require(1 <= metrics.get("validation_selected_epoch", 0) <= 130, f"{label}: validation-selected epoch invalid") require(metrics["primary_test_accuracy_percent"] == metrics["validation_selected_test_accuracy_percent"], f"{label}: primary is not the validation-selected checkpoint") component_sum = (metrics.get("training_wall_s", 0) + metrics.get("validation_wall_s", 0) + metrics.get("test_wall_s", 0)) require(math.isclose(component_sum, metrics["wall_s"], rel_tol=1e-12, abs_tol=1e-9), f"{label}: component wall times do not sum to total") def render(rows, protocols): burst = rows[PROTOCOL_IDS[0]] dual = rows[PROTOCOL_IDS[1]] bm, dm = burst["metrics"], dual["metrics"] bp, dp = protocols[PROTOCOL_IDS[0]], protocols[PROTOCOL_IDS[1]] return f"""# Native author-code baseline audit Both records passed the strict C4 artifact, source, protocol, environment, dataset, completeness, finiteness, selection, and cost-definition checks. These are method-native reproductions, not equal-compute comparisons with SDIL. | method | reproduced seed | primary test (%) | validation-selected test (%) | test-selected test (%) | published (%) | audited wall (s) | |:--|--:|--:|--:|--:|--:|--:| | BurstCCN | {bp['seed']} | {bm['primary_final_test_accuracy_percent']:.3f} | {bm['validation_selected_test_accuracy_percent']:.3f} | {bm['test_selected_test_accuracy_percent']:.3f} | {bp['published_accuracy_percent_mean']:.2f} ± {bp['published_accuracy_percent_sd']:.2f} | {bm['wall_s']:.1f} | | Dual Prop | {dp['seed']} | {dm['primary_test_accuracy_percent']:.3f} | {dm['validation_selected_test_accuracy_percent']:.3f} | — | {dp['published_accuracy_percent_mean']:.2f} ± {dp['published_accuracy_percent_sd']:.2f} | {dm['wall_s']:.1f} | BurstCCN's primary metric is its final epoch. Its validation-selected metric is the test accuracy from the minimum-validation-error epoch; its best test value is explicitly test-selected because the native trainer evaluates test every epoch. The `n_epochs=400` author configuration performs an epoch-0 evaluation and 399 optimization epochs. Dual Prop evaluates test once after restoring the best-validation checkpoint, so its primary and validation-selected values are identical and no test-selected value exists. The BurstCCN wall value is W&B run time through the final epoch log. The Dual Prop wall value sums author-measured train, validation, and final test time and excludes setup, checkpoint I/O, and diagnostic plotting. One reproduced seed establishes executable fidelity; published multi-seed uncertainty is retained for context and is not replaced by a one-seed error bar. """ def main(): parser = argparse.ArgumentParser() parser.add_argument("--protocols", default=DEFAULT_PROTOCOLS) parser.add_argument("--results", default=os.path.join(ROOT, "results", "native")) parser.add_argument("--output", default=os.path.join( ROOT, "results", "native_baseline_audit.md")) args = parser.parse_args() with open(args.protocols) as handle: ledger = json.load(handle) require(ledger.get("schema_version") == 1, "unsupported protocol ledger schema") protocols = ledger["protocols"] rows = {} for protocol_id in PROTOCOL_IDS: path = os.path.join(args.results, protocol_id + ".json") require(os.path.isfile(path), f"missing native result: {path}") with open(path) as handle: rows[protocol_id] = json.load(handle) validate_burst(rows[PROTOCOL_IDS[0]], protocols[PROTOCOL_IDS[0]]) validate_dualprop(rows[PROTOCOL_IDS[1]], protocols[PROTOCOL_IDS[1]]) text = render(rows, protocols) os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) with open(args.output, "w") as handle: handle.write(text) manifest = { "strict": True, "c4_pass": True, "protocol_ledger_sha256": file_hash(args.protocols), "result_sha256": { protocol_id: file_hash(os.path.join(args.results, protocol_id + ".json")) for protocol_id in PROTOCOL_IDS }, } print(json.dumps(manifest, indent=2, sort_keys=True)) if __name__ == "__main__": main()