"""Local worker benchmarking utilities."""

from __future__ import annotations

import json
import os
import platform
import re
import resource
import shutil
import statistics
import subprocess
import sys
import tempfile
import time
import wave
from collections.abc import Iterator
from contextlib import ExitStack, contextmanager
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from importlib import metadata
from importlib.resources import as_file, files
from pathlib import Path
from typing import Any

from hive_compute.handlers.whisper_handler import WhisperHandler
from hive_compute.worker.idle_detector import get_hardware_info

DEFAULT_BENCHMARK_MODELS = ("large-v3-turbo", "large-v3")
DEFAULT_BENCHMARK_OUTPUT_DIR = Path("~/.ambient/benchmarks").expanduser()
BUNDLED_BENCHMARK_FIXTURES = (
    ("short", "benchmark_audio_short.wav"),
    ("medium", "benchmark_audio_medium.wav"),
    ("full", "benchmark_audio.wav"),
)
DEFAULT_WHISPER_BENCHMARK_CASE = "mlx-large-v3-turbo-fp16"


@dataclass(frozen=True)
class BenchmarkFixture:
    name: str
    audio_path: str
    audio_seconds: float


@dataclass
class BenchmarkModelSummary:
    model: str
    runs: int
    failed_runs: int
    total_audio_seconds: float
    total_elapsed_seconds: float | None
    average_realtime_factor: float | None


@dataclass
class BenchmarkResult:
    fixture: str
    audio_path: str
    model: str
    backend: str
    language: str
    audio_seconds: float
    elapsed_seconds: float | None
    realtime_factor: float | None
    memory_rss_mb: float | None
    time_to_first_token_seconds: float | None
    word_count: int | None
    transcript_preview: str | None
    error: str | None = None


@dataclass(frozen=True)
class BenchmarkCase:
    name: str
    backend: str
    model: str
    quantization: str
    repo: str | None = None
    notes: str = ""


@dataclass
class WhisperBenchmarkResult:
    case: str
    backend: str
    model: str
    quantization: str
    repo: str | None
    status: str
    audio_duration_seconds: float | None = None
    cold_run_seconds: float | None = None
    warm_run_seconds: float | None = None
    warm_realtime_factor: float | None = None
    time_to_first_token_seconds: float | None = None
    word_error_rate: float | None = None
    transcript: str = ""
    notes: str = ""
    peak_rss_mb: float | None = None


@dataclass(frozen=True)
class WhisperBenchmarkPolicySummary:
    machine_class: str
    completed_policy_cases: tuple[str, ...]
    missing_policy_cases: tuple[str, ...]
    whisper_cpp_cases: tuple[str, ...]
    pareto_cases: tuple[str, ...]
    recommended_case: str | None
    decision: str
    reason: str


BUILT_IN_CASES: dict[str, BenchmarkCase] = {
    "mlx-large-v3-fp16": BenchmarkCase(
        name="mlx-large-v3-fp16",
        backend="mlx-whisper",
        model="large-v3",
        quantization="fp16",
        repo="mlx-community/whisper-large-v3-mlx",
    ),
    "mlx-large-v3-int8": BenchmarkCase(
        name="mlx-large-v3-int8",
        backend="mlx-whisper",
        model="large-v3",
        quantization="int8",
        repo="LibraxisAI/whisper-large-v3-mlx-q8",
    ),
    "mlx-large-v3-int4": BenchmarkCase(
        name="mlx-large-v3-int4",
        backend="mlx-whisper",
        model="large-v3",
        quantization="int4",
        repo="LibraxisAI/whisper-large-v3-q4",
    ),
    "mlx-large-v3-turbo-fp16": BenchmarkCase(
        name="mlx-large-v3-turbo-fp16",
        backend="mlx-whisper",
        model="large-v3-turbo",
        quantization="fp16",
        repo="mlx-community/whisper-large-v3-turbo",
    ),
    "mlx-large-v3-turbo-int8": BenchmarkCase(
        name="mlx-large-v3-turbo-int8",
        backend="mlx-whisper",
        model="large-v3-turbo",
        quantization="int8",
        repo="LibraxisAI/whisper-large-v3-turbo-mlx-q8",
    ),
    "mlx-large-v3-turbo-int4": BenchmarkCase(
        name="mlx-large-v3-turbo-int4",
        backend="mlx-whisper",
        model="large-v3-turbo",
        quantization="int4",
        repo="mlx-community/whisper-large-v3-turbo-q4",
    ),
    "mlx-distil-large-v3-fp16": BenchmarkCase(
        name="mlx-distil-large-v3-fp16",
        backend="mlx-whisper",
        model="distil-large-v3",
        quantization="fp16",
        repo="mlx-community/distil-whisper-large-v3",
        notes="No vetted INT8/INT4 MLX repo was found for distil-large-v3.",
    ),
}

