#!/usr/bin/env python3 """Validate the compact MLIR graph package without the removed source MLIR. The full source-reparse result is retained as ``source_validation_at_generation.json``. This validator checks that record, then independently verifies the current package identities, graph artifacts, CSV counts, and manifest hashes after repository-only naming changes. """ from __future__ import annotations import argparse import csv import hashlib import json import os import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def rows(path: Path) -> list[dict[str, str]]: with path.open(encoding="utf-8", newline="") as handle: return list(csv.DictReader(handle)) def csv_data_rows(path: Path) -> int: with path.open(encoding="utf-8", newline="") as handle: reader = csv.reader(handle) next(reader, None) return sum(1 for _ in reader) def atomic_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle: json.dump(value, handle, indent=2, sort_keys=True, ensure_ascii=False) handle.write("\n") temporary = Path(handle.name) os.replace(temporary, path) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(".")) parser.add_argument("--report-dir", type=Path, default=Path("reports/graphs/mlir")) parser.add_argument("--output", type=Path) args = parser.parse_args() root = args.repo_root.resolve() report_dir = args.report_dir if args.report_dir.is_absolute() else root / args.report_dir output = args.output or report_dir / "validation.json" if not output.is_absolute(): output = root / output checks = 0 errors: list[str] = [] def check(condition: bool, detail: str) -> None: nonlocal checks checks += 1 if not condition: errors.append(detail) historical_path = report_dir / "source_validation_at_generation.json" historical = json.loads(historical_path.read_text(encoding="utf-8")) check(historical.get("status") == "PASS", "historical source validation is not PASS") historical_counts = historical.get("counts", {}) check(historical_counts.get("failed_calls") == 0, "historical source validation has failures") check(historical_counts.get("independently_parsed_graphs") == 56, "historical source parse count is not 56") inventory = rows(report_dir / "operation_inventory.csv") check(len(inventory) == 56, f"inventory rows {len(inventory)} != 56") check(len({row["graph_id"] for row in inventory}) == 56, "duplicate graph_id") check(sum(row["stage"] == "ONNX" for row in inventory) == 42, "ONNX graph count != 42") check(sum(row["stage"] == "AFFINE_SCF_MEMREF" for row in inventory) == 14, "Affine graph count != 14") for row in inventory: check(row["analysis_status"] == "PASS", f"{row['graph_id']}: analysis status is not PASS") for path_field, hash_field in ( ("execution_dependency_graph_svg", "execution_dependency_graph_svg_sha256"), ("execution_dependency_graph_png", "execution_dependency_graph_png_sha256"), ("graph_record_json", "graph_record_json_sha256"), ): path = root / row[path_field] check(path.is_file(), f"{row['graph_id']}: missing {row[path_field]}") if path.is_file(): check(sha256(path) == row[hash_field], f"{row['graph_id']}: checksum mismatch {path_field}") record_path = root / row["graph_record_json"] if record_path.is_file(): record = json.loads(record_path.read_text(encoding="utf-8")) check(record.get("graph_id") == row["graph_id"], f"{row['graph_id']}: record identity mismatch") check(record.get("stage") == "T85_MLIR_IR_GRAPH_RECORD", f"{row['graph_id']}: old record stage") summary = json.loads((report_dir / "summary.json").read_text(encoding="utf-8")) check(summary.get("status") == "PASS", "summary status is not PASS") check(summary.get("stage") == "T85_MLIR_IR_GRAPH", "summary uses an old stage name") counts = summary.get("counts", {}) check(csv_data_rows(report_dir / "operation_order.csv") == counts.get("operation_rows"), "operation row count mismatch") check(csv_data_rows(report_dir / "ssa_edges.csv") == counts.get("ssa_edge_rows"), "SSA row count mismatch") check(csv_data_rows(report_dir / "control_flow_edges.csv") == counts.get("control_relation_rows"), "control row count mismatch") manifest = json.loads((report_dir / "artifact_manifest.json").read_text(encoding="utf-8")) check(manifest.get("status") == "PASS", "artifact manifest is not PASS") check(manifest.get("stage") == "T85_MLIR_IR_GRAPH_MANIFEST", "manifest uses an old stage name") manifest_paths: set[str] = set() for item in manifest.get("files", []): path = root / item["path"] manifest_paths.add(item["path"]) check(path.is_file(), f"manifest file missing: {item['path']}") if path.is_file(): check(path.stat().st_size == item["bytes"], f"manifest size mismatch: {item['path']}") check(sha256(path) == item["sha256"], f"manifest checksum mismatch: {item['path']}") check(len(manifest_paths) == manifest.get("file_count"), "manifest duplicate path or count mismatch") check(str(historical_path.relative_to(root)) in manifest_paths, "historical source validation is not pinned") status = "PASS" if not errors else "FAIL" result = { "schema_version": "1.0", "stage": "T85_RETAINED_MLIR_IR_GRAPH_VALIDATION", "status": status, "generated_at": datetime.now(timezone.utc).isoformat(), "checks": checks, "failed_checks": len(errors), "errors": errors, "validation_scope": "retained package integrity after repository naming cleanup", "source_reparse_status": "PASS_AT_GENERATION_NOT_REPEATED_AFTER_COMPACT_CLEANUP", "source_validation": { "path": str(historical_path.relative_to(root)), "sha256": sha256(historical_path), "independently_parsed_graphs": historical_counts.get("independently_parsed_graphs"), "check_calls": historical_counts.get("check_calls"), "failed_calls": historical_counts.get("failed_calls"), }, "policy": { "model_runtime_run": False, "compiler_or_lowering_run": False, "dataset_run": False, "allocator_run": False, }, } atomic_json(output, result) print(f"{status}: checks={checks} failures={len(errors)}") return 0 if status == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())