"""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 copy import time from pathlib import Path import pytest from meshnet_node.performance_contract import ( STOP_CONDITION, ContractThresholds, PerformanceContract, PerformanceContractError, _canonical_sha256, evaluate_contract, parse_contract, ) from meshnet_node.recipe_drivers import ( _artifact_sha256, _validate_config, 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") 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) 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) 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 return 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], }, stop_condition="test", ) 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_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] evaluation = evaluate_contract(contract, incomplete) assert evaluation.quality_lane_pass is False assert evaluation.verdict == "stop" 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" 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 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)