POLICY_MATRIX_CASES = (
    "mlx-large-v3-fp16",
    "mlx-large-v3-int8",
    "mlx-large-v3-int4",
    "mlx-large-v3-turbo-fp16",
    "mlx-large-v3-turbo-int8",
    "mlx-large-v3-turbo-int4",
    "mlx-distil-large-v3-fp16",
)

CASE_GROUPS: dict[str, tuple[str, ...]] = {
    "policy-matrix": POLICY_MATRIX_CASES,
    "mlx-policy": POLICY_MATRIX_CASES,
}


@contextmanager
def bundled_benchmark_fixtures() -> Iterator[list[BenchmarkFixture]]:
    fixtures: list[BenchmarkFixture] = []
    with ExitStack() as stack:
        for name, resource_name in BUNDLED_BENCHMARK_FIXTURES:
            resource = files("hive_compute.data").joinpath(resource_name)
            path = Path(stack.enter_context(as_file(resource)))
            fixtures.append(_build_benchmark_fixture(name=name, audio_path=path))
        yield fixtures


def resolve_benchmark_models(transcribe_config: dict[str, Any], models: str | None = None) -> list[str]:
    if models:
        parsed = [item.strip() for item in models.split(",") if item.strip()]
        if not parsed:
            raise ValueError("At least one benchmark model must be specified.")
        return list(dict.fromkeys(parsed))

    ordered = [transcribe_config.get("model", "large-v3-turbo"), *DEFAULT_BENCHMARK_MODELS]
    return [model for model in dict.fromkeys(ordered) if model]


def resolve_benchmark_fixtures(audio_path: str | Path | None = None) -> list[BenchmarkFixture]:
    if audio_path is None:
        raise ValueError("audio_path must be provided when benchmark fixtures are not supplied.")

    path = Path(audio_path).expanduser()
    return [_build_benchmark_fixture(name=path.stem, audio_path=path)]


def run_benchmark_suite(*args, **kwargs):
    if "reference_text" in kwargs or "cases" in kwargs:
        return run_whisper_benchmark_suite(**kwargs)
    return _run_local_benchmark_suite(*args, **kwargs)


def _run_local_benchmark_suite(
    transcribe_config: dict[str, Any],
    audio_path: str | Path | None,
    models: list[str],
    output_dir: str | Path | None = None,
    handler_factory=WhisperHandler,
    fixtures: list[BenchmarkFixture] | None = None,
) -> dict[str, Any]:
    if fixtures is None and audio_path is None:
        with bundled_benchmark_fixtures() as bundled:
            return _run_local_benchmark_suite(
                transcribe_config,
                audio_path=None,
                models=models,
                output_dir=output_dir,
                handler_factory=handler_factory,
                fixtures=bundled,
            )

    selected_fixtures = fixtures or resolve_benchmark_fixtures(audio_path)
    if not selected_fixtures:
        raise ValueError("At least one benchmark fixture must be specified.")

    output_root = Path(output_dir).expanduser() if output_dir else DEFAULT_BENCHMARK_OUTPUT_DIR
    output_root.mkdir(parents=True, exist_ok=True)

    created_at = datetime.now(UTC)
    machine = _normalize_machine_metadata(
        {
            "hostname": platform.node(),
            "platform": platform.platform(),
            **get_hardware_info(),
        }
    )
    environment = build_benchmark_environment(machine=machine, backend=transcribe_config.get("backend", "mlx-whisper"))

    results: list[BenchmarkResult] = []
    for fixture in selected_fixtures:
        fixture_path = Path(fixture.audio_path)
        for model in models:
            config = dict(transcribe_config)
            config["model"] = model
            handler = handler_factory(config)
            rss_before = _current_rss_mb()
            started_at = time.perf_counter()
            try:
                transcription = handler._transcribe(fixture_path)
            except Exception as exc:
                elapsed = time.perf_counter() - started_at
                results.append(
                    BenchmarkResult(
                        fixture=fixture.name,
                        audio_path=str(fixture_path),
                        model=model,
                        backend=config.get("backend", "mlx-whisper"),
                        language=config.get("language", "en"),
                        audio_seconds=fixture.audio_seconds,
                        elapsed_seconds=round(elapsed, 3),
                        realtime_factor=_compute_realtime_factor(fixture.audio_seconds, elapsed),
                        memory_rss_mb=_rss_delta_mb(rss_before, _current_rss_mb()),
                        time_to_first_token_seconds=None,
                        word_count=None,
                        transcript_preview=None,
                        error=str(exc),
                    )
                )
                continue

            elapsed = time.perf_counter() - started_at
            transcript = transcription.get("transcript", "")
            results.append(
                BenchmarkResult(
                    fixture=fixture.name,
                    audio_path=str(fixture_path),
                    model=model,
                    backend=transcription.get("backend", config.get("backend", "mlx-whisper")),
                    language=transcription.get("language", config.get("language", "en")),
                    audio_seconds=fixture.audio_seconds,
                    elapsed_seconds=round(elapsed, 3),
                    realtime_factor=_compute_realtime_factor(fixture.audio_seconds, elapsed),
                    memory_rss_mb=_rss_delta_mb(rss_before, _current_rss_mb()),
                    time_to_first_token_seconds=None,
                    word_count=transcription.get("word_count"),
                    transcript_preview=(transcript[:80] if transcript else None),
                )
            )

    summaries = summarize_benchmark_results(results)
    stamp = created_at.strftime("%Y%m%d-%H%M%S")
    json_path = output_root / f"benchmark-results-{stamp}.json"
    md_path = output_root / f"benchmark-results-{stamp}.md"
    markdown = render_benchmark_markdown(
        created_at=created_at,
        fixtures=selected_fixtures,
        machine=machine,
        environment=environment,
        results=results,
        summaries=summaries,
    )
    primary_audio_path = selected_fixtures[0].audio_path if len(selected_fixtures) == 1 else None

    payload = {
        "created_at": created_at.isoformat(),
        "audio_path": primary_audio_path,
        "fixtures": [asdict(fixture) for fixture in selected_fixtures],
        "machine": machine,
        "environment": environment,
        "results": [asdict(result) for result in results],
        "summaries": [asdict(summary) for summary in summaries],
        "markdown_path": str(md_path),
        "json_path": str(json_path),
    }
    json_path.write_text(json.dumps(payload, indent=2))
    md_path.write_text(markdown)

    return {
        "created_at": payload["created_at"],
        "audio_path": payload["audio_path"],
        "fixtures": payload["fixtures"],
        "machine": machine,
        "environment": environment,
        "results": payload["results"],
        "summaries": payload["summaries"],
        "markdown": markdown,
        "markdown_path": str(md_path),
        "json_path": str(json_path),
    }


