"""The versioned safetensors-versus-GGUF performance contract.

The contract is the decision rule the native GGUF track is judged by, written
down *before* the numbers arrive and consumed later by the release gate
(DGR-014).  Its thresholds are ratios against the Transformers/safetensors
reference recipe rather than absolute tokens/sec, because the absolute figure is
a property of whichever machine ran the benchmark and would have to be re-argued
on every host; a ratio is a claim about the runtime.

Three rules give the contract its teeth:

* **Thresholds are locked.**  ``CONTRACT_SCHEMA_VERSION`` and ``locked_at``
  travel with the document.  Moving a threshold after seeing results is a new
  contract version and a human decision, not a tweak.
* **Only like-for-like comparisons count.**  A recipe measured on a different
  device than the reference is marked non-comparable and is granted no benefit,
  so a GPU-versus-CPU mismatch can never be laundered into a speed win.
* **Quantized recipes never claim numerical equivalence.**  Quality is gated on
  the near-lossless quality lane; the performance-fit lane is judged on speed,
  memory and fit alone.

The verdict is one of ``promote``, ``optimize`` or ``stop`` — the three outcomes
the release gate is allowed to reach.
"""

from __future__ import annotations

import base64
import binascii
import hashlib
import json
from collections import Counter
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Mapping

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

from .recipe_benchmark import Lane, REPORT_SCHEMA_VERSION

# Layout of the contract document understood by this reader.
CONTRACT_SCHEMA_VERSION = 1
PROVENANCE_SCHEMA_VERSION = 1
REAL_REPORT_PRODUCER = "meshnet_node.recipe_drivers.run_configured_benchmark/v1"

VERDICT_PROMOTE = "promote"
VERDICT_OPTIMIZE = "optimize"
VERDICT_STOP = "stop"


class PerformanceContractError(ValueError):
    """Raised when a contract is missing, malformed, or of an unsupported version."""


@dataclass(frozen=True)
class ContractThresholds:
    """The locked decision thresholds.

    Every value is a ratio of a GGUF recipe's metric to the reference recipe's
    metric on the same machine, same device, same plan.

    A *meaningful speed benefit* means the GGUF recipe decodes at least 25%
    faster for a single request without making time-to-first-token materially
    worse, or sustains at least 25% more aggregate throughput under concurrency.
    Either route is a real win for the product: one helps a single user, the
    other helps a loaded node.

    A *meaningful fit benefit* means peak resident memory (RSS plus VRAM) drops
    by at least 25%.  Fit is the product thesis — models larger than one
    consumer node — so it is measured in resident bytes, not in how small the
    file on disk is.  Artifact size has its own reported threshold because a
    smaller download is a real but secondary good.

    25% is chosen to sit well clear of ordinary run-to-run variance on a busy
    developer machine while still being a benefit a user would notice.  A 5%
    edge would not justify owning a native runtime and a patch stack.
    """

    min_decode_speedup: float = 1.25
    max_ttft_ratio: float = 1.25
    min_aggregate_throughput_speedup: float = 1.25
    max_resident_memory_ratio: float = 0.75
    max_artifact_size_ratio: float = 0.60
    min_quality_exact_match_rate: float = 0.90
    min_quality_mean_similarity: float = 0.97
    max_failure_rate: float = 0.0

    def to_dict(self) -> dict:
        return asdict(self)


@dataclass(frozen=True)
class PerformanceContract:
    """A locked, versioned contract plus the baseline it was locked against."""

    contract_version: int
    locked_at: str
    locked_by: str
    plan_id: str
    thresholds: ContractThresholds
    baseline: Mapping[str, Any]
    stop_condition: str
    notes: str = ""
    schema_version: int = CONTRACT_SCHEMA_VERSION

    def to_dict(self) -> dict:
        return {
            "schema_version": self.schema_version,
            "contract_version": self.contract_version,
            "locked_at": self.locked_at,
            "locked_by": self.locked_by,
            "plan_id": self.plan_id,
            "thresholds": self.thresholds.to_dict(),
            "baseline": dict(self.baseline),
            "stop_condition": self.stop_condition,
            "notes": self.notes,
        }


STOP_CONDITION = (
    "Stop the native llama.cpp/GGUF track when, on the same machine and device "
    "as the Transformers/safetensors reference and under this plan, no "
    "performance-fit GGUF recipe delivers either a meaningful speed benefit "
    "(>=25% higher single-request decode tokens/sec without a >25% worse TTFT, "
    "or >=25% higher aggregate throughput under concurrency) or a meaningful fit "
    "benefit (>=25% lower peak resident memory), or when the near-lossless "
    "quality lane fails, which indicates a broken runtime rather than a "
    "quantization trade-off."
)


