scripts/run_mlir_batch.py

#!/usr/bin/env python3
"""Run a failure-isolated, checksum-guarded ONNX-MLIR T60 batch.

This runner never creates or changes model weights.  It consumes only existing
ONNX artifacts, emits textual IR, records every command and failure, and can be
rescanned after upstream T40 jobs finish.  The same import path covers every
active model, including AD01; optional native publication remains a separate
workflow.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import re
import shlex
import signal
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import jsonschema
import onnx


REPO_ROOT = Path(__file__).resolve().parents[1]
REGISTRY = REPO_ROOT / "model_registry.csv"
TOOLCHAIN_DIR = REPO_ROOT / "environment" / "toolchains" / "onnx_mlir"
TOOLCHAIN_LOCK = TOOLCHAIN_DIR / "toolchain.lock.json"
ONNX_MLIR = TOOLCHAIN_DIR / "bin" / "onnx-mlir"
ONNX_MLIR_OPT = TOOLCHAIN_DIR / "bin" / "onnx-mlir-opt"
MLIR_OPT = TOOLCHAIN_DIR / "bin" / "mlir-opt"
TIME = Path("/usr/bin/time")
CONFIG_DIR = REPO_ROOT / "configs" / "mlir" / "batch"
BATCH_LOG_DIR = REPO_ROOT / "logs" / "mlir" / "eligible_all"
CONFIG_SCHEMA = REPO_ROOT / "schemas" / "mlir_batch_config.schema.json"
RESULT_SCHEMA = REPO_ROOT / "schemas" / "mlir_batch_result.schema.json"

PIPELINE = [
    "onnx_to_onnx_dialect",
    "onnx_dialect_parse",
    "onnx_to_krnl",
    "krnl_parse",
    "krnl_to_affine_scf_arith_memref",
    "affine_scf_arith_memref_parse",
    "affine_scf_arith_memref_to_llvm",
    "llvm_reconcile_unrealized_casts",
    "llvm_dialect_parse",
    "optional_onnx_to_tosa",
    "optional_tosa_parse",
    "optional_onnx_to_stablehlo",
    "optional_stablehlo_strict_parse",
]

FAILURE_CODES = {
    "onnx_to_onnx_dialect": "FAIL_MLIR_IMPORT",
    "onnx_dialect_parse": "FAIL_MLIR_IMPORT",
    "onnx_to_krnl": "FAIL_MLIR_LOWERING",
    "krnl_parse": "FAIL_MLIR_LOWERING",
    "krnl_to_affine_scf_arith_memref": "FAIL_MLIR_LOWERING",
    "affine_scf_arith_memref_parse": "FAIL_MLIR_LOWERING",
    "affine_scf_arith_memref_to_llvm": "FAIL_MLIR_LOWERING",
    "llvm_reconcile_unrealized_casts": "FAIL_MLIR_LOWERING",
    "llvm_dialect_parse": "FAIL_MLIR_LOWERING",
    "optional_onnx_to_tosa": "FAIL_MLIR_IMPORT",
    "optional_tosa_parse": "FAIL_MLIR_IMPORT",
    "optional_onnx_to_stablehlo": "FAIL_MLIR_IMPORT",
    "optional_stablehlo_strict_parse": "FAIL_MLIR_IMPORT",
}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def json_hash(value: Any) -> str:
    encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(encoded).hexdigest()


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.with_name(path.name + f".tmp.{os.getpid()}")
    temp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
    os.replace(temp, path)


def file_record(path: Path) -> dict[str, Any]:
    exists = path.is_file()
    return {
        "path": str(path),
        "exists": exists,
        "sha256": sha256(path) if exists else None,
        "bytes": path.stat().st_size if exists else None,
    }


def load_schema(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text())


def validate(schema_path: Path, document: Any) -> None:
    jsonschema.Draft202012Validator(load_schema(schema_path)).validate(document)


def eligible_rows() -> list[dict[str, str]]:
    with REGISTRY.open(newline="") as handle:
        rows = list(csv.DictReader(handle))
    return [
        row
        for row in rows
        if row.get("eligibility") == "ELIGIBLE"
    ]


def model_dir(model_id: str) -> Path:
    matches = sorted((REPO_ROOT / "models").glob(f"*/{model_id}"))
    if len(matches) != 1:
        raise RuntimeError(f"expected one model directory for {model_id}, found {matches}")
    return matches[0]


def config_path(path: Path) -> str:
    """Store repository-owned paths portably inside generated configs."""
    return str(path.resolve().relative_to(REPO_ROOT.resolve()))


def resolve_config_path(value: str | Path) -> Path:
    """Resolve a portable config path while retaining legacy absolute support."""
    path = Path(value)
    return path if path.is_absolute() else (REPO_ROOT / path).resolve()


def registry_checksum(row: dict[str, str], variant: str) -> str | None:
    key = "paired_fp32_checksum" if variant == "fp32" else "public_quantized_checksum"
    value = row.get(key, "").strip()
    return value if re.fullmatch(r"[0-9a-f]{64}", value) else None


def resolve_input(row: dict[str, str], root: Path, variant: str) -> dict[str, Any]:
    filename = "model_fp32.onnx" if variant == "fp32" else "model_quantized.onnx"
    subdir = "fp32" if variant == "fp32" else "quantized"
    canonical = root / "onnx" / subdir / filename
    expected = registry_checksum(row, variant)
    if canonical.is_file():
        record = file_record(canonical)
        return {
            "input_path": config_path(canonical),
            "input_exists": True,
            "input_sha256": record["sha256"],
            "input_bytes": record["bytes"],
            "input_source": "CANONICAL_T40_ONNX",
            "registry_expected_sha256": expected,
        }

    artifact_key = "paired_fp32_artifact" if variant == "fp32" else "public_quantized_artifact"
    artifact = row.get(artifact_key, "").strip()
    fallback = REPO_ROOT / artifact if artifact else Path("/__missing_registry_artifact__")
    if fallback.suffix.lower() == ".onnx" and fallback.is_file() and expected:
        actual = sha256(fallback)
        if actual == expected:
            return {
                "input_path": config_path(fallback),
                "input_exists": True,
                "input_sha256": actual,
                "input_bytes": fallback.stat().st_size,
                "input_source": "CHECKSUM_VERIFIED_REGISTRY_ONNX",
                "registry_expected_sha256": expected,
            }

    return {
        "input_path": config_path(canonical),
        "input_exists": False,
        "input_sha256": None,
        "input_bytes": None,
        "input_source": "MISSING_T40_ONNX_PREREQUISITE",
        "registry_expected_sha256": expected,
    }


def make_config(row: dict[str, str], timeout_sec: int) -> tuple[Path, dict[str, Any]]:
    root = model_dir(row["model_id"])
    config = {
        "schema_version": "1.0",
        "model_id": row["model_id"],
        "task": row["task"],
        "priority": row.get("priority", "UNSPECIFIED"),
        "model_dir": config_path(root),
        "toolchain_lock": config_path(TOOLCHAIN_LOCK),
        "toolchain_lock_sha256": sha256(TOOLCHAIN_LOCK),
        "timeout_sec": timeout_sec,
        "pipeline": PIPELINE,
        "variants": {
            "fp32": resolve_input(row, root, "fp32"),
            "public_quantized": resolve_input(row, root, "public_quantized"),
        },
    }
    validate(CONFIG_SCHEMA, config)
    path = CONFIG_DIR / f"{row['model_id']}_mlir.json"
    write_json(path, config)
    return path, config


def capture_tool_versions() -> dict[str, Any]:
    lock = json.loads(TOOLCHAIN_LOCK.read_text())
    commands: dict[str, Any] = {}
    BATCH_LOG_DIR.mkdir(parents=True, exist_ok=True)
    for name, command in {
        "onnx_mlir": [str(ONNX_MLIR), "--version"],
        "onnx_mlir_opt": [str(ONNX_MLIR_OPT), "--version"],
        "mlir_opt": [str(MLIR_OPT), "--version"],
    }.items():
        proc = subprocess.run(command, cwd=REPO_ROOT, capture_output=True, text=True, check=False)
        (BATCH_LOG_DIR / f"tool_{name}.command.txt").write_text(shlex.join(command) + "\n")
        (BATCH_LOG_DIR / f"tool_{name}.stdout.log").write_text(proc.stdout)
        (BATCH_LOG_DIR / f"tool_{name}.stderr.log").write_text(proc.stderr)
        (BATCH_LOG_DIR / f"tool_{name}.exit_code").write_text(f"{proc.returncode}\n")
        commands[name] = {
            "command": command,
            "exit_code": proc.returncode,
            "stdout": proc.stdout.strip(),
        }
    return {
        "lock_sha256": sha256(TOOLCHAIN_LOCK),
        "versions": lock.get("versions", {}),
        "lock_status": lock.get("status", {}),
        "probes": commands,
    }


def inspect_onnx(path: Path) -> dict[str, Any]:
    if not path.is_file():
        return {"status": "MISSING"}
    model = onnx.load(str(path), load_external_data=False)
    ops: dict[str, int] = {}
    for node in model.graph.node:
        ops[node.op_type] = ops.get(node.op_type, 0) + 1
    qdq = {name: ops.get(name, 0) for name in ("QuantizeLinear", "DequantizeLinear")}
    qoperator_names = (
        "QLinearConv", "QLinearMatMul", "ConvInteger", "MatMulInteger",
        "DynamicQuantizeLinear", "QLinearAdd", "QLinearMul", "QLinearSigmoid",
        "QLinearLeakyRelu", "QLinearAveragePool", "QLinearGlobalAveragePool",
    )
    qoperators = {name: ops[name] for name in qoperator_names if ops.get(name)}
    integer_types = {2, 3, 4, 5, 6, 7, 12, 13}
    int_initializers = sum(1 for value in model.graph.initializer if value.data_type in integer_types)
    int_io = sum(
        1
        for value in list(model.graph.input) + list(model.graph.output)
        if value.type.HasField("tensor_type")
        and value.type.tensor_type.elem_type in integer_types
    )
    if sum(qdq.values()) and qoperators:
        representation = "MIXED_QDQ_QOPERATOR"
    elif sum(qdq.values()):
        representation = "QDQ"
    elif qoperators:
        representation = "QOPERATOR"
    elif int_initializers or int_io:
        representation = "INTEGER_TENSOR"
    else:
        representation = "FLOAT_ONLY"
    return {
        "status": "PASS",
        "onnx_ir_version": model.ir_version,
        "opset_imports": {item.domain or "ai.onnx": item.version for item in model.opset_import},
        "node_count": len(model.graph.node),
        "operator_counts": dict(sorted(ops.items())),
        "qdq_counts": qdq,
        "qoperator_counts": qoperators,
        "integer_initializer_count": int_initializers,
        "integer_graph_io_count": int_io,
        "representation": representation,
    }


def count_text_markers(path: Path) -> dict[str, int]:
    markers = {
        "onnx_quantize_linear_ops": 0,
        "onnx_dequantize_linear_ops": 0,
        "onnx_qoperator_ops": 0,
        "quant_dialect_ops": 0,
        "stablehlo_ops": 0,
        "tosa_ops": 0,
        "krnl_ops": 0,
        "llvm_float_compute_ops": 0,
        "arith_float_compute_ops": 0,
        "i8_mentions": 0,
        "ui8_mentions": 0,
        "f32_mentions": 0,
        "unrealized_conversion_casts": 0,
    }
    if not path.is_file():
        return markers
    patterns = {
        "onnx_quantize_linear_ops": re.compile(r'"onnx\.QuantizeLinear"'),
        "onnx_dequantize_linear_ops": re.compile(r'"onnx\.DequantizeLinear"'),
        "onnx_qoperator_ops": re.compile(r'"onnx\.(?:QLinear\w+|ConvInteger|MatMulInteger|DynamicQuantizeLinear)"'),
        "quant_dialect_ops": re.compile(r'(?<![A-Za-z0-9_])quant\.[A-Za-z_]'),
        "stablehlo_ops": re.compile(r'(?<![A-Za-z0-9_])stablehlo\.[A-Za-z_]'),
        "tosa_ops": re.compile(r'(?<![A-Za-z0-9_])tosa\.[A-Za-z_]'),
        "krnl_ops": re.compile(r'(?<![A-Za-z0-9_])krnl\.[A-Za-z_]'),
        "llvm_float_compute_ops": re.compile(r'(?<![A-Za-z0-9_])llvm\.f(?:add|sub|mul|div|rem|neg|cmp)'),
        "arith_float_compute_ops": re.compile(r'(?<![A-Za-z0-9_])arith\.(?:addf|subf|mulf|divf|remf|negf|cmpf)'),
        # MLIR shaped types spell an element separator as ``...x<i8>``.  The
        # leading ``x`` is therefore allowed in addition to a lexical boundary.
        "i8_mentions": re.compile(r"(?:(?<=x)|(?<![A-Za-z0-9_]))i8(?![A-Za-z0-9_])"),
        "ui8_mentions": re.compile(r"(?:(?<=x)|(?<![A-Za-z0-9_]))ui8(?![A-Za-z0-9_])"),
        "f32_mentions": re.compile(r"(?:(?<=x)|(?<![A-Za-z0-9_]))f32(?![A-Za-z0-9_])"),
        "unrealized_conversion_casts": re.compile(r"builtin\.unrealized_conversion_cast"),
    }
    with path.open(encoding="utf-8", errors="replace") as handle:
        for line in handle:
            for key, pattern in patterns.items():
                markers[key] += len(pattern.findall(line))
    return markers


def next_attempt_dir(log_root: Path) -> Path:
    log_root.mkdir(parents=True, exist_ok=True)
    numbers = []
    for child in log_root.glob("attempt_*" ):
        try:
            numbers.append(int(child.name.split("_", 1)[1]))
        except ValueError:
            pass
    path = log_root / f"attempt_{max(numbers, default=0) + 1:03d}"
    path.mkdir()
    return path


def reusable_stage(latest: Path, fingerprint: str, output: Path | None, reuse_failures: bool) -> dict[str, Any] | None:
    if not latest.is_file():
        return None
    try:
        prior = json.loads(latest.read_text())
    except (OSError, json.JSONDecodeError):
        return None
    if prior.get("fingerprint") != fingerprint:
        return None
    if prior.get("status") not in ({"PASS", "PASS_WITH_PATCH", "PARTIAL"} | ({"FAIL", "BLOCKED"} if reuse_failures else set())):
        return None
    if output is not None and prior.get("status") not in {"FAIL", "BLOCKED"}:
        current = file_record(output)
        expected = (prior.get("outputs") or [{}])[0]
        if not current["exists"] or current["sha256"] != expected.get("sha256"):
            return None
    reused = dict(prior)
    reused["reused_from_previous_run"] = True
    return reused


def classify_failure(stage: str, stderr: str, timed_out: bool) -> tuple[str, str]:
    code = FAILURE_CODES[stage]
    lower = stderr.lower()
    if "unsupported" in lower or "unimplemented" in lower or "failed to legalize operation" in lower:
        secondary = code
        code = "FAIL_UNSUPPORTED_OP"
        return code, secondary
    if timed_out:
        return code, "TIMEOUT"
    return code, ""


def run_command_stage(
    *,
    model_root: Path,
    variant: str,
    stage: str,
    command: list[str],
    inputs: list[Path],
    output: Path | None,
    command_output: Path | None,
    timeout_sec: int,
    settings: dict[str, Any],
    reuse_failures: bool,
    optional: bool = False,
) -> dict[str, Any]:
    input_records = [file_record(path) for path in inputs]
    fingerprint = json_hash({
        "command": command,
        "inputs": input_records,
        "toolchain_lock_sha256": settings["toolchain_lock_sha256"],
        "timeout_sec": timeout_sec,
        "optional": optional,
    })
    log_root = model_root / "mlir" / "logs" / variant / stage
    latest = log_root / "latest_result.json"
    if output is not None and output.is_file():
        prior: dict[str, Any] | None = None
        if latest.is_file():
            try:
                prior = json.loads(latest.read_text())
            except (OSError, json.JSONDecodeError):
                prior = None
        if prior is None or prior.get("fingerprint") != fingerprint:
            # Never replace an artifact produced by an unverified or different
            # setting.  Preserve it and emit a reproducible conflict record;
            # callers may select a separate run directory for the new setting.
            attempt_dir = next_attempt_dir(log_root)
            stdout_log = attempt_dir / "stdout.log"
            stderr_log = attempt_dir / "stderr.log"
            resource_log = attempt_dir / "resource.log"
            command_log = attempt_dir / "command.txt"
            execution = [str(TIME), "-v", "-o", str(resource_log), *command]
            command_log.write_text(shlex.join(execution) + "\n")
            stdout_log.write_text("")
            stderr_log.write_text(
                "Refused to overwrite an existing artifact produced by different "
                f"or unverified settings: {output}\n"
            )
            resource_log.write_text("Command intentionally not executed; output settings conflict.\n")
            now = utc_now()
            result = {
                "stage": stage,
                "optional": optional,
                "status": "BLOCKED",
                "failure_code": "FAIL_ENVIRONMENT",
                "secondary_failure_code": "OUTPUT_SETTINGS_CONFLICT",
                "command": shlex.join(execution),
                "command_argv": command,
                "execution_argv": execution,
                "working_directory": str(REPO_ROOT),
                "inputs": input_records,
                "outputs": [file_record(output)],
                "started_at": now,
                "ended_at": now,
                "duration_sec": 0.0,
                "exit_code": 125,
                "timed_out": False,
                "stdout_log": str(stdout_log),
                "stderr_log": str(stderr_log),
                "resource_log": str(resource_log),
                "command_log": str(command_log),
                "patch": None,
                "validation": {"overwrite_refused": True, "existing_output": file_record(output)},
                "fingerprint": fingerprint,
                "reused_from_previous_run": False,
            }
            write_json(attempt_dir / "stage_result.json", result)
            # Do not replace latest_result.json: it remains the provenance for
            # the protected artifact and enables later same-setting resume.
            return result
    reused = reusable_stage(latest, fingerprint, output, reuse_failures)
    if reused is not None:
        return reused

    attempt_dir = next_attempt_dir(log_root)
    stdout_log = attempt_dir / "stdout.log"
    stderr_log = attempt_dir / "stderr.log"
    resource_log = attempt_dir / "resource.log"
    command_log = attempt_dir / "command.txt"
    execution = [str(TIME), "-v", "-o", str(resource_log), *command]
    command_log.write_text(shlex.join(execution) + "\n")
    started_at = utc_now()
    started = time.monotonic()
    timed_out = False
    with stdout_log.open("wb") as stdout_handle, stderr_log.open("wb") as stderr_handle:
        proc = subprocess.Popen(
            execution,
            cwd=REPO_ROOT,
            stdout=stdout_handle,
            stderr=stderr_handle,
            start_new_session=True,
        )
        try:
            exit_code = proc.wait(timeout=timeout_sec)
        except subprocess.TimeoutExpired:
            timed_out = True
            os.killpg(proc.pid, signal.SIGTERM)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                os.killpg(proc.pid, signal.SIGKILL)
                proc.wait()
            exit_code = 124
    ended_at = utc_now()
    stderr = stderr_log.read_text(errors="replace")

    if exit_code == 0 and command_output is not None and output is not None:
        if command_output.is_file() and command_output.stat().st_size > 0:
            output.parent.mkdir(parents=True, exist_ok=True)
            os.replace(command_output, output)
        else:
            exit_code = 65
            with stderr_log.open("a") as handle:
                handle.write(f"\nExpected non-empty command output missing: {command_output}\n")
            stderr = stderr_log.read_text(errors="replace")

    output_records = [file_record(output)] if output is not None else []
    passed = exit_code == 0 and (output is None or output_records[0]["exists"])
    failure_code = None
    secondary = None
    if not passed:
        failure_code, secondary = classify_failure(stage, stderr, timed_out)
    status = "PASS" if passed else "FAIL"
    result = {
        "stage": stage,
        "optional": optional,
        "status": status,
        "failure_code": failure_code,
        "secondary_failure_code": secondary or None,
        "command": shlex.join(execution),
        "command_argv": command,
        "execution_argv": execution,
        "working_directory": str(REPO_ROOT),
        "inputs": input_records,
        "outputs": output_records,
        "started_at": started_at,
        "ended_at": ended_at,
        "duration_sec": time.monotonic() - started,
        "exit_code": exit_code,
        "timed_out": timed_out,
        "stdout_log": str(stdout_log),
        "stderr_log": str(stderr_log),
        "resource_log": str(resource_log),
        "command_log": str(command_log),
        "patch": None,
        "validation": {
            "output_nonempty": bool(output_records and output_records[0]["exists"]),
            "error_excerpt": stderr[-2000:] if stderr else "",
        },
        "fingerprint": fingerprint,
        "reused_from_previous_run": False,
    }
    write_json(attempt_dir / "stage_result.json", result)
    write_json(latest, result)
    return result


def dependency_failure(
    *, model_root: Path, variant: str, stage: str, dependency: str,
    config_sha: str, missing_path: Path, reuse_failures: bool,
) -> dict[str, Any]:
    log_root = model_root / "mlir" / "logs" / variant / stage
    fingerprint = json_hash({"config_sha": config_sha, "stage": stage, "missing": str(missing_path)})
    reused = reusable_stage(log_root / "latest_result.json", fingerprint, None, reuse_failures)
    if reused is not None:
        return reused
    attempt_dir = next_attempt_dir(log_root)
    stdout_log = attempt_dir / "stdout.log"
    stderr_log = attempt_dir / "stderr.log"
    resource_log = attempt_dir / "resource.log"
    command_log = attempt_dir / "command.txt"
    command = ["/usr/bin/test", "-s", str(missing_path)]
    execution = [str(TIME), "-v", "-o", str(resource_log), *command]
    command_log.write_text(shlex.join(execution) + "\n")
    started = utc_now()
    proc = subprocess.run(execution, cwd=REPO_ROOT, capture_output=True, text=True, check=False)
    stdout_log.write_text(proc.stdout)
    stderr_log.write_text(
        proc.stderr + f"MLIR prerequisite unavailable: {dependency}: {missing_path}\n"
    )
    result = {
        "stage": stage,
        "optional": False,
        "status": "BLOCKED",
        "failure_code": "FAIL_MLIR_IMPORT",
        "secondary_failure_code": "MISSING_ONNX_PREREQUISITE",
        "command": shlex.join(execution),
        "command_argv": command,
        "execution_argv": execution,
        "working_directory": str(REPO_ROOT),
        "inputs": [file_record(missing_path)],
        "outputs": [],
        "started_at": started,
        "ended_at": utc_now(),
        "duration_sec": 0.0,
        "exit_code": proc.returncode,
        "timed_out": False,
        "stdout_log": str(stdout_log),
        "stderr_log": str(stderr_log),
        "resource_log": str(resource_log),
        "command_log": str(command_log),
        "patch": None,
        "validation": {"dependency": dependency, "input_exists": False},
        "fingerprint": fingerprint,
        "reused_from_previous_run": False,
    }
    write_json(attempt_dir / "stage_result.json", result)
    write_json(log_root / "latest_result.json", result)
    return result


def run_variant(
    config: dict[str, Any], variant: str, *, reuse_failures: bool,
) -> dict[str, Any]:
    root = resolve_config_path(config["model_dir"])
    vcfg = config["variants"][variant]
    input_path = resolve_config_path(vcfg["input_path"])
    input_record = file_record(input_path)
    config_sha = json_hash(config)
    if not input_path.is_file():
        stage = dependency_failure(
            model_root=root,
            variant=variant,
            stage="onnx_to_onnx_dialect",
            dependency="T40 ONNX output or checksum-verified native ONNX",
            config_sha=config_sha,
            missing_path=input_path,
            reuse_failures=reuse_failures,
        )
        return {
            "status": "BLOCKED",
            "failure_code": "FAIL_MLIR_IMPORT",
            "secondary_failure_code": "MISSING_ONNX_PREREQUISITE",
            "required_path_status": "BLOCKED",
            "input": input_record,
            "input_source": vcfg["input_source"],
            "input_integrity_unchanged": True,
            "stages": [stage],
            "quantization": {"source": {"status": "MISSING"}, "low_level_status": "BLOCKED"},
        }

    initial_sha = sha256(input_path)
    if initial_sha != vcfg["input_sha256"]:
        raise RuntimeError(f"input changed since scan for {config['model_id']} {variant}")

    out_dir = root / "mlir" / ("fp32" if variant == "fp32" else "quantized")
    work_dir = root / "mlir" / "work" / variant
    out_dir.mkdir(parents=True, exist_ok=True)
    work_dir.mkdir(parents=True, exist_ok=True)
    timeout = config["timeout_sec"]
    stages: list[dict[str, Any]] = []

    onnx_mlir = out_dir / "onnx.mlir"
    onnx_temp_base = work_dir / "onnx_import"
    onnx_command_output = Path(str(onnx_temp_base) + ".onnx.mlir")
    stage = run_command_stage(
        model_root=root, variant=variant, stage="onnx_to_onnx_dialect",
        command=[str(ONNX_MLIR), "--EmitONNXIR", "-o", str(onnx_temp_base), str(input_path)],
        inputs=[input_path], output=onnx_mlir, command_output=onnx_command_output,
        timeout_sec=timeout, settings=config, reuse_failures=reuse_failures,
    )
    stages.append(stage)
    if stage["status"] != "PASS":
        return finish_variant(config, variant, input_path, initial_sha, stages, inspect_onnx(input_path))

    stages.append(run_command_stage(
        model_root=root, variant=variant, stage="onnx_dialect_parse",
        command=[str(ONNX_MLIR_OPT), str(onnx_mlir), "-o", "/dev/null"],
        inputs=[onnx_mlir], output=None, command_output=None, timeout_sec=timeout,
        settings=config, reuse_failures=reuse_failures,
    ))

    krnl = out_dir / "krnl.mlir"
    krnl_tmp = work_dir / "krnl.mlir"
    stages.append(run_command_stage(
        model_root=root, variant=variant, stage="onnx_to_krnl",
        command=[str(ONNX_MLIR_OPT), "--convert-onnx-to-krnl", str(onnx_mlir), "-o", str(krnl_tmp)],
        inputs=[onnx_mlir], output=krnl, command_output=krnl_tmp, timeout_sec=timeout,
        settings=config, reuse_failures=reuse_failures,
    ))
    if stages[-1]["status"] == "PASS":
        stages.append(run_command_stage(
            model_root=root, variant=variant, stage="krnl_parse",
            command=[str(ONNX_MLIR_OPT), str(krnl), "-o", "/dev/null"],
            inputs=[krnl], output=None, command_output=None, timeout_sec=timeout,
            settings=config, reuse_failures=reuse_failures,
        ))

    affine = out_dir / "affine_scf_memref.mlir"
    affine_tmp = work_dir / "affine_scf_memref.mlir"
    if krnl.is_file() and stages[-1]["status"] == "PASS":
        stages.append(run_command_stage(
            model_root=root, variant=variant, stage="krnl_to_affine_scf_arith_memref",
            command=[str(ONNX_MLIR_OPT), "--convert-krnl-to-affine", str(krnl), "-o", str(affine_tmp)],
            inputs=[krnl], output=affine, command_output=affine_tmp, timeout_sec=timeout,
            settings=config, reuse_failures=reuse_failures,
        ))
        if stages[-1]["status"] == "PASS":
            stages.append(run_command_stage(
                model_root=root, variant=variant, stage="affine_scf_arith_memref_parse",
                command=[str(ONNX_MLIR_OPT), str(affine), "-o", "/dev/null"],
                inputs=[affine], output=None, command_output=None, timeout_sec=timeout,
                settings=config, reuse_failures=reuse_failures,
            ))

    llvm_raw = work_dir / "llvm_raw.mlir"
    llvm_final = out_dir / "llvm.mlir"
    if affine.is_file() and any(s["stage"] == "krnl_to_affine_scf_arith_memref" and s["status"] == "PASS" for s in stages):
        stages.append(run_command_stage(
            model_root=root, variant=variant, stage="affine_scf_arith_memref_to_llvm",
            command=[str(ONNX_MLIR_OPT), "--convert-krnl-to-llvm", str(affine), "-o", str(llvm_raw)],
            inputs=[affine], output=llvm_raw, command_output=llvm_raw, timeout_sec=timeout,
            settings=config, reuse_failures=reuse_failures,
        ))
        if stages[-1]["status"] == "PASS":
            llvm_tmp = work_dir / "llvm_reconciled.mlir"
            stages.append(run_command_stage(
                model_root=root, variant=variant, stage="llvm_reconcile_unrealized_casts",
                command=[str(ONNX_MLIR_OPT), "--reconcile-unrealized-casts", str(llvm_raw), "-o", str(llvm_tmp)],
                inputs=[llvm_raw], output=llvm_final, command_output=llvm_tmp, timeout_sec=timeout,
                settings=config, reuse_failures=reuse_failures,
            ))
            if stages[-1]["status"] == "PASS":
                stages.append(run_command_stage(
                    model_root=root, variant=variant, stage="llvm_dialect_parse",
                    command=[str(MLIR_OPT), str(llvm_final), "-o", "/dev/null"],
                    inputs=[llvm_final], output=None, command_output=None, timeout_sec=timeout,
                    settings=config, reuse_failures=reuse_failures,
                ))

    # Optional paths are always attempted independently from the required path.
    for dialect, flag, converter_stage, parser_stage in (
        ("tosa", "--convert-onnx-to-tosa", "optional_onnx_to_tosa", "optional_tosa_parse"),
        ("stablehlo", "--convert-onnx-to-stablehlo", "optional_onnx_to_stablehlo", "optional_stablehlo_strict_parse"),
    ):
        output = out_dir / f"{dialect}.mlir"
        temp = work_dir / f"{dialect}.mlir"
        convert = run_command_stage(
            model_root=root, variant=variant, stage=converter_stage,
            command=[str(ONNX_MLIR_OPT), flag, str(onnx_mlir), "-o", str(temp)],
            inputs=[onnx_mlir], output=output, command_output=temp, timeout_sec=timeout,
            settings=config, reuse_failures=reuse_failures, optional=True,
        )
        stages.append(convert)
        if convert["status"] == "PASS":
            stages.append(run_command_stage(
                model_root=root, variant=variant, stage=parser_stage,
                command=[str(ONNX_MLIR_OPT), str(output), "-o", "/dev/null"],
                inputs=[output], output=None, command_output=None, timeout_sec=timeout,
                settings=config, reuse_failures=reuse_failures, optional=True,
            ))

    return finish_variant(config, variant, input_path, initial_sha, stages, inspect_onnx(input_path))


def quantization_assessment(
    source: dict[str, Any], stage_paths: dict[str, Path], required_path_status: str,
) -> dict[str, Any]:
    result: dict[str, Any] = {"source": source, "mlir_stage_markers": {}}
    if source.get("status") != "PASS":
        result.update({"low_level_status": "BLOCKED", "failure_code": "FAIL_MLIR_IMPORT"})
        return result
    representation = source.get("representation")
    if representation == "FLOAT_ONLY":
        result.update({
            "onnx_dialect_preservation": "NOT_APPLICABLE",
            "low_level_status": "NOT_APPLICABLE",
            "failure_code": None,
        })
        return result
    inspected = {name: count_text_markers(path) for name, path in stage_paths.items() if path.is_file()}
    result["mlir_stage_markers"] = inspected
    src_q = sum(source.get("qdq_counts", {}).values())
    src_qop = sum(source.get("qoperator_counts", {}).values())
    onnx_markers = inspected.get("onnx", {})
    explicit_match = (
        onnx_markers.get("onnx_quantize_linear_ops", 0) == source.get("qdq_counts", {}).get("QuantizeLinear", 0)
        and onnx_markers.get("onnx_dequantize_linear_ops", 0) == source.get("qdq_counts", {}).get("DequantizeLinear", 0)
        and onnx_markers.get("onnx_qoperator_ops", 0) == src_qop
    )
    explicit_after_import = sum(
        onnx_markers.get(name, 0)
        for name in (
            "onnx_quantize_linear_ops", "onnx_dequantize_linear_ops",
            "onnx_qoperator_ops",
        )
    )
    integer_types_after_import = bool(
        onnx_markers.get("i8_mentions", 0) or onnx_markers.get("ui8_mentions", 0)
    )
    if explicit_match:
        onnx_preservation = "PASS"
        onnx_observation = "Explicit Q/DQ and QOperator counts match the source ONNX graph."
    elif explicit_after_import and integer_types_after_import:
        onnx_preservation = "PARTIAL"
        onnx_observation = (
            "Explicit quantized operations and integer types remain, but ONNX import "
            "canonicalized/fused the representation so operation counts differ; this is "
            "not promoted to exact preservation PASS."
        )
    else:
        onnx_preservation = "FAIL" if "onnx" in inspected else "UNKNOWN"
        onnx_observation = "Explicit source quantization is not visible after ONNX dialect import."
    result.update({
        "onnx_dialect_preservation": onnx_preservation,
        "onnx_dialect_exact_count_match": explicit_match,
        "onnx_dialect_observation": onnx_observation,
        "onnx_dialect_explicit_quant_ops": explicit_after_import,
    })
    if required_path_status != "PASS":
        result.update({
            "low_level_status": "BLOCKED",
            "failure_code": "FAIL_MLIR_IMPORT" if "onnx" not in inspected else "FAIL_MLIR_LOWERING",
            "source_explicit_quant_ops": src_q + src_qop,
        })
        return result
    low = inspected.get("llvm") or inspected.get("affine_scf_memref") or inspected.get("krnl") or {}
    explicit_low = (
        low.get("quant_dialect_ops", 0)
        + low.get("onnx_quantize_linear_ops", 0)
        + low.get("onnx_dequantize_linear_ops", 0)
        + low.get("onnx_qoperator_ops", 0)
    )
    float_compute = low.get("llvm_float_compute_ops", 0) + low.get("arith_float_compute_ops", 0)
    preserved_low = explicit_low > 0 and float_compute == 0
    result.update({
        "source_explicit_quant_ops": src_q + src_qop,
        "low_level_explicit_quant_ops": explicit_low,
        "low_level_float_compute_markers": float_compute,
        "integer_storage_or_io_present_at_low_level": bool(low.get("i8_mentions", 0) or low.get("ui8_mentions", 0)),
        "low_level_status": "PASS" if preserved_low else "PARTIAL",
        "failure_code": None if preserved_low else "FAIL_QUANTIZATION_PRESERVATION",
        "observation": (
            "Explicit quantized operations/types remain through the lowest validated dialect without floating compute markers."
            if preserved_low
            else "Source quantization is explicit in ONNX dialect, but the lowest validated dialect does not prove an integer-only compute path; float compute lowering is recorded separately."
        ),
    })
    return result


def finish_variant(
    config: dict[str, Any], variant: str, input_path: Path, initial_sha: str,
    stages: list[dict[str, Any]], source: dict[str, Any],
) -> dict[str, Any]:
    root = resolve_config_path(config["model_dir"])
    subdir = "fp32" if variant == "fp32" else "quantized"
    out_dir = root / "mlir" / subdir
    required_names = {
        "onnx_to_onnx_dialect", "onnx_dialect_parse", "onnx_to_krnl", "krnl_parse",
        "krnl_to_affine_scf_arith_memref", "affine_scf_arith_memref_parse",
        "affine_scf_arith_memref_to_llvm", "llvm_reconcile_unrealized_casts", "llvm_dialect_parse",
    }
    required = [stage for stage in stages if stage["stage"] in required_names]
    required_ok = len(required) == len(required_names) and all(stage["status"] == "PASS" for stage in required)
    required_status = "PASS" if required_ok else "PARTIAL" if any(stage["status"] == "PASS" for stage in required) else "FAIL"
    quantization = quantization_assessment(
        source,
        {
            "onnx": out_dir / "onnx.mlir",
            "krnl": out_dir / "krnl.mlir",
            "affine_scf_memref": out_dir / "affine_scf_memref.mlir",
            "llvm": out_dir / "llvm.mlir",
            "tosa": out_dir / "tosa.mlir",
            "stablehlo": out_dir / "stablehlo.mlir",
        },
        required_status,
    )
    stage_by_name = {stage["stage"]: stage for stage in stages}
    optional_routes: dict[str, Any] = {}
    for dialect, converter, parser in (
        ("tosa", "optional_onnx_to_tosa", "optional_tosa_parse"),
        ("stablehlo", "optional_onnx_to_stablehlo", "optional_stablehlo_strict_parse"),
    ):
        convert_result = stage_by_name.get(converter)
        parse_result = stage_by_name.get(parser)
        if not convert_result or convert_result["status"] != "PASS":
            route_status = "FAIL"
            route_failure = convert_result.get("failure_code") if convert_result else "FAIL_MLIR_IMPORT"
        elif not parse_result or parse_result["status"] != "PASS":
            # A non-empty optional IR was emitted, but without a strict parser
            # pass it is not promoted to a validated dialect route.
            route_status = "PARTIAL"
            route_failure = parse_result.get("failure_code") if parse_result else "FAIL_MLIR_IMPORT"
        else:
            route_status = "PASS"
            route_failure = None
        markers = quantization.get("mlir_stage_markers", {}).get(dialect, {})
        residual_quant = sum(
            markers.get(name, 0)
            for name in (
                "onnx_quantize_linear_ops", "onnx_dequantize_linear_ops",
                "onnx_qoperator_ops",
            )
        )
        if variant == "public_quantized" and route_status == "PASS" and residual_quant:
            route_status = "PARTIAL"
            route_failure = "FAIL_MLIR_QUANT_LEGALIZATION"
        optional_routes[dialect] = {
            "status": route_status,
            "failure_code": route_failure,
            "conversion_status": convert_result.get("status") if convert_result else "NOT_RUN",
            "strict_parse_status": parse_result.get("status") if parse_result else "NOT_RUN",
            "residual_onnx_quant_ops": residual_quant,
            "artifact": file_record(out_dir / f"{dialect}.mlir"),
        }
    if not required_ok:
        failed = next((stage for stage in required if stage["status"] != "PASS"), None)
        failure = failed["failure_code"] if failed else "FAIL_MLIR_LOWERING"
        status = "PARTIAL" if required_status == "PARTIAL" else "FAIL"
    elif variant == "public_quantized" and quantization.get("low_level_status") != "PASS":
        status = "PARTIAL"
        failure = "FAIL_QUANTIZATION_PRESERVATION"
    else:
        status = "PASS"
        failure = None
    return {
        "status": status,
        "failure_code": failure,
        "required_path_status": required_status,
        "input": file_record(input_path),
        "input_source": config["variants"][variant]["input_source"],
        "input_integrity_unchanged": sha256(input_path) == initial_sha,
        "stages": stages,
        "quantization": quantization,
        "optional_routes": optional_routes,
        "artifacts": {
            name: file_record(out_dir / filename)
            for name, filename in {
                "onnx": "onnx.mlir", "krnl": "krnl.mlir",
                "affine_scf_memref": "affine_scf_memref.mlir", "llvm": "llvm.mlir",
                "tosa": "tosa.mlir", "stablehlo": "stablehlo.mlir",
            }.items()
        },
    }


def run_model(config_path: Path, config: dict[str, Any], tool_versions: dict[str, Any], reuse_failures: bool) -> dict[str, Any]:
    started = utc_now()
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ")
    variants: dict[str, Any] = {}
    for variant in ("fp32", "public_quantized"):
        try:
            variants[variant] = run_variant(config, variant, reuse_failures=reuse_failures)
        except Exception as error:  # one variant/model must not abort the batch
            root = resolve_config_path(config["model_dir"])
            error_log = root / "mlir" / "logs" / variant / "runner_exception.log"
            error_log.parent.mkdir(parents=True, exist_ok=True)
            error_log.write_text(f"{type(error).__name__}: {error}\n")
            variants[variant] = {
                "status": "FAIL", "failure_code": "FAIL_ENVIRONMENT",
                "required_path_status": "FAIL",
                "input": file_record(resolve_config_path(config["variants"][variant]["input_path"])),
                "stages": [],
                "quantization": {"low_level_status": "BLOCKED", "error": str(error)},
                "runner_exception_log": str(error_log),
            }
    statuses = {item["status"] for item in variants.values()}
    if statuses == {"PASS"}:
        overall = "PASS"
    elif "PASS" in statuses or "PARTIAL" in statuses:
        overall = "PARTIAL"
    elif statuses == {"BLOCKED"}:
        overall = "BLOCKED"
    else:
        overall = "FAIL"
    flattened_stages: list[dict[str, Any]] = []
    for variant_name, variant_result in variants.items():
        for stage in variant_result.get("stages", []):
            record = dict(stage)
            record.update({
                "model_id": config["model_id"],
                "artifact_id": f"{config['model_id']}-onnx-mlir-{variant_name}",
                "variant": variant_name,
                "stage_id": f"mlir_{stage['stage']}",
                "tool_versions": tool_versions,
                "random_seed": 20260806,
                "error_summary": stage.get("validation", {}).get("error_excerpt", ""),
                "config_sha256": sha256(config_path),
                "stage_config_sha256": stage["fingerprint"],
                "source_artifact": config["variants"][variant_name]["input_path"],
                "source_checksum": config["variants"][variant_name]["input_sha256"],
                "artifact": next(
                    (item["path"] for item in stage.get("outputs", []) if item.get("exists")),
                    None,
                ),
            })
            flattened_stages.append(record)
    result = {
        "schema_version": "1.0",
        "model_id": config["model_id"],
        "config_path": str(config_path),
        "config_sha256": sha256(config_path),
        "run_id": run_id,
        "started_at": started,
        "ended_at": utc_now(),
        "overall_status": overall,
        "tool_versions": tool_versions,
        "pipeline": PIPELINE,
        "variants": variants,
        # Flattened records make the independent result directly consumable by
        # the project-wide compatibility reporter without changing the richer
        # per-variant structure used by this runner.
        "stages": flattened_stages,
        "forbidden_operations_performed": [],
        "patch": None,
    }
    validate(RESULT_SCHEMA, result)
    result_path = resolve_config_path(config["model_dir"]) / "mlir" / "mlir_batch_run_result.json"
    write_json(result_path, result)
    publish_model_summaries(config_path, config, result, result_path)
    return result


def publish_model_summaries(
    config_path: Path,
    config: dict[str, Any],
    result: dict[str, Any],
    result_path: Path,
) -> None:
    """Publish task-specified matrix, quantization and conversion-log files."""
    mlir_root = resolve_config_path(config["model_dir"]) / "mlir"
    local_config = resolve_config_path(config["model_dir"]) / "config" / "mlir_batch_config.json"
    write_json(local_config, config)

    matrix_variants: dict[str, Any] = {}
    for variant_name, variant in result["variants"].items():
        matrix_variants[variant_name] = {
            "status": variant["status"],
            "failure_code": variant.get("failure_code"),
            "required_path_status": variant["required_path_status"],
            "input": variant["input"],
            "input_source": variant.get("input_source"),
            "input_integrity_unchanged": variant.get("input_integrity_unchanged"),
            "required_artifacts": variant.get("artifacts", {}),
            "optional_routes": variant.get("optional_routes", {}),
            "stages": [
                {
                    key: stage.get(key)
                    for key in (
                        "stage", "status", "failure_code", "secondary_failure_code",
                        "exit_code", "timed_out", "command", "inputs", "outputs",
                        "stdout_log", "stderr_log", "resource_log", "patch",
                        "validation", "reused_from_previous_run",
                    )
                }
                for stage in variant.get("stages", [])
            ],
        }
    matrix = {
        "schema_version": "1.0",
        "model_id": config["model_id"],
        "generated_at": utc_now(),
        "config": file_record(config_path),
        "model_local_config": file_record(local_config),
        "result": file_record(result_path),
        "toolchain_lock": file_record(TOOLCHAIN_LOCK),
        "required_pipeline": PIPELINE[:9],
        "optional_pipeline": PIPELINE[9:],
        "variants": matrix_variants,
        "forbidden_operations_performed": [],
    }
    write_json(mlir_root / "mlir_stage_matrix.json", matrix)

    quant_variant = result["variants"]["public_quantized"]
    quant_data = quant_variant.get("quantization", {})
    quant_stages: dict[str, Any] = {}
    stage_name_for_artifact = {
        "onnx_ir": "onnx",
        "krnl": "krnl",
        "affine_scf_memref": "affine_scf_memref",
        "llvm": "llvm",
    }
    for report_name, artifact_name in stage_name_for_artifact.items():
        artifact = quant_variant.get("artifacts", {}).get(artifact_name, {})
        if not artifact.get("exists"):
            status = "FAIL"
            code = quant_variant.get("failure_code") or "FAIL_MLIR_LOWERING"
        elif report_name == "onnx_ir":
            onnx_preservation = quant_data.get("onnx_dialect_preservation")
            status = "PASS" if onnx_preservation == "PASS" else "PARTIAL"
            code = "FAIL_QUANTIZATION_PRESERVATION" if onnx_preservation == "FAIL" else None
        elif quant_data.get("low_level_status") == "PASS":
            status, code = "PASS", None
        else:
            status, code = "PARTIAL", quant_data.get("failure_code") or "FAIL_QUANTIZATION_PRESERVATION"
        quant_stages[report_name] = {
            "status": status,
            "failure_code": code,
            "artifact": artifact,
            "markers": quant_data.get("mlir_stage_markers", {}).get(artifact_name, {}),
        }
    for dialect in ("tosa", "stablehlo"):
        route = quant_variant.get("optional_routes", {}).get(dialect, {})
        quant_stages[dialect] = {
            "status": route.get("status", "FAIL"),
            "failure_code": route.get("failure_code"),
            "artifact": route.get("artifact", {}),
            "markers": quant_data.get("mlir_stage_markers", {}).get(dialect, {}),
        }
    quant_report = {
        "schema_version": "1.0",
        "model_id": config["model_id"],
        "generated_at": utc_now(),
        "source": quant_data.get("source", {}),
        "onnx_dialect_preservation": quant_data.get("onnx_dialect_preservation", "UNKNOWN"),
        "low_level_status": quant_data.get("low_level_status", "BLOCKED"),
        "failure_code": quant_data.get("failure_code"),
        "observation": quant_data.get("observation", "MLIR import prerequisite or lowering failed."),
        "stages": quant_stages,
        "forbidden_operations_performed": [],
    }
    write_json(mlir_root / "mlir_quantization_preservation.json", quant_report)

    manifest_records: list[dict[str, Any]] = []
    required_names = (
        ("onnx", "onnx"),
        ("krnl", "krnl"),
        ("affine_scf_memref", "affine_scf_memref"),
        ("llvm_reconciled", "llvm"),
    )
    for variant_name, variant in result["variants"].items():
        for ir_stage, artifact_name in required_names:
            artifact = variant.get("artifacts", {}).get(artifact_name, {
                "path": str(mlir_root / ("fp32" if variant_name == "fp32" else "quantized") / f"{artifact_name}.mlir"),
                "exists": False, "sha256": None, "bytes": None,
            })
            if not artifact.get("exists"):
                status = "BLOCKED" if variant["required_path_status"] == "BLOCKED" else "FAIL"
                failure = variant.get("failure_code") or "FAIL_MLIR_LOWERING"
            elif (
                variant_name == "public_quantized"
                and ir_stage == "llvm_reconciled"
                and variant["required_path_status"] == "PASS"
                and quant_data.get("low_level_status") != "PASS"
            ):
                # Structural/compiler success remains PASS for earlier IR.  The
                # lowest required artifact carries the independent quantization
                # loss so project aggregation produces PARTIAL, not false PASS
                # or false total FAIL.
                status = "PARTIAL"
                failure = quant_data.get("failure_code") or "FAIL_QUANTIZATION_PRESERVATION"
            else:
                status, failure = "PASS", None
            manifest_records.append({
                **artifact,
                "variant": variant_name,
                "ir_stage": ir_stage,
                "status": status,
                "failure_code": failure,
            })
    manifest = {
        "schema_version": "1.0",
        "model_id": config["model_id"],
        "generated_at": utc_now(),
        "config_sha256": sha256(config_path),
        "toolchain_lock_sha256": sha256(TOOLCHAIN_LOCK),
        "artifacts": manifest_records,
        "forbidden_operations_performed": [],
    }
    write_json(mlir_root / "artifact_manifest.json", manifest)

    log_lines = [
        f"model_id={config['model_id']}",
        f"config={config_path}",
        f"config_sha256={sha256(config_path)}",
        f"toolchain_lock={TOOLCHAIN_LOCK}",
        f"toolchain_lock_sha256={sha256(TOOLCHAIN_LOCK)}",
        f"result={result_path}",
        "forbidden_operations_performed=NONE",
    ]
    for stage in result["stages"]:
        log_lines.extend([
            "",
            f"variant={stage['variant']}",
            f"stage={stage['stage']}",
            f"status={stage['status']}",
            f"failure_code={stage.get('failure_code') or 'NONE'}",
            f"exit_code={stage.get('exit_code')}",
            f"command={stage['command']}",
            f"stdout={stage['stdout_log']}",
            f"stderr={stage['stderr_log']}",
            f"resource={stage['resource_log']}",
            f"inputs={json.dumps(stage.get('inputs', []), sort_keys=True)}",
            f"outputs={json.dumps(stage.get('outputs', []), sort_keys=True)}",
            f"patch={stage.get('patch') or 'NONE'}",
        ])
    (mlir_root / "mlir_conversion.log").write_text("\n".join(log_lines) + "\n")


def summarize(configs: list[tuple[Path, dict[str, Any]]], results: list[dict[str, Any]], tool_versions: dict[str, Any]) -> dict[str, Any]:
    models: dict[str, Any] = {}
    stage_counts: dict[str, int] = {}
    variant_counts: dict[str, int] = {}
    reused = 0
    executed = 0
    failure_code_counts: dict[str, int] = {}
    variant_failure_code_counts: dict[str, int] = {}
    stage_failure_code_counts: dict[str, int] = {}
    required_route_counts: dict[str, int] = {}
    quant_source_representation_counts: dict[str, int] = {}
    quant_low_level_status_counts: dict[str, int] = {}
    artifact_records = 0
    artifact_existing = 0
    artifact_checksum_matches = 0
    actual_onnx_mlir_import_commands = 0
    for result in sorted(results, key=lambda item: item["model_id"]):
        variants = result["variants"]
        model_failures = sorted({item.get("failure_code") for item in variants.values() if item.get("failure_code")})
        models[result["model_id"]] = {
            "overall_status": result["overall_status"],
            "variant_statuses": {name: item["status"] for name, item in variants.items()},
            "required_path_statuses": {name: item["required_path_status"] for name, item in variants.items()},
            "failure_codes": model_failures,
            "result_path": str(
                resolve_config_path(
                    next(
                        c[1]["model_dir"]
                        for c in configs
                        if c[1]["model_id"] == result["model_id"]
                    )
                )
                / "mlir"
                / "mlir_batch_run_result.json"
            ),
        }
        for variant_name, item in variants.items():
            variant_counts[item["status"]] = variant_counts.get(item["status"], 0) + 1
            route_key = item["required_path_status"]
            required_route_counts[route_key] = required_route_counts.get(route_key, 0) + 1
            if item.get("failure_code"):
                failure_code_counts[item["failure_code"]] = failure_code_counts.get(item["failure_code"], 0) + 1
                variant_failure_code_counts[item["failure_code"]] = variant_failure_code_counts.get(item["failure_code"], 0) + 1
            if variant_name == "public_quantized":
                quant = item.get("quantization", {})
                representation = quant.get("source", {}).get("representation")
                if representation:
                    quant_source_representation_counts[representation] = quant_source_representation_counts.get(representation, 0) + 1
                low_status = quant.get("low_level_status")
                if low_status and low_status != "NOT_APPLICABLE":
                    quant_low_level_status_counts[low_status] = quant_low_level_status_counts.get(low_status, 0) + 1
            for artifact in item.get("artifacts", {}).values():
                artifact_records += 1
                if artifact.get("exists"):
                    artifact_existing += 1
                    path = Path(artifact["path"])
                    if path.is_file() and sha256(path) == artifact.get("sha256"):
                        artifact_checksum_matches += 1
            for stage in item["stages"]:
                stage_counts[stage["status"]] = stage_counts.get(stage["status"], 0) + 1
                if stage.get("failure_code"):
                    failure_code_counts[stage["failure_code"]] = failure_code_counts.get(stage["failure_code"], 0) + 1
                    stage_failure_code_counts[stage["failure_code"]] = stage_failure_code_counts.get(stage["failure_code"], 0) + 1
                if stage["stage"] == "onnx_to_onnx_dialect" and stage.get("command_argv", [""])[0] == str(ONNX_MLIR):
                    actual_onnx_mlir_import_commands += 1
                if stage.get("reused_from_previous_run"):
                    reused += 1
                else:
                    executed += 1
    return {
        "schema_version": "1.0",
        "generated_at": utc_now(),
        "registry_sha256": sha256(REGISTRY),
        "toolchain_lock_sha256": sha256(TOOLCHAIN_LOCK),
        "tool_versions": tool_versions,
        "eligible_model_count": len(configs),
        "models_with_both_onnx_inputs": sum(all(v["input_exists"] for v in c["variants"].values()) for _, c in configs),
        "models": models,
        "variant_status_counts": dict(sorted(variant_counts.items())),
        "required_route_status_counts": dict(sorted(required_route_counts.items())),
        "stage_status_counts": dict(sorted(stage_counts.items())),
        "failure_code_counts_variant_and_stage_records": dict(sorted(failure_code_counts.items())),
        "variant_failure_code_counts": dict(sorted(variant_failure_code_counts.items())),
        "stage_failure_code_counts": dict(sorted(stage_failure_code_counts.items())),
        "quant_source_representation_counts": dict(sorted(quant_source_representation_counts.items())),
        "quant_low_level_status_counts": dict(sorted(quant_low_level_status_counts.items())),
        "actual_onnx_mlir_import_commands": actual_onnx_mlir_import_commands,
        "mlir_prerequisite_attempt_records": len(configs) * 2,
        "artifact_records": artifact_records,
        "artifact_existing": artifact_existing,
        "artifact_checksum_matches": artifact_checksum_matches,
        "stage_executed_count": executed,
        "stage_reused_count": reused,
        "forbidden_operations_performed": [],
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--timeout-sec", type=int, default=300)
    parser.add_argument("--workers", type=int, default=1)
    parser.add_argument("--model-id", action="append", default=[])
    parser.add_argument("--scan-only", action="store_true")
    parser.add_argument("--reuse-failures", action="store_true")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    BATCH_LOG_DIR.mkdir(parents=True, exist_ok=True)
    tool_versions = capture_tool_versions()
    selected = eligible_rows()
    if args.model_id:
        wanted = set(args.model_id)
        selected = [row for row in selected if row["model_id"] in wanted]
    configs = [make_config(row, args.timeout_sec) for row in selected]
    scan = {
        "schema_version": "1.0",
        "generated_at": utc_now(),
        "registry_sha256": sha256(REGISTRY),
        "eligible_model_count": len(configs),
        "models": {
            config["model_id"]: {
                "config_path": str(path),
                "config_sha256": sha256(path),
                "variants": config["variants"],
            }
            for path, config in configs
        },
    }
    write_json(BATCH_LOG_DIR / "scan_manifest.json", scan)
    if args.scan_only:
        print(json.dumps({
            "scan_manifest": str(BATCH_LOG_DIR / "scan_manifest.json"),
            "models": len(configs),
            "with_both_inputs": sum(all(v["input_exists"] for v in c["variants"].values()) for _, c in configs),
        }, sort_keys=True))
        return 0

    results: list[dict[str, Any]] = []
    with ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
        futures = {
            pool.submit(run_model, path, config, tool_versions, args.reuse_failures): config["model_id"]
            for path, config in configs
        }
        for future in as_completed(futures):
            model_id = futures[future]
            try:
                result = future.result()
            except Exception as error:
                print(f"{model_id}: batch-level runner error: {error}", file=sys.stderr)
                continue
            results.append(result)
            print(f"{model_id}: {result['overall_status']}", flush=True)
    summary = summarize(configs, results, tool_versions)
    summary["command_argv"] = [sys.executable, *sys.argv]
    summary["command"] = shlex.join(summary["command_argv"])
    summary["working_directory"] = str(REPO_ROOT)
    write_json(BATCH_LOG_DIR / "batch_summary.json", summary)
    print(json.dumps({
        "summary": str(BATCH_LOG_DIR / "batch_summary.json"),
        "models_completed": len(results),
        "variant_status_counts": summary["variant_status_counts"],
    }, sort_keys=True))
    return 0 if len(results) == len(configs) else 1


if __name__ == "__main__":
    raise SystemExit(main())