818 lines
35 KiB
Python
818 lines
35 KiB
Python
"""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 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()}
|
|
|
|
reference_counts = token_counts(reference)
|
|
prompt_ids = {str(prompt["id"]) for prompt in plan["prompts"]}
|
|
repeats = int(plan["repeats"])
|
|
for prompt_id in prompt_ids:
|
|
for level in required_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"
|
|
)
|
|
for entry in recipes:
|
|
if entry.get("available") and token_counts(entry) != reference_counts:
|
|
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
|
|
if failure_rate > thresholds.max_failure_rate:
|
|
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
|