def _recipe_entries(report: Mapping[str, Any]) -> dict[str, Mapping[str, Any]]:
    return {entry["recipe"]["id"]: entry for entry in report["recipes"]}


def _cell(entry: Mapping[str, Any], concurrency: int) -> Mapping[str, Any] | None:
    return entry["concurrency"].get(str(concurrency))


def _resident_bytes(cell: Mapping[str, Any]) -> int:
    return int(cell["peak_rss_bytes"]) + int(cell["peak_vram_bytes"])


def _ratio(value: float, reference: float) -> float:
    """Ratio guarded against a zero reference, which means "not measured"."""
    if reference <= 0:
        return 0.0
    return round(value / reference, 4)


def _canonical_sha256(value: Any) -> str:
    payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def report_signing_payload(report: Mapping[str, Any]) -> bytes:
    """Canonical report bytes covered by the Ed25519 signature."""
    provenance = report.get("provenance")
    if not isinstance(provenance, Mapping):
        raise PerformanceContractError("real benchmark report lacks signed provenance")
    unsigned = dict(report)
    unsigned_provenance = dict(provenance)
    unsigned_provenance.pop("signature", None)
    unsigned["provenance"] = unsigned_provenance
    return json.dumps(
        unsigned, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")


def _decode_base64(value: Any, label: str) -> bytes:
    if not isinstance(value, str):
        raise PerformanceContractError(f"report lacks {label}")
    try:
        return base64.b64decode(value, validate=True)
    except (binascii.Error, ValueError) as exc:
        raise PerformanceContractError(f"report carries invalid {label}") from exc


def _verify_real_provenance(
    contract: PerformanceContract, report: Mapping[str, Any]
) -> None:
    provenance = report.get("provenance")
    if not isinstance(provenance, Mapping):
        raise PerformanceContractError("real benchmark report lacks signed provenance")
    if provenance.get("schema_version") != PROVENANCE_SCHEMA_VERSION:
        raise PerformanceContractError("report provenance schema is unsupported")
    if provenance.get("producer") != REAL_REPORT_PRODUCER:
        raise PerformanceContractError("report was not emitted by the canonical real runner")
    if provenance.get("signature_algorithm") != "ed25519":
        raise PerformanceContractError("report provenance is not Ed25519 signed")
    for field in ("run_id", "started_at", "completed_at"):
        if not provenance.get(field):
            raise PerformanceContractError(f"report provenance lacks {field}")

    required_config = contract.baseline.get("required_config_sha256")
    if not required_config or provenance.get("config_sha256") != required_config:
        raise PerformanceContractError("report config digest does not match the locked config")

    encoded_public_key = contract.baseline.get("required_signer_public_key")
    public_key_bytes = _decode_base64(encoded_public_key, "locked signer public key")
    if len(public_key_bytes) != 32:
        raise PerformanceContractError("locked Ed25519 public key must be 32 bytes")
    expected_fingerprint = hashlib.sha256(public_key_bytes).hexdigest()
    if provenance.get("signer_public_key_sha256") != expected_fingerprint:
        raise PerformanceContractError("report signer fingerprint does not match the contract")

    signature = _decode_base64(provenance.get("signature"), "Ed25519 signature")
    try:
        Ed25519PublicKey.from_public_bytes(public_key_bytes).verify(
            signature, report_signing_payload(report)
        )
    except (InvalidSignature, ValueError) as exc:
        raise PerformanceContractError("report Ed25519 signature verification failed") from exc


def _validate_report(contract: PerformanceContract, report: Mapping[str, Any]) -> None:
    """Fail closed when a report is not the experiment the contract locked."""
    try:
        schema_version = report["schema_version"]
        evidence_class = report["evidence_class"]
        plan = report["plan"]
        recipes = report["recipes"]
        reference_id = report["reference_recipe_id"]
        host = report["host"]
    except (KeyError, TypeError) as exc:
        raise PerformanceContractError("benchmark report is missing required structure") from exc

    if schema_version != REPORT_SCHEMA_VERSION:
        raise PerformanceContractError(
            f"report schema {schema_version!r} is not supported schema {REPORT_SCHEMA_VERSION}"
        )
    if plan.get("plan_id") != contract.plan_id:
        raise PerformanceContractError(
            f"report plan {plan.get('plan_id')!r} does not match locked plan {contract.plan_id!r}"
        )
    required_plan_sha256 = contract.baseline.get("required_plan_sha256")
    measured_plan_sha256 = _canonical_sha256(plan)
    if required_plan_sha256 and measured_plan_sha256 != required_plan_sha256:
        raise PerformanceContractError(
            f"report plan digest {measured_plan_sha256} does not match locked digest "
            f"{required_plan_sha256}"
        )
    minimum_repeats = int(contract.baseline.get("minimum_repeats", 0))
    minimum_prompts = int(contract.baseline.get("minimum_prompt_count", 0))
    if int(plan.get("repeats", 0)) < minimum_repeats:
        raise PerformanceContractError("report has too few repeats for the locked contract")
    if len(plan.get("prompts", ())) < minimum_prompts:
        raise PerformanceContractError("report has too few prompts for the locked contract")
    minimum_output_tokens = int(contract.baseline.get("minimum_output_tokens", 0))
    if int(plan.get("sampling", {}).get("max_output_tokens", 0)) < minimum_output_tokens:
        raise PerformanceContractError("report output length is below the locked contract")
    required_evidence = contract.baseline.get("required_evidence_class")
    if required_evidence and evidence_class != required_evidence:
        raise PerformanceContractError(
            f"report evidence class {evidence_class!r} does not satisfy {required_evidence!r}"
        )
    if evidence_class in {"local-real", "multi-machine-real"}:
        _verify_real_provenance(contract, report)
    if required_evidence and (
        not isinstance(host, Mapping)
        or any(key not in host for key in ("hostname", "platform", "python", "cpu_count"))
    ):
        raise PerformanceContractError("report lacks measured host provenance")
    if not isinstance(recipes, list) or not recipes:
        raise PerformanceContractError("report contains no recipes")

    recipe_ids = [entry.get("recipe", {}).get("id") for entry in recipes]
    if len(set(recipe_ids)) != len(recipe_ids) or None in recipe_ids:
        raise PerformanceContractError("report recipe IDs must be present and unique")
    required_recipes = set(contract.baseline.get("required_recipes", ()))
    missing_recipes = required_recipes - set(recipe_ids)
    if missing_recipes:
        raise PerformanceContractError(
            f"report is missing required recipes {sorted(missing_recipes)}"
        )
    if reference_id not in recipe_ids:
        raise PerformanceContractError("report reference recipe is absent")

    levels = {int(level) for level in plan.get("concurrency_levels", ())}
    required_levels = {
        int(level) for level in contract.baseline.get("required_concurrency_levels", ())
    }
    if not required_levels.issubset(levels):
        raise PerformanceContractError(
            f"report concurrency {sorted(levels)} lacks required levels {sorted(required_levels)}"
        )
    if not plan.get("prompts"):
        raise PerformanceContractError("report plan contains no prompts")

    model_id = plan.get("model_id")
    model_revision = plan.get("model_revision")
    required_device = contract.baseline.get("required_device")
    required_artifacts = dict(contract.baseline.get("required_artifact_sha256") or {})
    required_runtimes = dict(contract.baseline.get("required_recipe_runtime") or {})
    required_backends = dict(contract.baseline.get("required_backend_detail") or {})
    required_host = dict(contract.baseline.get("required_host_identity") or {})
    for field, expected in required_host.items():
        if host.get(field) != expected:
            raise PerformanceContractError(
                f"report host/runtime field {field!r} does not match the locked identity"
            )
    for entry in recipes:
        recipe = entry.get("recipe", {})
        recipe_id = recipe.get("id")
        if required_device and recipe.get("device") != required_device:
            raise PerformanceContractError(
                f"recipe {recipe.get('id')!r} did not run on locked device {required_device!r}"
            )
        if required_evidence:
            if recipe.get("source_model_id") != model_id:
                raise PerformanceContractError("report mixes source model IDs")
            if recipe.get("source_model_revision") != model_revision:
                raise PerformanceContractError("report mixes source model revisions")
            digest = recipe.get("artifact_sha256", "")
            if not isinstance(digest, str) or len(digest) != 64:
                raise PerformanceContractError("report lacks an artifact SHA-256 digest")
            expected_digest = required_artifacts.get(recipe_id)
            if not expected_digest or digest != expected_digest:
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} artifact digest does not match the contract"
                )
            expected_runtime = required_runtimes.get(recipe_id)
            if not isinstance(expected_runtime, Mapping):
                raise PerformanceContractError(
                    f"contract lacks runtime identity for recipe {recipe_id!r}"
                )
            actual_runtime = {
                field: recipe.get(field) for field in expected_runtime
            }
            if actual_runtime != dict(expected_runtime):
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} runtime identity does not match the contract"
                )
            if entry.get("available"):
                expected_backend = required_backends.get(recipe_id)
                actual_backend = entry.get("load", {}).get("backend_detail")
                if not expected_backend or actual_backend != expected_backend:
                    raise PerformanceContractError(
                        f"recipe {recipe_id!r} backend identity does not match the contract"
                    )
        if entry.get("available"):
            cells = entry.get("concurrency", {})
            missing_cells = required_levels - {int(level) for level in cells}
            if missing_cells:
                raise PerformanceContractError(
                    f"recipe {recipe.get('id')!r} lacks concurrency cells {sorted(missing_cells)}"
                )

    reference = next(entry for entry in recipes if entry["recipe"]["id"] == reference_id)
    if not reference.get("available"):
        raise PerformanceContractError("reference recipe is unavailable")
    reference_failures = sum(
        int(cell.get("failures", 0)) for cell in reference.get("concurrency", {}).values()
    )
    if reference_failures:
        raise PerformanceContractError("reference recipe contains failed requests")

    def token_counts(entry: Mapping[str, Any]) -> dict[tuple[str, int, int], list[tuple[int, int]]]:
        counts: dict[tuple[str, int, int], list[tuple[int, int]]] = {}
        for outcome in entry.get("outcomes", ()):
            if not outcome.get("ok"):
                continue
            key = (
                str(outcome.get("prompt_id")),
                int(outcome.get("concurrency", 0)),
                int(outcome.get("repeat", -1)),
            )
            counts.setdefault(key, []).append(
                (int(outcome.get("prompt_tokens", 0)), int(outcome.get("decode_tokens", 0)))
            )
        return {key: sorted(values) for key, values in counts.items()}

    def outcome_counts(entry: Mapping[str, Any]) -> Counter[tuple[str, int, int]]:
        return Counter(
            (
                str(outcome.get("prompt_id")),
                int(outcome.get("concurrency", 0)),
                int(outcome.get("repeat", -1)),
            )
            for outcome in entry.get("outcomes", ())
        )

    prompt_ids = {str(prompt["id"]) for prompt in plan["prompts"]}
    repeats = int(plan["repeats"])
    for entry in recipes:
        if not entry.get("available"):
            continue
        recipe_id = str(entry["recipe"]["id"])
        cells = entry.get("concurrency", {})
        actual_levels = {int(level) for level in cells}
        if actual_levels != levels:
            raise PerformanceContractError(
                f"recipe {recipe_id!r} has unexpected concurrency cells"
            )
        for outcome in entry.get("outcomes", ()):
            try:
                outcome_recipe_id = str(outcome["recipe_id"])
                prompt_id = str(outcome["prompt_id"])
                level = int(outcome["concurrency"])
                repeat = int(outcome["repeat"])
            except (KeyError, TypeError, ValueError) as exc:
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} contains a malformed raw outcome"
                ) from exc
            if outcome_recipe_id != recipe_id:
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} contains an outcome for {outcome_recipe_id!r}"
                )
            if prompt_id not in prompt_ids or level not in levels or not 0 <= repeat < repeats:
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} contains an out-of-domain raw outcome"
                )
            if not isinstance(outcome.get("ok"), bool):
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} contains an outcome without boolean ok status"
                )

        coverage = outcome_counts(entry)
        for prompt_id in prompt_ids:
            for level in levels:
                for repeat in range(repeats):
                    if coverage[(prompt_id, level, repeat)] != level:
                        raise PerformanceContractError(
                            f"recipe {recipe_id!r} lacks complete request coverage"
                        )
        for level in levels:
            cell = cells.get(str(level), cells.get(level))
            raw = [
                outcome
                for outcome in entry.get("outcomes", ())
                if int(outcome["concurrency"]) == level
            ]
            if int(cell.get("concurrency", -1)) != level:
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} cell identity does not match concurrency {level}"
                )
            if int(cell.get("requests", -1)) != len(raw):
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} aggregate requests do not match raw outcomes"
                )
            raw_failures = sum(not outcome["ok"] for outcome in raw)
            if int(cell.get("failures", -1)) != raw_failures:
                raise PerformanceContractError(
                    f"recipe {recipe_id!r} aggregate failures do not match raw outcomes"
                )

    reference_counts = token_counts(reference)
    for prompt_id in prompt_ids:
        for level in levels:
            for repeat in range(repeats):
                values = reference_counts.get((prompt_id, level, repeat), ())
                if len(values) != level:
                    raise PerformanceContractError(
                        "reference recipe lacks complete prompt/repeat/concurrency coverage"
                    )
    if contract.thresholds.max_failure_rate != 0:
        raise PerformanceContractError(
            "nonzero failure tolerance requires an explicit failed-request token policy"
        )
    for entry in recipes:
        if not entry.get("available"):
            continue
        for key, values in token_counts(entry).items():
            if Counter(values) - Counter(reference_counts.get(key, ())):
                raise PerformanceContractError(
                    f"recipe {entry['recipe']['id']!r} used different prompt/decode token counts"
                )


