scripts/capture_netron_graphs.py

#!/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())