"""Repeatable control-plane benchmark and comparison receipt."""

from __future__ import annotations

import asyncio
import statistics
import tempfile
import time
from pathlib import Path
from typing import Any

from .contracts import GenerationRequest
from .demo import build_demo_runtime
from .receipts import sha256_value


async def run_control_plane_benchmark(
    *,
    requests_per_tenant: int = 25,
    concurrency: int = 12,
) -> dict[str, Any]:
    prompts = {
        "support": "What is the refund window?",
        "operations": "Report line 7 temperature and interlock state.",
        "research": "Calculate 19 * 37",
    }
    with tempfile.TemporaryDirectory(prefix="backplane-benchmark-") as folder:
        runtime = build_demo_runtime(Path(folder) / "state.sqlite3")
        semaphore = asyncio.Semaphore(concurrency)

        async def one(tenant: str, ordinal: int):
            async with semaphore:
                return await runtime.run(
                    GenerationRequest(
                        tenant_id=tenant,
                        prompt=prompts[tenant],
                        request_id=f"bench-{tenant}-{ordinal}",
                    ),
                    authenticate=False,
                )

        started = time.perf_counter()
        results = await asyncio.gather(
            *[
                one(tenant, ordinal)
                for tenant in prompts
                for ordinal in range(requests_per_tenant)
            ]
        )
        elapsed = time.perf_counter() - started
        await runtime.close()

    latencies = [result.receipt.latency_ms for result in results]
    queues = [result.receipt.queue_ms for result in results]
    stacks: dict[str, set[tuple[str, ...]]] = {}
    for result in results:
        stacks.setdefault(result.receipt.tenant_id, set()).add(result.receipt.card_stack)
    isolation_failures = sum(len(values) != 1 for values in stacks.values())
    sorted_latencies = sorted(latencies)
    p95_index = max(0, round(0.95 * (len(sorted_latencies) - 1)))
    report = {
        "schema_version": "1.0",
        "backend": "control-plane-sandbox",
        "requests": len(results),
        "tenants": len(prompts),
        "concurrency": concurrency,
        "elapsed_seconds": round(elapsed, 4),
        "throughput_rps": round(len(results) / max(elapsed, 1e-9), 2),
        "latency_ms": {
            "p50": round(statistics.median(latencies), 3),
            "p95": round(sorted_latencies[p95_index], 3),
            "max": round(max(latencies), 3),
        },
        "queue_ms": {
            "median": round(statistics.median(queues), 3),
            "max": round(max(queues), 3),
        },
        "batch_sizes": sorted({result.receipt.batch_size for result in results}),
        "isolation_failures": isolation_failures,
        "status": "PASS" if isolation_failures == 0 else "FAIL",
        "scope": (
            "Control-plane isolation, scheduling, and receipts only; "
            "this does not benchmark neural inference quality."
        ),
    }
    return {**report, "sha256": sha256_value(report)}


def run_benchmark_sync(**kwargs: Any) -> dict[str, Any]:
    return asyncio.run(run_control_plane_benchmark(**kwargs))