@dataclass(frozen=True)
class RecipeEvaluation:
    """How one GGUF recipe fared against the reference under the contract."""

    recipe_id: str
    lane: str
    comparable: bool
    incomparable_reason: str
    speed_benefit: bool
    fit_benefit: bool
    quality_pass: bool | None
    failures: int
    measurements: Mapping[str, Any]
    reasons: tuple[str, ...]

    def to_dict(self) -> dict:
        data = asdict(self)
        data["measurements"] = dict(self.measurements)
        data["reasons"] = list(self.reasons)
        return data


@dataclass(frozen=True)
class ContractEvaluation:
    """The release-gate answer: a verdict plus every reason behind it."""

    contract_version: int
    plan_id: str
    verdict: str
    quality_lane_pass: bool
    speed_benefit: bool
    fit_benefit: bool
    stop_condition_met: bool
    recipes: tuple[RecipeEvaluation, ...]
    rationale: tuple[str, ...]

    def to_dict(self) -> dict:
        return {
            "contract_version": self.contract_version,
            "plan_id": self.plan_id,
            "verdict": self.verdict,
            "quality_lane_pass": self.quality_lane_pass,
            "speed_benefit": self.speed_benefit,
            "fit_benefit": self.fit_benefit,
            "stop_condition_met": self.stop_condition_met,
            "recipes": [recipe.to_dict() for recipe in self.recipes],
            "rationale": list(self.rationale),
        }