def render_benchmark_markdown(
    *,
    created_at: datetime,
    fixtures: list[BenchmarkFixture],
    machine: dict[str, Any],
    environment: dict[str, Any],
    results: list[BenchmarkResult],
    summaries: list[BenchmarkModelSummary] | None = None,
) -> str:
    rendered_summaries = summaries or summarize_benchmark_results(results)
    lines = [
        "# HiveCompute Benchmark",
        "",
        f"- Generated: {created_at.isoformat()}",
        f"- Machine: `{machine.get('chip', 'unknown chip')}`, {machine.get('memory_gb', 'unknown')} GB RAM",
        f"- Fixtures: {len(fixtures)}",
        f"- Machine Class: `{environment.get('machine_class', 'unknown')}`",
        f"- Runtime: `python {environment.get('python_version', 'unknown')}`, "
        f"`ambient {environment.get('hive_compute_version', 'unknown')}`, "
        f"`{environment.get('backend_package', 'backend')} {environment.get('backend_package_version', 'unknown')}`",
    ]
    if len(fixtures) == 1:
        lines.append(f"- Audio: `{fixtures[0].audio_path}`")

    lines.extend(
        [
            "",
            "## Fixtures",
            "",
            "| Fixture | Audio (s) | Path |",
            "| --- | ---: | --- |",
        ]
    )
    for fixture in fixtures:
        lines.append(f"| {fixture.name} | {_format_float(fixture.audio_seconds)} | `{fixture.audio_path}` |")

    lines.extend(
        [
            "",
            "## Summary by Model",
            "",
            "| Model | Runs | Failures | Audio (s) | Time (s) | Avg RTx |",
            "| --- | ---: | ---: | ---: | ---: | ---: |",
        ]
    )
    for summary in rendered_summaries:
        lines.append(
            f"| {summary.model} | {summary.runs} | {summary.failed_runs} | "
            f"{_format_float(summary.total_audio_seconds)} | "
            f"{_format_float(summary.total_elapsed_seconds)} | "
            f"{_format_float(summary.average_realtime_factor)} |"
        )

    lines.extend(
        [
            "",
            "## Detailed Results",
            "",
            "| Fixture | Model | Backend | Time (s) | RTx | RSS Δ (MB) | TTFT (s) | Status |",
            "| --- | --- | --- | ---: | ---: | ---: | ---: | --- |",
        ]
    )
    for result in results:
        status = result.error or "ok"
        lines.append(
            f"| {result.fixture} | {result.model} | {result.backend} | "
            f"{_format_float(result.elapsed_seconds)} | {_format_float(result.realtime_factor)} | "
            f"{_format_float(result.memory_rss_mb)} | {_format_float(result.time_to_first_token_seconds)} | "
            f"{status} |"
        )
    return "\n".join(lines) + "\n"


def summarize_benchmark_results(results: list[BenchmarkResult]) -> list[BenchmarkModelSummary]:
    grouped: dict[str, list[BenchmarkResult]] = {}
    for result in results:
        grouped.setdefault(result.model, []).append(result)

    summaries: list[BenchmarkModelSummary] = []
    for model, model_results in grouped.items():
        elapsed_values = [value for value in (item.elapsed_seconds for item in model_results) if value is not None]
        realtime_values = [value for value in (item.realtime_factor for item in model_results) if value is not None]
        summaries.append(
            BenchmarkModelSummary(
                model=model,
                runs=len(model_results),
                failed_runs=sum(1 for item in model_results if item.error),
                total_audio_seconds=round(sum(item.audio_seconds for item in model_results), 3),
                total_elapsed_seconds=round(sum(elapsed_values), 3) if elapsed_values else None,
                average_realtime_factor=(
                    round(sum(realtime_values) / len(realtime_values), 2) if realtime_values else None
                ),
            )
        )
    return summaries


