fix: harden DGR-001 performance contract evidence

.scratch/distributed-gguf-runtime/evidence/DGR-001/BLOCKED.md

@@ -0,0 +1,43 @@
+# DGR-001 downstream stop-condition handoff
+
+Status: **DGR-001 is complete; native-track promotion is blocked by the immutable v1 verdict.**
+
+This is no longer an execution-prerequisite blocker. The required real benchmark
+ran successfully, every recipe completed at concurrency 1 and 4, artifacts were
+verified, and deterministic/full test gates passed.
+
+## Locked result
+
+`contract-evaluation.json` records:
+
+```text
+verdict: stop
+quality_lane_pass: false
+speed_benefit: true
+fit_benefit: true
+stop_condition_met: true
+```
+
+The exact-revision BF16 GGUF quality lane compared every prompt but achieved
+`0.3333` exact match and `0.9471` mean similarity against the Transformers BF16
+reference. V1 requires `0.90` and `0.97`. Quantized Q4_K_M had substantial speed
+and fit benefits, but the contract explicitly forbids speed from redeeming a
+failed near-lossless quality lane.
+
+## Scope of this stop
+
+The measured baseline is Qwen2.5-0.5B on CPU using a CPU-only llama.cpp build.
+It is not a Radeon, large-model, distributed, or native-shard result. Therefore:
+
+1. Do not silently mark v1 promoted or weaken its thresholds after observing the
+   data.
+2. Do not let DGR-004 or later runtime stories treat DGR-001 completion as a
+   positive promotion signal.
+3. A human may choose one of these explicit paths:
+   - stop the native GGUF track as v1 directs;
+   - diagnose and fix the BF16 runtime divergence, then rerun the exact v1 plan;
+   - authorize a separately versioned GPU/large-model contract whose scope and
+     workload are locked before its measurements.
+
+All raw evidence, configuration, artifacts, hashes, and reproduction commands
+are in this directory and `README.md`.

.scratch/distributed-gguf-runtime/evidence/DGR-001/README.md

@@ -0,0 +1,153 @@
+# DGR-001 — Safetensors versus GGUF performance contract
+
+Status: **complete; immutable v1 verdict is `stop`.**
+
+DGR-001 successfully produced a controlled local-real CPU baseline. Completion
+means the experiment and decision contract are durable and verified; it does
+**not** mean the native GGUF track is approved to continue. The locked quality
+gate failed, so dependent runtime work requires a human decision or a new,
+explicitly versioned experiment/contract rather than silently weakening v1.
+
+## Controlled workload
+
+- Model: `Qwen/Qwen2.5-0.5B-Instruct`
+- Exact source revision: `7ae557604adf67be50417f59c2c2f167def9a775`
+- Machine: `fedora`, Linux `7.0.14-101.fc43.x86_64`, 32 logical CPUs
+- Device: CPU for every recipe; VRAM is therefore correctly reported as zero
+- Runtime reference: Transformers `5.13.0`, PyTorch
+  `2.10.0+rocm7.13.0a20260513`, BF16 safetensors
+- GGUF runtime: llama.cpp version 9991, commit
+  `e920c523e3b8a0163fe498af5bf90df35ff51d25`
+- Workload: three fixed short/medium/long prompts, greedy sampling, 32 output
+  tokens, three repeats, two warmups, concurrency 1 and 4, 16 CPU threads
+- Evidence class: `local-real`
+
+All artifacts are beneath `/run/media/popov/DATA/llm/`; no model artifact was
+created under `/home`.
+
+## Recipes and exact artifacts
+
+| Recipe | Artifact | SHA-256 |
+|---|---|---|
+| Transformers BF16 reference | complete mounted Hugging Face snapshot | `e596e9d6205fdc9177569cccd7f8b471b058f66e3630c8e4326d5aad52bd18b6` |
+| llama.cpp BF16 quality lane | `Qwen2.5-0.5B-Instruct-7ae5576-BF16.gguf` | `e842fdc35d7f00fda95a54e1b51731ba1d196aea45065cc9f46925fdc1d6f862` |
+| llama.cpp Q4_K_M performance/fit lane | `Qwen2.5-0.5B-Instruct-7ae5576-Q4_K_M.gguf` | `a88e3f570e2efeaf06b50df9859db2c70d8646aa3a2c94a14e14d5797a2921a5` |
+
+The snapshot digest covers every sorted relative path, resolved size, and file
+byte, so tokenizer/config drift is included. The BF16 GGUF was converted
+directly from the exact snapshot while preserving BF16 weights. Q4_K_M was
+quantized from an exact-revision F16 conversion with the pinned quantizer.
+Runtime validation recomputes every declared digest before model loading.
+
+## Real results
+
+All recipes completed every request with zero failures.
+
+| Metric | Transformers BF16 | llama.cpp BF16 | llama.cpp Q4_K_M |
+|---|---:|---:|---:|
+| Decode tok/s, c=1 | 46.1 | 88.0 | 170.1 |
+| Aggregate decode tok/s, c=4 | 47.1 | 211.4 | 206.4 |
+| TTFT p50, c=1 | 37.5 ms | 43.9 ms | 23.8 ms |
+| Peak resident memory, c=1 | 1.94 GB | 1.11 GB | 0.54 GB |
+| Artifact size | 1.00 GB | 0.99 GB | 0.40 GB |
+| Failures | 0 | 0 | 0 |
+
+Against the reference, the eligible Q4_K_M lane measured:
+
+- single-request decode speedup: **3.69×**;
+- concurrency-4 aggregate throughput speedup: **4.38×**;
+- resident-memory ratio: **0.279×**;
+- artifact-size ratio: **0.398×**.
+
+The near-lossless BF16 quality lane compared all three prompts but measured:
+
+- exact match: **0.3333** (v1 requires at least `0.90`);
+- mean text similarity: **0.9471** (v1 requires at least `0.97`).
+
+Tokenization and stopping were controlled: every runtime saw the same prompt
+token counts and reported 31 post-TTFT decode tokens. The mismatch is genuine
+greedy runtime divergence on two prompts, not missing coverage or a text-length
+artifact. Therefore `contract-evaluation.json` records:
+
+```text
+verdict: stop
+quality_lane_pass: false
+speed_benefit: true
+fit_benefit: true
+stop_condition_met: true
+```
+
+Thresholds were not changed after observing these results.
+
+## Implementation
+
+- `recipe_benchmark.py` provides the runtime-neutral measurement core, true
+  concurrency, continuous in-flight peak-memory sampling, percentile/throughput
+  aggregation, failures, and output drift.
+- `recipe_drivers.py` provides opt-in Transformers and llama-server drivers,
+  mounted-drive confinement, exact artifact/runtime verification, equal
+  device/thread budgets, greedy-only validation, measured host provenance, and
+  a CPU-only v1 guard until process VRAM can be measured honestly.
+- Peak RSS is runtime-scoped: Transformers reports growth above its pre-runtime
+  Python baseline, while llama.cpp reports its isolated server process tree.
+  Both are sampled continuously during in-flight requests.
+- TTFT uses each runtime's prompt/first-token compute boundary; end-to-end HTTP,
+  scheduling, and queue overhead remains in latency and `queue_wait_ms`.
+- The exact canonical plan SHA-256 locks prompts, model/revision, sampling,
+  output length, repeats, warmups, and concurrency. The evaluator also requires
+  equal prompt/decode token counts across recipes.
+- llama.cpp's `predicted_n` includes the first token while `predicted_ms` begins
+  after it; the driver subtracts that token so decode throughput matches the
+  Transformers inter-token convention.
+- `performance_contract.py` rejects wrong plans, synthetic evidence, missing
+  recipes/concurrency, mixed model revisions, incomplete quality coverage,
+  failed references, and missing artifact/host provenance.
+- Quantized drift remains advisory. Only the near-lossless lane can satisfy the
+  quality gate, and only performance-fit recipes can earn speed/fit benefits.
+
+## Evidence files
+
+- `performance-contract.json` — immutable v1 thresholds and stop condition
+- `benchmark-config.json` — exact real-run plan, drivers, artifacts, and hashes
+- `results.json` — raw machine-readable per-request and aggregate evidence
+- `results.txt` — human-readable benchmark summary
+- `baseline.json` — distilled measurements for later comparison
+- `contract-evaluation.json` — fail-closed v1 verdict
+- `commands.txt` — reproducible conversion, benchmark, evaluation, and test commands
+- `BLOCKED.md` — downstream stop-condition handoff
+- `known-unrelated-failure.md` — clean-base reproduction of the tracker race
+
+## Verification
+
+```text
+Targeted: 22 passed
+Full suite: 749 passed, 13 skipped
+Earlier cancellation retry matrix, DGR-001: 4/5 passed
+Earlier cancellation retry matrix, clean d904c40: 4/5 passed
+compileall: passed
+git diff --check: passed
+Evidence JSON parse/integrity checks: passed
+```
+
+The full-suite exception is documented in `known-unrelated-failure.md` and
+satisfies the issue's explicit clean-tree reproduction clause. DGR-001 changes
+no tracker/proxy files.
+
+The earlier Ralph claim that the full suite was blocked by Protobuf 6.33.6 was
+invalid: it used Hermes Agent's internal venv. Verification above used the
+project `.venv`, which has the DGR-002-compatible runtime. Real inference used
+`.venv-rocm` Python 3.12.
+
+## Limitations and dependent-story handoff
+
+- This is a **0.5B CPU baseline**, not evidence for a large model, Radeon GPU,
+  distributed execution, network transport, or native shard worker.
+- The installed llama.cpp build is CPU-only (`GGML_HIP=OFF`). No GPU comparison
+  is claimed.
+- Absolute timings are developer-machine measurements; locked ratios and raw
+  artifacts are provided for reproducibility.
+- DGR-014 may consume v1 only with the exact plan/evidence requirements enforced
+  by `performance_contract.py`.
+- DGR-004 and later native-runtime work must not treat DGR-001 completion as a
+  promotion. V1 says `stop`; proceeding requires a human decision backed by a
+  separately versioned GPU/large-model contract or a diagnosed quality fix.

.scratch/distributed-gguf-runtime/evidence/DGR-001/baseline.json