def _evaluate_recipe(
    entry: Mapping[str, Any],
    reference: Mapping[str, Any],
    drift_by_recipe: Mapping[str, Mapping[str, Any]],
    thresholds: ContractThresholds,
    concurrency_levels: list[int],
    expected_prompt_count: int,
) -> RecipeEvaluation:
    recipe = entry["recipe"]
    lane = recipe["lane"]
    reasons: list[str] = []

    if not entry["available"]:
        return RecipeEvaluation(
            recipe_id=recipe["id"], lane=lane, comparable=False,
            incomparable_reason=entry["unavailable_reason"] or "recipe was not measured",
            speed_benefit=False, fit_benefit=False, quality_pass=None, failures=0,
            measurements={}, reasons=("recipe unavailable; no benefit granted",),
        )

    if recipe["device"] != reference["recipe"]["device"]:
        return RecipeEvaluation(
            recipe_id=recipe["id"], lane=lane, comparable=False,
            incomparable_reason=(
                f"recipe ran on device {recipe['device']!r} but the reference ran on "
                f"{reference['recipe']['device']!r}; a cross-device ratio is not a runtime result"
            ),
            speed_benefit=False, fit_benefit=False, quality_pass=None, failures=0,
            measurements={}, reasons=("cross-device comparison; no benefit granted",),
        )

    single = _cell(entry, 1)
    reference_single = _cell(reference, 1)
    measurements: dict[str, Any] = {}
    speed_benefit = False

    if single and reference_single:
        decode_speedup = _ratio(
            single["decode_tokens_per_sec"], reference_single["decode_tokens_per_sec"]
        )
        ttft_ratio = _ratio(single["ttft_p50_ms"], reference_single["ttft_p50_ms"])
        measurements["decode_speedup"] = decode_speedup
        measurements["ttft_ratio"] = ttft_ratio
        single_request_win = (
            decode_speedup >= thresholds.min_decode_speedup
            and 0 < ttft_ratio <= thresholds.max_ttft_ratio
        )
        if single_request_win:
            speed_benefit = lane == Lane.PERFORMANCE_FIT.value
            reasons.append(
                f"single-request decode {decode_speedup:.2f}x reference "
                f"(>= {thresholds.min_decode_speedup:.2f}x) at TTFT ratio {ttft_ratio:.2f}"
            )
        else:
            reasons.append(
                f"no single-request speed win: decode {decode_speedup:.2f}x, TTFT {ttft_ratio:.2f}x"
            )

    concurrent = [level for level in concurrency_levels if level > 1]
    if concurrent:
        top = max(concurrent)
        cell, reference_cell = _cell(entry, top), _cell(reference, top)
        if cell and reference_cell:
            aggregate_speedup = _ratio(
                cell["aggregate_decode_tokens_per_sec"],
                reference_cell["aggregate_decode_tokens_per_sec"],
            )
            measurements["aggregate_throughput_speedup"] = aggregate_speedup
            measurements["aggregate_concurrency"] = top
            if aggregate_speedup >= thresholds.min_aggregate_throughput_speedup:
                speed_benefit = lane == Lane.PERFORMANCE_FIT.value
                reasons.append(
                    f"aggregate throughput at concurrency {top} is {aggregate_speedup:.2f}x reference "
                    f"(>= {thresholds.min_aggregate_throughput_speedup:.2f}x)"
                )
            else:
                reasons.append(
                    f"no concurrency speed win: aggregate throughput at {top} is "
                    f"{aggregate_speedup:.2f}x reference"
                )

    fit_benefit = False
    if single and reference_single:
        resident_ratio = _ratio(_resident_bytes(single), _resident_bytes(reference_single))
        artifact_ratio = _ratio(
            entry["load"]["artifact_bytes"], reference["load"]["artifact_bytes"]
        )
        measurements["resident_memory_ratio"] = resident_ratio
        measurements["artifact_size_ratio"] = artifact_ratio
        if 0 < resident_ratio <= thresholds.max_resident_memory_ratio:
            fit_benefit = lane == Lane.PERFORMANCE_FIT.value
            reasons.append(
                f"peak resident memory is {resident_ratio:.2f}x reference "
                f"(<= {thresholds.max_resident_memory_ratio:.2f}x)"
            )
        else:
            reasons.append(f"no fit win: peak resident memory is {resident_ratio:.2f}x reference")
        measurements["artifact_size_win"] = (
            0 < artifact_ratio <= thresholds.max_artifact_size_ratio
        )

    failures = sum(cell["failures"] for cell in entry["concurrency"].values())
    requests = sum(cell["requests"] for cell in entry["concurrency"].values())
    failure_rate = _ratio(failures, requests) if requests else 0.0
    measurements["failure_rate"] = failure_rate
    failure_limit_exceeded = requests == 0 or (
        failures > 0
        if thresholds.max_failure_rate == 0
        else failures / requests > thresholds.max_failure_rate
    )
    if failure_limit_exceeded:
        reasons.append(f"failure rate {failure_rate:.2%} exceeds the contract limit")
        speed_benefit = False
        fit_benefit = False

    # Quality is a claim only the near-lossless lane is allowed to make.  A
    # quantized recipe's drift is recorded elsewhere and deliberately not read
    # here: Q4 disagreeing with bf16 is the trade-off, not a failure.
    quality_pass: bool | None = None
    if lane == Lane.QUALITY.value:
        drift = drift_by_recipe.get(recipe["id"])
        if drift is None:
            quality_pass = False
            reasons.append("quality-lane recipe has no drift measurement against the reference")
        else:
            complete_coverage = drift.get("compared_prompts") == expected_prompt_count
            quality_pass = (
                complete_coverage
                and failures == 0
                and drift["exact_match_rate"] >= thresholds.min_quality_exact_match_rate
                and drift["mean_similarity"] >= thresholds.min_quality_mean_similarity
            )
            measurements["compared_prompts"] = drift.get("compared_prompts", 0)
            measurements["expected_prompts"] = expected_prompt_count
            measurements["exact_match_rate"] = drift["exact_match_rate"]
            measurements["mean_similarity"] = drift["mean_similarity"]
            if not complete_coverage:
                reasons.append(
                    f"quality lane compared {drift.get('compared_prompts', 0)} of "
                    f"{expected_prompt_count} required prompts"
                )
            if failures:
                reasons.append("quality lane contains failed requests")
            reasons.append(
                f"quality lane exact-match {drift['exact_match_rate']:.2f} / similarity "
                f"{drift['mean_similarity']:.3f} versus the reference "
                f"({'pass' if quality_pass else 'fail'})"
            )

    return RecipeEvaluation(
        recipe_id=recipe["id"], lane=lane, comparable=True, incomparable_reason="",
        speed_benefit=speed_benefit, fit_benefit=fit_benefit, quality_pass=quality_pass,
        failures=failures, measurements=measurements, reasons=tuple(reasons),
    )


