#!/usr/bin/env python3 """Build static operation-order and SSA graphs from ONNX-MLIR text IR.""" from __future__ import annotations import argparse import csv import json import os import platform import shlex import shutil import subprocess import sys import tempfile import time import traceback from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts.mlir_graph_common import ( AFFINE_PAIR_IDS, PARSER_SCHEMA_VERSION, VARIANTS, ParsedGraph, atomic_csv, atomic_json, atomic_text, file_record, graph_fingerprint, load_csv, parse_mlir, render_execution_dependency_svg, repo_path, resolve_coverage_path, sha256, ) SCHEMA_VERSION = "1.1" SUCCESS_EVIDENCE_STATUSES = {"PASS", "PASS_WITH_PATCH"} GRAPH_INVENTORY_FIELDS = [ "graph_id", "model_id", "model_name", "task", "variant", "graph_role", "stage", "source_coverage_path", "source_artifact", "source_sha256", "source_bytes", "source_matrix_status", "analysis_status", "failure_code", "failure_detail", "operation_row_count", "compute_operation_count", "module_metadata_operation_count", "function_count", "block_count", "ssa_edge_count", "operation_producer_edge_count", "function_arg_edge_count", "block_arg_edge_count", "external_unclassified_edge_count", "program_order_edge_count", "region_contains_edge_count", "region_yield_edge_count", "loop_backedge_count", "cfg_successor_edge_count", "unresolved_ssa_use_count", "duplicate_ssa_definition_count", "producer_after_consumer_count", "diagnostic_count", "quantization_related_operation_count", "order_semantics", "runtime_order_status", "parser_schema", "fingerprint", "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", "render_command", "render_exit_code", "render_stdout_log", "render_stdout_log_sha256", "render_stderr_log", "render_stderr_log_sha256", "resumed_render", ] PAIR_FIELDS = [ "model_id", "model_name", "task", "fp32_onnx_status", "public_quantized_onnx_status", "pair_common_stage", "common_stage_selection_status", "selection_reason", "fp32_common_source", "fp32_common_sha256", "public_quantized_common_source", "public_quantized_common_sha256", "primary_onnx_graphs", "supplemental_lower_graphs", "fp32_krnl_status", "fp32_krnl_artifact", "fp32_krnl_sha256", "public_quantized_krnl_status", "public_quantized_krnl_artifact", "public_quantized_krnl_sha256", "fp32_llvm_status", "fp32_llvm_artifact", "fp32_llvm_sha256", "public_quantized_llvm_status", "public_quantized_llvm_artifact", "public_quantized_llvm_sha256", "fp32_last_fully_successful_ir", "public_quantized_last_fully_successful_ir", "fp32_krnl_evidence_validation", "fp32_krnl_evidence_failure", "public_quantized_krnl_evidence_validation", "public_quantized_krnl_evidence_failure", "fp32_llvm_evidence_validation", "fp32_llvm_evidence_failure", "public_quantized_llvm_evidence_validation", "public_quantized_llvm_evidence_failure", ] OPERATION_FIELDS = [ "graph_id", "model_id", "task", "variant", "graph_role", "stage", "node_id", "node_kind", "function", "region_path", "block_id", "block_order", "static_order", "operation", "dialect", "results", "result_types", "operands", "onnx_node_name", "result_shapes", "result_dtypes", "result_bytes", "operand_types", "symbol_references", "quantization_role", "source_start_line", "source_end_line", "source_statement_sha256", "dense_payload_omitted", "parent_op_node_id", "opens_region", "rendered_in_graph", ] SSA_FIELDS = [ "graph_id", "model_id", "variant", "stage", "edge_id", "producer_kind", "producer_node_id", "producer_result", "consumer_node_id", "consumer_operand_index", "consumer_operand", "producer_static_order", "consumer_static_order", "producer_before_consumer", ] RELATION_FIELDS = [ "graph_id", "model_id", "variant", "stage", "relation_type", "source_node_id", "target_node_id", "source_block_id", "target_block_id", "detail", ] def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") def normalize_variant(value: str) -> str: return "public_quantized" if value in {"quantized", "public_quantized"} else value def stage_slug(stage: str) -> str: return stage.lower() def runtime_status(model_id: str, stage: str) -> str: if stage != "ONNX": return "RUNTIME_OVERLAY_NOT_PROJECTED_TO_LOWER_IR" return ( "ORT_RUNTIME_PROFILE_AUXILIARY" if model_id in {"LM04", "SP08", "VC13"} else "RUNTIME_ORDER_UNAVAILABLE" ) def supporting_evidence( row: dict[str, str], variant: str, stage: str, root: Path ) -> dict[str, str]: """Validate successful Krnl/LLVM evidence without using it as graph input. Successful artifacts are relocated to the current repository and checked against the matrix digest. PARTIAL/FAIL paths stay byte-for-byte as historical failure evidence and are never opened as successful inputs. """ status = row.get(f"{stage}_status", "UNKNOWN") original_artifact = row.get(f"{stage}_artifact", "UNKNOWN") expected_sha = row.get(f"{stage}_sha256", "UNKNOWN") result = { "status": status, "artifact": original_artifact, "sha256": expected_sha, "validation": "FAILURE_EVIDENCE_ONLY", "failure": "", } if status not in SUCCESS_EVIDENCE_STATUSES: return result context = f"{row.get('model_id', 'UNKNOWN')}:{variant}:{stage.upper()}" try: path = resolve_coverage_path(original_artifact, root) actual_sha = sha256(path) if actual_sha != expected_sha: raise ValueError( f"checksum mismatch: actual={actual_sha}, expected={expected_sha}" ) except (OSError, ValueError) as error: result["validation"] = "FAIL_ANALYSIS" result["failure"] = f"{context}: {type(error).__name__}: {error}" return result result["artifact"] = repo_path(path, root) result["validation"] = "CHECKSUM_VERIFIED" return result def require_matrix_rows(rows: list[dict[str, str]]) -> dict[tuple[str, str], dict[str, str]]: required = { "model_id", "task", "variant", "onnx_status", "onnx_artifact", "onnx_sha256", "affine_scf_memref_status", "affine_scf_memref_artifact", "affine_scf_memref_sha256", } if not rows: raise ValueError("IR coverage matrix is empty") missing = required - set(rows[0]) if missing: raise ValueError(f"IR coverage matrix missing fields: {sorted(missing)}") by_key: dict[tuple[str, str], dict[str, str]] = {} for source in rows: row = dict(source) variant = normalize_variant(row["variant"]) key = (row["model_id"], variant) if key in by_key: raise ValueError(f"duplicate IR coverage row: {key}") row["variant"] = variant by_key[key] = row if len(by_key) != 42: raise ValueError(f"expected 42 active variant rows, found {len(by_key)}") model_ids = sorted({model_id for model_id, _ in by_key}) if len(model_ids) != 21: raise ValueError(f"expected 21 active models, found {len(model_ids)}") for model_id in model_ids: if {(model_id, variant) for variant in VARIANTS} - by_key.keys(): raise ValueError(f"incomplete FP32/public-quantized pair: {model_id}") return by_key def build_specs( by_key: dict[tuple[str, str], dict[str, str]], root: Path, model_names: dict[str, str] | None = None, *, primary_only: bool = False, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: active_ids = {key[0] for key in by_key} names = model_names or {model_id: model_id for model_id in active_ids} if set(names) != active_ids: raise ValueError( f"eligible registry IDs differ from IR coverage: registry={sorted(names)}, coverage={sorted(active_ids)}" ) specs: list[dict[str, Any]] = [] pair_rows: list[dict[str, Any]] = [] for model_id in sorted({key[0] for key in by_key}): pair = {variant: by_key[(model_id, variant)] for variant in VARIANTS} use_affine = model_id in AFFINE_PAIR_IDS and not primary_only common_stage = "AFFINE_SCF_MEMREF" if use_affine else "ONNX" common_paths: dict[str, Path] = {} common_shas: dict[str, str] = {} evidence_by_variant: dict[str, dict[str, dict[str, str]]] = {} for variant in VARIANTS: row = pair[variant] evidence_by_variant[variant] = { stage: supporting_evidence(row, variant, stage, root) for stage in ("krnl", "llvm") } evidence_errors = [ evidence["failure"] for evidence in evidence_by_variant[variant].values() if evidence["failure"] ] if not primary_only else [] primary_resolution_error = "" try: primary_path = resolve_coverage_path(row["onnx_artifact"], root) except (OSError, ValueError) as error: primary_path = Path(row["onnx_artifact"]) primary_resolution_error = f"{type(error).__name__}: {error}" specs.append({ "graph_id": f"{model_id}:{variant}:ONNX", "model_id": model_id, "model_name": names[model_id], "task": row["task"], "variant": variant, "graph_role": "PRIMARY_ONNX_DIALECT", "stage": "ONNX", "coverage_path": row["onnx_artifact"], "source": primary_path, "expected_sha256": row["onnx_sha256"], "source_status": row["onnx_status"], "source_resolution_error": primary_resolution_error, "source_selection_error": ( "ONNX source status is not PASS: " + row["onnx_status"] if row["onnx_status"] != "PASS" else "" ), "supporting_evidence_errors": evidence_errors, }) if use_affine: lower_resolution_error = "" try: lower_path = resolve_coverage_path(row["affine_scf_memref_artifact"], root) except (OSError, ValueError) as error: lower_path = Path(row["affine_scf_memref_artifact"]) lower_resolution_error = f"{type(error).__name__}: {error}" specs.append({ "graph_id": f"{model_id}:{variant}:AFFINE_SCF_MEMREF", "model_id": model_id, "model_name": names[model_id], "task": row["task"], "variant": variant, "graph_role": "SUPPLEMENTAL_PAIR_COMMON_LOWER", "stage": "AFFINE_SCF_MEMREF", "coverage_path": row["affine_scf_memref_artifact"], "source": lower_path, "expected_sha256": row["affine_scf_memref_sha256"], "source_status": row["affine_scf_memref_status"], "source_resolution_error": lower_resolution_error, "source_selection_error": ( "AFFINE_SCF_MEMREF source status is not PASS: " + row["affine_scf_memref_status"] if row["affine_scf_memref_status"] != "PASS" else "" ), "supporting_evidence_errors": evidence_errors, }) common_paths[variant] = lower_path common_shas[variant] = row["affine_scf_memref_sha256"] else: common_paths[variant] = primary_path common_shas[variant] = row["onnx_sha256"] pair_rows.append({ "model_id": model_id, "model_name": names[model_id], "task": pair["fp32"]["task"], "fp32_onnx_status": pair["fp32"]["onnx_status"], "public_quantized_onnx_status": pair["public_quantized"]["onnx_status"], "pair_common_stage": common_stage, "common_stage_selection_status": ( "PASS" if all( pair[variant]["affine_scf_memref_status" if use_affine else "onnx_status"] == "PASS" for variant in VARIANTS ) else "FAIL_ANALYSIS" ), "selection_reason": ( "FP32/Q both PASS at AFFINE_SCF_MEMREF; supplemental same-stage comparison generated" if use_affine else ( "Public graph set uses ONNX Dialect for every model" if primary_only else "No lower stage is PASS for both variants; ONNX Dialect remains the common stage" ) ), "fp32_common_source": repo_path(common_paths["fp32"], root), "fp32_common_sha256": common_shas["fp32"], "public_quantized_common_source": repo_path(common_paths["public_quantized"], root), "public_quantized_common_sha256": common_shas["public_quantized"], "primary_onnx_graphs": "2", "supplemental_lower_graphs": "2" if use_affine else "0 (primary ONNX reused as common view)", "fp32_krnl_status": evidence_by_variant["fp32"]["krnl"]["status"], "fp32_krnl_artifact": evidence_by_variant["fp32"]["krnl"]["artifact"], "fp32_krnl_sha256": evidence_by_variant["fp32"]["krnl"]["sha256"], "public_quantized_krnl_status": evidence_by_variant["public_quantized"]["krnl"]["status"], "public_quantized_krnl_artifact": evidence_by_variant["public_quantized"]["krnl"]["artifact"], "public_quantized_krnl_sha256": evidence_by_variant["public_quantized"]["krnl"]["sha256"], "fp32_llvm_status": evidence_by_variant["fp32"]["llvm"]["status"], "fp32_llvm_artifact": evidence_by_variant["fp32"]["llvm"]["artifact"], "fp32_llvm_sha256": evidence_by_variant["fp32"]["llvm"]["sha256"], "public_quantized_llvm_status": evidence_by_variant["public_quantized"]["llvm"]["status"], "public_quantized_llvm_artifact": evidence_by_variant["public_quantized"]["llvm"]["artifact"], "public_quantized_llvm_sha256": evidence_by_variant["public_quantized"]["llvm"]["sha256"], "fp32_last_fully_successful_ir": pair["fp32"].get("last_fully_successful_ir", "UNKNOWN"), "public_quantized_last_fully_successful_ir": pair["public_quantized"].get("last_fully_successful_ir", "UNKNOWN"), **{ f"{variant}_{stage}_evidence_validation": evidence_by_variant[variant][stage]["validation"] for variant in VARIANTS for stage in ("krnl", "llvm") }, **{ f"{variant}_{stage}_evidence_failure": evidence_by_variant[variant][stage]["failure"] for variant in VARIANTS for stage in ("krnl", "llvm") }, }) expected_graphs = 42 if primary_only else 56 if len(specs) != expected_graphs or len(pair_rows) != 21: raise ValueError(f"selection invariant failed: {len(specs)} graphs, {len(pair_rows)} pairs") return specs, pair_rows def graph_output_paths(output_dir: Path, spec: dict[str, Any]) -> dict[str, Path]: directory = output_dir / "graphs" / spec["model_id"] / spec["variant"] / stage_slug(spec["stage"]) return { "directory": directory, "svg": directory / "execution_dependency_graph.svg", "png": directory / "execution_dependency_graph.png", "record": directory / "graph_record.json", } def resume_record_valid(record_path: Path, fingerprint: str, svg: Path, png: Path) -> bool: if not record_path.is_file() or not svg.is_file() or not png.is_file(): return False try: record = json.loads(record_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False outputs = record.get("outputs", {}) return ( record.get("status") == "PASS" and record.get("fingerprint") == fingerprint and isinstance(record.get("render"), dict) and record["render"].get("status") == "PASS" and outputs.get("svg", {}).get("sha256") == sha256(svg) and outputs.get("png", {}).get("sha256") == sha256(png) ) def load_resume_record( record_path: Path, fingerprint: str, svg: Path, png: Path ) -> tuple[dict[str, Any] | None, bool, str]: """Load a resume record while distinguishing retryable output drift. Malformed JSON, an I/O exception, or an incompatible fingerprint is a per-graph analysis failure. A well-formed record with incomplete or checksum-mismatched outputs is retryable from the render stage. """ try: record = json.loads(record_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: return None, False, f"RESUME_RECORD_READ: {type(error).__name__}: {error}" if not isinstance(record, dict): return None, False, "RESUME_RECORD_SCHEMA: top-level JSON value is not an object" if record.get("fingerprint") != fingerprint: return record, False, ( "OUTPUT_SETTINGS_CONFLICT: resume fingerprint differs; existing outputs " "were not overwritten" ) try: valid = resume_record_valid(record_path, fingerprint, svg, png) except OSError as error: return record, False, f"RESUME_OUTPUT_READ: {type(error).__name__}: {error}" return record, valid, "" def render_one( *, svg: Path, png: Path, log_dir: Path, inkscape: str, root: Path, ) -> dict[str, Any]: log_dir.mkdir(parents=True, exist_ok=True) stdout_path = log_dir / "render.stdout.log" stderr_path = log_dir / "render.stderr.log" command_path = log_dir / "render.command.txt" exit_path = log_dir / "render.exit_code.txt" temporary_png = png.with_name(f".{png.name}.rendering.png") command = [ inkscape, str(svg), "--export-type=png", f"--export-filename={temporary_png}", "--export-width=2400", ] atomic_text(command_path, shlex.join(command) + "\n") exit_code = 127 failure = "" output_replaced = False temporary_bytes = 0 render_started_ns = time.time_ns() try: if temporary_png.exists(): temporary_png.unlink() except OSError as error: failure = f"STALE_TEMP_REMOVE: {type(error).__name__}: {error}" with stdout_path.open("wb") as stdout_handle, stderr_path.open("wb") as stderr_handle: if not failure: try: completed = subprocess.run( command, cwd=root, stdout=stdout_handle, stderr=stderr_handle, check=False, timeout=300, ) exit_code = completed.returncode except (OSError, subprocess.TimeoutExpired) as error: failure = f"{type(error).__name__}: {error}" if failure: stderr_handle.write((failure + "\n").encode("utf-8", errors="replace")) atomic_text(exit_path, f"{exit_code}\n") if exit_code == 0 and temporary_png.is_file(): try: temporary_bytes = temporary_png.stat().st_size with temporary_png.open("rb") as stream: png_signature = stream.read(8) if temporary_bytes <= 8 or png_signature != b"\x89PNG\r\n\x1a\n": failure = ( "RENDER_OUTPUT_INVALID: newly created temporary output is not a " f"non-empty PNG ({temporary_bytes} bytes)" ) else: png.parent.mkdir(parents=True, exist_ok=True) os.replace(temporary_png, png) output_replaced = True except OSError as error: failure = f"RENDER_OUTPUT_REPLACE: {type(error).__name__}: {error}" elif exit_code == 0: failure = "RENDER_OUTPUT_MISSING: Inkscape exited 0 without creating a new temporary PNG" elif not failure: failure = f"INKSCAPE_EXIT_NONZERO: exit_code={exit_code}" if temporary_png.exists(): try: temporary_png.unlink() except OSError as error: failure = (failure + "; " if failure else "") + f"TEMP_CLEANUP: {type(error).__name__}: {error}" status = "PASS" if exit_code == 0 and output_replaced and png.is_file() else "FAIL" return { "status": status, "failure_code": None if status == "PASS" else "FAIL_ANALYSIS", "failure_detail": failure, "failure_stage": None if status == "PASS" else "RENDER", "command": shlex.join(command), "command_argv": command, "exit_code": exit_code, "render_started_ns": render_started_ns, "temporary_output_bytes": temporary_bytes, "output_replaced": output_replaced, "stdout_log": file_record(stdout_path, root), "stderr_log": file_record(stderr_path, root), "command_log": file_record(command_path, root), "exit_code_log": file_record(exit_path, root), } def checkpoint(path: Path, rows: list[dict[str, Any]], started_at: str) -> None: atomic_json(path, { "schema_version": SCHEMA_VERSION, "stage": "T85_MLIR_IR_GRAPH_BUILD_CHECKPOINT", "started_at": started_at, "updated_at": utc_now(), "completed_graph_count": sum(row.get("analysis_status") == "PASS" for row in rows), "failed_graph_count": sum(row.get("analysis_status") == "FAIL" for row in rows), "graphs": rows, "policy": { "existing_mlir_read_only": True, "converter_run": False, "mlir_toolchain_run": False, "model_runtime_run": False, "allocator_work_performed": False, }, }) def analysis_diagnostic( stage: str, detail: str, *, error: BaseException | None = None, source: Path | None = None, ) -> dict[str, Any]: line = getattr(error, "lineno", None) if error is not None else None return { "code": "FAIL_ANALYSIS", "stage": stage, "detail": detail, "exception_type": type(error).__name__ if error is not None else "", "source": str(source) if source is not None else "", "source_line": line if isinstance(line, int) else None, "traceback": traceback.format_exc() if error is not None else "", } def empty_parsed_graph(source: Path, diagnostic: dict[str, Any]) -> ParsedGraph: return ParsedGraph( source=source, operations=[], ssa_edges=[], relations=[], definitions=[], diagnostics=[diagnostic], functions=[], block_count=0, unresolved_use_count=0, duplicate_definition_count=0, producer_after_consumer_count=0, ) def failed_render(stage: str, detail: str) -> dict[str, Any]: return { "status": "FAIL", "failure_code": "FAIL_ANALYSIS", "failure_stage": stage, "failure_detail": detail, "command": f"NOT_RUN_{stage}", "command_argv": [], "exit_code": -1, "output_replaced": False, "stdout_log": {"path": "", "sha256": ""}, "stderr_log": {"path": "", "sha256": ""}, "command_log": {"path": "", "sha256": ""}, "exit_code_log": {"path": "", "sha256": ""}, } def source_record(item: dict[str, Any], root: Path) -> dict[str, Any]: spec = item["spec"] source = spec["source"] record: dict[str, Any] = { "coverage_path": spec["coverage_path"], "path": repo_path(source, root), "exists": source.is_file(), "bytes": None, "sha256": item.get("source_sha", ""), "matrix_status": spec["source_status"], "expected_sha256": spec["expected_sha256"], "checksum_matches_matrix": ( bool(item.get("source_sha")) and item.get("source_sha") == spec["expected_sha256"] ), } if source.is_file(): try: record["bytes"] = source.stat().st_size except OSError: pass return record def graph_counts(parsed: ParsedGraph) -> dict[str, Any]: relation_counts = Counter(row["relation_type"] for row in parsed.relations) operation_producer = sum(edge["producer_kind"] == "OPERATION" for edge in parsed.ssa_edges) function_arg_edges = sum(edge["producer_kind"] == "FUNCTION_ARG" for edge in parsed.ssa_edges) block_arg_edges = sum(edge["producer_kind"] == "BLOCK_ARG" for edge in parsed.ssa_edges) external_unclassified = ( len(parsed.ssa_edges) - operation_producer - function_arg_edges - block_arg_edges ) quantization_nodes = sum( op.get("quantization_role", "NONE") != "NONE" for op in parsed.operations if op["function"] != "module" ) return { "operation_rows": len(parsed.operations), "compute_operations": sum(op["function"] != "module" for op in parsed.operations), "module_metadata_operations": sum(op["function"] == "module" for op in parsed.operations), "functions": len(parsed.functions), "blocks": parsed.block_count, "ssa_edges": len(parsed.ssa_edges), "operation_producer_edges": operation_producer, "function_arg_edges": function_arg_edges, "block_arg_edges": block_arg_edges, "external_unclassified_edges": external_unclassified, "quantization_related_operations": quantization_nodes, "relations": dict(sorted(relation_counts.items())), "unresolved_ssa_uses": parsed.unresolved_use_count, "duplicate_ssa_definitions": parsed.duplicate_definition_count, "producer_after_consumer": parsed.producer_after_consumer_count, "diagnostics": len(parsed.diagnostics), } def build_graph_record(item: dict[str, Any], root: Path) -> dict[str, Any]: spec = item["spec"] outputs = item["outputs"] parsed: ParsedGraph = item["parsed"] render = item.get("render") status = "PASS" if render and render.get("status") == "PASS" else "FAIL" valid_outputs = status == "PASS" failure_detail = "" if status == "PASS" else (render or {}).get( "failure_detail", "unknown analysis failure" ) return { "schema_version": SCHEMA_VERSION, "stage": "T85_MLIR_IR_GRAPH_RECORD", "status": status, "failure_code": None if status == "PASS" else "FAIL_ANALYSIS", "failure_stage": None if status == "PASS" else (render or {}).get("failure_stage", "ANALYSIS"), "failure_detail": failure_detail, "analysis_diagnostics": item.get("analysis_diagnostics", []), "graph_id": spec["graph_id"], "model_id": spec["model_id"], "model_name": spec["model_name"], "task": spec["task"], "variant": spec["variant"], "graph_role": spec["graph_role"], "mlir_stage": spec["stage"], "fingerprint": item.get("fingerprint", "NOT_AVAILABLE"), "source": source_record(item, root), "counts": graph_counts(parsed), "order_semantics": "STATIC_MLIR_PROGRAM_ORDER", "runtime_order_status": runtime_status(spec["model_id"], spec["stage"]), "parser_diagnostics": parsed.diagnostics, "render_metadata": item.get("render_meta", {}), "render": render, "outputs": { "svg": file_record(outputs["svg"], root) if valid_outputs and outputs["svg"].is_file() else None, "png": file_record(outputs["png"], root) if valid_outputs and outputs["png"].is_file() else None, }, "retained_stale_outputs": { "svg_exists": outputs["svg"].is_file(), "png_exists": outputs["png"].is_file(), } if not valid_outputs else None, "resumed_render": item.get("resumed", False), "policy": { "existing_mlir_read_only": True, "converter_run": False, "mlir_toolchain_run": False, "model_runtime_run": False, "dataset_work_performed": False, "allocator_work_performed": False, "model_weight_architecture_modified": False, }, } def persist_graph_record(item: dict[str, Any], root: Path) -> None: if not item.get("write_record", True): return atomic_json(item["outputs"]["record"], build_graph_record(item, root)) item["record_written"] = True def mark_item_failure( item: dict[str, Any], root: Path, stage: str, detail: str, *, error: BaseException | None = None, ) -> None: diagnostic = analysis_diagnostic( stage, detail, error=error, source=item["spec"].get("source") ) item.setdefault("analysis_diagnostics", []).append(diagnostic) if item.get("parsed") is None: item["parsed"] = empty_parsed_graph(item["spec"]["source"], diagnostic) item["render"] = failed_render(stage, detail) item["resumed"] = False try: persist_graph_record(item, root) except OSError as record_error: # The checkpoint still captures this exact failure even when the # per-graph destination itself is temporarily unwritable. record_detail = f"GRAPH_RECORD_WRITE: {type(record_error).__name__}: {record_error}" item["analysis_diagnostics"].append( analysis_diagnostic( "GRAPH_RECORD_WRITE", record_detail, error=record_error, source=item["outputs"]["record"], ) ) item["render"] = failed_render("GRAPH_RECORD_WRITE", record_detail) item["record_written"] = False def checkpoint_rows(items: list[dict[str, Any]], root: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for item in items: render = item.get("render") rows.append({ "graph_id": item["spec"]["graph_id"], "analysis_status": render.get("status") if render else "RUNNING", "failure_code": (render or {}).get("failure_code", ""), "failure_stage": (render or {}).get("failure_stage", ""), "failure_detail": (render or {}).get("failure_detail", ""), "source": source_record(item, root), "graph_record": repo_path(item["outputs"]["record"], root), "graph_record_written": bool(item.get("record_written")), }) return rows def build_inventory_row(item: dict[str, Any], root: Path) -> dict[str, Any]: spec = item["spec"] outputs = item["outputs"] parsed: ParsedGraph = item["parsed"] render = item.get("render") or failed_render("ANALYSIS", "missing render result") status = "PASS" if render.get("status") == "PASS" else "FAIL" counts = graph_counts(parsed) relations = counts["relations"] source = source_record(item, root) valid_outputs = status == "PASS" return { "graph_id": spec["graph_id"], "model_id": spec["model_id"], "model_name": spec["model_name"], "task": spec["task"], "variant": spec["variant"], "graph_role": spec["graph_role"], "stage": spec["stage"], "source_coverage_path": spec["coverage_path"], "source_artifact": source["path"], "source_sha256": source["sha256"], "source_bytes": source["bytes"] if source["bytes"] is not None else "", "source_matrix_status": spec["source_status"], "analysis_status": status, "failure_code": "" if status == "PASS" else "FAIL_ANALYSIS", "failure_detail": "" if status == "PASS" else render.get("failure_detail", "unknown failure"), "operation_row_count": counts["operation_rows"], "compute_operation_count": counts["compute_operations"], "module_metadata_operation_count": counts["module_metadata_operations"], "function_count": counts["functions"], "block_count": counts["blocks"], "ssa_edge_count": counts["ssa_edges"], "operation_producer_edge_count": counts["operation_producer_edges"], "function_arg_edge_count": counts["function_arg_edges"], "block_arg_edge_count": counts["block_arg_edges"], "external_unclassified_edge_count": counts["external_unclassified_edges"], "program_order_edge_count": relations.get("PROGRAM_ORDER", 0), "region_contains_edge_count": relations.get("REGION_CONTAINS", 0), "region_yield_edge_count": relations.get("REGION_YIELD", 0), "loop_backedge_count": relations.get("LOOP_BACKEDGE", 0), "cfg_successor_edge_count": relations.get("CFG_SUCCESSOR", 0), "unresolved_ssa_use_count": counts["unresolved_ssa_uses"], "duplicate_ssa_definition_count": counts["duplicate_ssa_definitions"], "producer_after_consumer_count": counts["producer_after_consumer"], "diagnostic_count": counts["diagnostics"], "quantization_related_operation_count": counts["quantization_related_operations"], "order_semantics": "STATIC_MLIR_PROGRAM_ORDER", "runtime_order_status": runtime_status(spec["model_id"], spec["stage"]), "parser_schema": PARSER_SCHEMA_VERSION, "fingerprint": item.get("fingerprint", "NOT_AVAILABLE"), "execution_dependency_graph_svg": repo_path(outputs["svg"], root), "execution_dependency_graph_svg_sha256": ( sha256(outputs["svg"]) if valid_outputs and outputs["svg"].is_file() else "" ), "execution_dependency_graph_png": repo_path(outputs["png"], root), "execution_dependency_graph_png_sha256": ( sha256(outputs["png"]) if valid_outputs and outputs["png"].is_file() else "" ), "graph_record_json": repo_path(outputs["record"], root), "graph_record_json_sha256": sha256(outputs["record"]) if outputs["record"].is_file() else "", "render_command": render.get("command", ""), "render_exit_code": render.get("exit_code", ""), "render_stdout_log": render.get("stdout_log", {}).get("path", ""), "render_stdout_log_sha256": render.get("stdout_log", {}).get("sha256", ""), "render_stderr_log": render.get("stderr_log", {}).get("path", ""), "render_stderr_log_sha256": render.get("stderr_log", {}).get("sha256", ""), "resumed_render": item.get("resumed", False), } def decorate_parsed_rows( spec: dict[str, Any], parsed: ParsedGraph ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: base = { "graph_id": spec["graph_id"], "model_id": spec["model_id"], "task": spec["task"], "variant": spec["variant"], "graph_role": spec["graph_role"], "stage": spec["stage"], } operations = [] for source in parsed.operations: row = {**base, **source} row["results"] = ";".join(source["results"]) row["result_types"] = ";".join(source["result_types"]) row["result_shapes"] = ";".join(source["result_shapes"]) row["result_dtypes"] = ";".join(source["result_dtypes"]) row["result_bytes"] = ";".join(source["result_bytes"]) row["operands"] = ";".join(source["operands"]) row["operand_types"] = ";".join(source["operand_types"]) row["symbol_references"] = ";".join(source["symbol_references"]) row["rendered_in_graph"] = source["function"] != "module" operations.append(row) edges = [{**{k: base[k] for k in ("graph_id", "model_id", "variant", "stage")}, **row} for row in parsed.ssa_edges] relations = [{**{k: base[k] for k in ("graph_id", "model_id", "variant", "stage")}, **row} for row in parsed.relations] return operations, edges, relations def report_text( inventory: list[dict[str, Any]], pair_rows: list[dict[str, Any]], summary: dict[str, Any], ) -> str: by_graph = {(row["model_id"], row["variant"], row["stage"]): row for row in inventory} lines = [ "# MLIR 정적 실행순서·SSA 의존 그래프 보고서", "", "## 1. 결과", "", "활성 21개 모델의 FP32·공개 양자화 ONNX Dialect 42개를 모두 그래프로 만들었다. " "FP32와 양자화가 함께 성공한 7개 모델은 Affine/SCF/MemRef 보조 graph 14개도 추가했다. " "따라서 총 56개 SVG/PNG와 operation·SSA·제어흐름 CSV가 생성됐다.", "검증 범위: 42/42 ONNX primary graph, 14/14 lower supplemental graph.", "", "주 graph의 node는 MLIR operation, 파란 edge는 SSA 값의 정의에서 사용으로 이어지는 연결이다. " "node 번호와 배치는 MLIR text의 정적 program order를 따른다.", "", "## 2. IR 단계별 역할", "", "| IR 단계 | 이 보고서에서의 역할 | 모델 범위 |", "|---|---|---:|", "| ONNX Dialect | 전체 모델의 공통 주 graph | 21 pair / 42 variant |", "| Affine/SCF/MemRef | lowering 후 구조를 보는 보조 graph | 7 pair / 14 variant |", "| Krnl | lowering 성공·실패 evidence | 그림 생성 대상 아님 |", "| LLVM Dialect | codegen 가능성 evidence | 그림 생성 대상 아님 |", "", "## 3. 전체 21개 모델 결과", "", "표의 `Op`는 `func.return`을 포함한 main graph 함수의 operation 수다. `SSA`는 operation, " "함수 입력 또는 lower-IR region argument에서 consumer로 연결된 값 사용 수다.", "`공통 비교 단계`는 FP32와 공개 양자화 모델이 둘 다 성공한 가장 낮은 IR 단계다. " "`Q 관련 Op`는 Quantize, Dequantize, Requantize 또는 정수 양자화 연산 수다.", "", "| 모델 | 공통 비교 단계 | ONNX Op (FP32→Q) | Q 관련 Op (FP32→Q) | ONNX SSA (FP32→Q) | 보조 lower Op (FP32→Q) | 결과 | graph |", "|---|---|---:|---:|---:|---:|---|---|", ] for pair in pair_rows: model_id = pair["model_id"] fp = by_graph[(model_id, "fp32", "ONNX")] quant = by_graph[(model_id, "public_quantized", "ONNX")] if pair["pair_common_stage"] == "AFFINE_SCF_MEMREF": lower_fp = by_graph[(model_id, "fp32", "AFFINE_SCF_MEMREF")] lower_q = by_graph[(model_id, "public_quantized", "AFFINE_SCF_MEMREF")] lower_ops = f"{int(lower_fp['compute_operation_count']):,}→{int(lower_q['compute_operation_count']):,}" links = ( f"[ONNX FP32](graphs/{model_id}/fp32/onnx/execution_dependency_graph.svg) / " f"[ONNX Q](graphs/{model_id}/public_quantized/onnx/execution_dependency_graph.svg) / " f"[Lower FP32](graphs/{model_id}/fp32/affine_scf_memref/execution_dependency_graph.svg) / " f"[Lower Q](graphs/{model_id}/public_quantized/affine_scf_memref/execution_dependency_graph.svg)" ) else: lower_ops = "—" links = ( f"[FP32](graphs/{model_id}/fp32/onnx/execution_dependency_graph.svg) / " f"[Q](graphs/{model_id}/public_quantized/onnx/execution_dependency_graph.svg)" ) result = "PASS" if fp["analysis_status"] == quant["analysis_status"] == "PASS" else "FAIL" lines.append( f"| {model_id} — {pair['model_name']} | {pair['pair_common_stage']} | " f"{int(fp['compute_operation_count']):,}→{int(quant['compute_operation_count']):,} | " f"{int(fp['quantization_related_operation_count']):,}→{int(quant['quantization_related_operation_count']):,} | " f"{int(fp['ssa_edge_count']):,}→{int(quant['ssa_edge_count']):,} | " f"{lower_ops} | {result} | {links} |" ) lines.extend([ "", "## 4. Krnl·LLVM lowering evidence", "", "Krnl과 LLVM은 graph 입력으로 사용하지 않고 기존 T60의 stage status·artifact·checksum을 " "`pair_stage_selection.csv`에 연결했다.", "", "| 모델 | Krnl (FP32 / Q) | LLVM (FP32 / Q) | 마지막 성공 IR (FP32 / Q) |", "|---|---|---|---|", ]) for pair in pair_rows: lines.append( f"| {pair['model_id']} | {pair['fp32_krnl_status']} / {pair['public_quantized_krnl_status']} | " f"{pair['fp32_llvm_status']} / {pair['public_quantized_llvm_status']} | " f"{pair['fp32_last_fully_successful_ir']} / {pair['public_quantized_last_fully_successful_ir']} |" ) lines.extend([ "", "## 5. graph 읽는 방법", "", "- node 왼쪽 숫자: 해당 MLIR function 안에서의 정적 operation 순서", "- 파란 선: SSA value의 producer operation에서 consumer operation으로 향하는 연결", "- 회색 선: 같은 block에 연속해서 적힌 operation의 program order", "- 보라 선: loop 같은 operation이 내부 region을 포함하는 관계", "- 빨간 점선: loop back-edge, region yield 또는 명시적 control-flow successor", "- 주황색 테두리: quantize/dequantize/requantize 또는 정수 양자화 operator", "", "SVG는 확대해 각 node의 operation·block·source line을 확인할 수 있고, PNG는 빠른 검토용이다. " "정확한 전체 목록은 `operation_order.csv`, `ssa_edges.csv`, " "`control_flow_edges.csv`에 있다.", "", "## 6. 검증 결과", "", f"- source checksum 일치: {summary['counts']['source_checksum_pass_graphs']}/56", f"- 분석·렌더 PASS: {summary['counts']['graph_pass']}/56", f"- unresolved SSA use: {summary['counts']['unresolved_ssa_uses']}", f"- duplicate SSA definition: {summary['counts']['duplicate_ssa_definitions']}", f"- producer-after-consumer 위반: {summary['counts']['producer_after_consumer']}", f"- ONNX primary operation rows: {summary['counts']['primary_onnx_operation_rows']:,}", f"- Affine 보조 operation rows: {summary['counts']['supplemental_affine_operation_rows']:,}", "", "## 7. 결과 파일", "", "- 입력 목록·checksum: `operation_inventory.csv`", "- pair 공통 단계: `pair_stage_selection.csv`", "- operation/SSA/control 관계: `operation_order.csv`, `ssa_edges.csv`, `control_flow_edges.csv`", "- package checksum: `artifact_manifest.json`, `artifacts.sha256`", "", ]) return "\n".join(lines) def gallery_text(inventory: list[dict[str, Any]]) -> str: cards = [] for row in inventory: svg = row["execution_dependency_graph_svg"] png = row["execution_dependency_graph_png"] relative_svg = str(Path(svg).relative_to("reports/graphs/mlir")) relative_png = str(Path(png).relative_to("reports/graphs/mlir")) cards.append( f'