@@ -0,0 +1,122 @@
+{
+  "evidence_class": "local-real",
+  "host": {
+    "accelerator_name": "Radeon 8060S Graphics",
+    "accelerator_runtime": "7.13.26183",
+    "benchmark_lane": "cpu-controlled-baseline",
+    "converter_sha256": "c819f18fb22927b49fabc3b35d1c9e21ee638b3817eccd1bd4efbcc7116eeb4d",
+    "cpu_count": 32,
+    "cuda_available": true,
+    "hostname": "fedora",
+    "llama_cpp_commit": "e920c523e3b8a0163fe498af5bf90df35ff51d25",
+    "llama_cpp_version": "9991",
+    "llama_server_sha256": "fd8fe612970f23e447f2e717cfa51665be06b8d7315ba60556e010f6bca510dd",
+    "platform": "Linux-7.0.14-101.fc43.x86_64-x86_64-with-glibc2.42",
+    "python": "3.12.13",
+    "quantizer_sha256": "bd0cc8c7be6d48aad4755b31062e0e59a887cbadd43dbb8771853d5858bb198f",
+    "torch_version": "2.10.0+rocm7.13.0a20260513",
+    "transformers_version": "5.13.0"
+  },
+  "model_id": "Qwen/Qwen2.5-0.5B-Instruct",
+  "model_revision": "7ae557604adf67be50417f59c2c2f167def9a775",
+  "plan_sha256": "efe24690a9a7164bac6ab3fd0a6b22f078fc08aaefcfb96210ddf154e6050570",
+  "recipes": {
+    "llama-cpp-near-lossless-quality": {
+      "artifact_bytes": 994156448,
+      "available": true,
+      "concurrency": {
+        "1": {
+          "aggregate_decode_tokens_per_sec": 73.7861,
+          "decode_tokens_per_sec": 87.9728,
+          "failures": 0,
+          "latency_p50_ms": 385.2049,
+          "latency_p95_ms": 560.2939,
+          "peak_rss_bytes": 1110708224,
+          "peak_vram_bytes": 0,
+          "prefill_tokens_per_sec": 1427.2072,
+          "ttft_p50_ms": 43.929,
+          "ttft_p95_ms": 107.003
+        },
+        "4": {
+          "aggregate_decode_tokens_per_sec": 211.3515,
+          "decode_tokens_per_sec": 73.8932,
+          "failures": 0,
+          "latency_p50_ms": 467.5094,
+          "latency_p95_ms": 790.862,
+          "peak_rss_bytes": 1129578496,
+          "peak_vram_bytes": 0,
+          "prefill_tokens_per_sec": 1077.8162,
+          "ttft_p50_ms": 33.612,
+          "ttft_p95_ms": 128.501
+        }
+      },
+      "device": "cpu",
+      "lane": "quality"
+    },
+    "llama-cpp-quantized-performance-fit": {
+      "artifact_bytes": 397807520,
+      "available": true,
+      "concurrency": {
+        "1": {
+          "aggregate_decode_tokens_per_sec": 110.0458,
+          "decode_tokens_per_sec": 170.131,
+          "failures": 0,
+          "latency_p50_ms": 258.0681,
+          "latency_p95_ms": 465.8523,
+          "peak_rss_bytes": 542167040,
+          "peak_vram_bytes": 0,
+          "prefill_tokens_per_sec": 783.3775,
+          "ttft_p50_ms": 23.847,
+          "ttft_p95_ms": 237.696
+        },
+        "4": {
+          "aggregate_decode_tokens_per_sec": 206.377,
+          "decode_tokens_per_sec": 83.543,
+          "failures": 0,
+          "latency_p50_ms": 413.3897,
+          "latency_p95_ms": 910.253,
+          "peak_rss_bytes": 572788736,
+          "peak_vram_bytes": 0,
+          "prefill_tokens_per_sec": 474.3116,
+          "ttft_p50_ms": 67.945,
+          "ttft_p95_ms": 431.804
+        }
+      },
+      "device": "cpu",
+      "lane": "performance-fit"
+    },
+    "transformers-safetensors-reference": {
+      "artifact_bytes": 999586347,
+      "available": true,
+      "concurrency": {
+        "1": {
+          "aggregate_decode_tokens_per_sec": 40.3425,
+          "decode_tokens_per_sec": 46.1451,
+          "failures": 0,
+          "latency_p50_ms": 795.4807,
+          "latency_p95_ms": 930.9725,
+          "peak_rss_bytes": 1941213184,
+          "peak_vram_bytes": 0,
+          "prefill_tokens_per_sec": 671.8016,
+          "ttft_p50_ms": 37.4548,
+          "ttft_p95_ms": 193.4633
+        },
+        "4": {
+          "aggregate_decode_tokens_per_sec": 47.0903,
+          "decode_tokens_per_sec": 13.1337,
+          "failures": 0,
+          "latency_p50_ms": 2631.0031,
+          "latency_p95_ms": 3073.7389,
+          "peak_rss_bytes": 2177265664,
+          "peak_vram_bytes": 0,
+          "prefill_tokens_per_sec": 247.5617,
+          "ttft_p50_ms": 94.3995,
+          "ttft_p95_ms": 444.6749
+        }
+      },
+      "device": "cpu",
+      "lane": "quality"
+    }
+  },
+  "reference_recipe_id": "transformers-safetensors-reference"
+}

.scratch/distributed-gguf-runtime/evidence/DGR-001/benchmark-config.json

@@ -0,0 +1,118 @@
+{
+  "artifact_storage_root": "/run/media/popov/DATA/llm",
+  "evidence_class": "local-real",
+  "host": {
+    "benchmark_lane": "cpu-controlled-baseline",
+    "llama_cpp_commit": "e920c523e3b8a0163fe498af5bf90df35ff51d25",
+    "llama_cpp_version": "9991",
+    "llama_server_sha256": "fd8fe612970f23e447f2e717cfa51665be06b8d7315ba60556e010f6bca510dd",
+    "converter_sha256": "c819f18fb22927b49fabc3b35d1c9e21ee638b3817eccd1bd4efbcc7116eeb4d",
+    "quantizer_sha256": "bd0cc8c7be6d48aad4755b31062e0e59a887cbadd43dbb8771853d5858bb198f",
+    "transformers_version": "5.13.0"
+  },
+  "plan": {
+    "plan_id": "dgr-001-controlled-whole-model-baseline-v1",
+    "model_id": "Qwen/Qwen2.5-0.5B-Instruct",
+    "model_revision": "7ae557604adf67be50417f59c2c2f167def9a775",
+    "prompts": [
+      {
+        "id": "short-fact",
+        "text": "The capital of France is",
+        "context_class": "short"
+      },
+      {
+        "id": "medium-code",
+        "text": "Complete this Python function without commentary:\n\ndef fibonacci(n):\n    \"\"\"Return the nth Fibonacci number for n >= 0.\"\"\"\n",
+        "context_class": "medium"
+      },
+      {
+        "id": "long-summary",
+        "text": "A distributed inference service divides a transformer across consumer machines. The tracker owns admission, routing, cancellation, accounting, and telemetry, while workers own only model execution. Every request carries an immutable model identity and revision. Workers must reject incompatible protocol versions and resource demands before allocating large buffers. Activation tensors are chunked, checksummed, bounded by negotiated limits, and propagated with explicit flow-control credits. A caller may disconnect at any time, so cancellation must release queued work, in-flight transfers, and cache reservations without double billing. Retries can occur after network failures, requiring idempotent request identifiers and deterministic completion accounting. The system keeps the existing safetensors path as a correctness reference while a native GGUF path is measured. Benchmarks compare the same prompts, output lengths, sampling policy, device, and concurrency, and they separate near-lossless quality checks from quantized speed and fit claims. Summarize the design priorities in three concise bullet points.",
+        "context_class": "long"
+      }
+    ],
+    "sampling": {
+      "temperature": 0.0,
+      "top_p": 1.0,
+      "top_k": 1,
+      "seed": 1234,
+      "max_output_tokens": 32
+    },
+    "concurrency_levels": [1, 4],
+    "repeats": 3,
+    "warmup_requests": 2
+  },
+  "recipes": [
+    {
+      "id": "transformers-safetensors-reference",
+      "runtime": "transformers-5.13.0",
+      "weight_format": "safetensors",
+      "weight_quantization": "bfloat16",
+      "lane": "quality",
+      "device": "cpu",
+      "artifact_path": "/run/media/popov/DATA/llm/safetensor/models/models--Qwen--Qwen2.5-0.5B-Instruct/snapshots/7ae557604adf67be50417f59c2c2f167def9a775",
+      "artifact_sha256": "e596e9d6205fdc9177569cccd7f8b471b058f66e3630c8e4326d5aad52bd18b6",
+      "source_model_id": "Qwen/Qwen2.5-0.5B-Instruct",
+      "source_model_revision": "7ae557604adf67be50417f59c2c2f167def9a775",
+      "is_reference": true,
+      "notes": "artifact_sha256 is the deterministic digest of every snapshot path and file byte",
+      "driver": {
+        "type": "transformers",
+        "model_path": "/run/media/popov/DATA/llm/safetensor/models/models--Qwen--Qwen2.5-0.5B-Instruct/snapshots/7ae557604adf67be50417f59c2c2f167def9a775",
+        "device": "cpu",
+        "dtype": "bfloat16",
+        "threads": 16
+      }
+    },
+    {
+      "id": "llama-cpp-near-lossless-quality",
+      "runtime": "llama.cpp-9991-e920c523",
+      "weight_format": "gguf",
+      "weight_quantization": "bfloat16",
+      "lane": "quality",
+      "device": "cpu",
+      "artifact_path": "/run/media/popov/DATA/llm/dgr-001/Qwen2.5-0.5B-Instruct-7ae5576-BF16.gguf",
+      "artifact_sha256": "e842fdc35d7f00fda95a54e1b51731ba1d196aea45065cc9f46925fdc1d6f862",
+      "source_model_id": "Qwen/Qwen2.5-0.5B-Instruct",
+      "source_model_revision": "7ae557604adf67be50417f59c2c2f167def9a775",
+      "is_reference": false,
+      "notes": "Converted directly from the exact mounted safetensors revision while preserving BF16 weights with pinned llama.cpp",
+      "driver": {
+        "type": "llama-cpp-server",
+        "binary": "/run/media/popov/d/DEV/llamacpp/llama.cpp/build/bin/llama-server",
+        "binary_sha256": "fd8fe612970f23e447f2e717cfa51665be06b8d7315ba60556e010f6bca510dd",
+        "gguf_path": "/run/media/popov/DATA/llm/dgr-001/Qwen2.5-0.5B-Instruct-7ae5576-BF16.gguf",
+        "device": "cpu",
+        "threads": 16,
+        "n_parallel": 4,
+        "context_per_slot": 512,
+        "n_gpu_layers": 0
+      }
+    },
+    {
+      "id": "llama-cpp-quantized-performance-fit",
+      "runtime": "llama.cpp-9991-e920c523",
+      "weight_format": "gguf",
+      "weight_quantization": "Q4_K_M",
+      "lane": "performance-fit",
+      "device": "cpu",
+      "artifact_path": "/run/media/popov/DATA/llm/dgr-001/Qwen2.5-0.5B-Instruct-7ae5576-Q4_K_M.gguf",
+      "artifact_sha256": "a88e3f570e2efeaf06b50df9859db2c70d8646aa3a2c94a14e14d5797a2921a5",
+      "source_model_id": "Qwen/Qwen2.5-0.5B-Instruct",
+      "source_model_revision": "7ae557604adf67be50417f59c2c2f167def9a775",
+      "is_reference": false,
+      "notes": "Quantized from the exact-revision F16 GGUF with pinned llama-quantize",
+      "driver": {
+        "type": "llama-cpp-server",
+        "binary": "/run/media/popov/d/DEV/llamacpp/llama.cpp/build/bin/llama-server",
+        "binary_sha256": "fd8fe612970f23e447f2e717cfa51665be06b8d7315ba60556e010f6bca510dd",
+        "gguf_path": "/run/media/popov/DATA/llm/dgr-001/Qwen2.5-0.5B-Instruct-7ae5576-Q4_K_M.gguf",
+        "device": "cpu",
+        "threads": 16,
+        "n_parallel": 4,
+        "context_per_slot": 512,
+        "n_gpu_layers": 0
+      }
+    }
+  ]
+}

.scratch/distributed-gguf-runtime/evidence/DGR-001/commands.txt

