Compare RocJITsu and HotSwap gfx1250 B0-to-A0 translation

rj_gfx1250_b0_to_a0_compare.py

#!/usr/bin/env python3
# Copyright (c) 2026 Advanced Micro Devices, Inc.
# SPDX-License-Identifier: MIT

"""Compare RocJITsu and HotSwap gfx1250 B0-to-A0 translation resources.

Each translator runs in a fresh process. Linux wait4 resource accounting
captures user plus system CPU time and peak resident set size for that process.
The runner enables native translator timers by default and normalizes them into
exclusive common phase buckets using monotonic wall time.
HotSwap verbose decision logging is disabled by default so log I/O does not
distort the comparison.
"""

from __future__ import annotations

import argparse
import bisect
import csv
import datetime
import hashlib
import html
import io
import json
import math
import os
import pathlib
import platform
import random
import re
import shlex
import shutil
import signal
import statistics
import subprocess
import sys
import time
import uuid
from collections import Counter
from typing import Any, Sequence

SCHEMA_VERSION = 1
TOOLS = ("rocjitsu-dbt", "hotswap")
DEFAULT_SOURCE_ISA = "amdgcn-amd-amdhsa--gfx1250"
DEFAULT_TARGET_ISA = "amdgcn-amd-amdhsa--gfx1250"
NORMALIZED_PHASE_BUCKETS = (
    "input_object_parse",
    "setup",
    "decode",
    "analysis",
    "rewrite",
    "layout_fixups",
    "validation",
    "output",
    "residual",
)
HOTSWAP_TIMING_PATTERN = re.compile(
    r"^(?P<name>(?:phase|strat):\S+)\s+"
    r"(?P<calls>\d+)\s+calls\s+"
    r"(?P<total>[0-9.eE+-]+)\s+min\s+"
    r"(?P<minimum>[0-9.eE+-]+)\s+max\s+"
    r"(?P<maximum>[0-9.eE+-]+)\s+patches\s+"
    r"(?P<patches>\d+)\s+(?P<unit>ns|us|ms|s)$"
)


def utc_now() -> str:
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


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