def _read_wav_duration_seconds(path: Path) -> float:
    with wave.open(str(path), "rb") as audio_file:
        frames = audio_file.getnframes()
        rate = audio_file.getframerate()
    if rate <= 0:
        raise ValueError(f"Invalid WAV sample rate for benchmark audio: {path}")
    return round(frames / rate, 3)


def _build_benchmark_fixture(name: str, audio_path: Path) -> BenchmarkFixture:
    path = audio_path.expanduser()
    if not path.exists():
        raise FileNotFoundError(f"Benchmark audio not found: {path}")
    return BenchmarkFixture(name=name, audio_path=str(path), audio_seconds=_read_wav_duration_seconds(path))


def _compute_realtime_factor(audio_seconds: float, elapsed_seconds: float) -> float | None:
    if elapsed_seconds <= 0:
        return None
    return round(audio_seconds / elapsed_seconds, 2)


def _rss_delta_mb(before: float | None, after: float | None) -> float | None:
    if before is None or after is None:
        return None
    return round(max(after - before, 0.0), 1)


def _current_rss_mb() -> float | None:
    try:
        completed = subprocess.run(
            ["ps", "-o", "rss=", "-p", str(os.getpid())],
            check=True,
            capture_output=True,
            text=True,
        )
    except Exception:
        return None

    value = completed.stdout.strip()
    if not value:
        return None
    return round(int(value) / 1024, 1)


def build_benchmark_environment(*, machine: dict[str, Any], backend: str) -> dict[str, Any]:
    machine = _normalize_machine_metadata(machine)
    backend_package = _backend_package_name(backend)
    return {
        "normalization_version": 1,
        "machine_class": _machine_class(machine),
        "python_version": platform.python_version(),
        "hive_compute_version": _package_version("ambient"),
        "backend": backend,
        "backend_package": backend_package,
        "backend_package_version": _package_version(backend_package),
    }


def _normalize_machine_metadata(machine: dict[str, Any]) -> dict[str, Any]:
    normalized = dict(machine)
    if not normalized.get("chip"):
        normalized["chip"] = platform.machine() or platform.processor() or "unknown"
    if not isinstance(normalized.get("memory_gb"), int | float):
        normalized_memory = _fallback_memory_gb()
        if normalized_memory is not None:
            normalized["memory_gb"] = normalized_memory
    return normalized


def _machine_class(machine: dict[str, Any]) -> str:
    chip = str(machine.get("chip") or "unknown").strip().lower()
    memory = machine.get("memory_gb")
    slug = "-".join(part for part in chip.replace("/", " ").split() if part)
    if not slug:
        slug = "unknown"
    memory_suffix = f"-{int(memory)}gb" if isinstance(memory, int | float) else ""
    return f"{slug}{memory_suffix}"