@@ -0,0 +1,51 @@
+# Exact source snapshot (already present on mounted storage)
+SOURCE=/run/media/popov/DATA/llm/safetensor/models/models--Qwen--Qwen2.5-0.5B-Instruct/snapshots/7ae557604adf67be50417f59c2c2f167def9a775
+LLAMA=/run/media/popov/d/DEV/llamacpp/llama.cpp
+ROCM_PY=/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv-rocm/bin/python
+PROJECT_PY=/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python
+OUT=/run/media/popov/DATA/llm/dgr-001
+
+# Converter support check (no writes)
+$ROCM_PY $LLAMA/convert_hf_to_gguf.py "$SOURCE" --outtype f16 --outfile "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-F16.gguf" --dry-run
+
+# Exact-revision near-lossless and performance-fit artifacts
+$ROCM_PY $LLAMA/convert_hf_to_gguf.py "$SOURCE" --outtype f16 --outfile "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-F16.gguf"
+$LLAMA/build/bin/llama-quantize "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-F16.gguf" "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-Q4_K_M.gguf" Q4_K_M
+$ROCM_PY $LLAMA/convert_hf_to_gguf.py "$SOURCE" --outtype bf16 --outfile "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-BF16.gguf"
+
+# Runtime and artifact identity
+git -C "$LLAMA" rev-parse HEAD
+$LLAMA/build/bin/llama-server --version
+sha256sum "$LLAMA/build/bin/llama-server" "$LLAMA/convert_hf_to_gguf.py" "$LLAMA/build/bin/llama-quantize"
+sha256sum "$SOURCE/model.safetensors" "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-BF16.gguf" "$OUT/Qwen2.5-0.5B-Instruct-7ae5576-Q4_K_M.gguf"
+
+# Deterministic complete-snapshot digest used by benchmark-config.json
+PYTHONPATH=packages/node $ROCM_PY - <<'PY'
+from pathlib import Path
+from meshnet_node.recipe_drivers import _artifact_sha256
+print(_artifact_sha256(Path('/run/media/popov/DATA/llm/safetensor/models/models--Qwen--Qwen2.5-0.5B-Instruct/snapshots/7ae557604adf67be50417f59c2c2f167def9a775')))
+PY
+
+# Canonical opt-in local-real benchmark
+MESHNET_ENABLE_REAL_INFERENCE_TESTS=1 PYTHONPATH=packages/node $ROCM_PY -m meshnet_node.recipe_benchmark \
+  --config .scratch/distributed-gguf-runtime/evidence/DGR-001/benchmark-config.json \
+  --json-out .scratch/distributed-gguf-runtime/evidence/DGR-001/results.json \
+  --summary-out .scratch/distributed-gguf-runtime/evidence/DGR-001/results.txt
+
+# Distil the baseline and evaluate immutable v1
+PYTHONPATH=packages/node $PROJECT_PY - <<'PY'
+from pathlib import Path
+import json
+from meshnet_node.performance_contract import baseline_from_report, evaluate_contract, load_contract
+root = Path('.scratch/distributed-gguf-runtime/evidence/DGR-001')
+report = json.loads((root / 'results.json').read_text())
+contract = load_contract(root / 'performance-contract.json')
+(root / 'baseline.json').write_text(json.dumps(baseline_from_report(report), indent=2, sort_keys=True) + '\n')
+(root / 'contract-evaluation.json').write_text(json.dumps(evaluate_contract(contract, report).to_dict(), indent=2, sort_keys=True) + '\n')
+PY
+
+# Deterministic verification
+PYTHONPATH=packages/node $PROJECT_PY -m pytest -q tests/test_recipe_benchmark.py
+PYTHONPATH=packages/node $PROJECT_PY -m pytest -q
+PYTHONPATH=packages/node $PROJECT_PY -m compileall -q packages tests
+git diff --check

.scratch/distributed-gguf-runtime/evidence/DGR-001/contract-evaluation.json

@@ -0,0 +1,71 @@
+{
+  "contract_version": 1,
+  "fit_benefit": true,
+  "plan_id": "dgr-001-controlled-whole-model-baseline-v1",
+  "quality_lane_pass": false,
+  "rationale": [
+    "the near-lossless quality lane failed: the GGUF runtime disagrees with the safetensors reference beyond what near-lossless weights can explain",
+    "a meaningful speed benefit was measured",
+    "a meaningful fit benefit was measured"
+  ],
+  "recipes": [
+    {
+      "comparable": true,
+      "failures": 0,
+      "fit_benefit": false,
+      "incomparable_reason": "",
+      "lane": "quality",
+      "measurements": {
+        "aggregate_concurrency": 4,
+        "aggregate_throughput_speedup": 4.4882,
+        "artifact_size_ratio": 0.9946,
+        "artifact_size_win": false,
+        "compared_prompts": 3,
+        "decode_speedup": 1.9064,
+        "exact_match_rate": 0.3333,
+        "expected_prompts": 3,
+        "failure_rate": 0.0,
+        "mean_similarity": 0.9471,
+        "resident_memory_ratio": 0.5722,
+        "ttft_ratio": 1.1729
+      },
+      "quality_pass": false,
+      "reasons": [
+        "single-request decode 1.91x reference (>= 1.25x) at TTFT ratio 1.17",
+        "aggregate throughput at concurrency 4 is 4.49x reference (>= 1.25x)",
+        "peak resident memory is 0.57x reference (<= 0.75x)",
+        "quality lane exact-match 0.33 / similarity 0.947 versus the reference (fail)"
+      ],
+      "recipe_id": "llama-cpp-near-lossless-quality",
+      "speed_benefit": false
+    },
+    {
+      "comparable": true,
+      "failures": 0,
+      "fit_benefit": true,
+      "incomparable_reason": "",
+      "lane": "performance-fit",
+      "measurements": {
+        "aggregate_concurrency": 4,
+        "aggregate_throughput_speedup": 4.3826,
+        "artifact_size_ratio": 0.398,
+        "artifact_size_win": true,
+        "decode_speedup": 3.6869,
+        "failure_rate": 0.0,
+        "resident_memory_ratio": 0.2793,
+        "ttft_ratio": 0.6367
+      },
+      "quality_pass": null,
+      "reasons": [
+        "single-request decode 3.69x reference (>= 1.25x) at TTFT ratio 0.64",
+        "aggregate throughput at concurrency 4 is 4.38x reference (>= 1.25x)",
+        "peak resident memory is 0.28x reference (<= 0.75x)"
+      ],
+      "recipe_id": "llama-cpp-quantized-performance-fit",
+      "speed_benefit": true
+    }
+  ],
+  "speed_benefit": true,
+  "stop_condition_met": true,
+  "verdict": "stop"
+}

.scratch/distributed-gguf-runtime/evidence/DGR-001/known-unrelated-failure.md

@@ -0,0 +1,32 @@
+# Observed pre-existing intermittent tracker race
+
+This file records an earlier unrelated timing observation; it is **not** the
+final DGR-001 verification result.
+
+Test:
+
+```text
+tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy
+```
+
+One earlier full-suite run produced:
+
+```text
+1 failed, 745 passed, 13 skipped
+```
+
+A five-run isolated retry matrix reproduced the same rate on both branches:
+
+```text
+DGR-001 branch: 4/5 passed, 1/5 failed
+clean d904c40:  4/5 passed, 1/5 failed
+```
+
+The final full-suite run on the exact hardened DGR-001 state completed green:
+
+```text
+749 passed, 13 skipped in 251.42s
+```
+
+The earlier race was therefore timing-sensitive, pre-existing, and unrelated
+to the DGR-001 benchmark/contract files.

.scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json

@@ -0,0 +1,44 @@
+{
+  "schema_version": 1,
+  "contract_version": 1,
+  "locked_at": "2026-07-13T00:00:00Z",
+  "locked_by": "DGR-001",
+  "plan_id": "dgr-001-controlled-whole-model-baseline-v1",
+  "thresholds": {
+    "min_decode_speedup": 1.25,
+    "max_ttft_ratio": 1.25,
+    "min_aggregate_throughput_speedup": 1.25,
+    "max_resident_memory_ratio": 0.75,
+    "max_artifact_size_ratio": 0.6,
+    "min_quality_exact_match_rate": 0.9,
+    "min_quality_mean_similarity": 0.97,
+    "max_failure_rate": 0.0
+  },
+  "baseline": {
+    "status": "pending-real-evidence",
+    "required_evidence_class": "local-real",
+    "required_recipes": [
+      "transformers-safetensors-reference",
+      "llama-cpp-near-lossless-quality",
+      "llama-cpp-quantized-performance-fit"
+    ],
+    "required_concurrency_levels": [
+      1,
+      4
+    ],
+    "required_controlled_variables": [
+      "model architecture",
+      "model revision",
+      "machine and device",
+      "formatted prompts and context lengths",
+      "output length and greedy sampling policy"
+    ],
+    "required_plan_sha256": "efe24690a9a7164bac6ab3fd0a6b22f078fc08aaefcfb96210ddf154e6050570",
+    "minimum_prompt_count": 3,
+    "minimum_repeats": 3,
+    "minimum_output_tokens": 32,
+    "required_device": "cpu"
+  },
+  "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.",
+  "notes": "Quantized performance-fit output drift is reported as advisory only. It is not numerical-equivalence evidence. DGR-014 consumes this immutable v1 contract."
+}

.scratch/distributed-gguf-runtime/evidence/DGR-001/results.txt

@@ -0,0 +1,10 @@
+Recipe benchmark dgr-001-controlled-whole-model-baseline-v1 (local-real)
+model Qwen/Qwen2.5-0.5B-Instruct@7ae557604adf67be50417f59c2c2f167def9a775
+transformers-safetensors-reference     [quality         ] c= 1 ttft p50/p95     37.5/   193.5 ms; prefill   671.8 tok/s; decode   46.1 tok/s; aggregate    40.3 tok/s; rss  1.94 GB; vram  0.00 GB; artifact  1.00 GB; failures 0
+transformers-safetensors-reference     [quality         ] c= 4 ttft p50/p95     94.4/   444.7 ms; prefill   247.6 tok/s; decode   13.1 tok/s; aggregate    47.1 tok/s; rss  2.18 GB; vram  0.00 GB; artifact  1.00 GB; failures 0
+llama-cpp-near-lossless-quality        [quality         ] c= 1 ttft p50/p95     43.9/   107.0 ms; prefill  1427.2 tok/s; decode   88.0 tok/s; aggregate    73.8 tok/s; rss  1.11 GB; vram  0.00 GB; artifact  0.99 GB; failures 0
+llama-cpp-near-lossless-quality        [quality         ] c= 4 ttft p50/p95     33.6/   128.5 ms; prefill  1077.8 tok/s; decode   73.9 tok/s; aggregate   211.4 tok/s; rss  1.13 GB; vram  0.00 GB; artifact  0.99 GB; failures 0
+llama-cpp-quantized-performance-fit    [performance-fit ] c= 1 ttft p50/p95     23.8/   237.7 ms; prefill   783.4 tok/s; decode  170.1 tok/s; aggregate   110.0 tok/s; rss  0.54 GB; vram  0.00 GB; artifact  0.40 GB; failures 0
+llama-cpp-quantized-performance-fit    [performance-fit ] c= 4 ttft p50/p95     67.9/   431.8 ms; prefill   474.3 tok/s; decode   83.5 tok/s; aggregate   206.4 tok/s; rss  0.57 GB; vram  0.00 GB; artifact  0.40 GB; failures 0
+drift llama-cpp-near-lossless-quality  vs transformers-safetensors-reference exact 0.33; similarity 0.947 (gated)
+drift llama-cpp-quantized-performance-fit vs transformers-safetensors-reference exact 0.00; similarity 0.456 (advisory)

.scratch/distributed-gguf-runtime/issues/01-lock-the-safetensors-versus-gguf-performance-contract.md

@@ -1,6 +1,6 @@
 # 01 — Lock the safetensors-versus-GGUF performance contract
 
-Status: ready-for-agent
+Status: done
 
 ## Mandatory fresh-session context
 

.scratch/distributed-gguf-runtime/prd.json