def evaluate_contract(
    contract: PerformanceContract,
    report: Mapping[str, Any],
) -> ContractEvaluation:
    """Judge a benchmark report against the locked contract.

    Only performance-fit recipes can earn a speed or fit benefit; the quality
    lane decides only whether the GGUF runtime is numerically sane.  A GGUF
    runtime that fails the quality lane is broken, and no amount of speed
    redeems it, so the verdict is ``stop`` regardless of the other numbers.
    """
    _validate_report(contract, report)
    entries = _recipe_entries(report)
    reference_id = report["reference_recipe_id"]
    reference = entries.get(reference_id)
    if reference is None:
        raise PerformanceContractError(
            f"report names reference recipe {reference_id!r}, which it does not contain"
        )

    drift_by_recipe = {entry["recipe_id"]: entry for entry in report["drift"]}
    concurrency_levels = list(report["plan"]["concurrency_levels"])

    evaluations = tuple(
        _evaluate_recipe(
            entry,
            reference,
            drift_by_recipe,
            contract.thresholds,
            concurrency_levels,
            len(report["plan"]["prompts"]),
        )
        for recipe_id, entry in entries.items()
        if recipe_id != reference_id
    )

    quality_lane = [
        evaluation for evaluation in evaluations
        if evaluation.lane == Lane.QUALITY.value and evaluation.comparable
    ]
    quality_lane_pass = bool(quality_lane) and all(
        evaluation.quality_pass for evaluation in quality_lane
    )
    performance_lane = [
        evaluation for evaluation in evaluations
        if evaluation.lane == Lane.PERFORMANCE_FIT.value
    ]
    speed_benefit = any(evaluation.speed_benefit for evaluation in performance_lane)
    fit_benefit = any(evaluation.fit_benefit for evaluation in performance_lane)

    rationale: list[str] = []
    if not quality_lane:
        rationale.append(
            "no comparable near-lossless GGUF recipe was measured, so the runtime's "
            "numerical correctness is unproven"
        )
    elif not quality_lane_pass:
        rationale.append(
            "the near-lossless quality lane failed: the GGUF runtime disagrees with the "
            "safetensors reference beyond what near-lossless weights can explain"
        )
    else:
        rationale.append("the near-lossless quality lane passed against the safetensors reference")

    rationale.append(
        "a meaningful speed benefit was measured" if speed_benefit
        else "no performance-fit recipe delivered a meaningful speed benefit"
    )
    rationale.append(
        "a meaningful fit benefit was measured" if fit_benefit
        else "no performance-fit recipe delivered a meaningful fit benefit"
    )

    stop_condition_met = not quality_lane_pass or not (speed_benefit or fit_benefit)
    if stop_condition_met:
        verdict = VERDICT_STOP
    elif speed_benefit and fit_benefit:
        verdict = VERDICT_PROMOTE
    else:
        verdict = VERDICT_OPTIMIZE
        rationale.append(
            "exactly one of speed or fit cleared the contract: the benefit is real but partial, "
            "so the measured bottleneck needs a bounded optimization task before promotion"
        )

    return ContractEvaluation(
        contract_version=contract.contract_version,
        plan_id=contract.plan_id,
        verdict=verdict,
        quality_lane_pass=quality_lane_pass,
        speed_benefit=speed_benefit,
        fit_benefit=fit_benefit,
        stop_condition_met=stop_condition_met,
        recipes=evaluations,
        rationale=tuple(rationale),
    )


