scripts/build_mlir_ir_graphs.py

#!/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'<article><h2>{row["model_id"]} · {row["variant"]} · {row["stage"]}</h2>'
            f'<a href="{relative_svg}"><img loading="lazy" src="{relative_png}" alt="{row["graph_id"]}"></a>'
            f'<p>{int(row["compute_operation_count"]):,} ops · {int(row["ssa_edge_count"]):,} SSA uses</p></article>'
        )
    return """<!doctype html><html lang="ko"><head><meta charset="utf-8"><title>T85 MLIR graph gallery</title>
<style>body{font-family:sans-serif;margin:24px;background:#f8fafc}main{display:grid;grid-template-columns:repeat(auto-fit,minmax(360px,1fr));gap:18px}article{background:white;border:1px solid #cbd5e1;border-radius:8px;padding:12px}img{width:100%;max-height:420px;object-fit:contain;border:1px solid #e2e8f0}h2{font-size:16px}</style></head><body><h1>T85 MLIR graph gallery</h1><p>그림을 클릭하면 확대 가능한 SVG를 엽니다.</p><main>""" + "".join(cards) + "</main></body></html>\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())