{row["model_id"]} · {row["variant"]} · {row["stage"]}

' f'{row[' f'

{int(row["compute_operation_count"]):,} ops · {int(row["ssa_edge_count"]):,} SSA uses

' ) return """T85 MLIR graph gallery

T85 MLIR graph gallery

그림을 클릭하면 확대 가능한 SVG를 엽니다.

""" + "".join(cards) + "
\n" def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=REPO_ROOT) parser.add_argument("--coverage-matrix", type=Path, default=Path("reports/conversion/ir_stage_coverage.csv")) parser.add_argument("--output-dir", type=Path, default=Path("reports/graphs/mlir")) parser.add_argument("--render-log-dir", type=Path, required=True) parser.add_argument("--checkpoint", type=Path, required=True) parser.add_argument("--resume", action="store_true") parser.add_argument("--render-workers", type=int, default=1) parser.add_argument( "--primary-only", action="store_true", help="Generate only the 42 ONNX Dialect FP32/public-quantized graphs.", ) parser.add_argument( "--images-only", action="store_true", help="Publish only SVG/PNG graph images; omit internal CSV, JSON, report, and gallery files.", ) args = parser.parse_args() if args.images_only and not args.primary_only: parser.error("--images-only requires --primary-only") if args.images_only and args.resume: parser.error("--images-only does not use graph_record.json; --resume is unavailable") started_at = utc_now() root = args.repo_root.resolve() matrix_path = args.coverage_matrix.resolve() if args.coverage_matrix.is_absolute() else (root / args.coverage_matrix).resolve() output_dir = args.output_dir.resolve() if args.output_dir.is_absolute() else (root / args.output_dir).resolve() render_log_dir = args.render_log_dir.resolve() if args.render_log_dir.is_absolute() else (root / args.render_log_dir).resolve() checkpoint_path = args.checkpoint.resolve() if args.checkpoint.is_absolute() else (root / args.checkpoint).resolve() for path in (matrix_path, output_dir, render_log_dir, checkpoint_path.parent): try: path.relative_to(root) except ValueError as error: raise SystemExit(f"T85 path outside repository root: {path}") from error output_dir.mkdir(parents=True, exist_ok=True) render_log_dir.mkdir(parents=True, exist_ok=True) checkpoint_path.parent.mkdir(parents=True, exist_ok=True) inkscape = shutil.which("inkscape") if not inkscape: raise SystemExit("inkscape is required for T85 PNG rendering") version_result = subprocess.run([inkscape, "--version"], capture_output=True, text=True, check=False, timeout=15) inkscape_version = (version_result.stdout + version_result.stderr).strip() if version_result.returncode != 0: raise SystemExit(f"inkscape --version failed: {inkscape_version}") matrix_rows = load_csv(matrix_path) by_key = require_matrix_rows(matrix_rows) registry_rows = load_csv(root / "model_registry.csv") model_names = { row["model_id"]: row["model_name"] for row in registry_rows if row.get("eligibility") == "ELIGIBLE" } specs, pair_rows = build_specs( by_key, root, model_names, primary_only=args.primary_only ) operation_rows: list[dict[str, Any]] = [] ssa_rows: list[dict[str, Any]] = [] relation_rows: list[dict[str, Any]] = [] work: list[dict[str, Any]] = [] preliminary: list[dict[str, Any]] = [] for spec in specs: source = spec["source"] outputs = graph_output_paths(output_dir, spec) item: dict[str, Any] = { "spec": spec, "outputs": outputs, "fingerprint": "NOT_AVAILABLE", "parsed": None, "render_meta": {}, "render": None, "resumed": False, "source_sha": "", "analysis_diagnostics": [], "record_written": False, "write_record": not args.images_only, } preliminary.append(item) try: outputs["directory"].mkdir(parents=True, exist_ok=True) except OSError as error: detail = f"OUTPUT_DIRECTORY: {type(error).__name__}: {error}" mark_item_failure(item, root, "OUTPUT_DIRECTORY", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue selection_errors = [ spec.get("source_selection_error", ""), spec.get("source_resolution_error", ""), *spec.get("supporting_evidence_errors", []), ] selection_errors = [detail for detail in selection_errors if detail] if selection_errors: detail = "; ".join(selection_errors) mark_item_failure(item, root, "SOURCE_SELECTION", detail) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue try: actual_sha = sha256(source) item["source_sha"] = actual_sha except OSError as error: detail = f"SOURCE_READ: {type(error).__name__}: {error}" mark_item_failure(item, root, "SOURCE_READ", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue if actual_sha != spec["expected_sha256"]: detail = ( f"SOURCE_CHECKSUM: actual={actual_sha}, expected={spec['expected_sha256']}" ) mark_item_failure(item, root, "SOURCE_CHECKSUM", detail) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue graph_runtime_status = runtime_status(spec["model_id"], spec["stage"]) try: item["fingerprint"] = graph_fingerprint( source, actual_sha, spec["source_status"], spec["stage"], inkscape_version, graph_runtime_status, ) parsed = parse_mlir(source, spec["graph_id"]) item["parsed"] = parsed except Exception as error: detail = f"PARSE: {type(error).__name__}: {error}" mark_item_failure(item, root, "PARSE", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue try: ops, edges, relations = decorate_parsed_rows(spec, parsed) except Exception as error: detail = f"PARSE_DECORATE: {type(error).__name__}: {error}" mark_item_failure(item, root, "PARSE_DECORATE", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue operation_rows.extend(ops) ssa_rows.extend(edges) relation_rows.extend(relations) if ( parsed.unresolved_use_count or parsed.duplicate_definition_count or parsed.producer_after_consumer_count or parsed.diagnostics ): detail = ( f"PARSE_INVARIANT: unresolved={parsed.unresolved_use_count}, " f"duplicate={parsed.duplicate_definition_count}, " f"producer_after={parsed.producer_after_consumer_count}, " f"diagnostics={len(parsed.diagnostics)}" ) mark_item_failure(item, root, "PARSE_INVARIANT", detail) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue try: svg_text, item["render_meta"] = render_execution_dependency_svg( parsed, title=f"{spec['model_id']} · {spec['variant']} · {spec['stage']}", graph_id=spec["graph_id"], stage=spec["stage"], runtime_order_status=graph_runtime_status, ) except Exception as error: detail = f"SVG_BUILD: {type(error).__name__}: {error}" mark_item_failure(item, root, "SVG_BUILD", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue if args.resume and outputs["record"].exists(): previous, resume_valid, resume_error = load_resume_record( outputs["record"], item["fingerprint"], outputs["svg"], outputs["png"] ) if resume_error: # Preserve the incompatible/malformed canonical record as # evidence and write this attempt beside it. outputs["record"] = outputs["directory"] / "graph_record.resume_failure.json" mark_item_failure(item, root, "RESUME", resume_error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue if resume_valid and previous is not None: item["render"] = previous["render"] item["resumed"] = True try: persist_graph_record(item, root) except OSError as error: detail = f"GRAPH_RECORD_WRITE: {type(error).__name__}: {error}" mark_item_failure( item, root, "GRAPH_RECORD_WRITE", detail, error=error ) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue try: atomic_text(outputs["svg"], svg_text) except OSError as error: detail = f"SVG_WRITE: {type(error).__name__}: {error}" mark_item_failure(item, root, "SVG_WRITE", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) continue work.append(item) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) if work: by_id = {item["spec"]["graph_id"]: item for item in preliminary} with ThreadPoolExecutor(max_workers=max(1, min(args.render_workers, 4))) as executor: futures = {} for item in work: spec = item["spec"] paths = item["outputs"] log_dir = render_log_dir / spec["model_id"] / spec["variant"] / stage_slug(spec["stage"]) try: future = executor.submit( render_one, svg=paths["svg"], png=paths["png"], log_dir=log_dir, inkscape=inkscape, root=root, ) futures[future] = spec["graph_id"] except Exception as error: detail = f"RENDER_SUBMIT: {type(error).__name__}: {error}" mark_item_failure(item, root, "RENDER_SUBMIT", detail, error=error) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) for future in as_completed(futures): graph_id = futures[future] item = by_id[graph_id] try: item["render"] = future.result() except Exception as error: detail = f"RENDER_FUTURE: {type(error).__name__}: {error}" mark_item_failure(item, root, "RENDER_FUTURE", detail, error=error) else: if item["render"].get("status") != "PASS": detail = item["render"].get("failure_detail", "render failed") item.setdefault("analysis_diagnostics", []).append( analysis_diagnostic("RENDER", detail, source=item["outputs"]["svg"]) ) try: # Persist immediately after each future so interruption # resumes from the next unfinished graph. persist_graph_record(item, root) except OSError as error: detail = f"GRAPH_RECORD_WRITE: {type(error).__name__}: {error}" mark_item_failure( item, root, "GRAPH_RECORD_WRITE", detail, error=error ) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) # Re-write records deterministically after all workers completed. They # were already checkpointed atomically at each terminal transition. for item in preliminary: if not item.get("record_written"): try: persist_graph_record(item, root) except OSError as error: detail = f"GRAPH_RECORD_WRITE: {type(error).__name__}: {error}" mark_item_failure(item, root, "GRAPH_RECORD_WRITE", detail, error=error) inventory_rows = [build_inventory_row(item, root) for item in preliminary] inventory_rows.sort(key=lambda row: (row["model_id"], VARIANTS.index(row["variant"]), row["stage"])) operation_rows.sort(key=lambda row: (row["model_id"], VARIANTS.index(row["variant"]), row["stage"], row["static_order"])) ssa_rows.sort(key=lambda row: (row["model_id"], VARIANTS.index(row["variant"]), row["stage"], row["consumer_static_order"], row["consumer_operand_index"])) relation_rows.sort(key=lambda row: (row["model_id"], VARIANTS.index(row["variant"]), row["stage"], row["relation_type"], row["source_node_id"], row["target_node_id"])) # Detailed inventories are local analysis evidence. The public # primary/images-only mode intentionally publishes only the final graphs. if not args.images_only: atomic_csv(output_dir / "operation_inventory.csv", inventory_rows, GRAPH_INVENTORY_FIELDS) atomic_csv(output_dir / "pair_stage_selection.csv", pair_rows, PAIR_FIELDS) atomic_csv(output_dir / "operation_order.csv", operation_rows, OPERATION_FIELDS) atomic_csv(output_dir / "ssa_edges.csv", ssa_rows, SSA_FIELDS) atomic_csv(output_dir / "control_flow_edges.csv", relation_rows, RELATION_FIELDS) counts = { "active_models": 21, "active_variants": 42, "graph_total": len(inventory_rows), "primary_onnx_graphs": sum(row["stage"] == "ONNX" for row in inventory_rows), "supplemental_affine_graphs": sum(row["stage"] == "AFFINE_SCF_MEMREF" for row in inventory_rows), "pair_common_affine_models": 0 if args.primary_only else len(AFFINE_PAIR_IDS), "pair_common_onnx_models": 21 if args.primary_only else 21 - len(AFFINE_PAIR_IDS), "graph_pass": sum(row["analysis_status"] == "PASS" for row in inventory_rows), "graph_fail": sum(row["analysis_status"] != "PASS" for row in inventory_rows), "source_checksum_pass_graphs": sum(row["source_sha256"] == next( spec["expected_sha256"] for spec in specs if spec["graph_id"] == row["graph_id"] ) for row in inventory_rows), "operation_rows": len(operation_rows), "ssa_edge_rows": len(ssa_rows), "control_relation_rows": len(relation_rows), "supporting_evidence_failures": sum( row[field] == "FAIL_ANALYSIS" for row in pair_rows for field in ( "fp32_krnl_evidence_validation", "public_quantized_krnl_evidence_validation", "fp32_llvm_evidence_validation", "public_quantized_llvm_evidence_validation", ) ), "primary_onnx_operation_rows": sum(row["stage"] == "ONNX" and row["rendered_in_graph"] for row in operation_rows), "supplemental_affine_operation_rows": sum(row["stage"] == "AFFINE_SCF_MEMREF" and row["rendered_in_graph"] for row in operation_rows), "unresolved_ssa_uses": sum(int(row["unresolved_ssa_use_count"]) for row in inventory_rows), "duplicate_ssa_definitions": sum(int(row["duplicate_ssa_definition_count"]) for row in inventory_rows), "producer_after_consumer": sum(int(row["producer_after_consumer_count"]) for row in inventory_rows), "region_yield_edges": sum(int(row["region_yield_edge_count"]) for row in inventory_rows), "loop_backedges": sum(int(row["loop_backedge_count"]) for row in inventory_rows), "quantization_related_operation_rows": sum( row["quantization_role"] != "NONE" for row in operation_rows if row["rendered_in_graph"] ), "result_values_with_known_type": sum( type_value not in {"", "UNKNOWN"} for row in operation_rows for type_value in row["result_types"].split(";") if type_value ), "result_values_with_unknown_type": sum( type_value == "UNKNOWN" for row in operation_rows for type_value in row["result_types"].split(";") if type_value ), "resumed_renders": sum(bool(row["resumed_render"]) for row in inventory_rows), } lowering_evidence = { "krnl_status_counts": dict(Counter( status for row in pair_rows for status in (row["fp32_krnl_status"], row["public_quantized_krnl_status"]) )), "llvm_status_counts": dict(Counter( status for row in pair_rows for status in (row["fp32_llvm_status"], row["public_quantized_llvm_status"]) )), "source": repo_path(matrix_path, root), "role": "LOWERING_CODEGEN_EVIDENCE_NOT_GRAPH_INPUT", } expected_graphs = 42 if args.primary_only else 56 invariant_keys = [ "unresolved_ssa_uses", "duplicate_ssa_definitions", "producer_after_consumer", ] if not args.primary_only: invariant_keys.append("supporting_evidence_failures") overall_status = "PASS" if ( counts["graph_pass"] == expected_graphs and not any( counts[key] for key in invariant_keys ) and all(row["common_stage_selection_status"] == "PASS" for row in pair_rows) ) else "FAIL" summary = { "schema_version": SCHEMA_VERSION, "stage": "T85_MLIR_IR_GRAPH", "status": overall_status, "failure_code": None if overall_status == "PASS" else "FAIL_ANALYSIS", "generated_at": utc_now(), "started_at": started_at, "order_semantics": "STATIC_MLIR_PROGRAM_ORDER", "primary_graph_stage": "ONNX", "supplemental_graph_stage": ( "NOT_INCLUDED" if args.primary_only else "AFFINE_SCF_MEMREF_WHEN_PAIR_COMMON_PASS" ), "publication_mode": "IMAGES_ONLY" if args.images_only else "FULL_LOCAL_EVIDENCE", "counts": counts, "lowering_evidence": lowering_evidence, "tool_versions": { "python": platform.python_version(), "inkscape": inkscape_version, "parser_schema": PARSER_SCHEMA_VERSION, "converter": "NOT_RUN", "mlir_toolchain": "NOT_RUN", "model_runtime": "NOT_RUN", }, "inputs": { "ir_coverage_matrix": file_record(matrix_path, root), "model_registry": file_record(root / "model_registry.csv", root), "mlir_source_total_bytes": sum( int(source_record(item, root)["bytes"] or 0) for item in preliminary ), }, "policy": { "existing_mlir_read_only": True, "converter_run": False, "mlir_toolchain_run": False, "model_runtime_run": False, "dataset_work_performed": False, "allocator_work_performed": False, "model_weight_architecture_modified": False, "prohibited_operations_performed": [], }, } if not args.images_only: atomic_json(output_dir / "summary.json", summary) atomic_text(output_dir / "mlir_ir_graph_report.md", report_text(inventory_rows, pair_rows, summary)) atomic_text(output_dir / "mlir_ir_graph_gallery.html", gallery_text(inventory_rows)) checkpoint(checkpoint_path, checkpoint_rows(preliminary, root), started_at) print(json.dumps({ "status": overall_status, "failure_code": summary["failure_code"], "counts": counts, "output_dir": repo_path(output_dir, root), "checkpoint": repo_path(checkpoint_path, root), "allocator_work_performed": False, "model_runtime_run": False, "mlir_toolchain_run": False, }, ensure_ascii=False, sort_keys=True)) return 0 if overall_status == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())