def parse_contract(data: Any, source: str = "<memory>") -> PerformanceContract:
    """Validate an already-decoded contract document."""
    if not isinstance(data, Mapping):
        raise PerformanceContractError(f"contract root in {source} must be a JSON object")

    schema_version = data.get("schema_version")
    if not isinstance(schema_version, int) or isinstance(schema_version, bool):
        raise PerformanceContractError(f"'schema_version' in {source} must be an integer")
    if schema_version != CONTRACT_SCHEMA_VERSION:
        raise PerformanceContractError(
            f"{source} declares contract schema version {schema_version}, but this node reads "
            f"version {CONTRACT_SCHEMA_VERSION}; upgrade the node or use a supported contract"
        )

    for required in ("contract_version", "locked_at", "locked_by", "plan_id", "stop_condition"):
        if not data.get(required):
            raise PerformanceContractError(f"{source} is missing {required!r}")

    raw_thresholds = data.get("thresholds")
    if not isinstance(raw_thresholds, Mapping):
        raise PerformanceContractError(f"'thresholds' in {source} must be a JSON object")
    known = set(ContractThresholds().to_dict())
    unknown = set(raw_thresholds) - known
    missing = known - set(raw_thresholds)
    if unknown or missing:
        raise PerformanceContractError(
            f"{source} threshold keys differ from v1; unknown={sorted(unknown)}, "
            f"missing={sorted(missing)}"
        )
    thresholds = ContractThresholds(**{
        key: float(value) for key, value in raw_thresholds.items()
    })
    contract_version = int(data["contract_version"])
    if contract_version != 1 or thresholds != ContractThresholds():
        raise PerformanceContractError(
            f"{source} changes immutable v1 thresholds without a supported contract version"
        )
    if str(data["stop_condition"]) != STOP_CONDITION:
        raise PerformanceContractError(
            f"{source} stop condition differs from executable v1 semantics"
        )

    return PerformanceContract(
        contract_version=contract_version,
        locked_at=str(data["locked_at"]),
        locked_by=str(data["locked_by"]),
        plan_id=str(data["plan_id"]),
        thresholds=thresholds,
        baseline=dict(data.get("baseline") or {}),
        stop_condition=str(data["stop_condition"]),
        notes=str(data.get("notes", "")),
        schema_version=schema_version,
    )


