#!/usr/bin/env python3 """Export existing ONNX/TFLite graphs with official Netron, one failure-isolated slot at a time.""" from __future__ import annotations import argparse import importlib.metadata import json import os import platform import shlex import signal import socket import struct import subprocess import sys import tempfile import time import traceback import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any from netron_capture_common import ( REPO_ROOT, atomic_csv, atomic_json, discover_slots, relative, resolve, sha256, ) CAPTURE_FIELDS = [ "model_id", "task", "task_group", "architecture_family", "variant", "format", "pipeline_stage", "pipeline_stage_status", "validation_stage_id", "validation_stage_status", "validation_failure_code", "production_stage_id", "production_stage_status", "production_failure_code", "production_exit_code", "production_command", "production_stdout_log", "production_stderr_log", "production_run_result", "production_run_result_sha256", "source_run_result", "source_run_result_sha256", "source_artifact", "source_artifact_exists", "source_artifact_bytes", "recorded_source_sha256", "current_source_sha256", "source_checksum_match", "artifact_status", "canonical_s7_selected", "capture_status", "failure_code", "failure_detail", "reused", "capture_method", "netron_version", "playwright_version", "chromium_version", "output_png", "output_png_sha256", "output_png_bytes", "output_png_width", "output_png_height", "ui_proof_png", "ui_proof_png_sha256", "ui_proof_png_bytes", "page_http_status", "page_title", "body_class", "origin_child_count", "graph_node_count", "graph_edge_count", "suggested_download_filename", "metadata_json", "metadata_json_sha256", "netron_server_command", "netron_server_exit_code", "netron_server_stdout_log", "netron_server_stderr_log", "capture_stdout_log", "capture_stderr_log", "started_at", "finished_at", ] def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") def png_info(path: Path) -> tuple[int, int]: data = path.read_bytes()[:24] if len(data) != 24 or data[:8] != b"\x89PNG\r\n\x1a\n" or data[12:16] != b"IHDR": raise ValueError(f"not a PNG with IHDR: {path}") width, height = struct.unpack(">II", data[16:24]) if width <= 0 or height <= 0: raise ValueError(f"invalid PNG dimensions {width}x{height}: {path}") return width, height def free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as handle: handle.bind(("127.0.0.1", 0)) return int(handle.getsockname()[1]) def wait_server(url: str, process: subprocess.Popen[bytes], timeout: float) -> int: deadline = time.monotonic() + timeout error = "server did not respond" while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError(f"Netron server exited before readiness with code {process.returncode}") try: with urllib.request.urlopen(url, timeout=2) as response: return int(response.status) except (urllib.error.URLError, TimeoutError, ConnectionError) as exception: error = str(exception) time.sleep(0.1) raise TimeoutError(f"Netron server readiness timeout: {error}") def stop_server(process: subprocess.Popen[bytes]) -> tuple[int, str]: if process.poll() is not None: return int(process.returncode), "ALREADY_EXITED" process.send_signal(signal.SIGINT) try: return int(process.wait(timeout=15)), "SIGINT" except subprocess.TimeoutExpired: process.terminate() try: return int(process.wait(timeout=5)), "SIGTERM_AFTER_SIGINT_TIMEOUT" except subprocess.TimeoutExpired: process.kill() return int(process.wait(timeout=5)), "SIGKILL_AFTER_TERM_TIMEOUT" def file_result(path: Path, root: Path) -> dict[str, Any]: width, height = png_info(path) return { "path": relative(path, root), "bytes": path.stat().st_size, "sha256": sha256(path), "width": width, "height": height, } def reusable_metadata( metadata_path: Path, output_png: Path, ui_png: Path, source_sha: str, versions: dict[str, str], ) -> dict[str, Any] | None: if not metadata_path.is_file() and not output_png.exists() and not ui_png.exists(): return None if not metadata_path.is_file() or not output_png.is_file() or not ui_png.is_file(): raise RuntimeError("OUTPUT_SETTINGS_CONFLICT: incomplete prior Netron output set") metadata = json.loads(metadata_path.read_text(encoding="utf-8")) if metadata.get("status") != "PASS": raise RuntimeError("OUTPUT_SETTINGS_CONFLICT: prior metadata is not PASS") if metadata.get("source", {}).get("sha256") != source_sha: raise RuntimeError("OUTPUT_SETTINGS_CONFLICT: source checksum changed") for key in ("netron", "playwright", "chromium"): if metadata.get("tool_versions", {}).get(key) != versions[key]: raise RuntimeError(f"OUTPUT_SETTINGS_CONFLICT: {key} version changed") if metadata.get("netron_export", {}).get("sha256") != sha256(output_png): raise RuntimeError("OUTPUT_SETTINGS_CONFLICT: Netron PNG checksum changed") if metadata.get("ui_proof", {}).get("sha256") != sha256(ui_png): raise RuntimeError("OUTPUT_SETTINGS_CONFLICT: UI proof PNG checksum changed") png_info(output_png) png_info(ui_png) return metadata def write_capture_logs(stdout_path: Path, stderr_path: Path, stdout_lines: list[str], stderr_lines: list[str]) -> None: stdout_path.write_text("\n".join(stdout_lines) + "\n", encoding="utf-8") stderr_path.write_text("\n".join(stderr_lines) + ("\n" if stderr_lines else ""), encoding="utf-8") def capture_one( root: Path, slot: dict[str, Any], record_dir: Path, netron_bin: Path, chromium: Any, versions: dict[str, str], load_timeout_sec: int, export_timeout_sec: int, ) -> dict[str, Any]: key = f"{slot['model_id']}_{slot['variant']}_{slot['format']}" started_at = utc_now() source = resolve(root, slot["source_artifact"]) if slot["source_artifact"] else None output_png = resolve(root, slot["output_png"]) metadata_path = resolve(root, slot["metadata_json"]) ui_png = output_png.with_name(f"{slot['format']}_netron_ui.png") per_slot_dir = record_dir / key per_slot_dir.mkdir(parents=True, exist_ok=True) server_stdout = per_slot_dir / "netron_server.stdout.log" server_stderr = per_slot_dir / "netron_server.stderr.log" capture_stdout = per_slot_dir / "capture.stdout.log" capture_stderr = per_slot_dir / "capture.stderr.log" stdout_lines = [f"slot={key}", f"started_at={started_at}"] stderr_lines: list[str] = [] record: dict[str, Any] = { **slot, "capture_status": "FAIL", "failure_code": "FAIL_ANALYSIS", "failure_detail": "capture did not start", "reused": False, "capture_method": "NETRON_BROWSER_EXPORT_AS_PNG_CTRL_SHIFT_E", "netron_version": versions["netron"], "playwright_version": versions["playwright"], "chromium_version": versions["chromium"], "output_png_sha256": "", "output_png_bytes": 0, "output_png_width": 0, "output_png_height": 0, "ui_proof_png": relative(ui_png, root), "ui_proof_png_sha256": "", "ui_proof_png_bytes": 0, "page_http_status": "", "page_title": "", "body_class": "", "origin_child_count": 0, "graph_node_count": 0, "graph_edge_count": 0, "suggested_download_filename": "", "metadata_json_sha256": "", "netron_server_command": "", "netron_server_exit_code": "", "netron_server_stdout_log": relative(server_stdout, root), "netron_server_stderr_log": relative(server_stderr, root), "capture_stdout_log": relative(capture_stdout, root), "capture_stderr_log": relative(capture_stderr, root), "started_at": started_at, "finished_at": "", } if slot["artifact_status"] == "NOT_AVAILABLE": record.update( { "capture_status": "NOT_AVAILABLE", "failure_code": slot["production_failure_code"] or slot["validation_failure_code"] or "FAIL_SOURCE", "failure_detail": "validation-stage model artifact does not exist; no conversion retry permitted", "finished_at": utc_now(), } ) metadata = { "schema_version": "1.0", "stage": "T80_NETRON_VISUALIZATION", "status": "NOT_AVAILABLE", "failure_code": record["failure_code"], "failure_detail": record["failure_detail"], "model_id": slot["model_id"], "variant": slot["variant"], "format": slot["format"], "source": {"path": slot["source_artifact"], "exists": False}, "tool_versions": versions, "validation_evidence": { "run_result": slot["source_run_result"], "stage_id": slot["validation_stage_id"], "stage_status": slot["validation_stage_status"], "failure_code": slot["validation_failure_code"], "pipeline_stage_status": slot["pipeline_stage_status"], }, "production_evidence": { "run_result": slot["production_run_result"], "run_result_sha256": slot["production_run_result_sha256"], "stage_id": slot["production_stage_id"], "stage_status": slot["production_stage_status"], "failure_code": slot["production_failure_code"], "exit_code": slot["production_exit_code"], "command": slot["production_command"], "stdout_log": slot["production_stdout_log"], "stderr_log": slot["production_stderr_log"], }, "policy": { "conversion_retry_performed": False, "tflite_to_onnx_for_netron": False, "model_weight_architecture_modified": False, "allocator_work_performed": False, }, } atomic_json(metadata_path, metadata) record["metadata_json_sha256"] = sha256(metadata_path) write_capture_logs(capture_stdout, capture_stderr, stdout_lines + ["status=NOT_AVAILABLE"], stderr_lines) server_stdout.write_text("server_not_started=source_artifact_not_available\n", encoding="utf-8") server_stderr.write_text("", encoding="utf-8") return record if slot["artifact_status"] != "AVAILABLE" or not source or not source.is_file(): record.update( { "capture_status": "BLOCKED", "failure_code": "FAIL_CHECKSUM", "failure_detail": "source artifact checksum does not match validation evidence", "finished_at": utc_now(), } ) write_capture_logs(capture_stdout, capture_stderr, stdout_lines, [record["failure_detail"]]) server_stdout.write_text("server_not_started=checksum_guard\n", encoding="utf-8") server_stderr.write_text("", encoding="utf-8") return record output_png.parent.mkdir(parents=True, exist_ok=True) try: reused = reusable_metadata(metadata_path, output_png, ui_png, slot["current_source_sha256"], versions) if reused is not None: netron_image = file_result(output_png, root) ui_image = file_result(ui_png, root) record.update( { "capture_status": "PASS", "failure_code": "", "failure_detail": "", "reused": True, "output_png_sha256": netron_image["sha256"], "output_png_bytes": netron_image["bytes"], "output_png_width": netron_image["width"], "output_png_height": netron_image["height"], "ui_proof_png_sha256": ui_image["sha256"], "ui_proof_png_bytes": ui_image["bytes"], "page_http_status": reused["load_evidence"]["page_http_status"], "page_title": reused["load_evidence"]["page_title"], "body_class": reused["load_evidence"]["body_class"], "origin_child_count": reused["load_evidence"]["origin_child_count"], "graph_node_count": reused["load_evidence"]["graph_node_count"], "graph_edge_count": reused["load_evidence"]["graph_edge_count"], "suggested_download_filename": reused["netron_export"]["suggested_download_filename"], "metadata_json_sha256": sha256(metadata_path), "netron_server_command": reused["execution"]["netron_server_command"], "netron_server_exit_code": reused["execution"]["netron_server_exit_code"], "netron_server_stdout_log": reused["execution"]["server_stdout_log"], "netron_server_stderr_log": reused["execution"]["server_stderr_log"], "finished_at": utc_now(), } ) write_capture_logs(capture_stdout, capture_stderr, stdout_lines + ["status=PASS", "reused=true"], []) server_stdout.write_text("server_not_started=reused_validated_capture\n", encoding="utf-8") server_stderr.write_text("", encoding="utf-8") return record except RuntimeError as exception: record.update( { "capture_status": "BLOCKED", "failure_code": "OUTPUT_SETTINGS_CONFLICT", "failure_detail": str(exception), "finished_at": utc_now(), } ) write_capture_logs(capture_stdout, capture_stderr, stdout_lines, [str(exception)]) server_stdout.write_text("server_not_started=output_settings_conflict\n", encoding="utf-8") server_stderr.write_text("", encoding="utf-8") return record port = free_port() command = [ str(netron_bin), str(source), "--host", "127.0.0.1", "--port", str(port), "--verbosity", "debug", ] url = f"http://127.0.0.1:{port}/" record["netron_server_command"] = shlex.join(command) process: subprocess.Popen[bytes] | None = None temporary_paths: list[Path] = [] stop_method = "NOT_STARTED" try: with server_stdout.open("wb") as stdout_handle, server_stderr.open("wb") as stderr_handle: process = subprocess.Popen( command, cwd=root, stdout=stdout_handle, stderr=stderr_handle, start_new_session=False, ) page_status = wait_server(url, process, min(load_timeout_sec, 60)) browser = chromium.launch(headless=True) try: context = browser.new_context( accept_downloads=True, color_scheme="light", viewport={"width": 1920, "height": 1080}, ) page = context.new_page() console_errors: list[str] = [] page_errors: list[str] = [] page.on( "console", lambda message: console_errors.append(f"{message.type}: {message.text}") if message.type in {"error", "warning"} else None, ) page.on("pageerror", lambda error: page_errors.append(str(error))) response = page.goto( url, wait_until="networkidle", timeout=load_timeout_sec * 1000, ) page_status = int(response.status) if response else page_status page.locator("#origin > *").first.wait_for( state="attached", timeout=load_timeout_sec * 1000 ) body_class = page.locator("body").get_attribute("class") or "" origin_children = page.locator("#origin > *").count() graph_nodes = page.locator("#origin .graph-node").count() graph_edges = page.locator("#origin .edge-path").count() page_title = page.title() if "default" not in body_class.split(): message = page.locator(".message-text").text_content() or "" raise RuntimeError(f"Netron did not enter default graph view: {body_class!r} {message!r}") if origin_children <= 0 or graph_nodes <= 0: raise RuntimeError( f"Netron graph DOM is empty: origin_children={origin_children}, nodes={graph_nodes}" ) if page_errors: raise RuntimeError(f"Netron page errors: {page_errors}") ui_handle = tempfile.NamedTemporaryFile( prefix=f".{slot['format']}_netron_ui.", suffix=".png", dir=output_png.parent, delete=False ) ui_temp = Path(ui_handle.name) ui_handle.close() temporary_paths.append(ui_temp) page.screenshot(path=str(ui_temp), full_page=False) png_info(ui_temp) export_handle = tempfile.NamedTemporaryFile( prefix=f".{slot['format']}_netron.", suffix=".png", dir=output_png.parent, delete=False ) export_temp = Path(export_handle.name) export_handle.close() temporary_paths.append(export_temp) with page.expect_download(timeout=export_timeout_sec * 1000) as event: page.keyboard.press("Control+Shift+E") download = event.value download_failure = download.failure() if download_failure: raise RuntimeError(f"Netron PNG download failed: {download_failure}") download.save_as(export_temp) png_info(export_temp) suggested_filename = download.suggested_filename context.close() finally: browser.close() os.replace(export_temp, output_png) temporary_paths.remove(export_temp) os.replace(ui_temp, ui_png) temporary_paths.remove(ui_temp) server_exit_code, stop_method = stop_server(process) process = None netron_image = file_result(output_png, root) ui_image = file_result(ui_png, root) finished_at = utc_now() metadata = { "schema_version": "1.0", "stage": "T80_NETRON_VISUALIZATION", "status": "PASS", "failure_code": None, "model_id": slot["model_id"], "task": slot["task"], "variant": slot["variant"], "format": slot["format"], "source": { "path": slot["source_artifact"], "bytes": slot["source_artifact_bytes"], "sha256": slot["current_source_sha256"], "recorded_sha256": slot["recorded_source_sha256"], "checksum_match": slot["source_checksum_match"], "run_result": slot["source_run_result"], "run_result_sha256": slot["source_run_result_sha256"], "validation_stage_id": slot["validation_stage_id"], "validation_stage_status": slot["validation_stage_status"], "pipeline_stage_status": slot["pipeline_stage_status"], "canonical_s7_selected": slot["canonical_s7_selected"], }, "tool_versions": versions, "execution": { "netron_server_command_argv": command, "netron_server_command": shlex.join(command), "netron_server_exit_code": server_exit_code, "netron_server_stop_method": stop_method, "playwright_action": "page.keyboard.press('Control+Shift+E')", "working_directory": str(root), "server_stdout_log": relative(server_stdout, root), "server_stderr_log": relative(server_stderr, root), "capture_stdout_log": relative(capture_stdout, root), "capture_stderr_log": relative(capture_stderr, root), "started_at": started_at, "finished_at": finished_at, }, "load_evidence": { "url": url, "page_http_status": page_status, "page_title": page_title, "body_class": body_class, "origin_child_count": origin_children, "graph_node_count": graph_nodes, "graph_edge_count": graph_edges, "page_errors": page_errors, "console_errors_or_warnings": console_errors, }, "netron_export": { **netron_image, "method": "Netron browser File > Export as PNG accelerator", "accelerator": "Control+Shift+E", "suggested_download_filename": suggested_filename, }, "ui_proof": {**ui_image, "method": "Playwright 1920x1080 viewport screenshot after Netron graph readiness"}, "policy": { "conversion_performed": False, "converter_retry_performed": False, "tflite_to_onnx_for_netron": False, "model_weight_architecture_modified": False, "allocator_work_performed": False, "execution_order_inferred_from_netron": False, "prohibited_operations_performed": [], }, } atomic_json(metadata_path, metadata) stdout_lines.extend( [ "status=PASS", f"page_http_status={page_status}", f"body_class={body_class}", f"origin_child_count={origin_children}", f"graph_node_count={graph_nodes}", f"graph_edge_count={graph_edges}", f"output_png={netron_image['path']}", f"output_png_sha256={netron_image['sha256']}", f"output_png_dimensions={netron_image['width']}x{netron_image['height']}", f"ui_proof_png={ui_image['path']}", f"netron_server_exit_code={server_exit_code}", f"netron_server_stop_method={stop_method}", ] ) stderr_lines.extend(console_errors) record.update( { "capture_status": "PASS", "failure_code": "", "failure_detail": "", "output_png_sha256": netron_image["sha256"], "output_png_bytes": netron_image["bytes"], "output_png_width": netron_image["width"], "output_png_height": netron_image["height"], "ui_proof_png_sha256": ui_image["sha256"], "ui_proof_png_bytes": ui_image["bytes"], "page_http_status": page_status, "page_title": page_title, "body_class": body_class, "origin_child_count": origin_children, "graph_node_count": graph_nodes, "graph_edge_count": graph_edges, "suggested_download_filename": suggested_filename, "metadata_json_sha256": sha256(metadata_path), "netron_server_exit_code": server_exit_code, "finished_at": finished_at, } ) except Exception as exception: # noqa: BLE001 - batch must isolate every model failure. stderr_lines.extend([f"{type(exception).__name__}: {exception}", traceback.format_exc()]) record.update( { "capture_status": "FAIL", "failure_code": "FAIL_ANALYSIS", "failure_detail": f"{type(exception).__name__}: {exception}", "finished_at": utc_now(), } ) finally: if process is not None: server_exit_code, stop_method = stop_server(process) record["netron_server_exit_code"] = server_exit_code stdout_lines.append(f"netron_server_stop_method={stop_method}") for path in temporary_paths: path.unlink(missing_ok=True) write_capture_logs(capture_stdout, capture_stderr, stdout_lines, stderr_lines) return record def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=REPO_ROOT) parser.add_argument("--run-dir", type=Path, required=True) parser.add_argument( "--inventory-output", type=Path, default=Path("reports/graphs/netron/netron_capture_inventory.csv"), ) parser.add_argument("--scope", choices=("full", "canonical"), default="full") parser.add_argument("--model-id", action="append", default=[]) parser.add_argument("--limit", type=int) parser.add_argument("--load-timeout-sec", type=int, default=600) parser.add_argument("--export-timeout-sec", type=int, default=900) args = parser.parse_args() root = args.repo_root.resolve() run_dir = args.run_dir if args.run_dir.is_absolute() else root / args.run_dir run_dir = run_dir.resolve() record_dir = run_dir / "records" run_dir.mkdir(parents=True, exist_ok=True) inventory_output = args.inventory_output if args.inventory_output.is_absolute() else root / args.inventory_output netron_env = root / "environment/visualization/netron" netron_bin = netron_env / ".venv/bin/netron" browser_root = netron_env / "browsers" if not netron_bin.is_file(): raise SystemExit(f"missing isolated Netron binary: {netron_bin}") os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", str(browser_root)) from playwright.sync_api import sync_playwright netron_version_result = subprocess.run( [str(netron_bin), "--version"], capture_output=True, text=True, check=False ) if netron_version_result.returncode != 0: raise SystemExit(f"netron --version failed: {netron_version_result.stderr}") # Netron's argparse version action writes to stderr in v9.2.0. netron_version = (netron_version_result.stdout + netron_version_result.stderr).strip() if not netron_version: raise SystemExit("netron --version returned no version text") slots = discover_slots(root) if args.scope == "canonical": slots = [slot for slot in slots if slot["canonical_s7_selected"]] if args.model_id: requested = set(args.model_id) slots = [slot for slot in slots if slot["model_id"] in requested] # Capture the canonical pair first so an interrupted full run still leaves # the senior-report minimum in a complete, useful state. slots.sort( key=lambda slot: ( not slot["canonical_s7_selected"], slot["model_id"], slot["variant"], slot["format"], ) ) if args.limit is not None: slots = slots[: args.limit] results: list[dict[str, Any]] = [] toolchain: dict[str, Any] = {} with sync_playwright() as playwright: smoke_browser = playwright.chromium.launch(headless=True) chromium_version = smoke_browser.version smoke_browser.close() executable = Path(playwright.chromium.executable_path) versions = { "python": platform.python_version(), "netron": netron_version, "playwright": importlib.metadata.version("playwright"), "chromium": chromium_version, } toolchain = { **versions, "netron_binary": relative(netron_bin, root), "netron_binary_sha256": sha256(netron_bin), "chromium_executable": relative(executable, root), "chromium_executable_sha256": sha256(executable), "requirements_in": relative(netron_env / "requirements.in", root), "requirements_in_sha256": sha256(netron_env / "requirements.in"), "requirements_lock": relative(netron_env / "requirements.lock", root), "requirements_lock_sha256": sha256(netron_env / "requirements.lock"), "converter": "NOT_RUN", } atomic_json(run_dir / "toolchain.json", toolchain) for index, slot in enumerate(slots, start=1): key = f"{slot['model_id']}:{slot['variant']}:{slot['format']}" print(json.dumps({"event": "capture_start", "index": index, "total": len(slots), "slot": key}), flush=True) result = capture_one( root, slot, record_dir, netron_bin, playwright.chromium, versions, args.load_timeout_sec, args.export_timeout_sec, ) results.append(result) atomic_json(record_dir / f"{slot['model_id']}_{slot['variant']}_{slot['format']}.json", result) atomic_csv(inventory_output, results, CAPTURE_FIELDS) atomic_json( run_dir / "checkpoint.json", { "schema_version": "1.0", "stage": "T80_NETRON_VISUALIZATION", "scope": args.scope, "completed_records": len(results), "expected_records": len(slots), "status_counts": { status: sum(row["capture_status"] == status for row in results) for status in ("PASS", "NOT_AVAILABLE", "BLOCKED", "FAIL") }, "last_slot": key, "updated_at": utc_now(), }, ) print( json.dumps( { "event": "capture_finish", "slot": key, "status": result["capture_status"], "failure_code": result["failure_code"] or None, "output_png": result["output_png"] if result["capture_status"] == "PASS" else None, } ), flush=True, ) status_counts = { status: sum(row["capture_status"] == status for row in results) for status in ("PASS", "NOT_AVAILABLE", "BLOCKED", "FAIL") } expected_counts = { "PASS": sum(slot["artifact_status"] == "AVAILABLE" for slot in slots), "NOT_AVAILABLE": sum(slot["artifact_status"] == "NOT_AVAILABLE" for slot in slots), "BLOCKED": 0, "FAIL": 0, } status = "PASS" if status_counts == expected_counts else "PARTIAL" manifest = { "schema_version": "1.0", "stage": "T80_NETRON_VISUALIZATION", "status": status, "failure_code": None if status == "PASS" else "FAIL_ANALYSIS", "command_argv": [sys.executable, *sys.argv], "command": shlex.join([sys.executable, *sys.argv]), "working_directory": str(root), "scope": args.scope, "tool_versions": toolchain, "records": len(results), "status_counts": status_counts, "expected_status_counts": expected_counts, "inventory_output": relative(inventory_output, root), "inventory_output_sha256": sha256(inventory_output), "policy": { "conversion_performed": False, "converter_retry_performed": False, "tflite_to_onnx_for_netron": False, "model_weight_architecture_modified": False, "allocator_work_performed": False, "execution_order_inferred_from_netron": False, "prohibited_operations_performed": [], }, "finished_at": utc_now(), } atomic_json(run_dir / "execution_manifest.json", manifest) print(json.dumps({"status": status, **status_counts, "records": len(results)}, sort_keys=True)) return 0 if status == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())