@@ -6,7 +6,7 @@
     {
       "id": "DGR-001",
       "title": "Lock the safetensors-versus-GGUF performance contract",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/01-lock-the-safetensors-versus-gguf-performance-contract.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a runtime engineer, I need a controlled baseline so that GGUF work proceeds from measured speed, memory, and quality rather than reputation.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/01-lock-the-safetensors-versus-gguf-performance-contract.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a runtime engineer, I need a controlled baseline so that GGUF work proceeds from measured speed, memory, and quality rather than reputation.",
       "acceptanceCriteria": [
         "Benchmark the same model architecture/revision, machine, prompts, context lengths, output lengths, sampling policy, and concurrency across the current Transformers/safetensors recipe and whole-model llama.cpp recipes.",
         "Separate correctness/quality lanes from quantized performance/fit lanes instead of claiming BF16 and Q4 are numerically equivalent.",
@@ -27,14 +27,14 @@
         "Update only this story issue to Status: done after every acceptance criterion and quality gate passes"
       ],
       "priority": 2,
-      "passes": false,
+      "passes": true,
       "notes": "Source issue: .scratch/distributed-gguf-runtime/issues/01-lock-the-safetensors-versus-gguf-performance-contract.md",
       "dependsOn": []
     },
     {
       "id": "DGR-002",
       "title": "Adopt the versioned gRPC Shard protocol",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/02-adopt-the-versioned-grpc-shard-protocol.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a node developer, I need a battle-proven streaming protocol so that Python and C++ Shards communicate without a custom socket protocol.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/02-adopt-the-versioned-grpc-shard-protocol.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a node developer, I need a battle-proven streaming protocol so that Python and C++ Shards communicate without a custom socket protocol.",
       "acceptanceCriteria": [
         "Add a Protocol Buffers schema for capability, health, session stream, release, and cancellation operations.",
         "Define one long-lived bidirectional gRPC stream per Route Session Activation Seam with deadlines, cancellation, flow control, and structured errors.",
@@ -61,7 +61,7 @@
     {
       "id": "DGR-003",
       "title": "Define exact Artifact and runtime recipe identity",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs the Tracker, I need exact compatibility identity so that only numerically and operationally compatible Shards form an Inference Route.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs the Tracker, I need exact compatibility identity so that only numerically and operationally compatible Shards form an Inference Route.",
       "acceptanceCriteria": [
         "Separate weight quantization, activation dtype, compute dtype, KV dtype/layout, tokenizer revision, architecture adapter, backend, and runtime version.",
         "Bind derivative or split artifacts to an exact source Model Artifact hash and Shard range.",
@@ -89,7 +89,7 @@
     {
       "id": "DGR-004",
       "title": "Create the reproducible pinned llama.cpp patch stack",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/04-create-the-reproducible-pinned-llama-cpp-patch-stack.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a maintainer, I need a small auditable fork boundary so that upstream updates do not turn the runtime into an unmaintainable stitched codebase.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/04-create-the-reproducible-pinned-llama-cpp-patch-stack.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a maintainer, I need a small auditable fork boundary so that upstream updates do not turn the runtime into an unmaintainable stitched codebase.",
       "acceptanceCriteria": [
         "Pin one exact llama.cpp commit through a reproducible source dependency mechanism.",
         "Store a numbered minimal patch stack separately from Meshnet networking code.",
@@ -120,7 +120,7 @@
     {
       "id": "DGR-005",
       "title": "Implement dense-Llama range-aware GGUF ownership",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/05-implement-dense-llama-range-aware-gguf-ownership.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a node, I need to map only my assigned dense-Llama Shard so that aggregate consumer memory can hold a model larger than one node.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/05-implement-dense-llama-range-aware-gguf-ownership.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a node, I need to map only my assigned dense-Llama Shard so that aggregate consumer memory can hold a model larger than one node.",
       "acceptanceCriteria": [
         "Register and allocate only `blk.N.*` tensors in the assigned range.",
         "Load embeddings only for the head and final norm/LM head only for the tail, including tied embeddings.",
@@ -151,7 +151,7 @@
     {
       "id": "DGR-006",
       "title": "Implement architecture-defined boundary input/output",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/06-implement-architecture-defined-boundary-input-output.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a Shard, I need to consume and emit the correct transformer boundary state so that disjoint processes reproduce whole-model execution.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/06-implement-architecture-defined-boundary-input-output.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a Shard, I need to consume and emit the correct transformer boundary state so that disjoint processes reproduce whole-model execution.",
       "acceptanceCriteria": [
         "Head accepts token IDs and owns token embedding.",
         "Middle/tail bypass token embedding and accept the named boundary bundle.",
@@ -183,7 +183,7 @@
     {
       "id": "DGR-007",
       "title": "Add isolated concurrent local Hot KV State",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/07-add-isolated-concurrent-local-hot-kv-state.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a client, I need concurrent Route Sessions to retain independent per-Shard cache so that one request cannot clear or corrupt another.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/07-add-isolated-concurrent-local-hot-kv-state.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a client, I need concurrent Route Sessions to retain independent per-Shard cache so that one request cannot clear or corrupt another.",
       "acceptanceCriteria": [
         "Map `(Route Session ID, route epoch)` to an isolated llama sequence or bounded context.",
         "Allocate KV only for owned layers.",
@@ -214,7 +214,7 @@
     {
       "id": "DGR-008",
       "title": "Build the standalone C++ gRPC Shard worker",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/08-build-the-standalone-c-grpc-shard-worker.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a node runtime, I need one supervised native process so that llama.cpp internals remain behind a stable project-owned protocol.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/08-build-the-standalone-c-grpc-shard-worker.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a node runtime, I need one supervised native process so that llama.cpp internals remain behind a stable project-owned protocol.",
       "acceptanceCriteria": [
         "Worker exposes capability, health, session stream, release, cancellation, and metrics services from DGR-002.",
         "Worker loads one exact Artifact/recipe/Shard identity and refuses mismatched requests.",
@@ -249,7 +249,7 @@
     {
       "id": "DGR-009",
       "title": "Integrate the native worker with Meshnet",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/09-integrate-the-native-worker-with-meshnet.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs the existing node service, I need a GGUF Shard backend adapter so that the Tracker, relay, billing, telemetry, and capability admission remain the sole control plane.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/09-integrate-the-native-worker-with-meshnet.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs the existing node service, I need a GGUF Shard backend adapter so that the Tracker, relay, billing, telemetry, and capability admission remain the sole control plane.",
       "acceptanceCriteria": [
         "Implement the existing model-backend surface without changing Transformers behavior.",
         "Registration carries exact validated GGUF recipe, Shard, backend and concurrency/KV capacity.",
@@ -281,7 +281,7 @@
     {
       "id": "DGR-010",
       "title": "Pass local real-model two-process acceptance",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/10-pass-local-real-model-two-process-acceptance.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a release engineer, I need real local distributed parity before involving network variability.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/10-pass-local-real-model-two-process-acceptance.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a release engineer, I need real local distributed parity before involving network variability.",
       "acceptanceCriteria": [
         "Two local worker processes open disjoint dense-Llama ranges from the certified Artifact.",
         "Prefill and at least 32 greedy decode tokens match whole-model llama.cpp within the certified tolerance.",
@@ -314,7 +314,7 @@
     {
       "id": "DGR-011",
       "title": "Pass a real heterogeneous two-machine route",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/11-pass-a-real-heterogeneous-two-machine-route.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a consumer-hardware operator, I need two physical machines to execute one GGUF model so that the distributed claim is real.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/11-pass-a-real-heterogeneous-two-machine-route.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a consumer-hardware operator, I need two physical machines to execute one GGUF model so that the distributed claim is real.",
       "acceptanceCriteria": [
         "Tracker selects two physical nodes with disjoint Shards and one exact certified recipe/compatibility class.",
         "Actual CPU/GPU execution occurs on both nodes; synthetic workers do not satisfy acceptance.",
@@ -347,7 +347,7 @@
     {
       "id": "DGR-012",
       "title": "Implement continuous batching and bounded admission",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/12-implement-continuous-batching-and-bounded-admission.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a node operator, I need active sessions batched safely so that concurrency increases aggregate throughput rather than serializing every request.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/12-implement-continuous-batching-and-bounded-admission.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a node operator, I need active sessions batched safely so that concurrency increases aggregate throughput rather than serializing every request.",
       "acceptanceCriteria": [
         "Node scheduler admits sessions against weight, KV, scratch, and queue budgets.",
         "Compatible decode steps from multiple sessions form llama.cpp batches while preserving per-session positions and outputs.",
@@ -380,7 +380,7 @@
     {
       "id": "DGR-013",
       "title": "Harden failure, cancellation, and restart semantics",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/13-harden-failure-cancellation-and-restart-semantics.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a client, I need failures to be bounded and explicit so that distributed speed does not come with hanging or corrupted generations.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/13-harden-failure-cancellation-and-restart-semantics.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a client, I need failures to be bounded and explicit so that distributed speed does not come with hanging or corrupted generations.",
       "acceptanceCriteria": [
         "Deadlines and heartbeat/health loss terminate blocked stream operations.",
         "Cancellation propagates across every Shard and releases local KV and queued buffers.",
@@ -413,7 +413,7 @@
     {
       "id": "DGR-014",
       "title": "Enforce the GGUF-versus-safetensors release gate",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/14-enforce-the-gguf-versus-safetensors-release-gate.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs the product owner, I need an end-to-end comparison so that the native runtime ships only if it advances model access or performance.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/14-enforce-the-gguf-versus-safetensors-release-gate.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs the product owner, I need an end-to-end comparison so that the native runtime ships only if it advances model access or performance.",
       "acceptanceCriteria": [
         "Run current distributed safetensors and distributed GGUF routes on the same certified model/hardware/network scenario where technically comparable.",
         "Report quality, TTFT, prefill/decode throughput, aggregate concurrency throughput, p95 latency, seam cost, memory, KV pressure, failures, and cleanup.",
@@ -448,7 +448,7 @@
     {
       "id": "DGR-015",
       "title": "Add and certify a Qwen3/Qwen3-MoE adapter",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/15-add-and-certify-a-qwen3-qwen3-moe-adapter.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a client seeking top models, I need a separately certified MoE-capable architecture after the dense runtime proves stable.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/15-add-and-certify-a-qwen3-qwen3-moe-adapter.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a client seeking top models, I need a separately certified MoE-capable architecture after the dense runtime proves stable.",
       "acceptanceCriteria": [
         "Implement explicit tensor ownership, router/top-k, expert/shared-expert, Q/K normalization, boundary bundle, and cache semantics for the selected Qwen3 family recipe.",
         "Do not reuse the dense-Llama adapter through unchecked name substitutions.",
@@ -480,7 +480,7 @@
     {
       "id": "DGR-016",
       "title": "Produce the upstream llama.cpp collaboration package",
-      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/16-produce-the-upstream-llama-cpp-collaboration-package.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker\u2014not a stitched collection of runtimes.\n\nAs a maintainer, I need narrow upstreamable proposals so that our patch burden can shrink without asking llama.cpp to own Meshnet networking.",
+      "description": "MANDATORY FRESH-SESSION CONTEXT: Read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md` and `.scratch/distributed-gguf-runtime/issues/16-produce-the-upstream-llama-cpp-collaboration-package.md` completely before coding. Read the evidence handoff for every dependency. The global goal is performant concurrent inference for models larger than one consumer node, using Meshnet as the sole control plane, gRPC/Protobuf as the Shard protocol, and a small pinned llama.cpp worker—not a stitched collection of runtimes.\n\nAs a maintainer, I need narrow upstreamable proposals so that our patch burden can shrink without asking llama.cpp to own Meshnet networking.",
       "acceptanceCriteria": [
         "Separate generic llama.cpp hooks from Meshnet protocol/control-plane code.",
         "Prepare minimal reproducible examples and tests for range-aware loading, boundary input/output, and layer-filtered KV.",

packages/node/meshnet_node/performance_contract.py

@@ -0,0 +1,697 @@
+"""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 hashlib
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Mapping
+
+from .recipe_benchmark import Lane, REPORT_SCHEMA_VERSION
+
+# Layout of the contract document understood by this reader.
+CONTRACT_SCHEMA_VERSION = 1
+
+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 _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 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")
+    for entry in recipes:
+        recipe = entry.get("recipe", {})
+        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")
+        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"],
+        "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

packages/node/meshnet_node/recipe_benchmark.py

@@ -0,0 +1,674 @@
+"""Controlled safetensors-versus-GGUF recipe benchmark.
+
+This is a *model recipe* benchmark, unlike
+:mod:`meshnet_node.route_session_benchmark`, which is a transport harness.  It
+answers one question: on one machine, with one model revision and one fixed
+workload, what do the Transformers/safetensors recipe and the whole-model
+llama.cpp/GGUF recipes actually cost in speed, memory, fit, and output drift?
+
+Two ideas keep the answer honest.
+
+**Lanes.**  A recipe belongs to exactly one :class:`Lane`.  The quality lane
+holds near-lossless recipes (bf16/f16 weights) whose outputs may legitimately be
+compared for numerical agreement.  The performance-fit lane holds quantized
+recipes (Q8_0, Q4_K_M, ...).  Quantized recipes are judged on speed, memory, and
+artifact size only; their drift is *reported* but never read as evidence that
+Q4 and bf16 are numerically equivalent, because they are not.  The lane is a
+property of the recipe, so nothing downstream can quietly cross the boundary.
+
+**Drivers.**  The measurement core here is pure and runtime-free: it drives a
+:class:`RecipeDriver` and computes metrics.  Real runtimes live in
+:mod:`meshnet_node.recipe_drivers` and are imported only on demand, which keeps
+the default test suite deterministic, GPU-free and model-download-free while the
+same code path produces the real evidence.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import asdict, dataclass, field
+from difflib import SequenceMatcher
+from enum import Enum
+from pathlib import Path
+from typing import Any, Protocol, Sequence
+
+# Layout of the report document produced by :func:`build_report`.
+REPORT_SCHEMA_VERSION = 1
+
+
+class Lane(str, Enum):
+    """Why a recipe is being measured at all.
+
+    ``QUALITY`` recipes carry near-lossless weights, so comparing their output
+    with the reference recipe is a meaningful correctness signal.
+    ``PERFORMANCE_FIT`` recipes carry quantized weights: they exist to be faster
+    or to fit, and their drift is descriptive, never a pass/fail equivalence
+    claim.
+    """
+
+    QUALITY = "quality"
+    PERFORMANCE_FIT = "performance-fit"
+
+
+class BenchmarkError(RuntimeError):
+    """Raised when a benchmark cannot be run as specified."""
+
+
+@dataclass(frozen=True)
+class SamplingPolicy:
+    """The sampling policy every recipe must be given, identically.
+
+    Greedy by default: sampling noise would otherwise be indistinguishable from
+    quantization drift, and the whole point of the quality lane is to tell those
+    two apart.
+    """
+
+    temperature: float = 0.0
+    top_p: float = 1.0
+    top_k: int = 1
+    seed: int = 1234
+    max_output_tokens: int = 64
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class PromptSpec:
+    """One fixed prompt, tagged with the context length it is meant to exercise."""
+
+    id: str
+    text: str
+    context_class: str = "short"
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class BenchmarkPlan:
+    """The controlled variables: identical for every recipe in a report.
+
+    A plan is the experiment.  If two recipes were measured under different
+    plans, their numbers are not comparable and the report must not pretend they
+    are, so the plan is recorded once at the top of the document rather than
+    per-recipe.
+    """
+
+    plan_id: str
+    model_id: str
+    model_revision: str
+    prompts: tuple[PromptSpec, ...]
+    sampling: SamplingPolicy = SamplingPolicy()
+    concurrency_levels: tuple[int, ...] = (1, 4)
+    repeats: int = 1
+    warmup_requests: int = 1
+
+    def __post_init__(self) -> None:
+        if not self.prompts:
+            raise BenchmarkError("a benchmark plan needs at least one prompt")
+        if not self.concurrency_levels or any(level < 1 for level in self.concurrency_levels):
+            raise BenchmarkError("concurrency levels must all be >= 1")
+        if self.repeats < 1:
+            raise BenchmarkError("repeats must be >= 1")
+        if 1 not in self.concurrency_levels or 4 not in self.concurrency_levels:
+            raise BenchmarkError("a controlled baseline must include concurrency levels 1 and 4")
+
+    def to_dict(self) -> dict:
+        return {
+            "plan_id": self.plan_id,
+            "model_id": self.model_id,
+            "model_revision": self.model_revision,
+            "prompts": [prompt.to_dict() for prompt in self.prompts],
+            "sampling": self.sampling.to_dict(),
+            "concurrency_levels": list(self.concurrency_levels),
+            "repeats": self.repeats,
+            "warmup_requests": self.warmup_requests,
+        }
+
+
+@dataclass(frozen=True)
+class RecipeSpec:
+    """One runtime recipe under test.
+
+    ``is_reference`` marks the single recipe every other recipe's output drift is
+    measured against — the current Transformers/safetensors route, which
+    decision Gate 8 keeps as the correctness backend.
+    """
+
+    id: str
+    runtime: str
+    weight_format: str
+    weight_quantization: str
+    lane: Lane
+    device: str
+    artifact_path: str = ""
+    source_model_id: str = ""
+    source_model_revision: str = ""
+    artifact_sha256: str = ""
+    is_reference: bool = False
+    notes: str = ""
+
+    def to_dict(self) -> dict:
+        data = asdict(self)
+        data["lane"] = self.lane.value
+        return data
+
+
+@dataclass(frozen=True)
+class LoadStats:
+    """What loading the recipe cost, before any token is generated."""
+
+    artifact_bytes: int
+    load_ms: float
+    rss_bytes: int = 0
+    vram_bytes: int = 0
+    backend_detail: str = ""
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class GenerationSample:
+    """One completed generation as reported by a driver.
+
+    ``prefill_ms``/``decode_ms`` are the runtime's own split where it exposes one
+    (llama.cpp does); drivers that cannot split honestly report ``prefill_ms`` as
+    the time to the first token.  ``queue_wait_ms`` separates time spent waiting
+    for a runtime slot from time spent computing, so a concurrency-4 TTFT is not
+    silently read as a slower prefill.
+    """
+
+    text: str
+    prompt_tokens: int
+    decode_tokens: int
+    ttft_ms: float
+    prefill_ms: float
+    decode_ms: float
+    total_ms: float
+    queue_wait_ms: float = 0.0
+
+
+class RecipeDriver(Protocol):
+    """The seam every runtime implements; the measurement core knows nothing else."""
+
+    def load(self) -> LoadStats:
+        """Load the artifact and return its cost."""
+
+    def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample:
+        """Run one complete generation under the given sampling policy."""
+
+    def memory_probe(self) -> tuple[int, int]:
+        """Return ``(rss_bytes, vram_bytes)`` observed right now."""
+
+    def close(self) -> None:
+        """Release the runtime."""
+
+
+@dataclass(frozen=True)
+class RequestOutcome:
+    """One request attempt, successful or not.
+
+    A failure is a first-class result, not an exception that aborts the run: a
+    recipe that cannot sustain concurrency 4 has told us something, and the
+    report must carry it rather than lose it.
+    """
+
+    recipe_id: str
+    concurrency: int
+    prompt_id: str
+    repeat: int
+    ok: bool
+    latency_ms: float
+    ttft_ms: float = 0.0
+    prefill_ms: float = 0.0
+    decode_ms: float = 0.0
+    queue_wait_ms: float = 0.0
+    prompt_tokens: int = 0
+    decode_tokens: int = 0
+    text: str = ""
+    error: str = ""
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def _percentile(values: Sequence[float], percentile: float) -> float:
+    """Nearest-rank percentile; 0.0 for an empty sample."""
+    ordered = sorted(values)
+    if not ordered:
+        return 0.0
+    rank = max(1, -(-len(ordered) * percentile // 100))
+    return round(ordered[int(rank) - 1], 4)
+
+
+def _mean(values: Sequence[float]) -> float:
+    return round(statistics.fmean(values), 4) if values else 0.0
+
+
+@dataclass(frozen=True)
+class ConcurrencyMetrics:
+    """Aggregate metrics for one recipe at one concurrency level."""
+
+    concurrency: int
+    requests: int
+    failures: int
+    wall_ms: float
+    ttft_p50_ms: float
+    ttft_p95_ms: float
+    latency_p50_ms: float
+    latency_p95_ms: float
+    prefill_tokens_per_sec: float
+    decode_tokens_per_sec: float
+    aggregate_decode_tokens_per_sec: float
+    peak_rss_bytes: int
+    peak_vram_bytes: int
+    failure_reasons: tuple[str, ...] = ()
+
+    def to_dict(self) -> dict:
+        data = asdict(self)
+        data["failure_reasons"] = list(self.failure_reasons)
+        return data
+
+
+def summarize_concurrency(
+    outcomes: Sequence[RequestOutcome],
+    *,
+    concurrency: int,
+    wall_ms: float,
+    peak_rss_bytes: int,
+    peak_vram_bytes: int,
+) -> ConcurrencyMetrics:
+    """Aggregate one recipe/concurrency cell.
+
+    Per-request rates are averaged over successful requests; aggregate
+    throughput is total decoded tokens over the wall clock of the whole cell,
+    which is the only figure that credits a runtime for overlapping work.
+    """
+    ok = [outcome for outcome in outcomes if outcome.ok]
+    failures = [outcome for outcome in outcomes if not outcome.ok]
+    decode_tokens = sum(outcome.decode_tokens for outcome in ok)
+
+    prefill_rates = [
+        outcome.prompt_tokens / (outcome.prefill_ms / 1000)
+        for outcome in ok
+        if outcome.prefill_ms > 0 and outcome.prompt_tokens
+    ]
+    decode_rates = [
+        outcome.decode_tokens / (outcome.decode_ms / 1000)
+        for outcome in ok
+        if outcome.decode_ms > 0 and outcome.decode_tokens
+    ]
+    return ConcurrencyMetrics(
+        concurrency=concurrency,
+        requests=len(outcomes),
+        failures=len(failures),
+        wall_ms=round(wall_ms, 4),
+        ttft_p50_ms=_percentile([outcome.ttft_ms for outcome in ok], 50),
+        ttft_p95_ms=_percentile([outcome.ttft_ms for outcome in ok], 95),
+        latency_p50_ms=_percentile([outcome.latency_ms for outcome in ok], 50),
+        latency_p95_ms=_percentile([outcome.latency_ms for outcome in ok], 95),
+        prefill_tokens_per_sec=_mean(prefill_rates),
+        decode_tokens_per_sec=_mean(decode_rates),
+        aggregate_decode_tokens_per_sec=round(decode_tokens / max(1e-6, wall_ms / 1000), 4),
+        peak_rss_bytes=peak_rss_bytes,
+        peak_vram_bytes=peak_vram_bytes,
+        failure_reasons=tuple(sorted({outcome.error for outcome in failures if outcome.error})),
+    )
+
+
+@dataclass
+class RecipeMeasurement:
+    """Everything measured for one recipe across every concurrency level."""
+
+    recipe: RecipeSpec
+    load: LoadStats
+    metrics: dict[int, ConcurrencyMetrics] = field(default_factory=dict)
+    outcomes: list[RequestOutcome] = field(default_factory=list)
+    unavailable_reason: str = ""
+
+    @property
+    def available(self) -> bool:
+        return not self.unavailable_reason
+
+    def outputs_by_prompt(self) -> dict[str, str]:
+        """First successful output per prompt, at the lowest concurrency measured.
+
+        Drift is a property of the recipe, not of load: concurrency must not
+        change greedy output, so the least-contended sample is the fair one.
+        """
+        best: dict[str, tuple[int, str]] = {}
+        for outcome in self.outcomes:
+            if not outcome.ok:
+                continue
+            seen = best.get(outcome.prompt_id)
+            if seen is None or outcome.concurrency < seen[0]:
+                best[outcome.prompt_id] = (outcome.concurrency, outcome.text)
+        return {prompt_id: text for prompt_id, (_, text) in best.items()}
+
+    def to_dict(self) -> dict:
+        return {
+            "recipe": self.recipe.to_dict(),
+            "available": self.available,
+            "unavailable_reason": self.unavailable_reason,
+            "load": self.load.to_dict(),
+            "concurrency": {
+                str(level): metrics.to_dict() for level, metrics in sorted(self.metrics.items())
+            },
+            "outcomes": [outcome.to_dict() for outcome in self.outcomes],
+        }
+
+
+@dataclass(frozen=True)
+class DriftReport:
+    """Output drift of one recipe against the reference recipe.
+
+    ``advisory`` is true for every performance-fit recipe: the number is
+    published, but a Q4 recipe disagreeing with bf16 is expected behaviour, not a
+    defect, and no gate may read it as one.
+    """
+
+    recipe_id: str
+    lane: Lane
+    reference_id: str
+    compared_prompts: int
+    exact_match_rate: float
+    mean_similarity: float
+    advisory: bool
+    per_prompt: tuple[dict[str, Any], ...] = ()
+
+    def to_dict(self) -> dict:
+        return {
+            "recipe_id": self.recipe_id,
+            "lane": self.lane.value,
+            "reference_id": self.reference_id,
+            "compared_prompts": self.compared_prompts,
+            "exact_match_rate": self.exact_match_rate,
+            "mean_similarity": self.mean_similarity,
+            "advisory": self.advisory,
+            "per_prompt": list(self.per_prompt),
+        }
+
+
+def _first_divergence(left: str, right: str) -> int:
+    """Index of the first differing character, or -1 when the strings agree."""
+    for index, (a, b) in enumerate(zip(left, right)):
+        if a != b:
+            return index
+    return -1 if len(left) == len(right) else min(len(left), len(right))
+
+
+def compute_drift(
+    measurement: RecipeMeasurement,
+    reference: RecipeMeasurement,
+) -> DriftReport:
+    """Compare one recipe's greedy outputs with the reference recipe's."""
+    reference_outputs = reference.outputs_by_prompt()
+    outputs = measurement.outputs_by_prompt()
+    shared = sorted(set(outputs) & set(reference_outputs))
+
+    per_prompt: list[dict[str, Any]] = []
+    exact = 0
+    similarities: list[float] = []
+    for prompt_id in shared:
+        got, want = outputs[prompt_id], reference_outputs[prompt_id]
+        matches = got == want
+        exact += matches
+        similarity = round(SequenceMatcher(None, want, got).ratio(), 4)
+        similarities.append(similarity)
+        per_prompt.append({
+            "prompt_id": prompt_id,
+            "exact_match": matches,
+            "similarity": similarity,
+            "first_divergence_char": _first_divergence(want, got),
+            "reference_text": want,
+            "recipe_text": got,
+        })
+
+    return DriftReport(
+        recipe_id=measurement.recipe.id,
+        lane=measurement.recipe.lane,
+        reference_id=reference.recipe.id,
+        compared_prompts=len(shared),
+        exact_match_rate=round(exact / len(shared), 4) if shared else 0.0,
+        mean_similarity=_mean(similarities),
+        advisory=measurement.recipe.lane is Lane.PERFORMANCE_FIT,
+        per_prompt=tuple(per_prompt),
+    )
+
+
+class _PeakMemory:
+    """Continuously sample a driver's memory while requests are in flight."""
+
+    def __init__(self, driver: RecipeDriver) -> None:
+        self._driver = driver
+        self.peak_rss = 0
+        self.peak_vram = 0
+
+    def sample(self) -> None:
+        try:
+            rss, vram = self._driver.memory_probe()
+        except Exception:  # a probe must never fail a benchmark run
+            return
+        self.peak_rss = max(self.peak_rss, rss)
+        self.peak_vram = max(self.peak_vram, vram)
+
+    def sample_until(self, stop: threading.Event, interval_s: float = 0.01) -> None:
+        """Sample until ``stop`` is set, including one final post-request probe."""
+        while not stop.wait(interval_s):
+            self.sample()
+        self.sample()
+
+
+def _run_request(
+    driver: RecipeDriver,
+    recipe: RecipeSpec,
+    prompt: PromptSpec,
+    sampling: SamplingPolicy,
+    concurrency: int,
+    repeat: int,
+    memory: _PeakMemory,
+) -> RequestOutcome:
+    started = time.monotonic()
+    try:
+        sample = driver.generate(prompt.text, sampling)
+    except Exception as exc:  # a failed request is data, not a crashed benchmark
+        return RequestOutcome(
+            recipe_id=recipe.id,
+            concurrency=concurrency,
+            prompt_id=prompt.id,
+            repeat=repeat,
+            ok=False,
+            latency_ms=round((time.monotonic() - started) * 1000, 4),
+            error=f"{type(exc).__name__}: {exc}",
+        )
+    finally:
+        memory.sample()
+
+    return RequestOutcome(
+        recipe_id=recipe.id,
+        concurrency=concurrency,
+        prompt_id=prompt.id,
+        repeat=repeat,
+        ok=True,
+        latency_ms=round(sample.total_ms, 4),
+        ttft_ms=round(sample.ttft_ms, 4),
+        prefill_ms=round(sample.prefill_ms, 4),
+        decode_ms=round(sample.decode_ms, 4),
+        queue_wait_ms=round(sample.queue_wait_ms, 4),
+        prompt_tokens=sample.prompt_tokens,
+        decode_tokens=sample.decode_tokens,
+        text=sample.text,
+    )
+
+
+def measure_recipe(
+    driver: RecipeDriver,
+    recipe: RecipeSpec,
+    plan: BenchmarkPlan,
+) -> RecipeMeasurement:
+    """Load one recipe and run the whole plan against it.
+
+    The driver is closed exactly once, whatever happens, so a recipe that dies at
+    concurrency 4 still releases its weights before the next recipe loads.
+    """
+    load = driver.load()
+    measurement = RecipeMeasurement(recipe=recipe, load=load)
+
+    try:
+        for _ in range(plan.warmup_requests):
+            try:
+                driver.generate(plan.prompts[0].text, plan.sampling)
+            except Exception:  # a failing warmup is reported by the real requests
+                break
+
+        for concurrency in plan.concurrency_levels:
+            # Measure each cell independently and sample while requests are in
+            # flight; post-request probes alone miss transient KV/workspace peaks.
+            memory = _PeakMemory(driver)
+            memory.sample()
+            stop_sampling = threading.Event()
+            sampler = threading.Thread(
+                target=memory.sample_until,
+                args=(stop_sampling,),
+                name=f"recipe-memory-{recipe.id}-c{concurrency}",
+                daemon=True,
+            )
+            requests = [
+                (prompt, repeat)
+                for repeat in range(plan.repeats)
+                for prompt in plan.prompts
+                for _ in range(concurrency)
+            ]
+            started = time.monotonic()
+            sampler.start()
+            try:
+                with ThreadPoolExecutor(max_workers=concurrency) as pool:
+                    outcomes = list(pool.map(
+                        lambda item: _run_request(
+                            driver, recipe, item[0], plan.sampling, concurrency, item[1], memory
+                        ),
+                        requests,
+                    ))
+            finally:
+                stop_sampling.set()
+                sampler.join()
+            wall_ms = (time.monotonic() - started) * 1000
+
+            measurement.outcomes.extend(outcomes)
+            measurement.metrics[concurrency] = summarize_concurrency(
+                outcomes,
+                concurrency=concurrency,
+                wall_ms=wall_ms,
+                peak_rss_bytes=memory.peak_rss,
+                peak_vram_bytes=memory.peak_vram,
+            )
+    finally:
+        driver.close()
+
+    return measurement
+
+
+def build_report(
+    plan: BenchmarkPlan,
+    measurements: Sequence[RecipeMeasurement],
+    *,
+    host: dict[str, Any],
+    evidence_class: str,
+) -> dict:
+    """Assemble the machine-readable benchmark document.
+
+    ``evidence_class`` is one of ``synthetic``, ``local-real`` or
+    ``multi-machine-real`` and is never inferred: a report that cannot say how it
+    was produced cannot be trusted by a release gate.
+    """
+    if evidence_class not in {"synthetic", "local-real", "multi-machine-real"}:
+        raise BenchmarkError(f"unknown evidence class {evidence_class!r}")
+
+    references = [m for m in measurements if m.recipe.is_reference]
+    if len(references) != 1:
+        raise BenchmarkError(
+            f"exactly one reference recipe is required, got {len(references)}"
+        )
+    reference = references[0]
+    if reference.recipe.lane is not Lane.QUALITY:
+        raise BenchmarkError("the reference recipe must sit in the quality lane")
+
+    drift = [
+        compute_drift(measurement, reference).to_dict()
+        for measurement in measurements
+        if measurement is not reference and measurement.available
+    ]
+    return {
+        "schema_version": REPORT_SCHEMA_VERSION,
+        "evidence_class": evidence_class,
+        "plan": plan.to_dict(),
+        "host": host,
+        "reference_recipe_id": reference.recipe.id,
+        "recipes": [measurement.to_dict() for measurement in measurements],
+        "drift": drift,
+    }
+
+
+def format_summary(report: dict) -> str:
+    """Render the human-readable companion to the JSON artifact."""
+    plan = report["plan"]
+    lines = [
+        f"Recipe benchmark {plan['plan_id']} ({report['evidence_class']})",
+        f"model {plan['model_id']}@{plan['model_revision']}",
+    ]
+    for entry in report["recipes"]:
+        recipe = entry["recipe"]
+        if not entry["available"]:
+            lines.append(f"{recipe['id']:38} UNAVAILABLE: {entry['unavailable_reason']}")
+            continue
+        artifact_gb = entry["load"]["artifact_bytes"] / 1e9
+        for level, metrics in entry["concurrency"].items():
+            lines.append(
+                f"{recipe['id']:38} [{recipe['lane']:16}] c={level:>2} "
+                f"ttft p50/p95 {metrics['ttft_p50_ms']:8.1f}/{metrics['ttft_p95_ms']:8.1f} ms; "
+                f"prefill {metrics['prefill_tokens_per_sec']:7.1f} tok/s; "
+                f"decode {metrics['decode_tokens_per_sec']:6.1f} tok/s; "
+                f"aggregate {metrics['aggregate_decode_tokens_per_sec']:7.1f} tok/s; "
+                f"rss {metrics['peak_rss_bytes'] / 1e9:5.2f} GB; "
+                f"vram {metrics['peak_vram_bytes'] / 1e9:5.2f} GB; "
+                f"artifact {artifact_gb:5.2f} GB; failures {metrics['failures']}"
+            )
+    for entry in report["drift"]:
+        tag = "advisory" if entry["advisory"] else "gated"
+        lines.append(
+            f"drift {entry['recipe_id']:32} vs {entry['reference_id']:28} "
+            f"exact {entry['exact_match_rate']:.2f}; similarity {entry['mean_similarity']:.3f} ({tag})"
+        )
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Run the controlled safetensors-versus-GGUF recipe benchmark"
+    )
+    parser.add_argument("--config", type=Path, required=True, help="benchmark configuration JSON")
+    parser.add_argument("--json-out", type=Path, help="write the JSON report to this path")
+    parser.add_argument("--summary-out", type=Path, help="write the text summary to this path")
+    args = parser.parse_args(argv)
+
+    from .recipe_drivers import run_configured_benchmark  # heavy runtimes: import on demand
+
+    report = run_configured_benchmark(json.loads(args.config.read_text(encoding="utf-8")))
+    summary = format_summary(report)
+    if args.json_out:
+        args.json_out.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    if args.summary_out:
+        args.summary_out.write_text(summary + "\n", encoding="utf-8")
+    print(summary)
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover - CLI entry point
+    raise SystemExit(main())

packages/node/meshnet_node/recipe_drivers.py

@@ -0,0 +1,664 @@
+"""Real runtime drivers for the recipe benchmark.
+
+This module is the only place that imports torch, transformers, or spawns a
+llama.cpp server, and :mod:`meshnet_node.recipe_benchmark` imports it lazily.
+That keeps the default test suite deterministic, GPU-free and download-free
+while the real evidence runs through exactly the same measurement core.
+
+Fairness is the whole point of a baseline, so both drivers are held to the same
+rules:
+
+* They are handed a **pre-formatted prompt string**.  Neither applies a chat
+  template, because a template applied twice — or differently — by two runtimes
+  would show up as a speed and drift difference that has nothing to do with the
+  runtime.
+* They are given the **same CPU thread budget**, so the comparison measures
+  kernels rather than how many cores each runtime felt entitled to take.
+* They report the runtime's **own prefill/decode split** where it has one, and
+  say so honestly where it does not.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import platform
+import re
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any, Mapping
+
+from .recipe_benchmark import (
+    BenchmarkError,
+    BenchmarkPlan,
+    GenerationSample,
+    Lane,
+    LoadStats,
+    PromptSpec,
+    RecipeSpec,
+    SamplingPolicy,
+    build_report,
+    measure_recipe,
+)
+
+REAL_INFERENCE_ENV = "MESHNET_ENABLE_REAL_INFERENCE_TESTS"
+
+
+def real_inference_enabled() -> bool:
+    """Real runtimes stay off unless the operator opts in explicitly."""
+    return os.environ.get(REAL_INFERENCE_ENV) == "1"
+
+
+def require_real_inference() -> None:
+    if not real_inference_enabled():
+        raise BenchmarkError(
+            f"real model execution is opt-in: set {REAL_INFERENCE_ENV}=1 to run this benchmark"
+        )
+
+
+def _process_rss(pid: int | None = None) -> int:
+    """Resident bytes for a process and its children, or 0 when unobservable."""
+    try:
+        import psutil
+    except ImportError:
+        return 0
+    try:
+        process = psutil.Process(pid) if pid else psutil.Process()
+        total = process.memory_info().rss
+        for child in process.children(recursive=True):
+            try:
+                total += child.memory_info().rss
+            except psutil.Error:
+                continue
+        return int(total)
+    except Exception:
+        return 0
+
+
+def _directory_bytes(path: Path) -> int:
+    if path.is_file():
+        return path.stat().st_size
+    return sum(entry.stat().st_size for entry in path.rglob("*") if entry.is_file())
+
+
+def _artifact_sha256(path: Path) -> str:
+    """Hash an artifact file or a deterministic directory content manifest.
+
+    A file uses the ordinary SHA-256 digest. A directory hashes each sorted
+    relative path, resolved file size, and file bytes, so tokenizer/config drift
+    cannot hide behind a weight-only digest.
+    """
+    digest = hashlib.sha256()
+    if path.is_file():
+        entries = [(None, path)]
+    else:
+        entries = [
+            (entry.relative_to(path).as_posix(), entry)
+            for entry in sorted(path.rglob("*"))
+            if entry.is_file()
+        ]
+        if not entries:
+            raise BenchmarkError(f"artifact directory is empty: {path}")
+
+    for relative, entry in entries:
+        if relative is not None:
+            encoded = relative.encode("utf-8")
+            digest.update(len(encoded).to_bytes(8, "big"))
+            digest.update(encoded)
+            digest.update(entry.stat().st_size.to_bytes(8, "big"))
+        with entry.open("rb") as stream:
+            while chunk := stream.read(8 * 1024 * 1024):
+                digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _host_manifest() -> dict[str, Any]:
+    """Capture non-secret host facts with the report rather than trusting prose."""
+    manifest: dict[str, Any] = {
+        "hostname": socket.gethostname(),
+        "platform": platform.platform(),
+        "python": sys.version.split()[0],
+        "cpu_count": os.cpu_count(),
+    }
+    try:
+        import torch
+        import transformers
+
+        manifest["torch_version"] = torch.__version__
+        manifest["transformers_version"] = transformers.__version__
+        manifest["cuda_available"] = bool(torch.cuda.is_available())
+        if torch.cuda.is_available():
+            manifest["accelerator_name"] = torch.cuda.get_device_name(0)
+            manifest["accelerator_runtime"] = getattr(torch.version, "cuda", None) or getattr(
+                torch.version, "hip", None
+            )
+    except ImportError:
+        manifest["torch_version"] = None
+    return manifest
+
+
+def _validate_config(config: Mapping[str, Any]) -> None:
+    """Reject comparisons that mix models, artifacts, devices, or budgets."""
+    try:
+        plan = config["plan"]
+        root = Path(config["artifact_storage_root"]).resolve(strict=True)
+        recipes = config["recipes"]
+    except (KeyError, TypeError, OSError) as exc:
+        raise BenchmarkError(
+            "benchmark config needs an existing artifact_storage_root, plan, and recipes"
+        ) from exc
+    if not root.is_absolute() or root == Path("/home") or Path("/home") in root.parents:
+        raise BenchmarkError("model artifacts must use configured mounted-drive storage, never /home")
+    if not isinstance(recipes, list) or not recipes:
+        raise BenchmarkError("benchmark config needs at least one recipe")
+
+    sampling = plan.get("sampling", {})
+    if (
+        float(sampling.get("temperature", 0.0)) != 0.0
+        or int(sampling.get("top_k", 1)) != 1
+        or float(sampling.get("top_p", 1.0)) != 1.0
+    ):
+        raise BenchmarkError("the quality comparison requires greedy sampling")
+
+    if len(plan.get("prompts", ())) < 3 or int(plan.get("repeats", 0)) < 3:
+        raise BenchmarkError("contract-grade evidence requires at least 3 prompts and 3 repeats")
+    if int(plan.get("warmup_requests", 0)) < 1:
+        raise BenchmarkError("contract-grade evidence requires at least one warmup")
+    if int(sampling.get("max_output_tokens", 0)) < 32:
+        raise BenchmarkError("contract-grade evidence requires at least 32 output tokens")
+
+    thread_budgets: set[int] = set()
+    max_concurrency = max(int(level) for level in plan.get("concurrency_levels", (1, 4)))
+    for spec in recipes:
+        if spec.get("source_model_id") != plan.get("model_id"):
+            raise BenchmarkError("every recipe must declare the plan's exact source_model_id")
+        if spec.get("source_model_revision") != plan.get("model_revision"):
+            raise BenchmarkError("every recipe must declare the plan's exact source_model_revision")
+
+        digest = spec.get("artifact_sha256", "")
+        if not isinstance(digest, str) or re.fullmatch(r"[0-9a-f]{64}", digest) is None:
+            raise BenchmarkError("every recipe must declare a lowercase SHA-256 artifact digest")
+        artifact = Path(spec.get("artifact_path", "")).resolve(strict=True)
+        if artifact != root and root not in artifact.parents:
+            raise BenchmarkError("every model artifact must be beneath artifact_storage_root")
+        actual_digest = _artifact_sha256(artifact)
+        if not hmac.compare_digest(digest, actual_digest):
+            raise BenchmarkError(
+                f"artifact digest mismatch for {spec.get('id', '<unknown>')}: "
+                f"declared {digest}, measured {actual_digest}"
+            )
+
+        driver = spec.get("driver")
+        if not isinstance(driver, Mapping):
+            raise BenchmarkError("every recipe needs a driver object")
+        kind = driver.get("type")
+        if kind == "transformers":
+            driver_artifact = Path(driver.get("model_path", "")).resolve(strict=True)
+        elif kind == "llama-cpp-server":
+            driver_artifact = Path(driver.get("gguf_path", "")).resolve(strict=True)
+            binary = Path(driver.get("binary", "")).resolve(strict=True)
+            binary_digest = driver.get("binary_sha256", "")
+            if (
+                not isinstance(binary_digest, str)
+                or re.fullmatch(r"[0-9a-f]{64}", binary_digest) is None
+                or not hmac.compare_digest(binary_digest, _artifact_sha256(binary))
+            ):
+                raise BenchmarkError("llama.cpp binary SHA-256 mismatch")
+            if int(driver.get("n_parallel", max_concurrency)) < max_concurrency:
+                raise BenchmarkError("llama.cpp parallel slots must cover maximum concurrency")
+            if driver.get("device", "cpu") != "cpu" or int(driver.get("n_gpu_layers", 0)) != 0:
+                raise BenchmarkError(
+                    "v1 benchmark supports CPU-only llama.cpp until process VRAM is measurable"
+                )
+        else:
+            raise BenchmarkError(f"unknown driver type {kind!r}")
+        if driver_artifact != artifact:
+            raise BenchmarkError("driver artifact path must match the hashed recipe artifact")
+        if driver.get("device", "cpu") != spec.get("device"):
+            raise BenchmarkError("recipe and driver must declare the same device")
+        thread_budgets.add(int(driver.get("threads", 8)))
+
+    if len(thread_budgets) != 1:
+        raise BenchmarkError("every recipe must use the same CPU thread budget")
+
+
+class TransformersDriver:
+    """The current Transformers/safetensors recipe: the correctness reference.
+
+    Generation is a hand-written prefill-then-decode loop rather than
+    ``model.generate`` because the benchmark needs the two phases separated: one
+    forward over the prompt gives an exact prefill time and TTFT, and the cached
+    single-token steps that follow give an exact decode rate.  ``generate`` would
+    hand back one blended number.
+    """
+
+    def __init__(
+        self,
+        model_path: str,
+        *,
+        device: str = "cpu",
+        dtype: str = "bfloat16",
+        threads: int = 8,
+    ) -> None:
+        self.model_path = Path(model_path)
+        self.device = device
+        self.dtype = dtype
+        self.threads = threads
+        self._model: Any = None
+        self._tokenizer: Any = None
+        self._torch: Any = None
+        self._rss_baseline = 0
+
+    def load(self) -> LoadStats:
+        self._rss_baseline = _process_rss()
+        import torch
+        from transformers import AutoModelForCausalLM, AutoTokenizer
+
+        self._torch = torch
+        torch.set_num_threads(self.threads)
+        torch.manual_seed(0)
+
+        started = time.monotonic()
+        self._tokenizer = AutoTokenizer.from_pretrained(
+            str(self.model_path), local_files_only=True
+        )
+        self._model = AutoModelForCausalLM.from_pretrained(
+            str(self.model_path),
+            dtype=getattr(torch, self.dtype),
+            local_files_only=True,
+        )
+        self._model.to(self.device)
+        self._model.eval()
+        load_ms = (time.monotonic() - started) * 1000
+
+        return LoadStats(
+            artifact_bytes=_directory_bytes(self.model_path),
+            load_ms=round(load_ms, 4),
+            rss_bytes=max(0, _process_rss() - self._rss_baseline),
+            vram_bytes=self._vram_bytes(),
+            backend_detail=(
+                f"torch {torch.__version__}; dtype {self.dtype}; "
+                f"device {self.device}; intra-op threads {self.threads}"
+            ),
+        )
+
+    def _vram_bytes(self) -> int:
+        torch = self._torch
+        if torch is None or self.device == "cpu":
+            return 0
+        try:
+            if torch.cuda.is_available():
+                return int(torch.cuda.max_memory_allocated())
+        except Exception:
+            return 0
+        return 0
+
+    def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample:
+        if self._model is None:
+            raise BenchmarkError("TransformersDriver.generate called before load()")
+        torch = self._torch
+
+        # add_special_tokens=False: the plan owns the prompt format, and the
+        # llama.cpp recipe is given the identical string.
+        encoded = self._tokenizer(prompt, return_tensors="pt", add_special_tokens=False)
+        input_ids = encoded["input_ids"].to(self.device)
+        prompt_tokens = int(input_ids.shape[-1])
+        eos_ids = {self._tokenizer.eos_token_id} | set(
+            getattr(self._model.generation_config, "eos_token_id", None) or []
+            if isinstance(getattr(self._model.generation_config, "eos_token_id", None), list)
+            else []
+        )
+        eos_ids.discard(None)
+
+        started = time.monotonic()
+        with torch.inference_mode():
+            outputs = self._model(input_ids=input_ids, use_cache=True)
+            past = outputs.past_key_values
+            next_id = self._select(outputs.logits[:, -1, :], sampling)
+            ttft_ms = (time.monotonic() - started) * 1000
+
+            token_ids = [int(next_id.item())]
+            decode_started = time.monotonic()
+            while len(token_ids) < sampling.max_output_tokens and token_ids[-1] not in eos_ids:
+                outputs = self._model(
+                    input_ids=next_id.view(1, 1), past_key_values=past, use_cache=True
+                )
+                past = outputs.past_key_values
+                next_id = self._select(outputs.logits[:, -1, :], sampling)
+                token_ids.append(int(next_id.item()))
+            decode_ms = (time.monotonic() - decode_started) * 1000
+
+        total_ms = (time.monotonic() - started) * 1000
+        emitted = [token for token in token_ids if token not in eos_ids]
+        return GenerationSample(
+            text=self._tokenizer.decode(emitted, skip_special_tokens=True),
+            prompt_tokens=prompt_tokens,
+            # The first token is produced by the prefill forward, so the decode
+            # rate must not be credited with it.
+            decode_tokens=max(0, len(token_ids) - 1),
+            ttft_ms=ttft_ms,
+            prefill_ms=ttft_ms,
+            decode_ms=decode_ms,
+            total_ms=total_ms,
+        )
+
+    def _select(self, logits: Any, sampling: SamplingPolicy) -> Any:
+        if sampling.temperature > 0:
+            raise BenchmarkError(
+                "this benchmark is greedy-only: sampling noise is indistinguishable from "
+                "quantization drift, which is precisely what the quality lane must isolate"
+            )
+        return logits.argmax(dim=-1)
+
+    def memory_probe(self) -> tuple[int, int]:
+        return max(0, _process_rss() - self._rss_baseline), self._vram_bytes()
+
+    def close(self) -> None:
+        self._model = None
+        self._tokenizer = None
+        if self._torch is not None:
+            import gc
+
+            gc.collect()
+
+
+def _free_port() -> int:
+    with socket.socket() as probe:
+        probe.bind(("127.0.0.1", 0))
+        return int(probe.getsockname()[1])
+
+
+class LlamaCppServerDriver:
+    """The whole-model llama.cpp/GGUF recipe, driven through ``llama-server``.
+
+    ``llama-server`` is used rather than an in-process binding because it is the
+    shape llama.cpp is actually deployed in and the only one that offers
+    continuous batching across parallel slots — which is the runtime property
+    this project cares about most.  It also reports its own prefill/decode
+    timings per request, so the decode rate is the runtime's own number and not
+    an inference drawn from a client-side stopwatch.
+    """
+
+    def __init__(
+        self,
+        binary: str,
+        gguf_path: str,
+        *,
+        binary_sha256: str,
+        device: str = "cpu",
+        threads: int = 8,
+        n_parallel: int = 4,
+        context_per_slot: int = 1024,
+        n_gpu_layers: int = 0,
+        startup_timeout_s: float = 120.0,
+    ) -> None:
+        self.binary = Path(binary)
+        self.binary_sha256 = binary_sha256
+        self.gguf_path = Path(gguf_path)
+        self.device = device
+        self.threads = threads
+        self.n_parallel = n_parallel
+        self.context_per_slot = context_per_slot
+        self.n_gpu_layers = n_gpu_layers
+        self.startup_timeout_s = startup_timeout_s
+        self._process: subprocess.Popen | None = None
+        self._port = 0
+        self._log: Any = None
+
+    @property
+    def _url(self) -> str:
+        return f"http://127.0.0.1:{self._port}"
+
+    def _log_excerpt(self) -> str:
+        if self._log is None:
+            return ""
+        try:
+            self._log.flush()
+            self._log.seek(0)
+            return self._log.read()[-4096:].decode("utf-8", errors="replace").strip()
+        except Exception:
+            return ""
+
+    def load(self) -> LoadStats:
+        if not self.binary.exists():
+            raise BenchmarkError(f"llama-server binary not found at {self.binary}")
+        if not self.gguf_path.exists():
+            raise BenchmarkError(f"GGUF artifact not found at {self.gguf_path}")
+        measured_binary_sha256 = _artifact_sha256(self.binary)
+        if not hmac.compare_digest(self.binary_sha256, measured_binary_sha256):
+            raise BenchmarkError("llama-server binary changed after config validation")
+        version = " | ".join(
+            subprocess.run(
+                [str(self.binary), "--version"],
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                text=True,
+                timeout=10,
+            ).stdout.strip().splitlines()
+        )
+
+        self._port = _free_port()
+        command = [
+            str(self.binary),
+            "--model", str(self.gguf_path),
+            "--host", "127.0.0.1",
+            "--port", str(self._port),
+            "--threads", str(self.threads),
+            "--parallel", str(self.n_parallel),
+            # Every slot must hold a whole request, so the pool is sized for the
+            # worst case rather than letting llama.cpp silently truncate context.
+            "--ctx-size", str(self.context_per_slot * self.n_parallel),
+            "--n-gpu-layers", str(self.n_gpu_layers),
+            "--no-webui",
+        ]
+        started = time.monotonic()
+        self._log = tempfile.TemporaryFile(mode="w+b")
+        self._process = subprocess.Popen(
+            command, stdout=self._log, stderr=subprocess.STDOUT
+        )
+        self._await_health(started)
+        load_ms = (time.monotonic() - started) * 1000
+
+        return LoadStats(
+            artifact_bytes=self.gguf_path.stat().st_size,
+            load_ms=round(load_ms, 4),
+            rss_bytes=_process_rss(self._process.pid),
+            vram_bytes=0,
+            backend_detail=(
+                f"{version}; binary sha256 {measured_binary_sha256}; "
+                f"threads {self.threads}; parallel slots {self.n_parallel}; "
+                f"ctx/slot {self.context_per_slot}; gpu layers {self.n_gpu_layers}"
+            ),
+        )
+
+    def _await_health(self, started: float) -> None:
+        while time.monotonic() - started < self.startup_timeout_s:
+            if self._process is not None and self._process.poll() is not None:
+                raise BenchmarkError(
+                    f"llama-server exited with code {self._process.returncode} during startup; "
+                    f"log tail: {self._log_excerpt()}"
+                )
+            try:
+                with urllib.request.urlopen(f"{self._url}/health", timeout=2) as response:
+                    if response.status == 200:
+                        return
+            except (urllib.error.URLError, OSError):
+                time.sleep(0.25)
+        raise BenchmarkError(
+            f"llama-server did not become healthy within {self.startup_timeout_s:.0f}s; "
+            f"log tail: {self._log_excerpt()}"
+        )
+
+    def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample:
+        if self._process is None:
+            raise BenchmarkError("LlamaCppServerDriver.generate called before load()")
+        if sampling.temperature > 0:
+            raise BenchmarkError("this benchmark is greedy-only; see TransformersDriver._select")
+
+        body = json.dumps({
+            "prompt": prompt,
+            "n_predict": sampling.max_output_tokens,
+            "temperature": 0.0,
+            "top_k": 1,
+            "top_p": 1.0,
+            "seed": sampling.seed,
+            # Prompt cache reuse across repeats would measure the cache, not the
+            # prefill, and the safetensors recipe has no equivalent.
+            "cache_prompt": False,
+            "stream": True,
+        }).encode()
+        request = urllib.request.Request(
+            f"{self._url}/completion", data=body,
+            headers={"Content-Type": "application/json"}, method="POST",
+        )
+
+        started = time.monotonic()
+        chunks: list[str] = []
+        timings: Mapping[str, Any] = {}
+        with urllib.request.urlopen(request, timeout=600) as response:
+            for raw in response:
+                line = raw.decode("utf-8").strip()
+                if not line.startswith("data:"):
+                    continue
+                payload = json.loads(line[len("data:"):].strip())
+                content = payload.get("content", "")
+                chunks.append(content)
+                if payload.get("stop"):
+                    timings = payload.get("timings") or {}
+        total_ms = (time.monotonic() - started) * 1000
+
+        if not timings:
+            raise BenchmarkError("llama-server returned no timings; cannot report an honest split")
+
+        prefill_ms = float(timings.get("prompt_ms", 0.0))
+        decode_ms = float(timings.get("predicted_ms", 0.0))
+        return GenerationSample(
+            text="".join(chunks),
+            prompt_tokens=int(timings.get("prompt_n", 0)),
+            # llama.cpp starts predicted_ms after sampling the first token while
+            # predicted_n includes it. Exclude that token to match the
+            # Transformers inter-token decode metric.
+            decode_tokens=max(0, int(timings.get("predicted_n", 0)) - 1),
+            # Use the runtime's prompt/first-token timing, matching the
+            # in-process Transformers boundary. HTTP/SSE and slot delay remain
+            # represented by total latency and queue_wait_ms.
+            ttft_ms=prefill_ms,
+            prefill_ms=prefill_ms,
+            decode_ms=decode_ms,
+            total_ms=total_ms,
+            # Whatever the wall clock saw but the runtime did not attribute to
+            # compute is time this request spent waiting for a slot.
+            queue_wait_ms=max(0.0, total_ms - prefill_ms - decode_ms),
+        )
+
+    def memory_probe(self) -> tuple[int, int]:
+        if self._process is None:
+            return 0, 0
+        return _process_rss(self._process.pid), 0
+
+    def close(self) -> None:
+        if self._process is not None:
+            if self._process.poll() is None:
+                self._process.terminate()
+                try:
+                    self._process.wait(timeout=20)
+                except subprocess.TimeoutExpired:
+                    self._process.kill()
+                    self._process.wait(timeout=10)
+            self._process = None
+        if self._log is not None:
+            self._log.close()
+            self._log = None
+
+
+def build_driver(spec: Mapping[str, Any], plan: BenchmarkPlan) -> RecipeDriverBundle:
+    """Construct the driver named by a recipe's ``driver`` block."""
+    driver_spec = dict(spec["driver"])
+    kind = driver_spec.pop("type")
+    if kind == "transformers":
+        return TransformersDriver(**driver_spec)
+    if kind == "llama-cpp-server":
+        driver_spec.setdefault("n_parallel", max(plan.concurrency_levels))
+        return LlamaCppServerDriver(**driver_spec)
+    raise BenchmarkError(f"unknown driver type {kind!r}")
+
+
+RecipeDriverBundle = Any  # a RecipeDriver; named for readability at the call site
+
+
+def _plan_from_config(config: Mapping[str, Any]) -> BenchmarkPlan:
+    raw = config["plan"]
+    return BenchmarkPlan(
+        plan_id=raw["plan_id"],
+        model_id=raw["model_id"],
+        model_revision=raw["model_revision"],
+        prompts=tuple(PromptSpec(**prompt) for prompt in raw["prompts"]),
+        sampling=SamplingPolicy(**raw.get("sampling", {})),
+        concurrency_levels=tuple(raw.get("concurrency_levels", (1, 4))),
+        repeats=int(raw.get("repeats", 1)),
+        warmup_requests=int(raw.get("warmup_requests", 1)),
+    )
+
+
+def _recipe_from_config(spec: Mapping[str, Any]) -> RecipeSpec:
+    return RecipeSpec(
+        id=spec["id"],
+        runtime=spec["runtime"],
+        weight_format=spec["weight_format"],
+        weight_quantization=spec["weight_quantization"],
+        lane=Lane(spec["lane"]),
+        device=spec["device"],
+        artifact_path=spec.get("artifact_path", ""),
+        source_model_id=spec.get("source_model_id", ""),
+        source_model_revision=spec.get("source_model_revision", ""),
+        artifact_sha256=spec.get("artifact_sha256", ""),
+        is_reference=bool(spec.get("is_reference", False)),
+        notes=spec.get("notes", ""),
+    )
+
+
+def run_configured_benchmark(config: Mapping[str, Any]) -> dict:
+    """Run every recipe in ``config`` against one shared plan and return the report.
+
+    A recipe whose runtime cannot start is recorded as unavailable with the real
+    reason rather than dropped: a report that silently omits the recipe that
+    crashed would read as a clean result.
+    """
+    require_real_inference()
+    _validate_config(config)
+    plan = _plan_from_config(config)
+
+    from .recipe_benchmark import RecipeMeasurement  # local import keeps the seam obvious
+
+    measurements = []
+    for spec in config["recipes"]:
+        recipe = _recipe_from_config(spec)
+        driver = None
+        try:
+            driver = build_driver(spec, plan)
+            measurements.append(measure_recipe(driver, recipe, plan))
+        except Exception as exc:
+            measurements.append(RecipeMeasurement(
+                recipe=recipe,
+                load=LoadStats(artifact_bytes=0, load_ms=0.0),
+                unavailable_reason=f"{type(exc).__name__}: {exc}",
+            ))
+        finally:
+            if driver is not None:
+                driver.close()
+
+    return build_report(
+        plan,
+        measurements,
+        host={**dict(config.get("host", {})), **_host_manifest()},
+        evidence_class=config.get("evidence_class", "local-real"),
+    )

tests/test_recipe_benchmark.py

@@ -0,0 +1,564 @@
+"""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)