def atomic_text(path: pathlib.Path, value: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
    try:
        with temporary.open("w", encoding="utf-8", newline="") as stream:
            stream.write(value)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def atomic_json(path: pathlib.Path, value: Any) -> None:
    atomic_text(path, json.dumps(value, indent=2, sort_keys=True) + "\n")


def append_jsonl(path: pathlib.Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as stream:
        stream.write(json.dumps(value, sort_keys=True, separators=(",", ":")))
        stream.write("\n")
        stream.flush()
        os.fsync(stream.fileno())


def read_jsonl(path: pathlib.Path) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    if not path.exists():
        return records
    with path.open("r", encoding="utf-8") as stream:
        for line_number, line in enumerate(stream, 1):
            if not line.strip():
                continue
            try:
                value = json.loads(line)
            except json.JSONDecodeError as error:
                raise ValueError(
                    f"{path}:{line_number}: invalid JSON: {error}"
                ) from error
            if not isinstance(value, dict):
                raise ValueError(f"{path}:{line_number}: record is not an object")
            records.append(value)
    return records


def shell_command(arguments: Sequence[str]) -> str:
    return shlex.join(str(argument) for argument in arguments)


def file_identity(path: pathlib.Path) -> dict[str, Any]:
    resolved = path.resolve(strict=True)
    stat = resolved.stat()
    return {
        "path": str(resolved),
        "sha256": sha256_file(resolved),
        "size": stat.st_size,
        "mtime_ns": stat.st_mtime_ns,
    }


def source_revision(path: pathlib.Path) -> dict[str, Any]:
    directory = path.resolve().parent
    try:
        top_level = subprocess.run(
            ["git", "-C", str(directory), "rev-parse", "--show-toplevel"],
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError:
        return {"git_root": None, "git_commit": None}
    if top_level.returncode != 0:
        return {"git_root": None, "git_commit": None}
    root = pathlib.Path(top_level.stdout.strip())
    commit = subprocess.run(
        ["git", "-C", str(root), "rev-parse", "HEAD"],
        capture_output=True,
        text=True,
        check=False,
    )
    return {
        "git_root": str(root),
        "git_commit": commit.stdout.strip() if commit.returncode == 0 else None,
    }


def repository_root() -> pathlib.Path:
    script = pathlib.Path(__file__).resolve()
    for parent in script.parents:
        if (parent / "emulation" / "rocjitsu").is_dir():
            return parent
    return script.parent


def first_executable(candidates: Sequence[pathlib.Path | str]) -> str:
    for candidate in candidates:
        raw = str(candidate)
        if "/" not in raw:
            found = shutil.which(raw)
            if found:
                return found
            continue
        path = pathlib.Path(raw).expanduser()
        if path.is_file() and os.access(path, os.X_OK):
            return str(path.resolve())
    return ""


def default_dbt() -> str:
    root = repository_root()
    workspace = root.parent
    candidates: list[pathlib.Path | str] = []
    if value := os.environ.get("RJ_DBT_TRANSLATE"):
        candidates.append(value)
    candidates.extend(
        [
            workspace / "build" / "tools" / "rj_dbt_translate",
            root / "build" / "tools" / "rj_dbt_translate",
            "rj_dbt_translate",
        ]
    )
    return first_executable(candidates)


def default_hotswap() -> str:
    root = repository_root()
    workspace = root.parent
    candidates: list[pathlib.Path | str] = []
    if value := os.environ.get("HOTSWAP_REWRITE"):
        candidates.append(value)
    candidates.extend(
        [
            workspace
            / "build-amd-comgr"
            / "tools"
            / "amd_comgr"
            / "test-lit"
            / "hotswap-rewrite",
            workspace / "build-comgr" / "test-lit" / "hotswap-rewrite",
            workspace
            / "build-comgr"
            / "tools"
            / "amd_comgr"
            / "test-lit"
            / "hotswap-rewrite",
            "hotswap-rewrite",
        ]
    )
    return first_executable(candidates)


def resolve_executable(raw: str, label: str) -> pathlib.Path:
    if not raw:
        raise ValueError(f"{label} was not found; pass --{label}")
    found = raw
    if "/" not in raw:
        found = shutil.which(raw) or ""
    if not found:
        raise ValueError(f"{label} executable was not found: {raw}")
    path = pathlib.Path(found).expanduser().resolve(strict=True)
    if not path.is_file() or not os.access(path, os.X_OK):
        raise ValueError(f"{label} is not executable: {path}")
    return path


def discover_hotswap_library_dirs(binary: pathlib.Path) -> list[pathlib.Path]:
    for ancestor in list(binary.parents)[:7]:
        for candidate in (ancestor, ancestor / "lib"):
            if candidate.is_dir() and any(candidate.glob("libamd_comgr.so*")):
                return [candidate.resolve()]
    return []


def resolve_library_dirs(
    raw_directories: Sequence[str], hotswap: pathlib.Path
) -> list[pathlib.Path]:
    values = list(raw_directories)
    if not values and (environment := os.environ.get("HOTSWAP_LIBRARY_DIR")):
        values.append(environment)
    if not values:
        return discover_hotswap_library_dirs(hotswap)
    resolved: list[pathlib.Path] = []
    for raw in values:
        path = pathlib.Path(raw).expanduser().resolve(strict=True)
        if not path.is_dir():
            raise ValueError(f"HotSwap library path is not a directory: {path}")
        if path not in resolved:
            resolved.append(path)
    return resolved


def discover_inputs(
    paths: Sequence[str], recursive: bool, include_glob: str
) -> list[pathlib.Path]:
    discovered: set[pathlib.Path] = set()
    for raw_path in paths:
        path = pathlib.Path(raw_path).expanduser().resolve(strict=True)
        if path.is_file():
            if path.suffix != ".hsaco":
                raise ValueError(f"input does not end in .hsaco: {path}")
            discovered.add(path)
            continue
        if not path.is_dir():
            raise ValueError(f"input is neither a regular file nor directory: {path}")
        iterator = path.rglob(include_glob) if recursive else path.glob(include_glob)
        discovered.update(
            candidate.resolve() for candidate in iterator if candidate.is_file()
        )
    return sorted(discovered)


def positive_int(value: str) -> int:
    parsed = int(value)
    if parsed <= 0:
        raise argparse.ArgumentTypeError("must be greater than zero")
    return parsed


def nonnegative_int(value: str) -> int:
    parsed = int(value)
    if parsed < 0:
        raise argparse.ArgumentTypeError("must be nonnegative")
    return parsed


def nonnegative_float(value: str) -> float:
    parsed = float(value)
    if parsed < 0:
        raise argparse.ArgumentTypeError("must be nonnegative")
    return parsed


def positive_float(value: str) -> float:
    parsed = float(value)
    if parsed <= 0:
        raise argparse.ArgumentTypeError("must be greater than zero")
    return parsed


def input_token(path: pathlib.Path) -> str:
    safe_stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", path.stem)[:80]
    path_hash = hashlib.sha256(str(path).encode()).hexdigest()[:10]
    return f"{safe_stem}-{path_hash}"


def terminate_process_group(
    process: subprocess.Popen[bytes], grace: float
) -> tuple[int, Any] | None:
    if process.returncode is not None:
        return None
    try:
        os.killpg(process.pid, signal.SIGTERM)
    except ProcessLookupError:
        pass
    deadline = time.monotonic() + grace
    while time.monotonic() < deadline:
        try:
            waited, status, usage = os.wait4(process.pid, os.WNOHANG)
        except InterruptedError:
            continue
        except ChildProcessError:
            return None
        if waited:
            process.returncode = os.waitstatus_to_exitcode(status)
            return status, usage
        time.sleep(0.01)
    try:
        os.killpg(process.pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    return None


def timed_process(
    command: Sequence[str],
    *,
    stdout: Any,
    stderr: Any,
    environment: dict[str, str],
    timeout_seconds: float,
) -> dict[str, Any]:
    started_ns = time.monotonic_ns()
    process = subprocess.Popen(
        list(command),
        stdout=stdout,
        stderr=stderr,
        env=environment,
        start_new_session=True,
    )
    deadline = time.monotonic() + timeout_seconds if timeout_seconds > 0 else None
    timed_out = False
    status = 0
    usage = None
    try:
        while True:
            try:
                waited, status, usage = os.wait4(process.pid, os.WNOHANG)
            except InterruptedError:
                continue
            if waited:
                break
            if deadline is not None and time.monotonic() >= deadline:
                timed_out = True
                reaped = terminate_process_group(process, 1.0)
                if reaped is not None:
                    status, usage = reaped
                else:
                    try:
                        _, status, usage = os.wait4(process.pid, 0)
                    except ChildProcessError:
                        usage = None
                break
            time.sleep(0.005)
    except KeyboardInterrupt:
        reaped = terminate_process_group(process, 1.0)
        if reaped is not None:
            status, _ = reaped
        elif process.returncode is None:
            try:
                _, status, _ = os.wait4(process.pid, 0)
                process.returncode = os.waitstatus_to_exitcode(status)
            except ChildProcessError:
                pass
        raise

    if process.returncode is None:
        process.returncode = os.waitstatus_to_exitcode(status)
    elapsed_seconds = (time.monotonic_ns() - started_ns) / 1_000_000_000
    result: dict[str, Any] = {
        "return_code": process.returncode,
        "timed_out": timed_out,
        "elapsed_seconds": elapsed_seconds,
        "user_cpu_seconds": None,
        "system_cpu_seconds": None,
        "cpu_seconds": None,
        "max_rss_kib": None,
    }
    if usage is not None:
        user_seconds = float(usage.ru_utime)
        system_seconds = float(usage.ru_stime)
        result.update(
            {
                "user_cpu_seconds": user_seconds,
                "system_cpu_seconds": system_seconds,
                "cpu_seconds": user_seconds + system_seconds,
                "max_rss_kib": int(usage.ru_maxrss),
            }
        )
    return result


def dbt_command(binary: pathlib.Path, source: pathlib.Path) -> list[str]:
    return [
        str(binary),
        str(source),
        "--input-target",
        "gfx1250",
        "--output-target",
        "gfx1250",
        "--input-revision",
        "b0",
        "--output-revision",
        "a0",
        "--output-mode",
        "code-object",
    ]


def hotswap_command(
    binary: pathlib.Path,
    source: pathlib.Path,
    output: pathlib.Path,
    source_isa: str,
    target_isa: str,
) -> list[str]:
    return [
        str(binary),
        str(source),
        source_isa,
        target_isa,
        "--output",
        str(output),
        "--entry-trampolines",
        "--strict-mode",
    ]


def measurement_key(record: dict[str, Any]) -> tuple[str, str, str, int]:
    return (
        record["input"]["path"],
        record["tool"],
        record["phase"],
        int(record["repetition"]),
    )


def latest_measurements(
    records: Sequence[dict[str, Any]],
) -> dict[tuple[str, str, str, int], dict[str, Any]]:
    latest: dict[tuple[str, str, str, int], dict[str, Any]] = {}
    for record in records:
        if record.get("type") == "measurement":
            latest[measurement_key(record)] = record
    return latest


def output_identity(path: pathlib.Path) -> dict[str, Any] | None:
    if not path.is_file():
        return None
    stat = path.stat()
    with path.open("rb") as stream:
        is_elf = stream.read(4) == b"\x7fELF"
    return {
        "size": stat.st_size,
        "sha256": sha256_file(path),
        "is_elf": is_elf,
    }


def status_for_result(
    process_result: dict[str, Any], output: dict[str, Any] | None
) -> str:
    if process_result["timed_out"]:
        return "timeout"
    return_code = process_result["return_code"]
    if return_code is not None and return_code < 0:
        return "signal"
    if return_code != 0:
        return "failed"
    if output is None:
        return "missing_output"
    if not output["is_elf"]:
        return "invalid_output"
    return "success"


def parse_hotswap_timing_rows(path: pathlib.Path) -> list[dict[str, Any]]:
    if not path.is_file():
        return []
    unit_seconds = {"ns": 1e-9, "us": 1e-6, "ms": 1e-3, "s": 1.0}
    rows: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8", errors="replace") as stream:
        for line in stream:
            match = HOTSWAP_TIMING_PATTERN.match(line.strip())
            if match is None:
                continue
            scale = unit_seconds[match.group("unit")]
            rows.append(
                {
                    "name": match.group("name"),
                    "calls": int(match.group("calls")),
                    "wall_seconds": float(match.group("total")) * scale,
                    "minimum_wall_seconds": float(match.group("minimum")) * scale,
                    "maximum_wall_seconds": float(match.group("maximum")) * scale,
                    "patches": int(match.group("patches")),
                }
            )
    return rows


def load_rocjitsu_timing(
    path: pathlib.Path,
    *,
    process_cpu_seconds: float | None,
    process_elapsed_seconds: float | None,
) -> dict[str, Any]:
    if not path.is_file():
        return unavailable_phase_timing(
            "rocjitsu-dbt", path, "RocJITsu emitted no phase profile"
        )
    try:
        profile = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        return unavailable_phase_timing(
            "rocjitsu-dbt", path, f"could not read RocJITsu phase profile: {error}"
        )
    if not isinstance(profile, dict):
        return unavailable_phase_timing(
            "rocjitsu-dbt", path, "RocJITsu phase profile is not a JSON object"
        )

    warnings: list[str] = []

    def seconds(timing: Any, clock: str) -> float:
        if not isinstance(timing, dict):
            return 0.0
        value = timing.get(f"{clock}_ns", 0)
        if not isinstance(value, (int, float)) or value < 0:
            warnings.append(f"invalid {clock}_ns timing value: {value!r}")
            return 0.0
        return float(value) * 1e-9

    buckets = profile.get("buckets")
    if not isinstance(buckets, list):
        return unavailable_phase_timing(
            "rocjitsu-dbt", path, "RocJITsu phase profile has no buckets array"
        )

    normalized_wall = {bucket: 0.0 for bucket in NORMALIZED_PHASE_BUCKETS}
    normalized_cpu = {bucket: 0.0 for bucket in NORMALIZED_PHASE_BUCKETS}
    seen_buckets: set[str] = set()
    for row in buckets:
        if not isinstance(row, dict) or not isinstance(row.get("name"), str):
            warnings.append("ignored malformed RocJITsu bucket row")
            continue
        name = row["name"]
        if name not in normalized_wall:
            warnings.append(f"ignored unknown RocJITsu bucket {name!r}")
            continue
        if name in seen_buckets:
            warnings.append(f"duplicate RocJITsu bucket {name!r}")
        seen_buckets.add(name)
        normalized_wall[name] += seconds(row.get("timing"), "wall")
        normalized_cpu[name] += seconds(row.get("timing"), "cpu")
    missing = sorted(set(NORMALIZED_PHASE_BUCKETS) - seen_buckets)
    if missing:
        warnings.append("missing RocJITsu buckets: " + ", ".join(missing))

    native: list[dict[str, Any]] = []
    native_by_bucket = {bucket: [] for bucket in NORMALIZED_PHASE_BUCKETS}
    details = profile.get("details")
    if isinstance(details, list):
        for row in details:
            if not isinstance(row, dict):
                warnings.append("ignored malformed RocJITsu detail row")
                continue
            name = row.get("name")
            bucket = row.get("bucket")
            timing = row.get("timing")
            if not isinstance(name, str) or bucket not in normalized_wall:
                warnings.append("ignored malformed RocJITsu detail row")
                continue
            native.append(
                {
                    "name": name,
                    "bucket": bucket,
                    "calls": (
                        int(timing.get("count", 0)) if isinstance(timing, dict) else 0
                    ),
                    "wall_seconds": seconds(timing, "wall"),
                    "cpu_seconds": seconds(timing, "cpu"),
                }
            )
            native_by_bucket[bucket].append(name)
    else:
        warnings.append("RocJITsu phase profile has no details array")
    if "residual" in seen_buckets:
        native_by_bucket["residual"].append("outer_total residual")

    outer = profile.get("outer_total")
    envelope_wall = seconds(outer, "wall")
    envelope_cpu = seconds(outer, "cpu")
    normalized_wall_sum = sum(normalized_wall.values())
    normalized_cpu_sum = sum(normalized_cpu.values())
    wall_error = envelope_wall - normalized_wall_sum
    cpu_error = envelope_cpu - normalized_cpu_sum
    wall_tolerance = max(1e-9, envelope_wall * 1e-4)
    cpu_tolerance = max(1e-9, envelope_cpu * 1e-4)
    normalization_valid = (
        profile.get("schema_version") == 1
        and profile.get("tool") == "rocjitsu-dbt"
        and profile.get("complete") is True
        and not missing
        and abs(wall_error) <= wall_tolerance
        and abs(cpu_error) <= cpu_tolerance
    )
    if profile.get("schema_version") != 1:
        warnings.append(
            f"unsupported RocJITsu phase schema {profile.get('schema_version')!r}"
        )
    if profile.get("tool") != "rocjitsu-dbt":
        warnings.append(f"unexpected RocJITsu profile tool {profile.get('tool')!r}")
    if profile.get("complete") is not True:
        warnings.append("RocJITsu phase profile reports incomplete scopes")
    if abs(wall_error) > wall_tolerance:
        warnings.append(
            "RocJITsu wall phases differ from outer_total by "
            f"{wall_error:.9f} seconds"
        )
    if abs(cpu_error) > cpu_tolerance:
        warnings.append(
            "RocJITsu CPU phases differ from outer_total by " f"{cpu_error:.9f} seconds"
        )

    return {
        "schema_version": 1,
        "available": True,
        "source": "rj_dbt_phase_profile",
        "profile_mode": "exclusive",
        "clock": profile.get("clocks"),
        "raw_path": str(path),
        "native": native,
        "normalized_wall_seconds": normalized_wall,
        "normalized_cpu_seconds": normalized_cpu,
        "native_by_bucket": native_by_bucket,
        "envelope_wall_seconds": envelope_wall,
        "envelope_cpu_seconds": envelope_cpu,
        "normalized_sum_wall_seconds": normalized_wall_sum,
        "normalized_sum_cpu_seconds": normalized_cpu_sum,
        "reconciliation_error_seconds": wall_error,
        "cpu_reconciliation_error_seconds": cpu_error,
        "normalization_valid": normalization_valid,
        "process_elapsed_residual_seconds": (
            process_elapsed_seconds - envelope_wall
            if process_elapsed_seconds is not None
            else None
        ),
        "process_cpu_residual_seconds": (
            process_cpu_seconds - envelope_cpu
            if process_cpu_seconds is not None
            else None
        ),
        "profile_complete": profile.get("complete"),
        "warnings": warnings,
    }


def normalize_hotswap_timing(
    rows: Sequence[dict[str, Any]],
    *,
    raw_path: pathlib.Path,
    process_elapsed_seconds: float | None,
) -> dict[str, Any]:
    by_name = {row["name"]: row for row in rows}

    def total(name: str) -> float:
        return float(by_name.get(name, {}).get("wall_seconds", 0.0))

    input_metrics = ("phase:input_copy", "phase:elf_parse")
    setup_metrics = ("phase:initLLVM",)
    decode_metrics = ("phase:decode",)
    dispatch_analysis_metrics = (
        "phase:b0a0_dispatch/nop_sled_scan",
        "phase:b0a0_dispatch/cfg_build",
        "phase:b0a0_dispatch/liveness",
        "phase:b0a0_dispatch/site_control_flow",
    )
    entry_analysis_metrics = (
        "phase:entry_displacement_analysis",
        "phase:entry_displacement/decode",
    )
    dispatch_layout_metrics = (
        "phase:b0a0_dispatch/trampoline_layout",
        "phase:b0a0_dispatch/resource_metadata",
    )
    layout_metrics = (
        "phase:entry_displacement_emit",
        "phase:pool_setup",
        "phase:fixup_trampolines",
        "phase:entry_trampolines",
        "phase:prefetch_guard",
        "phase:grow_elf",
        "phase:debug_sections",
        "phase:kd_rewrite",
        "phase:symbol_insert",
        "phase:revision_metadata",
    )
    validation_metrics = ("phase:scratch_verify",)
    output_metrics = ("phase:output_copy",)

    normalized = {bucket: 0.0 for bucket in NORMALIZED_PHASE_BUCKETS}
    native_by_bucket: dict[str, list[str]] = {
        bucket: [] for bucket in NORMALIZED_PHASE_BUCKETS
    }

    def add(bucket: str, names: Sequence[str]) -> float:
        value = sum(total(name) for name in names)
        normalized[bucket] += value
        native_by_bucket[bucket].extend(name for name in names if name in by_name)
        return value

    add("input_object_parse", input_metrics)
    add("setup", setup_metrics)
    add("decode", decode_metrics)
    dispatch_analysis = add("analysis", dispatch_analysis_metrics)
    add("analysis", entry_analysis_metrics)
    dispatch_layout = add("layout_fixups", dispatch_layout_metrics)
    add("layout_fixups", layout_metrics)
    add("validation", validation_metrics)
    add("output", output_metrics)

    warnings: list[str] = []
    dispatch_total = total("phase:b0a0_dispatch")
    dispatch_rewrite = dispatch_total - dispatch_analysis - dispatch_layout
    tolerance = max(1e-9, dispatch_total * 1e-4)
    normalization_valid = True
    if dispatch_rewrite < -tolerance:
        normalization_valid = False
        warnings.append(
            "HotSwap analysis/layout children exceed b0a0_dispatch by "
            f"{-dispatch_rewrite:.9f} seconds"
        )
    normalized["rewrite"] += max(0.0, dispatch_rewrite)
    if "phase:b0a0_dispatch" in by_name:
        native_by_bucket["rewrite"].append("phase:b0a0_dispatch complement")
    unaccounted = total("phase:unaccounted")
    normalized["residual"] += unaccounted
    if "phase:unaccounted" in by_name:
        native_by_bucket["residual"].append("phase:unaccounted")

    envelope = total("phase:rewrite_total")
    normalized_sum = sum(normalized.values())
    reconciliation_error = envelope - normalized_sum
    envelope_tolerance = max(1e-9, envelope * 1e-4)
    if abs(reconciliation_error) > envelope_tolerance:
        normalization_valid = False
        warnings.append(
            "HotSwap normalized phases differ from rewrite_total by "
            f"{reconciliation_error:.9f} seconds"
        )
    process_residual = (
        process_elapsed_seconds - envelope
        if process_elapsed_seconds is not None and envelope > 0
        else None
    )
    return {
        "schema_version": 1,
        "available": bool(rows),
        "source": "amd_comgr_time_statistics",
        "clock": "steady_wall",
        "raw_path": str(raw_path),
        "native": list(rows),
        "normalized_wall_seconds": normalized,
        "normalized_cpu_seconds": None,
        "native_by_bucket": native_by_bucket,
        "envelope_wall_seconds": envelope or None,
        "normalized_sum_wall_seconds": normalized_sum,
        "reconciliation_error_seconds": reconciliation_error,
        "normalization_valid": normalization_valid and bool(rows),
        "process_elapsed_residual_seconds": process_residual,
        "warnings": warnings,
    }


def unavailable_phase_timing(
    tool: str, raw_path: pathlib.Path, reason: str
) -> dict[str, Any]:
    return {
        "schema_version": 1,
        "available": False,
        "source": tool,
        "raw_path": str(raw_path),
        "normalized_wall_seconds": {bucket: 0.0 for bucket in NORMALIZED_PHASE_BUCKETS},
        "normalized_cpu_seconds": None,
        "normalization_valid": False,
        "unavailable_reason": reason,
        "warnings": [] if reason == "phase timing disabled" else [reason],
    }


def run_measurement(
    *,
    tool: str,
    source: dict[str, Any],
    phase: str,
    repetition: int,
    attempt: int,
    order_in_pair: int,
    timeout_seconds: float,
    timeout_basis: str,
    args: argparse.Namespace,
    dbt: pathlib.Path,
    hotswap: pathlib.Path,
    library_dirs: Sequence[pathlib.Path],
    output_directory: pathlib.Path,
) -> dict[str, Any]:
    source_path = pathlib.Path(source["path"])
    token = input_token(source_path)
    label = f"{token}.{phase}-{repetition}.{tool}.attempt-{attempt}"
    logs = output_directory / "logs"
    work = output_directory / "work"
    retained_outputs = output_directory / "outputs"
    logs.mkdir(parents=True, exist_ok=True)
    work.mkdir(parents=True, exist_ok=True)
    if args.keep_outputs and phase == "measurement":
        retained_outputs.mkdir(parents=True, exist_ok=True)
        output = retained_outputs / f"{label}.co"
    else:
        output = work / f"{label}.co"
    output.unlink(missing_ok=True)

    stdout_log = logs / f"{label}.stdout.log"
    stderr_log = logs / f"{label}.stderr.log"
    phase_timing_log = logs / (
        f"{label}.phase-timing.json"
        if tool == "rocjitsu-dbt"
        else f"{label}.phase-timing.log"
    )
    stdout_log.unlink(missing_ok=True)
    stderr_log.unlink(missing_ok=True)
    phase_timing_log.unlink(missing_ok=True)
    environment = dict(os.environ)
    environment["LC_ALL"] = "C"
    for variable in (
        "RJ_DBT_PHASE_PROFILE",
        "AMD_COMGR_TIME_STATISTICS",
        "AMD_COMGR_TIME_STATISTICS_GRANULARITY",
        "AMD_COMGR_REDIRECT_LOGS",
        "AMD_COMGR_HOTSWAP_PROFILE_MODE",
    ):
        environment.pop(variable, None)

    if tool == "rocjitsu-dbt":
        command = dbt_command(dbt, source_path)
        if args.phase_timing:
            environment["RJ_DBT_PHASE_PROFILE"] = str(phase_timing_log)
        stdout_target = output.open("wb")
        recorded_stdout: str | None = None
    else:
        command = hotswap_command(
            hotswap,
            source_path,
            output,
            args.hotswap_source_isa,
            args.hotswap_target_isa,
        )
        if library_dirs:
            previous = environment.get("LD_LIBRARY_PATH")
            prefix = os.pathsep.join(str(path) for path in library_dirs)
            environment["LD_LIBRARY_PATH"] = (
                prefix + os.pathsep + previous if previous else prefix
            )
        if args.hotswap_verbose_logs:
            environment["AMD_COMGR_EMIT_VERBOSE_LOGS"] = "1"
        else:
            environment.pop("AMD_COMGR_EMIT_VERBOSE_LOGS", None)
        if args.phase_timing:
            environment["AMD_COMGR_TIME_STATISTICS"] = "1"
            environment["AMD_COMGR_TIME_STATISTICS_GRANULARITY"] = "ns"
            environment["AMD_COMGR_REDIRECT_LOGS"] = str(phase_timing_log)
            if args.hotswap_phase_profile_mode == "coarse":
                environment["AMD_COMGR_HOTSWAP_PROFILE_MODE"] = "coarse"
        stdout_target = stdout_log.open("wb")
        recorded_stdout = str(stdout_log)

    started_at = utc_now()
    try:
        with stdout_target, stderr_log.open("wb") as stderr_target:
            process_result = timed_process(
                command,
                stdout=stdout_target,
                stderr=stderr_target,
                environment=environment,
                timeout_seconds=timeout_seconds,
            )
    except OSError as error:
        process_result = {
            "return_code": None,
            "timed_out": False,
            "elapsed_seconds": 0.0,
            "user_cpu_seconds": None,
            "system_cpu_seconds": None,
            "cpu_seconds": None,
            "max_rss_kib": None,
            "spawn_error": str(error),
        }

    output_info = output_identity(output)
    status = (
        "spawn_error"
        if "spawn_error" in process_result
        else status_for_result(process_result, output_info)
    )
    retained_path: str | None = None
    if status == "success" and args.keep_outputs and phase == "measurement":
        retained_path = str(output)
    else:
        output.unlink(missing_ok=True)
    if output_info is not None:
        output_info["retained_path"] = retained_path
    if args.phase_timing and tool == "hotswap":
        timing_rows = parse_hotswap_timing_rows(phase_timing_log)
        phase_timing = normalize_hotswap_timing(
            timing_rows,
            raw_path=phase_timing_log,
            process_elapsed_seconds=process_result.get("elapsed_seconds"),
        )
        phase_timing["profile_mode"] = args.hotswap_phase_profile_mode
        if not timing_rows:
            phase_timing["warnings"].append("HotSwap emitted no time-statistics rows")
        if args.hotswap_phase_profile_mode == "coarse" and any(
            row["name"].startswith("strat:") for row in timing_rows
        ):
            phase_timing["warnings"].append(
                "HotSwap emitted detailed strategy rows in coarse mode"
            )
    elif args.phase_timing:
        phase_timing = load_rocjitsu_timing(
            phase_timing_log,
            process_cpu_seconds=process_result.get("cpu_seconds"),
            process_elapsed_seconds=process_result.get("elapsed_seconds"),
        )
    else:
        phase_timing = unavailable_phase_timing(
            tool, phase_timing_log, "phase timing disabled"
        )

    return {
        "schema_version": SCHEMA_VERSION,
        "type": "measurement",
        "recorded_at": utc_now(),
        "started_at": started_at,
        "input": source,
        "tool": tool,
        "phase": phase,
        "repetition": repetition,
        "attempt": attempt,
        "order_in_pair": order_in_pair,
        "timeout_seconds": timeout_seconds,
        "timeout_basis": timeout_basis,
        "command": command,
        "command_shell": shell_command(command),
        "status": status,
        **process_result,
        "output": output_info,
        "stdout_log": recorded_stdout,
        "stderr_log": str(stderr_log),
        "phase_timing": phase_timing,
    }


def median_value(records: Sequence[dict[str, Any]], field: str) -> float | None:
    values = [
        float(record[field]) for record in records if record.get(field) is not None
    ]
    return statistics.median(values) if values else None


def safe_ratio(numerator: float | None, denominator: float | None) -> float | None:
    if numerator is None or denominator is None or denominator <= 0:
        return None
    return numerator / denominator


def build_summary(
    metadata: dict[str, Any], records: Sequence[dict[str, Any]]
) -> dict[str, Any]:
    config = metadata["benchmark_config"]
    repetitions = int(config["repetitions"])
    latest = latest_measurements(records)
    per_input: list[dict[str, Any]] = []
    failure_counts: Counter[str] = Counter()
    successful_measurements = 0

    for source in config["inputs"]:
        entry: dict[str, Any] = {
            "input_path": source["path"],
            "input_sha256": source["sha256"],
            "input_size": source["size"],
        }
        for tool in TOOLS:
            values: list[dict[str, Any]] = []
            observed: list[dict[str, Any]] = []
            statuses: Counter[str] = Counter()
            for repetition in range(1, repetitions + 1):
                key = (source["path"], tool, "measurement", repetition)
                record = latest.get(key)
                if record is None:
                    failure_counts["missing"] += 1
                    statuses["missing"] += 1
                elif record["status"] == "success":
                    observed.append(record)
                    values.append(record)
                    statuses["success"] += 1
                    successful_measurements += 1
                else:
                    observed.append(record)
                    failure_counts[record["status"]] += 1
                    statuses[record["status"]] += 1
            prefix = "dbt" if tool == "rocjitsu-dbt" else "hotswap"
            entry[f"{prefix}_successes"] = len(values)
            entry[f"{prefix}_statuses"] = ";".join(
                f"{status}:{count}" for status, count in sorted(statuses.items())
            )
            entry[f"{prefix}_cpu_median_seconds"] = median_value(values, "cpu_seconds")
            entry[f"{prefix}_elapsed_median_seconds"] = median_value(
                values, "elapsed_seconds"
            )
            entry[f"{prefix}_peak_rss_median_kib"] = median_value(values, "max_rss_kib")
            observed_cpu = [
                float(record["cpu_seconds"])
                for record in observed
                if record.get("cpu_seconds") is not None
            ]
            observed_rss = [
                int(record["max_rss_kib"])
                for record in observed
                if record.get("max_rss_kib") is not None
            ]
            timeout_limits = [
                float(record["timeout_seconds"])
                for record in observed
                if record.get("timeout_seconds") is not None
            ]
            entry[f"{prefix}_observed_cpu_max_seconds"] = (
                max(observed_cpu) if observed_cpu else None
            )
            entry[f"{prefix}_observed_peak_rss_max_kib"] = (
                max(observed_rss) if observed_rss else None
            )
            entry[f"{prefix}_timeout_limit_max_seconds"] = (
                max(timeout_limits) if timeout_limits else None
            )

        entry["cpu_speedup_hotswap_over_dbt"] = safe_ratio(
            entry["hotswap_cpu_median_seconds"],
            entry["dbt_cpu_median_seconds"],
        )
        entry["rss_ratio_hotswap_over_dbt"] = safe_ratio(
            entry["hotswap_peak_rss_median_kib"],
            entry["dbt_peak_rss_median_kib"],
        )
        if (
            entry["hotswap_peak_rss_median_kib"] is not None
            and entry["hotswap_peak_rss_median_kib"] > 0
            and entry["dbt_peak_rss_median_kib"] is not None
        ):
            entry["dbt_rss_reduction_percent"] = (
                100.0
                * (
                    entry["hotswap_peak_rss_median_kib"]
                    - entry["dbt_peak_rss_median_kib"]
                )
                / entry["hotswap_peak_rss_median_kib"]
            )
        else:
            entry["dbt_rss_reduction_percent"] = None
        per_input.append(entry)

    paired = [
        value
        for value in per_input
        if value["dbt_successes"] == repetitions
        and value["hotswap_successes"] == repetitions
    ]
    dbt_cpu = sum(value["dbt_cpu_median_seconds"] for value in paired)
    hotswap_cpu = sum(value["hotswap_cpu_median_seconds"] for value in paired)
    speedups = [
        value["cpu_speedup_hotswap_over_dbt"]
        for value in paired
        if value["cpu_speedup_hotswap_over_dbt"] is not None
        and value["cpu_speedup_hotswap_over_dbt"] > 0
    ]

    measured_records = [
        record for record in latest.values() if record["phase"] == "measurement"
    ]
    peak_rss: dict[str, int | None] = {}
    for tool in TOOLS:
        values = [
            int(record["max_rss_kib"])
            for record in measured_records
            if record["tool"] == tool and record.get("max_rss_kib") is not None
        ]
        peak_rss[tool] = max(values) if values else None

    expected_measurements = len(config["inputs"]) * len(TOOLS) * repetitions
    complete = successful_measurements == expected_measurements
    corpus = {
        "paired_inputs": len(paired),
        "rocjitsu_dbt_cpu_seconds": dbt_cpu if paired else None,
        "hotswap_cpu_seconds": hotswap_cpu if paired else None,
        "cpu_speedup_hotswap_over_dbt": safe_ratio(hotswap_cpu, dbt_cpu),
        "per_input_cpu_speedup_geomean": (
            math.exp(sum(math.log(value) for value in speedups) / len(speedups))
            if speedups
            else None
        ),
        "rocjitsu_dbt_peak_rss_kib": peak_rss["rocjitsu-dbt"],
        "hotswap_peak_rss_kib": peak_rss["hotswap"],
        "rss_ratio_hotswap_over_dbt": safe_ratio(
            peak_rss["hotswap"], peak_rss["rocjitsu-dbt"]
        ),
    }
    if peak_rss["hotswap"] and peak_rss["rocjitsu-dbt"] is not None:
        corpus["dbt_rss_reduction_percent"] = (
            100.0
            * (peak_rss["hotswap"] - peak_rss["rocjitsu-dbt"])
            / peak_rss["hotswap"]
        )
    else:
        corpus["dbt_rss_reduction_percent"] = None

    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": utc_now(),
        "complete": complete,
        "input_count": len(config["inputs"]),
        "repetitions": repetitions,
        "expected_measurements": expected_measurements,
        "successful_measurements": successful_measurements,
        "failure_counts": dict(sorted(failure_counts.items())),
        "definitions": {
            "cpu_seconds": "Linux wait4 user CPU plus system CPU",
            "corpus_cpu_seconds": "sum of per-input medians over paired inputs",
            "peak_rss_kib": "maximum per-process Linux wait4 ru_maxrss over all measured runs, including failures",
            "cpu_speedup_hotswap_over_dbt": "HotSwap CPU divided by RocJITsu DBT CPU; values above one favor DBT",
            "rss_ratio_hotswap_over_dbt": "HotSwap RSS divided by RocJITsu DBT RSS; values above one favor DBT",
        },
        "corpus": corpus,
        "per_input": sorted(per_input, key=lambda value: value["input_path"]),
    }


def csv_text(fieldnames: Sequence[str], rows: Sequence[dict[str, Any]]) -> str:
    stream = io.StringIO(newline="")
    writer = csv.DictWriter(stream, fieldnames=fieldnames, extrasaction="ignore")
    writer.writeheader()
    writer.writerows(rows)
    return stream.getvalue()


def write_measurements_csv(
    path: pathlib.Path, records: Sequence[dict[str, Any]]
) -> None:
    rows: list[dict[str, Any]] = []
    for record in sorted(
        latest_measurements(records).values(),
        key=lambda value: (
            value["input"]["path"],
            value["phase"],
            value["repetition"],
            value["tool"],
        ),
    ):
        output = record.get("output") or {}
        rows.append(
            {
                "input_path": record["input"]["path"],
                "input_sha256": record["input"]["sha256"],
                "input_size": record["input"]["size"],
                "tool": record["tool"],
                "phase": record["phase"],
                "repetition": record["repetition"],
                "attempt": record["attempt"],
                "order_in_pair": record["order_in_pair"],
                "timeout_seconds": record.get("timeout_seconds"),
                "timeout_basis": record.get("timeout_basis"),
                "status": record["status"],
                "return_code": record.get("return_code"),
                "user_cpu_seconds": record.get("user_cpu_seconds"),
                "system_cpu_seconds": record.get("system_cpu_seconds"),
                "cpu_seconds": record.get("cpu_seconds"),
                "elapsed_seconds": record.get("elapsed_seconds"),
                "max_rss_kib": record.get("max_rss_kib"),
                "output_size": output.get("size"),
                "output_sha256": output.get("sha256"),
                "command": record["command_shell"],
                "stdout_log": record.get("stdout_log"),
                "stderr_log": record.get("stderr_log"),
            }
        )
    fields = [
        "input_path",
        "input_sha256",
        "input_size",
        "tool",
        "phase",
        "repetition",
        "attempt",
        "order_in_pair",
        "timeout_seconds",
        "timeout_basis",
        "status",
        "return_code",
        "user_cpu_seconds",
        "system_cpu_seconds",
        "cpu_seconds",
        "elapsed_seconds",
        "max_rss_kib",
        "output_size",
        "output_sha256",
        "command",
        "stdout_log",
        "stderr_log",
    ]
    atomic_text(path, csv_text(fields, rows))


def write_comparison_csv(path: pathlib.Path, summary: dict[str, Any]) -> None:
    fields = [
        "input_path",
        "input_sha256",
        "input_size",
        "dbt_successes",
        "dbt_statuses",
        "hotswap_successes",
        "hotswap_statuses",
        "dbt_cpu_median_seconds",
        "hotswap_cpu_median_seconds",
        "cpu_speedup_hotswap_over_dbt",
        "dbt_elapsed_median_seconds",
        "hotswap_elapsed_median_seconds",
        "dbt_peak_rss_median_kib",
        "hotswap_peak_rss_median_kib",
        "rss_ratio_hotswap_over_dbt",
        "dbt_rss_reduction_percent",
        "dbt_observed_cpu_max_seconds",
        "hotswap_observed_cpu_max_seconds",
        "dbt_observed_peak_rss_max_kib",
        "hotswap_observed_peak_rss_max_kib",
        "dbt_timeout_limit_max_seconds",
        "hotswap_timeout_limit_max_seconds",
    ]
    atomic_text(path, csv_text(fields, summary["per_input"]))


def phase_measurement_rows(
    records: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    measured = [
        record
        for record in latest_measurements(records).values()
        if record["phase"] == "measurement"
    ]
    for record in sorted(
        measured,
        key=lambda value: (
            value["input"]["path"],
            value["repetition"],
            value["tool"],
        ),
    ):
        timing = record.get("phase_timing") or {}
        available = timing.get("available") is True
        wall = timing.get("normalized_wall_seconds")
        cpu = timing.get("normalized_cpu_seconds")
        envelope_wall = timing.get("envelope_wall_seconds")
        envelope_cpu = timing.get("envelope_cpu_seconds")
        for bucket in NORMALIZED_PHASE_BUCKETS:
            wall_seconds = (
                float(wall.get(bucket, 0.0))
                if available and isinstance(wall, dict)
                else None
            )
            cpu_seconds = (
                float(cpu.get(bucket, 0.0))
                if available and isinstance(cpu, dict)
                else None
            )
            rows.append(
                {
                    "input_path": record["input"]["path"],
                    "input_sha256": record["input"]["sha256"],
                    "tool": record["tool"],
                    "repetition": record["repetition"],
                    "attempt": record["attempt"],
                    "status": record["status"],
                    "profile_available": available,
                    "profile_mode": timing.get("profile_mode"),
                    "normalization_valid": timing.get("normalization_valid") is True,
                    "bucket": bucket,
                    "wall_seconds": wall_seconds,
                    "cpu_seconds": cpu_seconds,
                    "percent_of_wall_envelope": (
                        100.0 * wall_seconds / float(envelope_wall)
                        if wall_seconds is not None
                        and envelope_wall is not None
                        and float(envelope_wall) > 0
                        else None
                    ),
                    "percent_of_cpu_envelope": (
                        100.0 * cpu_seconds / float(envelope_cpu)
                        if cpu_seconds is not None
                        and envelope_cpu is not None
                        and float(envelope_cpu) > 0
                        else None
                    ),
                    "envelope_wall_seconds": envelope_wall,
                    "envelope_cpu_seconds": envelope_cpu,
                    "reconciliation_error_seconds": timing.get(
                        "reconciliation_error_seconds"
                    ),
                    "cpu_reconciliation_error_seconds": timing.get(
                        "cpu_reconciliation_error_seconds"
                    ),
                    "process_elapsed_residual_seconds": timing.get(
                        "process_elapsed_residual_seconds"
                    ),
                    "process_cpu_residual_seconds": timing.get(
                        "process_cpu_residual_seconds"
                    ),
                    "raw_path": timing.get("raw_path"),
                    "warnings": "; ".join(timing.get("warnings") or []),
                }
            )
    return rows


def write_phase_measurements_csv(
    path: pathlib.Path, records: Sequence[dict[str, Any]]
) -> None:
    fields = [
        "input_path",
        "input_sha256",
        "tool",
        "repetition",
        "attempt",
        "status",
        "profile_available",
        "profile_mode",
        "normalization_valid",
        "bucket",
        "wall_seconds",
        "cpu_seconds",
        "percent_of_wall_envelope",
        "percent_of_cpu_envelope",
        "envelope_wall_seconds",
        "envelope_cpu_seconds",
        "reconciliation_error_seconds",
        "cpu_reconciliation_error_seconds",
        "process_elapsed_residual_seconds",
        "process_cpu_residual_seconds",
        "raw_path",
        "warnings",
    ]
    atomic_text(path, csv_text(fields, phase_measurement_rows(records)))


def phase_native_rows(records: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
    result: list[dict[str, Any]] = []
    measured = [
        record
        for record in latest_measurements(records).values()
        if record["phase"] == "measurement"
    ]
    for record in sorted(
        measured,
        key=lambda value: (
            value["input"]["path"],
            value["repetition"],
            value["tool"],
        ),
    ):
        timing = record.get("phase_timing") or {}
        native_by_bucket = timing.get("native_by_bucket") or {}
        bucket_by_name = {
            name: bucket
            for bucket, names in native_by_bucket.items()
            for name in names
            if isinstance(name, str)
        }
        for native in timing.get("native") or []:
            name = native.get("name")
            result.append(
                {
                    "input_path": record["input"]["path"],
                    "input_sha256": record["input"]["sha256"],
                    "tool": record["tool"],
                    "repetition": record["repetition"],
                    "attempt": record["attempt"],
                    "status": record["status"],
                    "native_name": name,
                    "normalized_bucket": (
                        native.get("bucket") or bucket_by_name.get(name)
                    ),
                    "calls": native.get("calls"),
                    "wall_seconds": native.get("wall_seconds"),
                    "cpu_seconds": native.get("cpu_seconds"),
                    "minimum_wall_seconds": native.get("minimum_wall_seconds"),
                    "maximum_wall_seconds": native.get("maximum_wall_seconds"),
                    "patches": native.get("patches"),
                    "raw_path": timing.get("raw_path"),
                }
            )
    return result


def write_phase_native_csv(
    path: pathlib.Path, records: Sequence[dict[str, Any]]
) -> None:
    fields = [
        "input_path",
        "input_sha256",
        "tool",
        "repetition",
        "attempt",
        "status",
        "native_name",
        "normalized_bucket",
        "calls",
        "wall_seconds",
        "cpu_seconds",
        "minimum_wall_seconds",
        "maximum_wall_seconds",
        "patches",
        "raw_path",
    ]
    atomic_text(path, csv_text(fields, phase_native_rows(records)))


def build_phase_summary(
    metadata: dict[str, Any], records: Sequence[dict[str, Any]]
) -> dict[str, Any]:
    config = metadata["benchmark_config"]
    repetitions = int(config["repetitions"])
    measured = [
        record
        for record in latest_measurements(records).values()
        if record["phase"] == "measurement"
    ]
    coverage: dict[str, dict[str, Any]] = {}
    warning_counts: Counter[str] = Counter()
    complete_profiles: dict[tuple[str, str], list[dict[str, Any]]] = {}

    for tool in TOOLS:
        tool_records = [record for record in measured if record["tool"] == tool]
        available = [
            record
            for record in tool_records
            if (record.get("phase_timing") or {}).get("available") is True
        ]
        valid = [
            record
            for record in available
            if (record.get("phase_timing") or {}).get("normalization_valid") is True
        ]
        eligible = [record for record in valid if record["status"] == "success"]
        for record in tool_records:
            warning_counts.update(
                (record.get("phase_timing") or {}).get("warnings") or []
            )
        by_input: dict[str, list[dict[str, Any]]] = {}
        for record in eligible:
            by_input.setdefault(record["input"]["path"], []).append(record)
        for input_path, input_records in by_input.items():
            if len(input_records) == repetitions:
                complete_profiles[(tool, input_path)] = input_records
        coverage[tool] = {
            "expected_profiles": len(config["inputs"]) * repetitions,
            "available_profiles": len(available),
            "valid_profiles": len(valid),
            "successful_valid_profiles": len(eligible),
            "complete_inputs": sum(1 for key in complete_profiles if key[0] == tool),
        }

    paired_inputs = [
        source["path"]
        for source in config["inputs"]
        if ("rocjitsu-dbt", source["path"]) in complete_profiles
        and ("hotswap", source["path"]) in complete_profiles
    ]

    def input_median(tool: str, input_path: str, bucket: str, clock: str) -> float:
        values = []
        for record in complete_profiles[(tool, input_path)]:
            normalized = (record.get("phase_timing") or {}).get(
                f"normalized_{clock}_seconds"
            )
            if isinstance(normalized, dict):
                values.append(float(normalized.get(bucket, 0.0)))
        return statistics.median(values) if values else 0.0

    def tool_distribution(tool: str, input_paths: Sequence[str]) -> dict[str, Any]:
        rows: list[dict[str, Any]] = []
        for bucket in NORMALIZED_PHASE_BUCKETS:
            wall_seconds = sum(
                input_median(tool, path, bucket, "wall") for path in input_paths
            )
            cpu_seconds = (
                sum(input_median(tool, path, bucket, "cpu") for path in input_paths)
                if tool == "rocjitsu-dbt"
                else None
            )
            rows.append(
                {
                    "bucket": bucket,
                    "wall_seconds": wall_seconds if input_paths else None,
                    "cpu_seconds": cpu_seconds if input_paths else None,
                }
            )
        wall_total = sum(row["wall_seconds"] or 0.0 for row in rows)
        cpu_total = sum(row["cpu_seconds"] or 0.0 for row in rows)
        for row in rows:
            row["wall_percent"] = (
                100.0 * row["wall_seconds"] / wall_total
                if row["wall_seconds"] is not None and wall_total > 0
                else None
            )
            row["cpu_percent"] = (
                100.0 * row["cpu_seconds"] / cpu_total
                if row["cpu_seconds"] is not None and cpu_total > 0
                else None
            )
        return {
            "input_count": len(input_paths),
            "input_paths": list(input_paths),
            "wall_seconds": wall_total if input_paths else None,
            "cpu_seconds": (
                cpu_total if input_paths and tool == "rocjitsu-dbt" else None
            ),
            "per_bucket": rows,
        }

    complete_input_paths = {
        tool: [
            source["path"]
            for source in config["inputs"]
            if (tool, source["path"]) in complete_profiles
        ]
        for tool in TOOLS
    }
    tool_distributions = {
        tool: tool_distribution(tool, paths)
        for tool, paths in complete_input_paths.items()
    }
    paired_distributions = {
        tool: tool_distribution(tool, paired_inputs) for tool in TOOLS
    }

    per_bucket: list[dict[str, Any]] = []
    for index, bucket in enumerate(NORMALIZED_PHASE_BUCKETS):
        dbt = paired_distributions["rocjitsu-dbt"]["per_bucket"][index]
        hotswap = paired_distributions["hotswap"]["per_bucket"][index]
        per_bucket.append(
            {
                "bucket": bucket,
                "rocjitsu_dbt_wall_seconds": dbt["wall_seconds"],
                "rocjitsu_dbt_wall_percent": dbt["wall_percent"],
                "hotswap_wall_seconds": hotswap["wall_seconds"],
                "hotswap_wall_percent": hotswap["wall_percent"],
                "hotswap_over_dbt_wall_ratio": safe_ratio(
                    hotswap["wall_seconds"], dbt["wall_seconds"]
                ),
                "rocjitsu_dbt_cpu_seconds": dbt["cpu_seconds"],
                "rocjitsu_dbt_cpu_percent": dbt["cpu_percent"],
            }
        )

    dbt_wall_total = paired_distributions["rocjitsu-dbt"]["wall_seconds"]
    hotswap_wall_total = paired_distributions["hotswap"]["wall_seconds"]
    dbt_cpu_total = paired_distributions["rocjitsu-dbt"]["cpu_seconds"]

    return {
        "schema_version": 1,
        "generated_at": utc_now(),
        "enabled": bool(config.get("phase_timing", False)),
        "comparison_clock": "steady/monotonic wall time",
        "aggregation": (
            "sum of per-input medians over inputs with every requested repetition "
            "successfully profiled by both tools"
        ),
        "tool_distribution_aggregation": (
            "sum of per-input medians over every input with all requested "
            "repetitions successfully profiled by that tool"
        ),
        "normalized_buckets": list(NORMALIZED_PHASE_BUCKETS),
        "coverage": coverage,
        "tool_distributions": tool_distributions,
        "paired_inputs": len(paired_inputs),
        "paired_input_paths": paired_inputs,
        "totals": {
            "rocjitsu_dbt_wall_seconds": dbt_wall_total if paired_inputs else None,
            "hotswap_wall_seconds": hotswap_wall_total if paired_inputs else None,
            "hotswap_over_dbt_wall_ratio": (
                safe_ratio(hotswap_wall_total, dbt_wall_total)
                if paired_inputs
                else None
            ),
            "rocjitsu_dbt_cpu_seconds": dbt_cpu_total if paired_inputs else None,
        },
        "per_bucket": per_bucket,
        "warning_counts": dict(sorted(warning_counts.items())),
    }


def phase_summary_csv_text(summary: dict[str, Any]) -> str:
    fields = [
        "scope",
        "tool",
        "input_count",
        "bucket",
        "wall_seconds",
        "wall_percent",
        "cpu_seconds",
        "cpu_percent",
        "rocjitsu_dbt_wall_seconds",
        "rocjitsu_dbt_wall_percent",
        "hotswap_wall_seconds",
        "hotswap_wall_percent",
        "hotswap_over_dbt_wall_ratio",
        "rocjitsu_dbt_cpu_seconds",
        "rocjitsu_dbt_cpu_percent",
    ]
    rows = [
        {
            "scope": "paired_comparison",
            "input_count": summary["paired_inputs"],
            **row,
        }
        for row in summary["per_bucket"]
    ]
    for tool in TOOLS:
        distribution = summary["tool_distributions"][tool]
        rows.extend(
            {
                "scope": "tool_distribution",
                "tool": tool,
                "input_count": distribution["input_count"],
                **row,
            }
            for row in distribution["per_bucket"]
        )
    return csv_text(fields, rows)


def compact_phase_seconds(value: float | None) -> str:
    if value is None:
        return "n/a"
    if value >= 60:
        return f"{value / 60:.2f}m"
    if value >= 1:
        return f"{value:.3f}s"
    if value >= 0.001:
        return f"{value * 1000:.2f}ms"
    return f"{value * 1_000_000:.1f}us"


def terminal_phase_summary(summary: dict[str, Any]) -> str:
    lines = [
        (
            "Phase timing (paired inputs, sum of per-input medians; "
            "monotonic wall clock)"
        ),
        (
            f"{'Bucket':<22} {'DBT':>11} {'DBT %':>7} "
            f"{'HotSwap':>11} {'HS %':>7} {'HS/DBT':>9}"
        ),
        f"{'-' * 22} {'-' * 11} {'-' * 7} {'-' * 11} {'-' * 7} {'-' * 9}",
    ]
    for row in summary["per_bucket"]:
        ratio = row["hotswap_over_dbt_wall_ratio"]
        dbt_percent = row["rocjitsu_dbt_wall_percent"]
        hotswap_percent = row["hotswap_wall_percent"]
        lines.append(
            f"{row['bucket']:<22} "
            f"{compact_phase_seconds(row['rocjitsu_dbt_wall_seconds']):>11} "
            f"{('n/a' if dbt_percent is None else f'{dbt_percent:.1f}%'):>7} "
            f"{compact_phase_seconds(row['hotswap_wall_seconds']):>11} "
            f"{('n/a' if hotswap_percent is None else f'{hotswap_percent:.1f}%'):>7} "
            f"{('n/a' if ratio is None else f'{ratio:.2f}x'):>9}"
        )
    dbt_distribution = summary["tool_distributions"]["rocjitsu-dbt"]
    hotswap_distribution = summary["tool_distributions"]["hotswap"]
    lines.extend(
        [
            "",
            (
                "Per-tool distribution over every complete profile "
                "(input sets may differ)"
            ),
            (
                f"{'Bucket':<22} {'DBT':>11} {'DBT %':>7} "
                f"{'HotSwap':>11} {'HS %':>7}"
            ),
            f"{'-' * 22} {'-' * 11} {'-' * 7} {'-' * 11} {'-' * 7}",
        ]
    )
    for index, bucket in enumerate(NORMALIZED_PHASE_BUCKETS):
        dbt = dbt_distribution["per_bucket"][index]
        hotswap = hotswap_distribution["per_bucket"][index]
        dbt_percent = dbt["wall_percent"]
        hotswap_percent = hotswap["wall_percent"]
        lines.append(
            f"{bucket:<22} "
            f"{compact_phase_seconds(dbt['wall_seconds']):>11} "
            f"{('n/a' if dbt_percent is None else f'{dbt_percent:.1f}%'):>7} "
            f"{compact_phase_seconds(hotswap['wall_seconds']):>11} "
            f"{('n/a' if hotswap_percent is None else f'{hotswap_percent:.1f}%'):>7}"
        )
    lines.extend(
        [
            "",
            f"Paired inputs: {summary['paired_inputs']}",
            (
                "All-profile inputs: "
                f"DBT={dbt_distribution['input_count']}, "
                f"HotSwap={hotswap_distribution['input_count']}"
            ),
        ]
    )
    for tool in TOOLS:
        value = summary["coverage"][tool]
        lines.append(
            f"{tool}: {value['valid_profiles']}/{value['expected_profiles']} "
            f"valid profiles; {value['complete_inputs']} complete inputs"
        )
    if summary["warning_counts"]:
        lines.append("Timer warnings:")
        lines.extend(
            f"  {count}x {warning}"
            for warning, count in summary["warning_counts"].items()
        )
    lines.append("")
    return "\n".join(lines)


def write_phase_reports(
    output_directory: pathlib.Path,
    metadata: dict[str, Any],
    records: Sequence[dict[str, Any]],
) -> dict[str, Any]:
    summary = build_phase_summary(metadata, records)
    atomic_json(output_directory / "phase_summary.json", summary)
    atomic_text(output_directory / "phase_summary.csv", phase_summary_csv_text(summary))
    atomic_text(
        output_directory / "phase_breakdown.txt", terminal_phase_summary(summary)
    )
    write_phase_measurements_csv(output_directory / "phase_measurements.csv", records)
    write_phase_native_csv(output_directory / "phase_native.csv", records)
    return summary


def histogram_data(records: Sequence[dict[str, Any]]) -> dict[str, Any]:
    measured = [
        record
        for record in latest_measurements(records).values()
        if record["phase"] == "measurement"
    ]
    definitions = (
        ("cpu_time", "CPU time", "seconds", "cpu_seconds", 1.0),
        ("peak_rss", "Peak RSS", "MiB", "max_rss_kib", 1.0 / 1024.0),
    )
    metrics: dict[str, Any] = {}
    for name, title, unit, field, scale in definitions:
        values_by_tool: dict[str, list[float]] = {}
        statuses_by_tool: dict[str, dict[str, int]] = {}
        all_values: list[float] = []
        for tool in TOOLS:
            tool_records = [record for record in measured if record["tool"] == tool]
            values = [
                float(record[field]) * scale
                for record in tool_records
                if record.get(field) is not None and float(record[field]) * scale > 0
            ]
            values_by_tool[tool] = values
            all_values.extend(values)
            statuses_by_tool[tool] = dict(
                sorted(Counter(record["status"] for record in tool_records).items())
            )
        if not all_values:
            continue
        bin_count = min(20, max(6, math.ceil(math.sqrt(len(all_values)))))
        lower = min(all_values)
        upper = max(all_values)
        if lower == upper:
            lower *= 0.9
            upper *= 1.1
            if lower <= 0:
                lower = upper / 10
        log_lower = math.log10(lower)
        log_upper = math.log10(upper)
        edges = [
            10 ** (log_lower + (log_upper - log_lower) * index / bin_count)
            for index in range(bin_count + 1)
        ]
        series = []
        for tool in TOOLS:
            counts = [0] * bin_count
            for value in values_by_tool[tool]:
                index = bisect.bisect_right(edges, value) - 1
                index = max(0, min(bin_count - 1, index))
                counts[index] += 1
            series.append(
                {
                    "tool": tool,
                    "sample_count": len(values_by_tool[tool]),
                    "status_counts": statuses_by_tool[tool],
                    "counts": counts,
                }
            )
        metrics[name] = {
            "title": title,
            "unit": unit,
            "scale": "log10",
            "source_field": field,
            "edges": edges,
            "series": series,
        }
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": utc_now(),
        "scope": (
            "latest measured process for every input/tool/repetition; "
            "includes successful, failed, and timed-out processes"
        ),
        "metrics": metrics,
    }


def histogram_csv_text(histograms: dict[str, Any]) -> str:
    rows: list[dict[str, Any]] = []
    for metric_name, metric in histograms["metrics"].items():
        edges = metric["edges"]
        for series in metric["series"]:
            for index, count in enumerate(series["counts"]):
                rows.append(
                    {
                        "metric": metric_name,
                        "unit": metric["unit"],
                        "scale": metric["scale"],
                        "tool": series["tool"],
                        "bin_index": index,
                        "lower_bound": edges[index],
                        "upper_bound": edges[index + 1],
                        "count": count,
                    }
                )
    fields = [
        "metric",
        "unit",
        "scale",
        "tool",
        "bin_index",
        "lower_bound",
        "upper_bound",
        "count",
    ]
    return csv_text(fields, rows)


def histogram_axis_label(value: float, unit: str) -> str:
    def compact(number: float) -> str:
        if number >= 10:
            return f"{number:.0f}"
        if number >= 1:
            return f"{number:.1f}".rstrip("0").rstrip(".")
        return f"{number:.2f}".rstrip("0").rstrip(".")

    if unit == "seconds":
        if value >= 60:
            return f"{compact(value / 60)} min"
        if value >= 1:
            return f"{compact(value)} s"
        if value >= 0.001:
            return f"{compact(value * 1000)} ms"
        return f"{compact(value * 1_000_000)} µs"
    if value >= 1024:
        gib = f"{value / 1024:.1f}".rstrip("0").rstrip(".")
        return f"{gib} GiB"
    return f"{compact(value)} MiB"


def histogram_svg(metric: dict[str, Any]) -> str:
    width = 1200
    height = 650
    left = 90
    right = 35
    top = 120
    bottom = 115
    plot_width = width - left - right
    plot_height = height - top - bottom
    edges = metric["edges"]
    bin_count = len(edges) - 1
    series = metric["series"]
    maximum = max(1, max(count for value in series for count in value["counts"]))
    colors = {"rocjitsu-dbt": "#ED1C24", "hotswap": "#3478B8"}
    labels = {"rocjitsu-dbt": "RocJITsu DBT", "hotswap": "HotSwap"}
    group_width = plot_width / bin_count
    bar_width = group_width * 0.38
    pieces = [
        '<?xml version="1.0" encoding="UTF-8"?>',
        (
            f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" '
            f'height="{height}" viewBox="0 0 {width} {height}" role="img">'
        ),
        f"<title>{html.escape(metric['title'])} histogram</title>",
        (
            "<desc>Log-scaled shared bins comparing RocJITsu DBT and HotSwap. "
            "All measured outcomes are included.</desc>"
        ),
        '<rect width="100%" height="100%" fill="white"/>',
        (
            '<g font-family="system-ui, sans-serif" fill="#1f2937" '
            'stroke-linecap="square">'
        ),
        (
            f'<text x="{width / 2}" y="38" text-anchor="middle" '
            f'font-size="25" font-weight="650">{html.escape(metric["title"])}'
            " distribution</text>"
        ),
        (
            f'<text x="{width / 2}" y="63" text-anchor="middle" '
            'font-size="14" fill="#5b6472">log-scaled shared bins; '
            "successful, failed, and timed-out processes</text>"
        ),
    ]
    tick_count = min(5, maximum)
    for tick_index in range(tick_count + 1):
        count = round(maximum * tick_index / tick_count)
        y = top + plot_height * (1 - count / maximum)
        pieces.extend(
            [
                (
                    f'<line x1="{left}" y1="{y:.2f}" x2="{width - right}" '
                    f'y2="{y:.2f}" stroke="#e5e7eb" stroke-width="1"/>'
                ),
                (
                    f'<text x="{left - 12}" y="{y + 5:.2f}" text-anchor="end" '
                    f'font-size="13">{count}</text>'
                ),
            ]
        )
    for series_index, value in enumerate(series):
        tool = value["tool"]
        for bin_index, count in enumerate(value["counts"]):
            if count == 0:
                continue
            x = (
                left
                + bin_index * group_width
                + group_width * 0.1
                + series_index * bar_width
            )
            bar_height = plot_height * count / maximum
            y = top + plot_height - bar_height
            tooltip = (
                f"{labels[tool]}: {count}; "
                f"{histogram_axis_label(edges[bin_index], metric['unit'])}–"
                f"{histogram_axis_label(edges[bin_index + 1], metric['unit'])}"
            )
            pieces.append(
                (
                    f'<rect x="{x:.2f}" y="{y:.2f}" width="{bar_width:.2f}" '
                    f'height="{bar_height:.2f}" fill="{colors[tool]}" '
                    'fill-opacity="0.82">'
                    f"<title>{html.escape(tooltip)}</title></rect>"
                )
            )
    pieces.extend(
        [
            (
                f'<line x1="{left}" y1="{top + plot_height}" '
                f'x2="{width - right}" y2="{top + plot_height}" '
                'stroke="#374151" stroke-width="1.5"/>'
            ),
            (
                f'<line x1="{left}" y1="{top}" x2="{left}" '
                f'y2="{top + plot_height}" stroke="#374151" stroke-width="1.5"/>'
            ),
        ]
    )
    x_tick_count = min(6, bin_count)
    for tick_index in range(x_tick_count + 1):
        edge_index = round(bin_count * tick_index / x_tick_count)
        x = left + plot_width * edge_index / bin_count
        label = histogram_axis_label(edges[edge_index], metric["unit"])
        pieces.extend(
            [
                (
                    f'<line x1="{x:.2f}" y1="{top + plot_height}" '
                    f'x2="{x:.2f}" y2="{top + plot_height + 7}" '
                    'stroke="#374151"/>'
                ),
                (
                    f'<text x="{x:.2f}" y="{top + plot_height + 27}" '
                    f'text-anchor="middle" font-size="12">{html.escape(label)}</text>'
                ),
            ]
        )
    pieces.extend(
        [
            (
                f'<text x="{width / 2}" y="{height - 35}" text-anchor="middle" '
                f'font-size="15">{html.escape(metric["title"])} '
                f'({html.escape(metric["unit"])}, log scale)</text>'
            ),
            (
                f'<text x="25" y="{top + plot_height / 2}" text-anchor="middle" '
                'font-size="15" transform="rotate(-90 25 '
                f'{top + plot_height / 2})">Process count</text>'
            ),
        ]
    )
    legend_x = 350
    for index, value in enumerate(series):
        tool = value["tool"]
        x = legend_x + index * 300
        failure_count = sum(
            count
            for status, count in value["status_counts"].items()
            if status != "success"
        )
        legend = f"{labels[tool]} (n={value['sample_count']}"
        if failure_count:
            legend += f", {failure_count} non-success"
        legend += ")"
        pieces.extend(
            [
                (
                    f'<rect x="{x}" y="82" width="16" height="16" '
                    f'fill="{colors[tool]}" fill-opacity="0.82"/>'
                ),
                (
                    f'<text x="{x + 23}" y="95" font-size="12">'
                    f"{html.escape(legend)}</text>"
                ),
            ]
        )
    pieces.extend(["</g>", "</svg>", ""])
    return "\n".join(pieces)


def terminal_histograms(histograms: dict[str, Any]) -> str:
    lines = [
        "Histogram scope: all measured outcomes (success, failure, and timeout)",
        "Shared logarithmic bins; bar lengths share one scale within each metric.",
        "",
    ]
    for metric in histograms["metrics"].values():
        edges = metric["edges"]
        by_tool = {series["tool"]: series for series in metric["series"]}
        dbt = by_tool["rocjitsu-dbt"]
        hotswap = by_tool["hotswap"]
        maximum = max(1, *dbt["counts"], *hotswap["counts"])
        width = 24

        def bar(count: int) -> str:
            if count == 0:
                return ""
            length = max(1, round(width * count / maximum))
            return "#" * length

        def statuses(series: dict[str, Any]) -> str:
            return ", ".join(
                f"{status}={count}" for status, count in series["status_counts"].items()
            )

        lines.extend(
            [
                metric["title"],
                (f"  RocJITsu DBT: n={dbt['sample_count']} " f"({statuses(dbt)})"),
                (
                    f"  HotSwap:      n={hotswap['sample_count']} "
                    f"({statuses(hotswap)})"
                ),
                "",
                (
                    f"{'Range':<25} {'DBT':>4} {'DBT distribution':<24} "
                    f"{'HS':>4} {'HotSwap distribution':<24}"
                ),
                f"{'-' * 25} {'-' * 4} {'-' * 24} {'-' * 4} {'-' * 24}",
            ]
        )
        for index, (dbt_count, hotswap_count) in enumerate(
            zip(dbt["counts"], hotswap["counts"])
        ):
            if dbt_count == 0 and hotswap_count == 0:
                continue
            interval = (
                f"{histogram_axis_label(edges[index], metric['unit'])}–"
                f"{histogram_axis_label(edges[index + 1], metric['unit'])}"
            )
            lines.append(
                f"{interval:<25} {dbt_count:>4} {bar(dbt_count):<24} "
                f"{hotswap_count:>4} {bar(hotswap_count):<24}"
            )
        lines.append("")
    return "\n".join(lines)


def write_histograms(
    output_directory: pathlib.Path, records: Sequence[dict[str, Any]]
) -> None:
    histograms = histogram_data(records)
    atomic_json(output_directory / "histograms.json", histograms)
    atomic_text(output_directory / "histograms.csv", histogram_csv_text(histograms))
    atomic_text(output_directory / "histograms.txt", terminal_histograms(histograms))
    filenames = {
        "cpu_time": "cpu_time_histogram.svg",
        "peak_rss": "peak_rss_histogram.svg",
    }
    for name, metric in histograms["metrics"].items():
        atomic_text(output_directory / filenames[name], histogram_svg(metric))


def format_seconds(value: float | None) -> str:
    return "n/a" if value is None else f"{value:.3f} s"


def format_rss(value: int | float | None) -> str:
    if value is None:
        return "n/a"
    mib = float(value) / 1024
    if mib >= 1024:
        return f"{mib / 1024:.2f} GiB"
    return f"{mib:.1f} MiB"


def format_ratio(value: float | None) -> str:
    return "n/a" if value is None else f"{value:.2f}x"


def markdown_summary(metadata: dict[str, Any], summary: dict[str, Any]) -> str:
    config = metadata["benchmark_config"]
    corpus = summary["corpus"]
    lines = [
        "# gfx1250 B0-to-A0 translator comparison",
        "",
        (
            f"- Inputs: {summary['input_count']} DeepSeek HSACO files; "
            f"{corpus['paired_inputs']} have complete paired measurements"
        ),
        (
            f"- Repetitions: {config['repetitions']} measured, "
            f"{config['warmups']} warmup"
        ),
        (
            "- HotSwap verbose decisions: "
            + ("enabled" if config["hotswap_verbose_logs"] else "disabled")
        ),
        (
            "- Translator phase timing: "
            + ("enabled" if config.get("phase_timing") else "disabled")
        ),
        (
            "- HotSwap phase profile mode: "
            + config.get("hotswap_phase_profile_mode", "detailed")
        ),
        (
            "- HotSwap timeout: "
            f"max({config['hotswap_timeout_min_seconds']:g} seconds, "
            f"{config['hotswap_timeout_multiplier']:g}x paired DBT elapsed time) "
            "(subject to the global cap)"
        ),
        "- CPU: user plus system time from Linux `wait4`",
        "- Peak RSS: maximum per-process `ru_maxrss` across all measured runs",
        "",
    ]
    if not summary["complete"]:
        lines.extend(
            [
                (
                    f"> Incomplete: {summary['successful_measurements']} of "
                    f"{summary['expected_measurements']} measurements succeeded."
                ),
                "",
            ]
        )
    lines.extend(
        [
            "| Metric | RocJITsu DBT | HotSwap | HotSwap / DBT |",
            "| --- | ---: | ---: | ---: |",
            (
                "| Corpus CPU (sum of input medians) | "
                f"{format_seconds(corpus['rocjitsu_dbt_cpu_seconds'])} | "
                f"{format_seconds(corpus['hotswap_cpu_seconds'])} | "
                f"{format_ratio(corpus['cpu_speedup_hotswap_over_dbt'])} |"
            ),
            (
                "| Peak RSS | "
                f"{format_rss(corpus['rocjitsu_dbt_peak_rss_kib'])} | "
                f"{format_rss(corpus['hotswap_peak_rss_kib'])} | "
                f"{format_ratio(corpus['rss_ratio_hotswap_over_dbt'])} |"
            ),
            "",
            (
                "Per-input geometric-mean CPU ratio (HotSwap / DBT): "
                f"{format_ratio(corpus['per_input_cpu_speedup_geomean'])}."
            ),
        ]
    )
    if corpus["dbt_rss_reduction_percent"] is not None:
        direction = "lower" if corpus["dbt_rss_reduction_percent"] >= 0 else "higher"
        lines.append(
            "RocJITsu DBT peak RSS is "
            f"{abs(corpus['dbt_rss_reduction_percent']):.1f}% {direction}."
        )
    if summary["failure_counts"]:
        lines.extend(["", "## Failures", "", "| Status | Count |", "| --- | ---: |"])
        lines.extend(
            f"| {status} | {count} |"
            for status, count in summary["failure_counts"].items()
        )
    lines.extend(
        [
            "",
            "## Histograms",
            "",
            (
                "Histograms use shared log-scaled bins and include all measured "
                "outcomes, including failures and timeouts."
            ),
            "",
            "![CPU time histogram](cpu_time_histogram.svg)",
            "",
            "![Peak RSS histogram](peak_rss_histogram.svg)",
            "",
            "## Translator phases",
            "",
            (
                "Comparable phase timing uses monotonic wall time. See "
                "`phase_breakdown.txt` for the terminal table, "
                "`phase_measurements.csv` for normalized buckets, and "
                "`phase_native.csv` for native timer rows."
            ),
            "",
            "## Commands",
            "",
            "RocJITsu DBT:",
            "",
            "```text",
            config["dbt_command_template"],
            "```",
            "",
            "HotSwap:",
            "",
            "```text",
            config["hotswap_command_template"],
            "```",
            "",
            "See `comparison.csv` for every input and `measurements.csv` for raw runs.",
            "",
        ]
    )
    return "\n".join(lines)


def render_results(
    output_directory: pathlib.Path, metadata: dict[str, Any]
) -> dict[str, Any]:
    records = read_jsonl(output_directory / "results.jsonl")
    summary = build_summary(metadata, records)
    summary["phase_timing"] = write_phase_reports(output_directory, metadata, records)
    write_histograms(output_directory, records)
    atomic_json(output_directory / "summary.json", summary)
    atomic_text(output_directory / "summary.md", markdown_summary(metadata, summary))
    write_measurements_csv(output_directory / "measurements.csv", records)
    write_comparison_csv(output_directory / "comparison.csv", summary)
    return summary


def prepare_run(
    args: argparse.Namespace,
    dbt: pathlib.Path,
    hotswap: pathlib.Path,
    library_dirs: Sequence[pathlib.Path],
    sources: Sequence[pathlib.Path],
) -> tuple[pathlib.Path, dict[str, Any], list[dict[str, Any]]]:
    output_directory = pathlib.Path(args.output_dir).expanduser().resolve()
    metadata_path = output_directory / "run.json"
    records_path = output_directory / "results.jsonl"

    input_identities = [file_identity(source) for source in sources]
    dbt_template = dbt_command(dbt, pathlib.Path("INPUT.hsaco"))
    hotswap_template = hotswap_command(
        hotswap,
        pathlib.Path("INPUT.hsaco"),
        pathlib.Path("OUTPUT.co"),
        args.hotswap_source_isa,
        args.hotswap_target_isa,
    )
    config = {
        "inputs": input_identities,
        "repetitions": args.repetitions,
        "warmups": args.warmups,
        "timeout_seconds": args.timeout_seconds,
        "hotswap_timeout_multiplier": args.hotswap_timeout_multiplier,
        "hotswap_timeout_min_seconds": args.hotswap_timeout_min_seconds,
        "seed": args.seed,
        "include_glob": args.include_glob,
        "keep_outputs": args.keep_outputs,
        "phase_timing": args.phase_timing,
        "hotswap_phase_profile_mode": args.hotswap_phase_profile_mode,
        "hotswap_verbose_logs": args.hotswap_verbose_logs,
        "hotswap_source_isa": args.hotswap_source_isa,
        "hotswap_target_isa": args.hotswap_target_isa,
        "hotswap_library_dirs": [str(path) for path in library_dirs],
        "rocjitsu_dbt": file_identity(dbt),
        "hotswap": file_identity(hotswap),
        "dbt_command_template": shell_command(dbt_template),
        "hotswap_command_template": shell_command(hotswap_template),
    }
    digest = hashlib.sha256(
        json.dumps(config, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()

    if args.resume:
        if not metadata_path.is_file():
            raise ValueError(f"resume metadata does not exist: {metadata_path}")
        metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
        if metadata.get("config_sha256") != digest:
            raise ValueError("resume configuration does not match run.json")
        return output_directory, metadata, read_jsonl(records_path)

    if output_directory.exists() and any(output_directory.iterdir()):
        raise ValueError(
            f"output directory is not empty: {output_directory}; use --resume"
        )
    output_directory.mkdir(parents=True, exist_ok=True)
    metadata = {
        "schema_version": SCHEMA_VERSION,
        "type": "run",
        "created_at": utc_now(),
        "config_sha256": digest,
        "benchmark_config": config,
        "runner": file_identity(pathlib.Path(__file__)),
        "source_revision": source_revision(pathlib.Path(__file__)),
        "host": {
            "platform": platform.platform(),
            "uname": list(platform.uname()),
            "python": sys.version,
        },
    }
    atomic_json(metadata_path, metadata)
    return output_directory, metadata, []


def benchmark(args: argparse.Namespace) -> int:
    if args.render_only and not args.resume:
        raise ValueError("--render-only requires --resume")
    dbt = resolve_executable(args.dbt, "dbt")
    hotswap = resolve_executable(args.hotswap, "hotswap")
    library_dirs = resolve_library_dirs(args.hotswap_library_dir, hotswap)
    raw_inputs = list(args.inputs)
    if not raw_inputs:
        raw_inputs = [
            os.environ.get("DEEPSEEK_HSACO_DIR", "~/rocjitsu/dsv4-flash-code-objs")
        ]
    sources = discover_inputs(raw_inputs, args.recursive, args.include_glob)
    if not sources:
        raise ValueError("no .hsaco files found")
    random.Random(args.seed).shuffle(sources)
    if args.limit:
        sources = sources[: args.limit]

    output_directory, metadata, prior_records = prepare_run(
        args, dbt, hotswap, library_dirs, sources
    )
    if args.render_only:
        summary = render_results(output_directory, metadata)
        if not args.quiet and metadata["benchmark_config"].get("phase_timing"):
            print((output_directory / "phase_breakdown.txt").read_text())
        print(output_directory / "summary.md")
        return 0 if summary["complete"] else 1
    records_path = output_directory / "results.jsonl"
    latest = latest_measurements(prior_records)
    attempts = Counter(
        measurement_key(record)
        for record in prior_records
        if record.get("type") == "measurement"
    )
    total = len(sources) * (args.warmups + args.repetitions) * len(TOOLS)
    completed = 0
    interrupted = False

    try:
        for source_path in sources:
            source = next(
                value
                for value in metadata["benchmark_config"]["inputs"]
                if value["path"] == str(source_path)
            )
            rounds = [
                ("warmup", repetition) for repetition in range(1, args.warmups + 1)
            ]
            rounds.extend(
                ("measurement", repetition)
                for repetition in range(1, args.repetitions + 1)
            )
            for phase, repetition in rounds:
                for order_in_pair, tool in enumerate(TOOLS, 1):
                    completed += 1
                    key = (str(source_path), tool, phase, repetition)
                    previous = latest.get(key)
                    if previous is not None and previous["status"] == "success":
                        if not args.quiet:
                            print(
                                f"[{completed}/{total}] {tool} "
                                f"{source_path.name} ({phase} {repetition}) resume"
                            )
                        continue
                    timeout_seconds = args.timeout_seconds
                    timeout_basis = "global_cap"
                    if tool == "hotswap":
                        dbt_key = (
                            str(source_path),
                            "rocjitsu-dbt",
                            phase,
                            repetition,
                        )
                        dbt_record = latest.get(dbt_key)
                        if (
                            dbt_record is not None
                            and dbt_record["status"] == "success"
                            and dbt_record.get("elapsed_seconds") is not None
                        ):
                            multiplied_timeout = (
                                args.hotswap_timeout_multiplier
                                * float(dbt_record["elapsed_seconds"])
                            )
                            candidate_timeout = max(
                                args.hotswap_timeout_min_seconds,
                                multiplied_timeout,
                            )
                            if (
                                args.timeout_seconds > 0
                                and args.timeout_seconds < candidate_timeout
                            ):
                                timeout_seconds = args.timeout_seconds
                            else:
                                timeout_seconds = candidate_timeout
                                timeout_basis = (
                                    "hotswap_minimum"
                                    if args.hotswap_timeout_min_seconds
                                    >= multiplied_timeout
                                    else "rocjitsu_dbt_elapsed_multiplier"
                                )
                        else:
                            timeout_basis = "global_cap_without_dbt_baseline"
                    attempt = attempts[key] + 1
                    record = run_measurement(
                        tool=tool,
                        source=source,
                        phase=phase,
                        repetition=repetition,
                        attempt=attempt,
                        order_in_pair=order_in_pair,
                        timeout_seconds=timeout_seconds,
                        timeout_basis=timeout_basis,
                        args=args,
                        dbt=dbt,
                        hotswap=hotswap,
                        library_dirs=library_dirs,
                        output_directory=output_directory,
                    )
                    append_jsonl(records_path, record)
                    attempts[key] = attempt
                    latest[key] = record
                    if not args.quiet:
                        cpu = record.get("cpu_seconds")
                        rss = record.get("max_rss_kib")
                        cpu_text = "n/a" if cpu is None else f"{cpu:.4f}s"
                        rss_text = "n/a" if rss is None else f"{rss / 1024:.1f}MiB"
                        print(
                            f"[{completed}/{total}] {tool} "
                            f"{source_path.name} ({phase} {repetition}) "
                            f"{record['status']} cpu={cpu_text} rss={rss_text}",
                            flush=True,
                        )
    except KeyboardInterrupt:
        interrupted = True
        print("interrupted; writing partial summaries", file=sys.stderr)

    summary = render_results(output_directory, metadata)
    if not args.quiet and metadata["benchmark_config"].get("phase_timing"):
        print((output_directory / "phase_breakdown.txt").read_text())
    print(output_directory / "summary.md")
    if interrupted:
        return 130
    return 0 if summary["complete"] else 1


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    result.add_argument(
        "inputs",
        nargs="*",
        help=(
            ".hsaco files or directories; defaults to $DEEPSEEK_HSACO_DIR or "
            "~/rocjitsu/dsv4-flash-code-objs"
        ),
    )
    result.add_argument("--output-dir", required=True)
    result.add_argument(
        "--dbt",
        default=default_dbt(),
        help="rj_dbt_translate executable (default: auto-detected)",
    )
    result.add_argument(
        "--hotswap",
        default=default_hotswap(),
        help="hotswap-rewrite executable (default: auto-detected)",
    )
    result.add_argument(
        "--hotswap-library-dir",
        action="append",
        default=[],
        help="prepend a directory to HotSwap LD_LIBRARY_PATH; repeatable",
    )
    result.add_argument("--hotswap-source-isa", default=DEFAULT_SOURCE_ISA)
    result.add_argument("--hotswap-target-isa", default=DEFAULT_TARGET_ISA)
    result.add_argument(
        "--hotswap-verbose-logs",
        action="store_true",
        help="set AMD_COMGR_EMIT_VERBOSE_LOGS=1 (changes measured log-I/O cost)",
    )
    result.add_argument(
        "--phase-timing",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="collect comparable translator phase timers (default: true)",
    )
    result.add_argument(
        "--hotswap-phase-profile-mode",
        choices=("coarse", "detailed"),
        default="coarse",
        help=(
            "HotSwap timer detail; coarse avoids per-instruction strategy timer "
            "overhead (default: coarse)"
        ),
    )
    result.add_argument(
        "--repetitions",
        type=positive_int,
        default=1,
        help="measured fresh-process runs per input and tool (default: 1)",
    )
    result.add_argument(
        "--warmups",
        type=nonnegative_int,
        default=0,
        help="unreported warmup pairs per input (default: 0)",
    )
    result.add_argument(
        "--timeout-seconds",
        type=nonnegative_float,
        default=900.0,
        help=(
            "DBT timeout and global HotSwap cap; zero disables the global cap "
            "(default: 900)"
        ),
    )
    result.add_argument(
        "--hotswap-timeout-multiplier",
        type=positive_float,
        default=4.0,
        help="bound HotSwap to this multiple of paired DBT elapsed time (default: 4)",
    )
    result.add_argument(
        "--hotswap-timeout-min-seconds",
        type=nonnegative_float,
        default=60.0,
        help="minimum HotSwap timeout before applying the global cap (default: 60)",
    )
    result.add_argument(
        "--recursive",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="recurse into input directories (default: true)",
    )
    result.add_argument(
        "--include-glob",
        default="dsv4-flash-original-*.hsaco",
        help=(
            "filename glob within directories " "(default: dsv4-flash-original-*.hsaco)"
        ),
    )
    result.add_argument("--seed", type=int, default=1250)
    result.add_argument(
        "--limit",
        type=nonnegative_int,
        default=0,
        help="benchmark a seeded subset; zero selects all inputs",
    )
    result.add_argument(
        "--keep-outputs",
        action="store_true",
        help="retain measured rewritten code objects",
    )
    result.add_argument(
        "--resume",
        action="store_true",
        help="resume an exactly matching run and retry failures",
    )
    result.add_argument(
        "--render-only",
        action="store_true",
        help="with --resume, regenerate summaries and histograms without translation",
    )
    result.add_argument("--quiet", action="store_true")
    result.set_defaults(function=benchmark)
    return result


def main(argv: Sequence[str] | None = None) -> int:
    args = parser().parse_args(argv)
    try:
        return int(args.function(args))
    except (OSError, ValueError, json.JSONDecodeError) as error:
        print(f"error: {error}", file=sys.stderr)
        return 2


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