"""The recipe benchmark's measurement core, driven by a scripted fake runtime. These tests never load a model, touch a GPU, or open a socket: the core is deliberately runtime-free so the arithmetic and the lane rules can be pinned down exactly, and the real drivers only have to be honest about what they report. """ from __future__ import annotations import base64 import copy import hashlib import json import time from dataclasses import replace from pathlib import Path import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshnet_node import recipe_benchmark as recipe_benchmark_module from meshnet_node import recipe_drivers as recipe_drivers_module from meshnet_node.performance_contract import ( PROVENANCE_SCHEMA_VERSION, REAL_REPORT_PRODUCER, STOP_CONDITION, ContractThresholds, PerformanceContract, PerformanceContractError, _canonical_sha256, evaluate_contract, parse_contract, report_signing_payload, ) from meshnet_node.recipe_drivers import ( CONTRACT_V1_PROFILE, GPU_DIAGNOSTIC_PROFILE, GPU_DIAGNOSTIC_REPORT_PRODUCER, _artifact_sha256, _gpu_offload_evidence, _gpu_layer_config_detail, _producer_for_profile, _validate_config, build_driver, require_real_inference, ) from meshnet_node.recipe_benchmark import ( BenchmarkError, BenchmarkPlan, GenerationSample, Lane, LoadStats, PromptSpec, RecipeSpec, SamplingPolicy, build_report, compute_drift, measure_recipe, summarize_concurrency, RequestOutcome, ) PROMPTS = ( PromptSpec(id="short", text="Say hello.", context_class="short"), PromptSpec(id="long", text="Summarize the following. " * 40, context_class="long"), ) def plan(**overrides) -> BenchmarkPlan: defaults = dict( plan_id="test-plan", model_id="test/model", model_revision="revision-1", prompts=PROMPTS, sampling=SamplingPolicy(max_output_tokens=8), concurrency_levels=(1, 4), repeats=1, warmup_requests=0, ) defaults.update(overrides) return BenchmarkPlan(**defaults) class FakeDriver: """A runtime with fixed timings, so every metric below has one right answer.""" def __init__( self, *, decode_ms_per_token: float = 10.0, prefill_ms: float = 100.0, artifact_bytes: int = 1_000_000, rss_bytes: int = 4_000_000, vram_bytes: int = 0, texts: dict[str, str] | None = None, fail_at_concurrency: int | None = None, decode_tokens: int = 8, generation_delay_s: float = 0.0, ) -> None: self.decode_ms_per_token = decode_ms_per_token self.prefill_ms = prefill_ms self.artifact_bytes = artifact_bytes self.rss_bytes = rss_bytes self.vram_bytes = vram_bytes self.texts = texts or {} self.fail_at_concurrency = fail_at_concurrency self.decode_tokens = decode_tokens self.generation_delay_s = generation_delay_s self.in_flight = 0 self.max_in_flight = 0 self.loads = 0 self.closes = 0 self.generations = 0 def load(self) -> LoadStats: self.loads += 1 return LoadStats( artifact_bytes=self.artifact_bytes, load_ms=50.0, rss_bytes=self.rss_bytes, vram_bytes=self.vram_bytes, ) def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample: self.in_flight += 1 self.max_in_flight = max(self.max_in_flight, self.in_flight) try: if self.generation_delay_s: time.sleep(self.generation_delay_s) if self.fail_at_concurrency and self.in_flight >= self.fail_at_concurrency: raise RuntimeError("slot exhausted") self.generations += 1 decode_ms = self.decode_ms_per_token * self.decode_tokens return GenerationSample( text=self.texts.get(prompt, "hello world"), prompt_tokens=10, decode_tokens=self.decode_tokens, ttft_ms=self.prefill_ms, prefill_ms=self.prefill_ms, decode_ms=decode_ms, total_ms=self.prefill_ms + decode_ms, ) finally: self.in_flight -= 1 def memory_probe(self) -> tuple[int, int]: return self.rss_bytes, self.vram_bytes def close(self) -> None: self.closes += 1 def recipe(recipe_id: str, lane: Lane, *, reference: bool = False, device: str = "cpu") -> RecipeSpec: return RecipeSpec( id=recipe_id, runtime="fake", weight_format="fake", weight_quantization="bf16", lane=lane, device=device, is_reference=reference, ) def test_plan_rejects_an_experiment_it_cannot_run(): with pytest.raises(BenchmarkError): plan(prompts=()) with pytest.raises(BenchmarkError): plan(concurrency_levels=(0,)) with pytest.raises(BenchmarkError): plan(repeats=0) def test_measure_runs_every_prompt_at_every_concurrency_level(): driver = FakeDriver() measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan()) # 2 prompts x (1 + 4) requests-per-level. assert len(measurement.outcomes) == 2 * 1 + 2 * 4 assert sorted(measurement.metrics) == [1, 4] assert driver.loads == 1 assert driver.closes == 1 assert measurement.available def test_concurrency_level_actually_overlaps_requests(): driver = FakeDriver(decode_ms_per_token=5.0, generation_delay_s=0.02) measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan(concurrency_levels=(1, 4))) assert driver.max_in_flight > 1, "concurrency 4 must run requests in parallel, not serially" def test_driver_is_closed_even_when_every_request_fails(): driver = FakeDriver(fail_at_concurrency=1) measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan()) assert driver.closes == 1 assert all(not outcome.ok for outcome in measurement.outcomes) assert measurement.metrics[1].failures == 2 assert measurement.metrics[1].failure_reasons == ("RuntimeError: slot exhausted",) def test_failed_requests_are_reported_not_raised(): driver = FakeDriver(fail_at_concurrency=4, generation_delay_s=0.02) measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan()) assert measurement.metrics[1].failures == 0 assert measurement.metrics[4].failures > 0 assert measurement.metrics[4].requests == 8 def test_summary_arithmetic_is_exact(): outcomes = [ RequestOutcome( recipe_id="r", concurrency=2, prompt_id="p", repeat=0, ok=True, latency_ms=200.0, ttft_ms=100.0, prefill_ms=100.0, decode_ms=100.0, prompt_tokens=10, decode_tokens=10, ), RequestOutcome( recipe_id="r", concurrency=2, prompt_id="p", repeat=1, ok=True, latency_ms=400.0, ttft_ms=200.0, prefill_ms=200.0, decode_ms=200.0, prompt_tokens=10, decode_tokens=10, ), ] metrics = summarize_concurrency( outcomes, concurrency=2, wall_ms=1000.0, peak_rss_bytes=7, peak_vram_bytes=9 ) assert metrics.latency_p50_ms == 200.0 assert metrics.latency_p95_ms == 400.0 # 10 tok / 0.1 s = 100 tok/s and 10 tok / 0.2 s = 50 tok/s, averaged. assert metrics.decode_tokens_per_sec == 75.0 # 20 decoded tokens over a 1 s wall clock, regardless of per-request rates. assert metrics.aggregate_decode_tokens_per_sec == 20.0 assert (metrics.peak_rss_bytes, metrics.peak_vram_bytes) == (7, 9) def test_aggregate_throughput_credits_overlap_but_per_request_rate_does_not(): """Two runtimes with identical per-request speed must be told apart by overlap.""" serial = summarize_concurrency( [ RequestOutcome(recipe_id="s", concurrency=4, prompt_id="p", repeat=i, ok=True, latency_ms=100.0, decode_ms=100.0, decode_tokens=10) for i in range(4) ], concurrency=4, wall_ms=400.0, peak_rss_bytes=0, peak_vram_bytes=0, ) batched = summarize_concurrency( [ RequestOutcome(recipe_id="b", concurrency=4, prompt_id="p", repeat=i, ok=True, latency_ms=100.0, decode_ms=100.0, decode_tokens=10) for i in range(4) ], concurrency=4, wall_ms=100.0, peak_rss_bytes=0, peak_vram_bytes=0, ) assert serial.decode_tokens_per_sec == batched.decode_tokens_per_sec == 100.0 assert serial.aggregate_decode_tokens_per_sec == 100.0 assert batched.aggregate_decode_tokens_per_sec == 400.0 def test_drift_against_the_reference_is_exact_for_an_identical_runtime(): texts = {prompt.text: f"answer for {prompt.id}" for prompt in PROMPTS} reference = measure_recipe( FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan() ) twin = measure_recipe(FakeDriver(texts=texts), recipe("twin", Lane.QUALITY), plan()) drift = compute_drift(twin, reference) assert drift.compared_prompts == 2 assert drift.exact_match_rate == 1.0 assert drift.mean_similarity == 1.0 assert drift.advisory is False def test_quantized_drift_is_advisory_and_never_an_equivalence_claim(): reference = measure_recipe( FakeDriver(texts={prompt.text: "the capital is Paris" for prompt in PROMPTS}), recipe("ref", Lane.QUALITY, reference=True), plan(), ) quantized = measure_recipe( FakeDriver(texts={prompt.text: "the capital is Lyon" for prompt in PROMPTS}), recipe("q4", Lane.PERFORMANCE_FIT), plan(), ) drift = compute_drift(quantized, reference) assert drift.advisory is True, "a quantized recipe's drift must be advisory" assert drift.exact_match_rate == 0.0 assert 0.0 < drift.mean_similarity < 1.0 assert drift.per_prompt[0]["first_divergence_char"] > 0 def test_report_needs_exactly_one_quality_lane_reference(): measurement = measure_recipe(FakeDriver(), recipe("a", Lane.QUALITY, reference=True), plan()) second = measure_recipe(FakeDriver(), recipe("b", Lane.QUALITY, reference=True), plan()) quantized = measure_recipe(FakeDriver(), recipe("q", Lane.PERFORMANCE_FIT), plan()) with pytest.raises(BenchmarkError, match="exactly one reference"): build_report(plan(), [measurement, second], host={}, evidence_class="synthetic") with pytest.raises(BenchmarkError, match="exactly one reference"): build_report(plan(), [quantized], host={}, evidence_class="synthetic") def test_reference_recipe_may_not_be_quantized(): quantized_reference = measure_recipe( FakeDriver(), recipe("q", Lane.PERFORMANCE_FIT, reference=True), plan() ) with pytest.raises(BenchmarkError, match="quality lane"): build_report(plan(), [quantized_reference], host={}, evidence_class="synthetic") def test_report_must_declare_how_it_was_produced(): measurement = measure_recipe(FakeDriver(), recipe("a", Lane.QUALITY, reference=True), plan()) with pytest.raises(BenchmarkError, match="evidence class"): build_report(plan(), [measurement], host={}, evidence_class="probably-real") def test_report_carries_every_metric_the_contract_reads(): reference = measure_recipe(FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan()) quantized = measure_recipe( FakeDriver(decode_ms_per_token=4.0, artifact_bytes=400_000, rss_bytes=1_000_000), recipe("q4", Lane.PERFORMANCE_FIT), plan(), ) report = build_report( plan(), [reference, quantized], host={"cpu": "test"}, evidence_class="synthetic" ) assert report["schema_version"] == 1 assert report["reference_recipe_id"] == "ref" entry = next(e for e in report["recipes"] if e["recipe"]["id"] == "q4") cell = entry["concurrency"]["1"] for metric in ( "ttft_p50_ms", "ttft_p95_ms", "latency_p50_ms", "latency_p95_ms", "prefill_tokens_per_sec", "decode_tokens_per_sec", "aggregate_decode_tokens_per_sec", "peak_rss_bytes", "peak_vram_bytes", "failures", ): assert metric in cell, f"the contract reads {metric}, so the report must carry it" assert entry["load"]["artifact_bytes"] == 400_000 assert [d["recipe_id"] for d in report["drift"]] == ["q4"] def test_unavailable_recipes_are_recorded_rather_than_dropped(): from meshnet_node.recipe_benchmark import RecipeMeasurement reference = measure_recipe(FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan()) missing = RecipeMeasurement( recipe=recipe("q4", Lane.PERFORMANCE_FIT), load=LoadStats(artifact_bytes=0, load_ms=0.0), unavailable_reason="BenchmarkError: GGUF artifact not found", ) report = build_report(plan(), [reference, missing], host={}, evidence_class="synthetic") entry = next(e for e in report["recipes"] if e["recipe"]["id"] == "q4") assert entry["available"] is False assert "not found" in entry["unavailable_reason"] assert report["drift"] == [], "an unmeasured recipe has no drift to report" def test_contract_requires_a_quality_lane_then_allows_quantized_fit_benefit(): texts = {prompt.text: "same greedy answer" for prompt in PROMPTS} reference = measure_recipe( FakeDriver(texts=texts, rss_bytes=4_000_000), recipe("safetensors", Lane.QUALITY, reference=True), plan() ) quality = measure_recipe( FakeDriver(texts=texts), recipe("gguf-f16", Lane.QUALITY), plan() ) q4 = measure_recipe( FakeDriver(texts={prompt.text: "different quantized answer" for prompt in PROMPTS}, rss_bytes=1_000_000, decode_ms_per_token=20.0), recipe("gguf-q4", Lane.PERFORMANCE_FIT), plan() ) report = build_report(plan(), [reference, quality, q4], host={}, evidence_class="synthetic") reference_entry = next( entry for entry in report["recipes"] if entry["recipe"]["id"] == "safetensors" ) q4_entry = next( entry for entry in report["recipes"] if entry["recipe"]["id"] == "gguf-q4" ) for level, q4_cell in q4_entry["concurrency"].items(): reference_cell = reference_entry["concurrency"][level] q4_cell["decode_tokens_per_sec"] = reference_cell["decode_tokens_per_sec"] / 2 q4_cell["aggregate_decode_tokens_per_sec"] = ( reference_cell["aggregate_decode_tokens_per_sec"] / 2 ) q4_cell["ttft_p50_ms"] = reference_cell["ttft_p50_ms"] * 2 contract = PerformanceContract( contract_version=1, locked_at="2026-07-13T00:00:00Z", locked_by="test", plan_id="test-plan", thresholds=ContractThresholds(), baseline={}, stop_condition="test", ) evaluation = evaluate_contract(contract, report) assert evaluation.quality_lane_pass is True assert evaluation.fit_benefit is True assert evaluation.verdict == "optimize" def test_peak_memory_is_sampled_while_requests_are_in_flight(): class TransientMemoryDriver(FakeDriver): def memory_probe(self) -> tuple[int, int]: return (99_000_000 if self.in_flight else 1_000_000), 0 measurement = measure_recipe( TransientMemoryDriver(generation_delay_s=0.05), recipe("transient", Lane.QUALITY, reference=True), plan(warmup_requests=0), ) assert measurement.metrics[1].peak_rss_bytes == 99_000_000 assert measurement.metrics[4].peak_rss_bytes == 99_000_000 def test_real_inference_requires_explicit_opt_in(monkeypatch): monkeypatch.delenv("MESHNET_ENABLE_REAL_INFERENCE_TESTS", raising=False) with pytest.raises(BenchmarkError, match="opt-in"): require_real_inference() def test_config_rejects_an_artifact_digest_mismatch(tmp_path: Path): artifact = tmp_path / "model.gguf" artifact.write_bytes(b"real model bytes") config = { "artifact_storage_root": str(tmp_path), "plan": { "model_id": "test/model", "model_revision": "revision-1", "prompts": [ {"id": "p1", "text": "one"}, {"id": "p2", "text": "two"}, {"id": "p3", "text": "three"}, ], "concurrency_levels": [1, 4], "repeats": 3, "warmup_requests": 1, "sampling": { "temperature": 0.0, "top_k": 1, "top_p": 1.0, "max_output_tokens": 32, }, }, "recipes": [{ "id": "recipe", "source_model_id": "test/model", "source_model_revision": "revision-1", "artifact_path": str(artifact), "artifact_sha256": _artifact_sha256(artifact), "device": "cpu", "driver": { "type": "llama-cpp-server", "binary": str(artifact), "binary_sha256": _artifact_sha256(artifact), "gguf_path": str(artifact), "device": "cpu", "threads": 8, "n_parallel": 4, }, }], } _validate_config(config) nested_digest = copy.deepcopy(config) nested_digest["recipes"][0]["driver"]["artifact_sha256"] = "f" * 64 with pytest.raises(BenchmarkError, match="artifact_sha256 is forbidden"): _validate_config(nested_digest) driver = build_driver(nested_digest["recipes"][0], plan()) assert driver.artifact_sha256 == config["recipes"][0]["artifact_sha256"] cuda_transformers = copy.deepcopy(config) cuda_transformers["recipes"][0]["device"] = "cuda" cuda_transformers["recipes"][0]["driver"] = { "type": "transformers", "model_path": str(artifact), "device": "cuda", "threads": 8, } with pytest.raises(BenchmarkError, match="every recipe to run on CPU"): _validate_config(cuda_transformers) config["recipes"][0]["artifact_sha256"] = "0" * 64 with pytest.raises(BenchmarkError, match="digest mismatch"): _validate_config(config) config["recipes"][0]["artifact_sha256"] = _artifact_sha256(artifact) config["recipes"][0]["driver"]["binary_sha256"] = "0" * 64 with pytest.raises(BenchmarkError, match="binary SHA-256 mismatch"): _validate_config(config) config["recipes"][0]["driver"]["binary_sha256"] = _artifact_sha256(artifact) config["recipes"][0]["driver"]["n_gpu_layers"] = 1 with pytest.raises(BenchmarkError, match="CPU-only"): _validate_config(config) config["recipes"][0]["device"] = "cuda" config["recipes"][0]["driver"]["device"] = "cuda" with pytest.raises(BenchmarkError, match="host marker"): _validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE) config["host"] = [] with pytest.raises(BenchmarkError, match="host metadata must be an object"): _validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE) config["host"] = {"benchmark_lane": "rocm-gpu-diagnostic"} _validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE) config["recipes"][0]["driver"]["n_gpu_layers"] = 0 with pytest.raises(BenchmarkError, match="positive n_gpu_layers"): _validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE) def test_gpu_offload_evidence_requires_measured_rocm_placement(): assert _gpu_layer_config_detail("cpu", 0) == "gpu layers 0" assert _gpu_layer_config_detail("cuda", 99) == "requested gpu layers 99" measured = """ common_param: - ROCm0 : Radeon 8060S Graphics (63963 MiB, 33058 MiB free) load_tensors: offloaded 25/25 layers to GPU """ assert _gpu_offload_evidence(measured, 99) == ( "measured accelerator ROCm0: Radeon 8060S Graphics; " "measured offload 25/25 layers" ) with pytest.raises(BenchmarkError, match="no measured ROCm device"): _gpu_offload_evidence("offloaded 25/25 layers to GPU", 99) with pytest.raises(BenchmarkError, match="no measured layer offload"): _gpu_offload_evidence( "- ROCm0 : Radeon 8060S Graphics (63963 MiB, 33058 MiB free)", 99 ) with pytest.raises(BenchmarkError, match="measured only 4/25"): _gpu_offload_evidence(measured.replace("25/25", "4/25"), 99) def test_profiles_derive_fixed_producers_and_cli_dispatch(monkeypatch, tmp_path: Path): assert _producer_for_profile(CONTRACT_V1_PROFILE) == REAL_REPORT_PRODUCER assert _producer_for_profile(GPU_DIAGNOSTIC_PROFILE) == GPU_DIAGNOSTIC_REPORT_PRODUCER with pytest.raises(BenchmarkError, match="unknown benchmark validation profile"): _producer_for_profile("caller-selected-producer") profiled_calls = [] def profiled_runner(config, *, profile): profiled_calls.append((config, profile)) return {"profile": profile} monkeypatch.setattr(recipe_drivers_module, "_run_profiled_benchmark", profiled_runner) assert recipe_drivers_module.run_configured_benchmark({"source": "v1"}) == { "profile": CONTRACT_V1_PROFILE } assert recipe_drivers_module.run_configured_gpu_diagnostic({"source": "gpu"}) == { "profile": GPU_DIAGNOSTIC_PROFILE } assert profiled_calls == [ ({"source": "v1"}, CONTRACT_V1_PROFILE), ({"source": "gpu"}, GPU_DIAGNOSTIC_PROFILE), ] config_path = tmp_path / "config.json" config_path.write_text(json.dumps({"test": "config"}), encoding="utf-8") calls = [] def contract_runner(config): calls.append((CONTRACT_V1_PROFILE, config)) return {"profile": CONTRACT_V1_PROFILE} def gpu_runner(config): calls.append((GPU_DIAGNOSTIC_PROFILE, config)) return {"profile": GPU_DIAGNOSTIC_PROFILE} monkeypatch.setattr(recipe_drivers_module, "run_configured_benchmark", contract_runner) monkeypatch.setattr(recipe_drivers_module, "run_configured_gpu_diagnostic", gpu_runner) monkeypatch.setattr( recipe_benchmark_module, "format_summary", lambda report: report["profile"] ) recipe_benchmark_module.main(["--config", str(config_path)]) recipe_benchmark_module.main( ["--profile", GPU_DIAGNOSTIC_PROFILE, "--config", str(config_path)] ) assert calls == [ (CONTRACT_V1_PROFILE, {"test": "config"}), (GPU_DIAGNOSTIC_PROFILE, {"test": "config"}), ] def test_committed_config_digest_matches_immutable_contract(): evidence = ( Path(__file__).parents[1] / ".scratch" / "distributed-gguf-runtime" / "evidence" / "DGR-001" ) contract = parse_contract( json.loads((evidence / "performance-contract.json").read_text()) ) config = json.loads((evidence / "benchmark-config.json").read_text()) assert _canonical_sha256(config) == contract.baseline["required_config_sha256"] trusted = json.loads( (evidence.parents[1] / "trusted-evidence-signers.json").read_text() ) public_key = base64.b64decode(contract.baseline["required_signer_public_key"]) fingerprint = hashlib.sha256(public_key).hexdigest() assert any( signer["algorithm"] == "ed25519" and signer["fingerprint_sha256"] == fingerprint and signer["status"] == "active" for signer in trusted["signers"] ) expected_cpu_gpu_detail = _gpu_layer_config_detail("cpu", 0) llama_backends = [ detail for recipe_id, detail in contract.baseline["required_backend_detail"].items() if recipe_id.startswith("llama-cpp-") ] assert llama_backends assert all(expected_cpu_gpu_detail in detail for detail in llama_backends) assert all("requested gpu layers" not in detail for detail in llama_backends) _TEST_SIGNING_KEY = Ed25519PrivateKey.generate() def _resign_test_report(report: dict) -> None: report["provenance"].pop("signature", None) report["provenance"]["signature"] = base64.b64encode( _TEST_SIGNING_KEY.sign(report_signing_payload(report)) ).decode("ascii") def _lock_real_report(report: dict) -> PerformanceContract: report["evidence_class"] = "local-real" report["host"] = { "hostname": "test-host", "platform": "test-platform", "python": "3.12", "cpu_count": 8, } for entry in report["recipes"]: entry["recipe"]["source_model_id"] = report["plan"]["model_id"] entry["recipe"]["source_model_revision"] = report["plan"]["model_revision"] entry["recipe"]["artifact_sha256"] = "a" * 64 entry["load"]["backend_detail"] = f"fake-backend-{entry['recipe']['id']}" public_key = _TEST_SIGNING_KEY.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw ) config_sha256 = "c" * 64 report["provenance"] = { "schema_version": PROVENANCE_SCHEMA_VERSION, "producer": REAL_REPORT_PRODUCER, "run_id": "test-run-id", "started_at": "2026-07-13T00:00:00Z", "completed_at": "2026-07-13T00:01:00Z", "config_sha256": config_sha256, "signature_algorithm": "ed25519", "signer_public_key_sha256": hashlib.sha256(public_key).hexdigest(), } contract = PerformanceContract( contract_version=1, locked_at="2026-07-13T00:00:00Z", locked_by="test", plan_id=report["plan"]["plan_id"], thresholds=ContractThresholds(), baseline={ "required_evidence_class": "local-real", "required_recipes": [entry["recipe"]["id"] for entry in report["recipes"]], "required_concurrency_levels": [1, 4], "required_config_sha256": config_sha256, "required_signer_public_key": base64.b64encode(public_key).decode("ascii"), "required_artifact_sha256": { entry["recipe"]["id"]: entry["recipe"]["artifact_sha256"] for entry in report["recipes"] }, "required_recipe_runtime": { entry["recipe"]["id"]: { field: entry["recipe"].get(field) for field in ("runtime", "weight_format", "weight_quantization", "device") } for entry in report["recipes"] }, "required_backend_detail": { entry["recipe"]["id"]: entry["load"]["backend_detail"] for entry in report["recipes"] }, "required_host_identity": {"python": "3.12"}, }, stop_condition="test", ) _resign_test_report(report) return contract def test_locked_contract_rejects_synthetic_evidence(): reference = measure_recipe( FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan() ) quality = measure_recipe(FakeDriver(), recipe("quality", Lane.QUALITY), plan()) q4 = measure_recipe(FakeDriver(), recipe("q4", Lane.PERFORMANCE_FIT), plan()) report = build_report( plan(), [reference, quality, q4], host={}, evidence_class="synthetic" ) contract = _lock_real_report(report) report["evidence_class"] = "synthetic" with pytest.raises(PerformanceContractError, match="evidence class"): evaluate_contract(contract, report) def test_non_synthetic_report_requires_canonical_provenance(): reference = measure_recipe( FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan() ) with pytest.raises(BenchmarkError, match="signed provenance"): build_report( plan(), [reference], host={}, evidence_class="local-real" ) def test_signed_real_report_rejects_tampering_and_rebinding(): texts = {prompt.text: "same greedy answer" for prompt in PROMPTS} measurements = [ measure_recipe(FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan()), measure_recipe(FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan()), measure_recipe(FakeDriver(texts=texts), recipe("q4", Lane.PERFORMANCE_FIT), plan()), ] report = build_report(plan(), measurements, host={}, evidence_class="synthetic") contract = _lock_real_report(report) assert evaluate_contract(contract, report).verdict in {"promote", "optimize", "stop"} unsigned = copy.deepcopy(report) unsigned["provenance"].pop("signature") with pytest.raises(PerformanceContractError, match="signature"): evaluate_contract(contract, unsigned) gpu_diagnostic = copy.deepcopy(report) gpu_diagnostic["provenance"]["producer"] = GPU_DIAGNOSTIC_REPORT_PRODUCER _resign_test_report(gpu_diagnostic) with pytest.raises(PerformanceContractError, match="canonical real runner"): evaluate_contract(contract, gpu_diagnostic) tampered = copy.deepcopy(report) tampered["recipes"][0]["recipe"]["artifact_sha256"] = "b" * 64 with pytest.raises(PerformanceContractError, match="signature verification failed"): evaluate_contract(contract, tampered) rebound_artifact = copy.deepcopy(tampered) _resign_test_report(rebound_artifact) with pytest.raises(PerformanceContractError, match="artifact digest"): evaluate_contract(contract, rebound_artifact) rebound_config = copy.deepcopy(report) rebound_config["provenance"]["config_sha256"] = "d" * 64 _resign_test_report(rebound_config) with pytest.raises(PerformanceContractError, match="config digest"): evaluate_contract(contract, rebound_config) rebound_runtime = copy.deepcopy(report) rebound_runtime["recipes"][1]["recipe"]["runtime"] = "other-runtime" _resign_test_report(rebound_runtime) with pytest.raises(PerformanceContractError, match="runtime identity"): evaluate_contract(contract, rebound_runtime) rebound_backend = copy.deepcopy(report) rebound_backend["recipes"][1]["load"]["backend_detail"] = "other-backend" _resign_test_report(rebound_backend) with pytest.raises(PerformanceContractError, match="backend identity"): evaluate_contract(contract, rebound_backend) rebound_host = copy.deepcopy(report) rebound_host["host"]["python"] = "9.9" _resign_test_report(rebound_host) with pytest.raises(PerformanceContractError, match="host/runtime field"): evaluate_contract(contract, rebound_host) def test_quality_lane_requires_every_prompt_to_be_compared(): texts = {prompt.text: "same greedy answer" for prompt in PROMPTS} reference = measure_recipe( FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan() ) quality = measure_recipe( FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan() ) q4 = measure_recipe( FakeDriver(texts=texts, rss_bytes=1_000_000), recipe("q4", Lane.PERFORMANCE_FIT), plan(), ) report = build_report( plan(), [reference, quality, q4], host={}, evidence_class="synthetic" ) contract = _lock_real_report(report) incomplete = copy.deepcopy(report) quality_drift = next( drift for drift in incomplete["drift"] if drift["recipe_id"] == "quality" ) quality_drift["compared_prompts"] = 1 quality_drift["per_prompt"] = quality_drift["per_prompt"][:1] _resign_test_report(incomplete) evaluation = evaluate_contract(contract, incomplete) assert evaluation.quality_lane_pass is False assert evaluation.verdict == "stop" def test_failed_request_reaches_zero_tolerance_gate_instead_of_validation_error(): texts = {prompt.text: "same greedy answer" for prompt in PROMPTS} measurements = [ measure_recipe( FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan() ), measure_recipe(FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan()), measure_recipe(FakeDriver(texts=texts), recipe("q4", Lane.PERFORMANCE_FIT), plan()), ] report = build_report(plan(), measurements, host={}, evidence_class="synthetic") contract = _lock_real_report(report) failed = copy.deepcopy(report) quality_entry = next( entry for entry in failed["recipes"] if entry["recipe"]["id"] == "quality" ) outcome = quality_entry["outcomes"][0] outcome["ok"] = False outcome["error"] = "simulated runtime failure" cell_key = next( key for key in quality_entry["concurrency"] if int(key) == outcome["concurrency"] ) quality_entry["concurrency"][cell_key]["failures"] = 1 _resign_test_report(failed) evaluation = evaluate_contract(contract, failed) quality_evaluation = next( item for item in evaluation.recipes if item.recipe_id == "quality" ) assert quality_evaluation.failures == 1 assert quality_evaluation.quality_pass is False assert evaluation.verdict == "stop" permissive_contract = replace( contract, thresholds=replace(contract.thresholds, max_failure_rate=0.01), ) with pytest.raises(PerformanceContractError, match="explicit failed-request token policy"): evaluate_contract(permissive_contract, failed) inconsistent = copy.deepcopy(failed) quality_entry = next( entry for entry in inconsistent["recipes"] if entry["recipe"]["id"] == "quality" ) quality_entry["concurrency"][cell_key]["failures"] = 0 _resign_test_report(inconsistent) with pytest.raises(PerformanceContractError, match="failures do not match raw outcomes"): evaluate_contract(contract, inconsistent) wrong_requests = copy.deepcopy(report) quality_entry = next( entry for entry in wrong_requests["recipes"] if entry["recipe"]["id"] == "quality" ) quality_entry["concurrency"][cell_key]["requests"] += 1 _resign_test_report(wrong_requests) with pytest.raises(PerformanceContractError, match="requests do not match raw outcomes"): evaluate_contract(contract, wrong_requests) misidentified = copy.deepcopy(report) quality_entry = next( entry for entry in misidentified["recipes"] if entry["recipe"]["id"] == "quality" ) quality_entry["outcomes"][0]["recipe_id"] = "other-recipe" _resign_test_report(misidentified) with pytest.raises(PerformanceContractError, match="contains an outcome for"): evaluate_contract(contract, misidentified) missing = copy.deepcopy(failed) quality_entry = next( entry for entry in missing["recipes"] if entry["recipe"]["id"] == "quality" ) quality_entry["outcomes"].pop() _resign_test_report(missing) with pytest.raises(PerformanceContractError, match="complete request coverage"): evaluate_contract(contract, missing) def test_locked_contract_rejects_changed_plan_and_token_counts(): texts = {prompt.text: "same greedy answer" for prompt in PROMPTS} reference = measure_recipe( FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan() ) quality = measure_recipe( FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan() ) q4 = measure_recipe( FakeDriver(texts=texts), recipe("q4", Lane.PERFORMANCE_FIT), plan() ) report = build_report( plan(), [reference, quality, q4], host={}, evidence_class="synthetic" ) contract = _lock_real_report(report) contract.baseline["required_plan_sha256"] = _canonical_sha256(report["plan"]) changed_plan = copy.deepcopy(report) changed_plan["plan"]["prompts"][0]["text"] = "changed after locking" _resign_test_report(changed_plan) with pytest.raises(PerformanceContractError, match="plan digest"): evaluate_contract(contract, changed_plan) changed_tokens = copy.deepcopy(report) quality_entry = next( entry for entry in changed_tokens["recipes"] if entry["recipe"]["id"] == "quality" ) quality_entry["outcomes"][0]["decode_tokens"] += 1 _resign_test_report(changed_tokens) with pytest.raises(PerformanceContractError, match="different prompt/decode token counts"): evaluate_contract(contract, changed_tokens) def test_v1_contract_thresholds_and_stop_condition_are_immutable(): raw = PerformanceContract( contract_version=1, locked_at="2026-07-13T00:00:00Z", locked_by="test", plan_id="test-plan", thresholds=ContractThresholds(), baseline={}, stop_condition=STOP_CONDITION, ).to_dict() parse_contract(raw) changed_threshold = copy.deepcopy(raw) changed_threshold["thresholds"]["min_decode_speedup"] = 1.01 with pytest.raises(PerformanceContractError, match="immutable v1 thresholds"): parse_contract(changed_threshold) changed_stop = copy.deepcopy(raw) changed_stop["stop_condition"] = "promote everything" with pytest.raises(PerformanceContractError, match="stop condition differs"): parse_contract(changed_stop)