| |
| """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()) |
|
|