def _fallback_memory_gb() -> int | None:
    try:
        page_size = os.sysconf("SC_PAGE_SIZE")
        physical_pages = os.sysconf("SC_PHYS_PAGES")
    except (AttributeError, OSError, ValueError):
        return None

    if not isinstance(page_size, int) or not isinstance(physical_pages, int):
        return None
    if page_size <= 0 or physical_pages <= 0:
        return None
    return int((page_size * physical_pages) // (1024**3))


def _backend_package_name(backend: str) -> str:
    if backend == "mlx-whisper":
        return "mlx-whisper"
    if backend == "parakeet-mlx":
        return "parakeet-mlx"
    return backend


def _package_version(name: str) -> str:
    try:
        return metadata.version(name)
    except Exception:
        return "unknown"


def _format_float(value: float | None) -> str:
    if value is None:
        return "n/a"
    if value == int(value):
        return f"{int(value)}"
    return f"{value:.2f}"


def normalize_transcript(text: str) -> list[str]:
    normalized = re.sub(r"[^a-z0-9']+", " ", text.lower()).strip()
    return [token for token in normalized.split() if token]


def word_error_rate(reference: str, hypothesis: str) -> float:
    ref_tokens = normalize_transcript(reference)
    hyp_tokens = normalize_transcript(hypothesis)
    if not ref_tokens:
        return 0.0 if not hyp_tokens else 1.0

    rows = len(ref_tokens) + 1
    cols = len(hyp_tokens) + 1
    dp = [[0] * cols for _ in range(rows)]
    for i in range(rows):
        dp[i][0] = i
    for j in range(cols):
        dp[0][j] = j

    for i in range(1, rows):
        for j in range(1, cols):
            cost = 0 if ref_tokens[i - 1] == hyp_tokens[j - 1] else 1
            dp[i][j] = min(
                dp[i - 1][j] + 1,
                dp[i][j - 1] + 1,
                dp[i - 1][j - 1] + cost,
            )

    return dp[-1][-1] / len(ref_tokens)


def detect_hardware_profile() -> dict[str, str]:
    profile = {
        "platform": platform.platform(),
        "machine": platform.machine(),
        "processor": platform.processor() or "unknown",
    }
    if platform.system() != "Darwin":
        return profile

    try:
        result = subprocess.run(
            ["system_profiler", "SPHardwareDataType"],
            capture_output=True,
            check=True,
            text=True,
        )
    except Exception:
        return profile

    output = result.stdout
    chip = re.search(r"Chip:\s*(.+)", output)
    memory = re.search(r"Memory:\s*(.+)", output)
    model = re.search(r"Model Identifier:\s*(.+)", output)
    if chip:
        profile["chip"] = chip.group(1).strip()
    if memory:
        profile["memory"] = memory.group(1).strip()
    if model:
        profile["model_identifier"] = model.group(1).strip()
    return profile


def probe_audio_duration_seconds(audio_path: Path) -> float:
    ffprobe = shutil.which("ffprobe")
    if ffprobe:
        try:
            result = subprocess.run(
                [
                    ffprobe,
                    "-v",
                    "error",
                    "-show_entries",
                    "format=duration",
                    "-of",
                    "default=noprint_wrappers=1:nokey=1",
                    str(audio_path),
                ],
                capture_output=True,
                check=True,
                text=True,
            )
            return float(result.stdout.strip())
        except Exception:
            if audio_path.suffix.lower() != ".wav":
                raise

    if audio_path.suffix.lower() == ".wav":
        return _read_wav_duration_seconds(audio_path)
    raise RuntimeError("ffprobe is required to measure non-WAV benchmark audio duration")


def parse_case_names(
    requested_cases: list[str],
    *,
    all_builtins: bool,
    repo_cases: list[str],
    cpp_cases: list[str],
) -> list[BenchmarkCase]:
    selected: list[BenchmarkCase] = []
    seen_case_names: set[str] = set()
    names = requested_cases or (sorted(BUILT_IN_CASES) if all_builtins else [DEFAULT_WHISPER_BENCHMARK_CASE])
    expanded_names: list[str] = []
    for name in names:
        expanded_names.extend(CASE_GROUPS.get(name, (name,)))

    for name in expanded_names:
        case = BUILT_IN_CASES.get(name)
        if case is None:
            available = sorted((*BUILT_IN_CASES, *CASE_GROUPS))
            raise ValueError(f"Unknown benchmark case '{name}'. Available cases: {', '.join(available)}")
        if case.name in seen_case_names:
            continue
        seen_case_names.add(case.name)
        selected.append(case)

    selected.extend(_parse_repo_cases(repo_cases))
    selected.extend(_parse_cpp_cases(cpp_cases))
    return selected


def _parse_repo_cases(entries: list[str]) -> list[BenchmarkCase]:
    cases: list[BenchmarkCase] = []
    for entry in entries:
        name, repo = _split_case_entry(entry)
        cases.append(
            BenchmarkCase(
                name=name,
                backend="mlx-whisper",
                model=name,
                quantization="custom",
                repo=repo,
                notes="Custom MLX repo override",
            )
        )
    return cases


def _parse_cpp_cases(entries: list[str]) -> list[BenchmarkCase]:
    cases: list[BenchmarkCase] = []
    for entry in entries:
        name, model_path = _split_case_entry(entry)
        cases.append(
            BenchmarkCase(
                name=name,
                backend="whisper-cpp",
                model=model_path,
                quantization="custom",
                notes="Custom whisper.cpp model path",
            )
        )
    return cases


def _split_case_entry(entry: str) -> tuple[str, str]:
    if "=" not in entry:
        raise ValueError(f"Custom case '{entry}' must use NAME=VALUE format")
    name, value = entry.split("=", 1)
    name = name.strip()
    value = value.strip()
    if not name or not value:
        raise ValueError(f"Custom case '{entry}' must use NAME=VALUE format")
    return name, value


def run_whisper_benchmark_suite(
    *,
    audio_path: Path,
    reference_text: str,
    language: str,
    repeats: int,
    cases: list[BenchmarkCase],
) -> tuple[dict[str, str], list[WhisperBenchmarkResult]]:
    hardware = detect_hardware_profile()
    results = [
        run_whisper_benchmark_case(
            case,
            audio_path=audio_path,
            reference_text=reference_text,
            language=language,
            repeats=repeats,
        )
        for case in cases
    ]
    return hardware, results


def run_whisper_benchmark_case(
    case: BenchmarkCase,
    *,
    audio_path: Path,
    reference_text: str,
    language: str,
    repeats: int,
) -> WhisperBenchmarkResult:
    if repeats < 1:
        raise ValueError("repeats must be at least 1")

    try:
        audio_duration = probe_audio_duration_seconds(audio_path)
    except Exception as exc:
        return WhisperBenchmarkResult(
            case=case.name,
            backend=case.backend,
            model=case.model,
            quantization=case.quantization,
            repo=case.repo,
            status="error",
            notes=f"Failed to measure audio duration: {exc}",
        )

    if case.backend == "mlx-whisper":
        return _run_mlx_case(
            case,
            audio_path=audio_path,
            audio_duration=audio_duration,
            reference_text=reference_text,
            language=language,
            repeats=repeats,
        )
    if case.backend == "whisper-cpp":
        return _run_cpp_case(
            case,
            audio_path=audio_path,
            audio_duration=audio_duration,
            reference_text=reference_text,
            language=language,
            repeats=repeats,
        )
    return WhisperBenchmarkResult(
        case=case.name,
        backend=case.backend,
        model=case.model,
        quantization=case.quantization,
        repo=case.repo,
        status="error",
        notes=f"Unsupported backend '{case.backend}'",
    )


def _run_mlx_case(
    case: BenchmarkCase,
    *,
    audio_path: Path,
    audio_duration: float,
    reference_text: str,
    language: str,
    repeats: int,
) -> WhisperBenchmarkResult:
    handler = WhisperHandler(
        {
            "backend": "mlx-whisper",
            "model": case.model,
            "quantization": "none",
            "language": language,
            "repo": case.repo,
        }
    )
    timings: list[float] = []
    transcript = ""
    note_parts = [case.notes] if case.notes else []

    try:
        for _ in range(repeats):
            started_at = time.perf_counter()
            result = handler._transcribe_mlx(audio_path)
            timings.append(time.perf_counter() - started_at)
            transcript = result.get("transcript", "")
    except Exception as exc:
        return WhisperBenchmarkResult(
            case=case.name,
            backend=case.backend,
            model=case.model,
            quantization=case.quantization,
            repo=case.repo,
            status="error",
            audio_duration_seconds=audio_duration,
            notes=f"MLX benchmark failed: {exc}",
            peak_rss_mb=_peak_rss_mb(),
        )

    warm_samples = timings[1:] or timings
    warm_run = statistics.mean(warm_samples)
    return WhisperBenchmarkResult(
        case=case.name,
        backend=case.backend,
        model=case.model,
        quantization=case.quantization,
        repo=case.repo,
        status="ok",
        audio_duration_seconds=audio_duration,
        cold_run_seconds=timings[0],
        warm_run_seconds=warm_run,
        warm_realtime_factor=(audio_duration / warm_run) if warm_run else None,
        word_error_rate=word_error_rate(reference_text, transcript),
        transcript=transcript,
        notes="; ".join(part for part in note_parts if part),
        peak_rss_mb=_peak_rss_mb(),
    )


def _run_cpp_case(
    case: BenchmarkCase,
    *,
    audio_path: Path,
    audio_duration: float,
    reference_text: str,
    language: str,
    repeats: int,
) -> WhisperBenchmarkResult:
    binary = shutil.which("whisper-cpp")
    if not binary:
        return WhisperBenchmarkResult(
            case=case.name,
            backend=case.backend,
            model=case.model,
            quantization=case.quantization,
            repo=case.repo,
            status="skipped",
            audio_duration_seconds=audio_duration,
            notes="whisper-cpp is not installed on PATH",
        )

    model_path = Path(case.model).expanduser()
    if not model_path.exists():
        return WhisperBenchmarkResult(
            case=case.name,
            backend=case.backend,
            model=case.model,
            quantization=case.quantization,
            repo=case.repo,
            status="skipped",
            audio_duration_seconds=audio_duration,
            notes=f"Model file is missing: {model_path}",
        )

    timings: list[float] = []
    transcript = ""
    try:
        for _ in range(repeats):
            with tempfile.TemporaryDirectory(prefix="hive-bench-cpp-") as tmpdir:
                output_root = Path(tmpdir) / "transcript"
                command = [
                    binary,
                    "--model",
                    str(model_path),
                    "--language",
                    language,
                    "--output-txt",
                    "--output-file",
                    str(output_root),
                    str(audio_path),
                ]
                started_at = time.perf_counter()
                subprocess.run(command, capture_output=True, check=True, text=True)
                timings.append(time.perf_counter() - started_at)
                transcript = output_root.with_suffix(".txt").read_text(encoding="utf-8").strip()
    except Exception as exc:
        return WhisperBenchmarkResult(
            case=case.name,
            backend=case.backend,
            model=case.model,
            quantization=case.quantization,
            repo=case.repo,
            status="error",
            audio_duration_seconds=audio_duration,
            notes=f"whisper.cpp benchmark failed: {exc}",
            peak_rss_mb=_peak_rss_mb(),
        )

    warm_samples = timings[1:] or timings
    warm_run = statistics.mean(warm_samples)
    return WhisperBenchmarkResult(
        case=case.name,
        backend=case.backend,
        model=case.model,
        quantization=case.quantization,
        repo=case.repo,
        status="ok",
        audio_duration_seconds=audio_duration,
        cold_run_seconds=timings[0],
        warm_run_seconds=warm_run,
        warm_realtime_factor=(audio_duration / warm_run) if warm_run else None,
        word_error_rate=word_error_rate(reference_text, transcript),
        transcript=transcript,
        notes=case.notes,
        peak_rss_mb=_peak_rss_mb(),
    )


def _peak_rss_mb() -> float:
    rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    if sys.platform == "darwin":
        return rss / 1024 / 1024
    return rss / 1024


def format_results_table(results: list[WhisperBenchmarkResult]) -> str:
    lines = [
        "case\tstatus\tbackend\tmodel\tquant\tcold_s\twarm_s\trtf\tttft_s\twer\tpeak_rss_mb",
    ]
    for result in results:
        lines.append(
            "\t".join(
                [
                    result.case,
                    result.status,
                    result.backend,
                    result.model,
                    result.quantization,
                    _format_whisper_metric(result.cold_run_seconds),
                    _format_whisper_metric(result.warm_run_seconds),
                    _format_whisper_metric(result.warm_realtime_factor),
                    _format_whisper_metric(result.time_to_first_token_seconds),
                    _format_whisper_metric(result.word_error_rate),
                    _format_whisper_metric(result.peak_rss_mb),
                ]
            )
        )
    return "\n".join(lines)


def write_json_report(path: Path, hardware: dict[str, str], results: list[WhisperBenchmarkResult]) -> None:
    policy_summary = summarize_whisper_policy_results(hardware, results)
    payload = {
        "hardware": hardware,
        "policy_summary": asdict(policy_summary),
        "results": [asdict(result) for result in results],
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")


def write_markdown_report(path: Path, hardware: dict[str, str], results: list[WhisperBenchmarkResult]) -> None:
    policy_summary = summarize_whisper_policy_results(hardware, results)
    lines = [
        "# Whisper Apple Silicon Benchmarks",
        "",
        "## Hardware",
        "",
    ]
    for key, value in hardware.items():
        lines.append(f"- {key}: {value}")

    lines.extend(
        [
            "",
            "## Policy Summary",
            "",
            f"- Machine class: `{policy_summary.machine_class}`",
            (
                f"- Built-in MLX policy coverage: {len(policy_summary.completed_policy_cases)}/"
                f"{len(POLICY_MATRIX_CASES)} completed"
            ),
            (
                "- Missing MLX policy cases: "
                + (
                    ", ".join(f"`{name}`" for name in policy_summary.missing_policy_cases)
                    if policy_summary.missing_policy_cases
                    else "none"
                )
            ),
            (
                "- whisper.cpp cases: "
                + (
                    ", ".join(f"`{name}`" for name in policy_summary.whisper_cpp_cases)
                    if policy_summary.whisper_cpp_cases
                    else "none"
                )
            ),
            (
                "- Pareto-optimal cases: "
                + (
                    ", ".join(f"`{name}`" for name in policy_summary.pareto_cases)
                    if policy_summary.pareto_cases
                    else "none"
                )
            ),
            "",
            f"## Recommendation for `{policy_summary.machine_class}`",
            "",
            f"- Decision: `{policy_summary.decision}`",
            (
                f"- Recommended case: `{policy_summary.recommended_case}`"
                if policy_summary.recommended_case
                else "- Recommended case: `none`"
            ),
            f"- Reason: {policy_summary.reason}",
            "",
            "## Results",
            "",
            (
                "- Time-to-first-token remains `n/a` for MLX and whisper.cpp in this harness "
                "because neither path exposes a streaming token callback yet."
            ),
            "",
            (
                "| Case | Status | Backend | Quantization | Cold run (s) | Warm run (s) | "
                "Warm realtime factor | TTFT (s) | WER | Peak RSS (MB) | Notes |"
            ),
            "| --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |",
        ]
    )
    for result in results:
        lines.append(
            "| "
            + " | ".join(
                [
                    result.case,
                    result.status,
                    result.backend,
                    result.quantization,
                    _format_whisper_metric(result.cold_run_seconds),
                    _format_whisper_metric(result.warm_run_seconds),
                    _format_whisper_metric(result.warm_realtime_factor),
                    _format_whisper_metric(result.time_to_first_token_seconds),
                    _format_whisper_metric(result.word_error_rate),
                    _format_whisper_metric(result.peak_rss_mb),
                    (result.notes or "").replace("|", "\\|"),
                ]
            )
            + " |"
        )

    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")


def _format_whisper_metric(value: float | None) -> str:
    return "—" if value is None else f"{value:.3f}"


def summarize_whisper_policy_results(
    hardware: dict[str, str],
    results: list[WhisperBenchmarkResult],
) -> WhisperBenchmarkPolicySummary:
    completed_policy_cases = tuple(
        case
        for case in POLICY_MATRIX_CASES
        if any(result.case == case and result.status == "ok" for result in results)
    )
    missing_policy_cases = tuple(case for case in POLICY_MATRIX_CASES if case not in completed_policy_cases)
    whisper_cpp_cases = tuple(result.case for result in results if result.backend == "whisper-cpp")
    pareto_cases = tuple(result.case for result in _pareto_optimal_results(results))
    recommended = _recommended_policy_result(results)
    machine = _normalize_machine_metadata(
        {
            "chip": hardware.get("chip"),
            "memory_gb": _parse_memory_gb(hardware.get("memory")),
        }
    )
    machine_class = _machine_class(machine)

    if recommended is None:
        decision = "defer"
        reason = "No successful cases produced comparable warm-path throughput and WER metrics."
    elif missing_policy_cases:
        decision = "defer"
        reason = (
            "Do not change worker defaults yet. The MLX policy sweep is incomplete for this machine class: "
            + ", ".join(missing_policy_cases)
        )
    else:
        runner_up = next(
            (result for result in _recommended_policy_candidates(results) if result.case != recommended.case),
            None,
        )
        if runner_up and runner_up.warm_realtime_factor and recommended.warm_realtime_factor:
            speedup = recommended.warm_realtime_factor / runner_up.warm_realtime_factor
            if speedup < 1.1:
                decision = "defer"
                reason = (
                    f"Do not change worker defaults yet. `{recommended.case}` only leads `{runner_up.case}` by "
                    f"{((speedup - 1) * 100):.1f}% on warm throughput for `{machine_class}`."
                )
            else:
                decision = "adopt"
                reason = (
                    f"`{recommended.case}` is the fastest case within the acceptable WER window for `{machine_class}`."
                )
        else:
            decision = "adopt"
            reason = f"`{recommended.case}` is the strongest measured case for `{machine_class}`."

    return WhisperBenchmarkPolicySummary(
        machine_class=machine_class,
        completed_policy_cases=completed_policy_cases,
        missing_policy_cases=missing_policy_cases,
        whisper_cpp_cases=whisper_cpp_cases,
        pareto_cases=pareto_cases,
        recommended_case=recommended.case if recommended else None,
        decision=decision,
        reason=reason,
    )


def _pareto_optimal_results(results: list[WhisperBenchmarkResult]) -> list[WhisperBenchmarkResult]:
    candidates = [
        result
        for result in results
        if result.status == "ok"
        and result.warm_realtime_factor is not None
        and result.word_error_rate is not None
    ]
    pareto: list[WhisperBenchmarkResult] = []
    for candidate in candidates:
        dominated = False
        for other in candidates:
            if other is candidate:
                continue
            if (
                other.warm_realtime_factor is not None
                and other.word_error_rate is not None
                and other.warm_realtime_factor >= candidate.warm_realtime_factor
                and other.word_error_rate <= candidate.word_error_rate
                and (
                    other.warm_realtime_factor > candidate.warm_realtime_factor
                    or other.word_error_rate < candidate.word_error_rate
                )
            ):
                dominated = True
                break
        if not dominated:
            pareto.append(candidate)

    return sorted(
        pareto,
        key=lambda result: (
            -(result.warm_realtime_factor or 0.0),
            result.word_error_rate if result.word_error_rate is not None else float("inf"),
            result.case,
        ),
    )


def _recommended_policy_candidates(results: list[WhisperBenchmarkResult]) -> list[WhisperBenchmarkResult]:
    successful = [
        result
        for result in results
        if result.status == "ok"
        and _is_vetted_policy_result(result)
        and result.warm_realtime_factor is not None
        and result.word_error_rate is not None
    ]
    if not successful:
        return []

    best_wer = min(result.word_error_rate for result in successful if result.word_error_rate is not None)
    acceptable_wer_ceiling = best_wer + 0.02
    acceptable = [
        result
        for result in successful
        if result.word_error_rate is not None and result.word_error_rate <= acceptable_wer_ceiling
    ]
    return sorted(
        acceptable,
        key=lambda result: (
            -(result.warm_realtime_factor or 0.0),
            result.word_error_rate if result.word_error_rate is not None else float("inf"),
            result.case,
        ),
    )


def _recommended_policy_result(results: list[WhisperBenchmarkResult]) -> WhisperBenchmarkResult | None:
    candidates = _recommended_policy_candidates(results)
    return candidates[0] if candidates else None


def _is_vetted_policy_result(result: WhisperBenchmarkResult) -> bool:
    case = BUILT_IN_CASES.get(result.case)
    if case is None or result.case not in POLICY_MATRIX_CASES:
        return False
    return (
        getattr(result, "backend", None) == case.backend
        and getattr(result, "quantization", None) == case.quantization
        and getattr(result, "repo", None) == case.repo
    )


def _parse_memory_gb(value: Any) -> int | None:
    if isinstance(value, int | float):
        return int(value)
    if not value:
        return None

    match = re.search(r"(\d+)", str(value))
    if not match:
        return None
    return int(match.group(1))