def load_contract(path: Path) -> PerformanceContract:
    """Load and validate the contract at ``path``."""
    try:
        raw = path.read_text(encoding="utf-8")
    except OSError as exc:
        raise PerformanceContractError(
            f"cannot read performance contract {path}: {exc.strerror or exc}"
        ) from exc
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise PerformanceContractError(
            f"{path} is not valid JSON: {exc.msg} at line {exc.lineno} column {exc.colno}"
        ) from exc
    return parse_contract(data, source=str(path))


def baseline_from_report(report: Mapping[str, Any]) -> dict[str, Any]:
    """Distil the reference numbers a later gate needs to compare against."""
    entries = _recipe_entries(report)
    baseline: dict[str, Any] = {
        "evidence_class": report["evidence_class"],
        "model_id": report["plan"]["model_id"],
        "model_revision": report["plan"]["model_revision"],
        "plan_sha256": _canonical_sha256(report["plan"]),
        "reference_recipe_id": report["reference_recipe_id"],
        "host": report["host"],
        "provenance": dict(report.get("provenance") or {}),
        "artifact_sha256": {
            recipe_id: entry["recipe"]["artifact_sha256"]
            for recipe_id, entry in entries.items()
        },
        "recipe_runtime": {
            recipe_id: {
                field: entry["recipe"].get(field)
                for field in ("runtime", "weight_format", "weight_quantization", "device")
            }
            for recipe_id, entry in entries.items()
        },
        "backend_detail": {
            recipe_id: entry.get("load", {}).get("backend_detail")
            for recipe_id, entry in entries.items()
            if entry.get("available")
        },
        "recipes": {},
    }
    for recipe_id, entry in entries.items():
        if not entry["available"]:
            baseline["recipes"][recipe_id] = {"available": False,
                                              "reason": entry["unavailable_reason"]}
            continue
        baseline["recipes"][recipe_id] = {
            "available": True,
            "lane": entry["recipe"]["lane"],
            "device": entry["recipe"]["device"],
            "artifact_bytes": entry["load"]["artifact_bytes"],
            "concurrency": {
                level: {
                    "ttft_p50_ms": cell["ttft_p50_ms"],
                    "ttft_p95_ms": cell["ttft_p95_ms"],
                    "latency_p50_ms": cell["latency_p50_ms"],
                    "latency_p95_ms": cell["latency_p95_ms"],
                    "prefill_tokens_per_sec": cell["prefill_tokens_per_sec"],
                    "decode_tokens_per_sec": cell["decode_tokens_per_sec"],
                    "aggregate_decode_tokens_per_sec": cell["aggregate_decode_tokens_per_sec"],
                    "peak_rss_bytes": cell["peak_rss_bytes"],
                    "peak_vram_bytes": cell["peak_vram_bytes"],
                    "failures": cell["failures"],
                }
                for level, cell in entry["concurrency"].items()
            },
        }
    return baseline