Hands-On: Build a Fail-Closed Inference Benchmark Comparator
This lab accompanies Part 10: Squeezing More Inference from Apple Silicon: llama.cpp Today, MLXForge Later. The main article explains why a performance comparison needs identity and correctness before arithmetic. Here, we are going to make that rule executable.
I deliberately kept this lab away from a real inference engine. It uses invented JSON fixtures and the Python standard library. You do not need MLXForge, a model, a GPU, Apple Silicon, a provider account, or a network connection. Nothing in the synthetic results says how fast any real machine or engine is.
What the lab does prove is narrower and more useful: it can reject an invalid document, distinguish a repeatability run from an optimization run, stop an incompatible comparison, block a speed conclusion after correctness drift, calculate median direction-aware deltas, and render the same bounded result as JSON, TSV, and Markdown.
What You Will Build
The companion contains exactly eight files:
README.md
benchmark_compare.py
run_lab.py
test_lab.py
fixtures/baseline.json
fixtures/candidate-optimization.json
fixtures/candidate-ineligible.json
requirements.txt
benchmark_compare.py owns validation, eligibility, aggregation, and report rendering. run_lab.py performs the bounded walkthrough and removes its temporary reports. test_lab.py exercises the failure boundaries. The three fixtures are invented and small enough to inspect by eye. requirements.txt documents that there are no third-party dependencies.
Download the complete eight-file Hands-On 10A package. Every file is also available below through the site’s standard collapsed source viewer.
README.md markdown View source
# Fail-Closed Inference Benchmark Comparator
This small lab compares invented benchmark bundles without loading a model, contacting a service, or reading private evidence. It is a teaching implementation, not an export from MLXForge.
Run the bounded walkthrough:
```bash
python run_lab.py
```
Run the tests:
```bash
python -m unittest -v test_lab.py
```
Compare either candidate directly:
```bash
python benchmark_compare.py fixtures/baseline.json fixtures/candidate-optimization.json
python benchmark_compare.py fixtures/baseline.json fixtures/candidate-ineligible.json
```
The second command exits with status 2 because a warm baseline and cold candidate are not comparable. That is the intended result.
The schema is closed. It contains fingerprints, execution conditions, correctness state, and numeric observations, but no prompt, generated output, hostname, user, address, token, credential, or private path. A normalized output hash only proves equality under the chosen normalization; it does not prove answer quality.
Generated projections live only in a temporary directory during `run_lab.py` and are removed before the runner reports success. The eight package files are the complete end state.
benchmark_compare.py python View source
#!/usr/bin/env python3
"""Validate and compare synthetic inference benchmark bundles."""
from __future__ import annotations
import argparse
import csv
import io
import json
import math
import re
import statistics
from dataclasses import dataclass
from pathlib import Path
from typing import Any
HEX64 = re.compile(r"^[0-9a-f]{64}$")
TOP_KEYS = {"schema_version", "identity", "correctness", "observations"}
IDENTITY_TYPES = {
"run_id": str,
"engine": str,
"source_fingerprint": str,
"artifact_fingerprint": str,
"workload_fingerprint": str,
"machine_class": str,
"memory_gb": int,
"warm_cold": str,
"profiled": bool,
"concurrency": int,
"batch_size": int,
"context_tokens": int,
"sampling_fingerprint": str,
}
CORRECTNESS_TYPES = {"passed": bool, "normalized_output_hash": str}
REQUIRED_METRICS = {
"ttft_ms": ("lower", "ms"),
"prompt_tokens_per_s": ("higher", "tokens/s"),
"decode_tokens_per_s": ("higher", "tokens/s"),
"peak_wired_mb": ("lower", "MiB"),
}
OPTIONAL_METRICS = {
"ssd_read_bytes": ("lower", "bytes"),
"warm_restore_latency_ms": ("lower", "ms"),
}
ALL_METRICS = REQUIRED_METRICS | OPTIONAL_METRICS
STRICT_IDENTITY_FIELDS = (
"artifact_fingerprint",
"workload_fingerprint",
"machine_class",
"memory_gb",
"warm_cold",
"profiled",
"concurrency",
"batch_size",
"context_tokens",
"sampling_fingerprint",
)
class SchemaError(ValueError):
"""Raised when a bundle violates the closed teaching schema."""
@dataclass(frozen=True)
class Comparison:
classification: str
experiment_kind: str | None
reasons: tuple[str, ...]
metrics: tuple[dict[str, Any], ...]
def as_dict(self) -> dict[str, Any]:
return {
"classification": self.classification,
"experiment_kind": self.experiment_kind,
"reasons": list(self.reasons),
"metrics": list(self.metrics),
}
def _expect_exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None:
actual = set(value)
if actual != expected:
missing = sorted(expected - actual)
unknown = sorted(actual - expected)
raise SchemaError(f"{label} keys differ; missing={missing}, unknown={unknown}")
def _expect_type(value: Any, expected: type, label: str) -> None:
if expected is int:
if isinstance(value, bool) or not isinstance(value, int):
raise SchemaError(f"{label} must be an integer")
elif type(value) is not expected:
raise SchemaError(f"{label} must be {expected.__name__}")
def validate_bundle(bundle: Any) -> dict[str, Any]:
"""Return a validated bundle or raise SchemaError."""
if not isinstance(bundle, dict):
raise SchemaError("bundle must be an object")
_expect_exact_keys(bundle, TOP_KEYS, "bundle")
if type(bundle["schema_version"]) is not int or bundle["schema_version"] != 1:
raise SchemaError("schema_version must be integer 1")
identity = bundle["identity"]
correctness = bundle["correctness"]
observations = bundle["observations"]
if not isinstance(identity, dict) or not isinstance(correctness, dict):
raise SchemaError("identity and correctness must be objects")
_expect_exact_keys(identity, set(IDENTITY_TYPES), "identity")
_expect_exact_keys(correctness, set(CORRECTNESS_TYPES), "correctness")
for key, expected in IDENTITY_TYPES.items():
_expect_type(identity[key], expected, f"identity.{key}")
for key, expected in CORRECTNESS_TYPES.items():
_expect_type(correctness[key], expected, f"correctness.{key}")
for key in ("run_id", "engine", "machine_class"):
if not identity[key].strip():
raise SchemaError(f"identity.{key} must not be empty")
for key in (
"source_fingerprint",
"artifact_fingerprint",
"workload_fingerprint",
"sampling_fingerprint",
):
if not HEX64.fullmatch(identity[key]):
raise SchemaError(f"identity.{key} must be lowercase 64-hex")
if not HEX64.fullmatch(correctness["normalized_output_hash"]):
raise SchemaError("correctness.normalized_output_hash must be lowercase 64-hex")
if identity["warm_cold"] not in {"warm", "cold"}:
raise SchemaError("identity.warm_cold must be warm or cold")
for key in ("memory_gb", "concurrency", "batch_size", "context_tokens"):
if identity[key] <= 0:
raise SchemaError(f"identity.{key} must be positive")
if not isinstance(observations, list) or len(observations) < 3:
raise SchemaError("observations must contain at least three rows")
metric_keys: set[str] | None = None
for index, row in enumerate(observations):
if not isinstance(row, dict):
raise SchemaError(f"observations[{index}] must be an object")
keys = set(row)
if not set(REQUIRED_METRICS).issubset(keys) or not keys.issubset(ALL_METRICS):
raise SchemaError(f"observations[{index}] has missing or unknown metrics")
if metric_keys is None:
metric_keys = keys
elif keys != metric_keys:
raise SchemaError("every observation must contain the same metric keys")
for key, value in row.items():
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise SchemaError(f"observations[{index}].{key} must be numeric")
if not math.isfinite(value) or value < 0:
raise SchemaError(f"observations[{index}].{key} must be finite and nonnegative")
return bundle
def load_bundle(path: str | Path) -> dict[str, Any]:
with Path(path).open(encoding="utf-8") as handle:
return validate_bundle(json.load(handle))
def compare_bundles(baseline: dict[str, Any], candidate: dict[str, Any]) -> Comparison:
baseline = validate_bundle(baseline)
candidate = validate_bundle(candidate)
mismatches = tuple(
field
for field in STRICT_IDENTITY_FIELDS
if baseline["identity"][field] != candidate["identity"][field]
)
baseline_keys = set(baseline["observations"][0])
candidate_keys = set(candidate["observations"][0])
if baseline_keys != candidate_keys:
mismatches += ("metric_set",)
if mismatches:
return Comparison("ineligible", None, mismatches, ())
if baseline["identity"]["profiled"]:
return Comparison("ineligible", None, ("profiled_runs_excluded",), ())
correctness_reasons: list[str] = []
if not baseline["correctness"]["passed"]:
correctness_reasons.append("baseline_correctness_failed")
if not candidate["correctness"]["passed"]:
correctness_reasons.append("candidate_correctness_failed")
if (
baseline["correctness"]["normalized_output_hash"]
!= candidate["correctness"]["normalized_output_hash"]
):
correctness_reasons.append("normalized_output_hash_mismatch")
if correctness_reasons:
return Comparison("correctness_failed", None, tuple(correctness_reasons), ())
experiment_kind = (
"repeatability"
if baseline["identity"]["source_fingerprint"]
== candidate["identity"]["source_fingerprint"]
else "optimization"
)
rows: list[dict[str, Any]] = []
for name in sorted(baseline_keys):
direction, unit = ALL_METRICS[name]
baseline_value = statistics.median(row[name] for row in baseline["observations"])
candidate_value = statistics.median(row[name] for row in candidate["observations"])
delta = candidate_value - baseline_value
percent_delta = None if baseline_value == 0 else (delta / baseline_value) * 100
directed_delta = delta if direction == "higher" else -delta
outcome = "unchanged" if delta == 0 else ("improved" if directed_delta > 0 else "regressed")
rows.append(
{
"metric": name,
"unit": unit,
"direction": direction,
"baseline_median": baseline_value,
"candidate_median": candidate_value,
"absolute_delta": delta,
"percentage_delta": percent_delta,
"outcome": outcome,
}
)
return Comparison("comparable", experiment_kind, (), tuple(rows))
def render_json(result: Comparison) -> str:
return json.dumps(result.as_dict(), indent=2, sort_keys=True) + "\n"
def render_tsv(result: Comparison) -> str:
output = io.StringIO(newline="")
writer = csv.writer(output, delimiter="\t", lineterminator="\n")
writer.writerow(("classification", result.classification))
writer.writerow(("experiment_kind", result.experiment_kind or ""))
writer.writerow(("reasons", ",".join(result.reasons)))
writer.writerow(())
writer.writerow(("metric", "unit", "direction", "baseline_median", "candidate_median", "absolute_delta", "percentage_delta", "outcome"))
for row in result.metrics:
writer.writerow(row[key] if row[key] is not None else "" for key in ("metric", "unit", "direction", "baseline_median", "candidate_median", "absolute_delta", "percentage_delta", "outcome"))
return output.getvalue()
def render_markdown(result: Comparison) -> str:
lines = [
f"Classification: `{result.classification}`",
f"Experiment: `{result.experiment_kind or 'none'}`",
f"Reasons: `{', '.join(result.reasons) if result.reasons else 'none'}`",
]
if result.metrics:
lines.extend(("", "| Metric | Unit | Direction | Baseline median | Candidate median | Delta | Percent delta | Outcome |", "| --- | --- | --- | ---: | ---: | ---: | ---: | --- |"))
for row in result.metrics:
percent = "n/a" if row["percentage_delta"] is None else f'{row["percentage_delta"]:.3f}%'
lines.append(f'| {row["metric"]} | {row["unit"]} | {row["direction"]} | {row["baseline_median"]:.3f} | {row["candidate_median"]:.3f} | {row["absolute_delta"]:.3f} | {percent} | {row["outcome"]} |')
return "\n".join(lines) + "\n"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("baseline", type=Path)
parser.add_argument("candidate", type=Path)
parser.add_argument("--format", choices=("json", "tsv", "markdown"), default="markdown")
args = parser.parse_args()
result = compare_bundles(load_bundle(args.baseline), load_bundle(args.candidate))
print({"json": render_json, "tsv": render_tsv, "markdown": render_markdown}[args.format](result), end="")
return 0 if result.classification == "comparable" else 2
if __name__ == "__main__":
raise SystemExit(main())
run_lab.py python View source
#!/usr/bin/env python3
"""Run the bounded, synthetic benchmark-comparison lab."""
from __future__ import annotations
import copy
import json
import tempfile
from pathlib import Path
from benchmark_compare import SchemaError, compare_bundles, load_bundle, render_json, render_markdown, render_tsv, validate_bundle
ROOT = Path(__file__).resolve().parent
FIXTURES = ROOT / "fixtures"
def main() -> int:
baseline = load_bundle(FIXTURES / "baseline.json")
candidate = load_bundle(FIXTURES / "candidate-optimization.json")
ineligible = load_bundle(FIXTURES / "candidate-ineligible.json")
result = compare_bundles(baseline, candidate)
rejected = compare_bundles(baseline, ineligible)
correctness = copy.deepcopy(candidate)
correctness["correctness"]["passed"] = False
correctness_result = compare_bundles(baseline, correctness)
unknown = copy.deepcopy(candidate)
unknown["identity"]["hostname"] = "must-not-be-accepted"
boolean_numeric = copy.deepcopy(candidate)
boolean_numeric["identity"]["memory_gb"] = True
def rejected_schema(value: dict) -> bool:
try:
validate_bundle(value)
except SchemaError:
return True
return False
with tempfile.TemporaryDirectory(prefix="benchmark-comparator-") as temporary:
report_dir = Path(temporary)
projections = {
"json": render_json(result),
"tsv": render_tsv(result),
"markdown": render_markdown(result),
}
for extension, content in projections.items():
(report_dir / f"comparison.{extension if extension != 'markdown' else 'md'}").write_text(content, encoding="utf-8")
json_payload = json.loads(projections["json"])
projection_agreement = (
json_payload["classification"] == result.classification
and all(result.classification in projections[name] for name in projections)
and all((result.experiment_kind or "") in projections[name] for name in projections)
and all(row["metric"] in projections["tsv"] and row["metric"] in projections["markdown"] for row in result.metrics)
)
metric_map = {row["metric"]: row for row in result.metrics}
checks = {
"baseline_schema_valid": True,
"candidate_schema_valid": True,
"artifact_identity_matches": baseline["identity"]["artifact_fingerprint"] == candidate["identity"]["artifact_fingerprint"],
"workload_identity_matches": baseline["identity"]["workload_fingerprint"] == candidate["identity"]["workload_fingerprint"],
"experiment_kind_is_optimization": result.experiment_kind == "optimization",
"correctness_passes": result.classification == "comparable",
"comparison_is_eligible": result.classification == "comparable",
"ttft_direction_is_lower": metric_map["ttft_ms"]["direction"] == "lower",
"decode_direction_is_higher": metric_map["decode_tokens_per_s"]["direction"] == "higher",
"median_aggregation_used": metric_map["ttft_ms"]["baseline_median"] == 210.0,
"incompatible_cache_state_rejected": rejected.classification == "ineligible" and "warm_cold" in rejected.reasons,
"profiled_mismatch_rejected": _mismatch_rejected(baseline, candidate, "profiled", True),
"incompatible_workload_rejected": _mismatch_rejected(baseline, candidate, "workload_fingerprint", "9" * 64),
"context_mismatch_rejected": _mismatch_rejected(baseline, candidate, "context_tokens", 8192),
"correctness_failure_blocks_speed_claim": correctness_result.classification == "correctness_failed" and not correctness_result.metrics,
"unknown_field_rejected": rejected_schema(unknown),
"boolean_numeric_rejected": rejected_schema(boolean_numeric),
"json_tsv_markdown_agree": projection_agreement,
"prompt_and_output_text_absent": _privacy_check(baseline, candidate, ineligible, projections),
"cleanup_complete": not report_dir.exists(),
}
print(json.dumps({"conditions": checks, "passed": sum(checks.values()), "total": len(checks), "ok": all(checks.values())}, indent=2, sort_keys=True))
return 0 if all(checks.values()) else 1
def _mismatch_rejected(baseline: dict, candidate: dict, field: str, value: object) -> bool:
changed = copy.deepcopy(candidate)
changed["identity"][field] = value
result = compare_bundles(baseline, changed)
return result.classification == "ineligible" and field in result.reasons
def _privacy_check(*values: object) -> bool:
text = json.dumps(values, sort_keys=True).lower()
forbidden = ('"prompt"', '"completion"', '"output_text"', '"hostname"', '"user"', '"ip_address"', '"token"')
return not any(term in text for term in forbidden)
if __name__ == "__main__":
raise SystemExit(main())
test_lab.py python View source
#!/usr/bin/env python3
"""Tests for the synthetic inference benchmark comparator."""
from __future__ import annotations
import copy
import json
import math
import unittest
from pathlib import Path
from benchmark_compare import SchemaError, compare_bundles, load_bundle, render_json, render_markdown, render_tsv, validate_bundle
ROOT = Path(__file__).resolve().parent
class ComparatorTests(unittest.TestCase):
def setUp(self) -> None:
self.baseline = load_bundle(ROOT / "fixtures" / "baseline.json")
self.candidate = load_bundle(ROOT / "fixtures" / "candidate-optimization.json")
def changed(self, section: str, field: str, value: object) -> dict:
bundle = copy.deepcopy(self.candidate)
bundle[section][field] = value
return bundle
def test_optimization_and_repeatability(self) -> None:
result = compare_bundles(self.baseline, self.candidate)
self.assertEqual((result.classification, result.experiment_kind), ("comparable", "optimization"))
repeat = self.changed("identity", "source_fingerprint", self.baseline["identity"]["source_fingerprint"])
self.assertEqual(compare_bundles(self.baseline, repeat).experiment_kind, "repeatability")
def test_identity_mismatches_are_ineligible(self) -> None:
cases = {
"artifact_fingerprint": "6" * 64,
"workload_fingerprint": "7" * 64,
"warm_cold": "cold",
"profiled": True,
"context_tokens": 8192,
"sampling_fingerprint": "8" * 64,
}
for field, value in cases.items():
with self.subTest(field=field):
result = compare_bundles(self.baseline, self.changed("identity", field, value))
self.assertEqual(result.classification, "ineligible")
self.assertIn(field, result.reasons)
self.assertFalse(result.metrics)
def test_engine_label_may_differ_when_contract_matches(self) -> None:
result = compare_bundles(self.baseline, self.changed("identity", "engine", "engine-b"))
self.assertEqual(result.classification, "comparable")
def test_matching_profiled_runs_remain_ineligible(self) -> None:
baseline = copy.deepcopy(self.baseline)
candidate = copy.deepcopy(self.candidate)
baseline["identity"]["profiled"] = True
candidate["identity"]["profiled"] = True
result = compare_bundles(baseline, candidate)
self.assertEqual(result.classification, "ineligible")
self.assertIn("profiled_runs_excluded", result.reasons)
def test_correctness_failure_or_hash_drift_blocks_metrics(self) -> None:
for field, value in (("passed", False), ("normalized_output_hash", "9" * 64)):
with self.subTest(field=field):
result = compare_bundles(self.baseline, self.changed("correctness", field, value))
self.assertEqual(result.classification, "correctness_failed")
self.assertFalse(result.metrics)
def test_closed_schema_rejects_unknown_missing_and_bad_types(self) -> None:
unknown = self.changed("identity", "engine", "engine-a")
unknown["identity"]["hostname"] = "private-host"
missing = copy.deepcopy(self.candidate)
del missing["identity"]["artifact_fingerprint"]
boolean = self.changed("identity", "memory_gb", True)
wrong_version_type = copy.deepcopy(self.candidate)
wrong_version_type["schema_version"] = 1.0
nonfinite = copy.deepcopy(self.candidate)
nonfinite["observations"][0]["ttft_ms"] = math.inf
for bundle in (unknown, missing, boolean, wrong_version_type, nonfinite):
with self.assertRaises(SchemaError):
validate_bundle(bundle)
def test_medians_and_direction_aware_outcomes(self) -> None:
rows = {row["metric"]: row for row in compare_bundles(self.baseline, self.candidate).metrics}
self.assertEqual(rows["ttft_ms"]["baseline_median"], 210.0)
self.assertEqual(rows["ttft_ms"]["outcome"], "improved")
self.assertEqual(rows["decode_tokens_per_s"]["outcome"], "improved")
self.assertEqual(rows["peak_wired_mb"]["outcome"], "improved")
def test_zero_baseline_percentage_is_none(self) -> None:
baseline = copy.deepcopy(self.baseline)
candidate = copy.deepcopy(self.candidate)
for bundle in (baseline, candidate):
for row in bundle["observations"]:
row["ttft_ms"] = 0.0
row = next(row for row in compare_bundles(baseline, candidate).metrics if row["metric"] == "ttft_ms")
self.assertIsNone(row["percentage_delta"])
def test_projections_are_deterministic_and_bounded(self) -> None:
result = compare_bundles(self.baseline, self.candidate)
projections = (render_json(result), render_tsv(result), render_markdown(result))
self.assertEqual(projections, (render_json(result), render_tsv(result), render_markdown(result)))
self.assertEqual(json.loads(projections[0])["classification"], "comparable")
joined = "".join(projections).lower()
for forbidden in ('"prompt":', '"completion":', '"output_text":', "private-host"):
self.assertNotIn(forbidden, joined)
if __name__ == "__main__":
unittest.main()
requirements.txt text View source
# Python 3.10 or newer; standard library only.
baseline.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "synthetic-baseline-01",
"engine": "engine-a",
"source_fingerprint": "1111111111111111111111111111111111111111111111111111111111111111",
"artifact_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"workload_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"machine_class": "synthetic-apple-silicon-lab",
"memory_gb": 32,
"warm_cold": "warm",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"sampling_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444"
},
"correctness": {
"passed": true,
"normalized_output_hash": "5555555555555555555555555555555555555555555555555555555555555555"
},
"observations": [
{"ttft_ms": 212.0, "prompt_tokens_per_s": 498.0, "decode_tokens_per_s": 29.8, "peak_wired_mb": 12040.0},
{"ttft_ms": 210.0, "prompt_tokens_per_s": 500.0, "decode_tokens_per_s": 30.0, "peak_wired_mb": 12000.0},
{"ttft_ms": 208.0, "prompt_tokens_per_s": 502.0, "decode_tokens_per_s": 30.2, "peak_wired_mb": 11960.0}
]
}
candidate-optimization.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "synthetic-candidate-01",
"engine": "engine-a",
"source_fingerprint": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"artifact_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"workload_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"machine_class": "synthetic-apple-silicon-lab",
"memory_gb": 32,
"warm_cold": "warm",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"sampling_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444"
},
"correctness": {
"passed": true,
"normalized_output_hash": "5555555555555555555555555555555555555555555555555555555555555555"
},
"observations": [
{"ttft_ms": 190.0, "prompt_tokens_per_s": 546.0, "decode_tokens_per_s": 32.2, "peak_wired_mb": 11820.0},
{"ttft_ms": 188.0, "prompt_tokens_per_s": 550.0, "decode_tokens_per_s": 32.5, "peak_wired_mb": 11800.0},
{"ttft_ms": 186.0, "prompt_tokens_per_s": 554.0, "decode_tokens_per_s": 32.8, "peak_wired_mb": 11780.0}
]
}
candidate-ineligible.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "synthetic-candidate-ineligible-01",
"engine": "engine-a",
"source_fingerprint": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"artifact_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"workload_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"machine_class": "synthetic-apple-silicon-lab",
"memory_gb": 32,
"warm_cold": "cold",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"sampling_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444"
},
"correctness": {
"passed": true,
"normalized_output_hash": "5555555555555555555555555555555555555555555555555555555555555555"
},
"observations": [
{"ttft_ms": 760.0, "prompt_tokens_per_s": 480.0, "decode_tokens_per_s": 32.0, "peak_wired_mb": 12100.0},
{"ttft_ms": 750.0, "prompt_tokens_per_s": 482.0, "decode_tokens_per_s": 32.1, "peak_wired_mb": 12080.0},
{"ttft_ms": 740.0, "prompt_tokens_per_s": 484.0, "decode_tokens_per_s": 32.2, "peak_wired_mb": 12060.0}
]
}
Start in a copy of the companion directory and inspect the files before running them:
find . -maxdepth 2 -type f -print | sort
python --version
Python 3.10 or newer is sufficient. The lab does not install anything and does not read configuration outside its directory.
Step 1: Inspect the Closed Schema
Open fixtures/baseline.json. The document has four top-level fields: schema_version, identity, correctness, and observations. Unknown fields are rejected rather than ignored.
The identity block carries the conditions that determine whether the arithmetic has meaning. A model label is present for human orientation, but the artifact fingerprint is the comparison identity. The workload and sampling fingerprints represent the semantic request and generation controls without storing the prompt itself. Machine class, memory, context, concurrency, batch, warm or cold state, and profiling state describe the execution conditions.
The correctness block contains a Boolean result and a normalized output hash. This is an equality contract for the lab, not a quality score. The observations contain four metrics:
| Metric | Direction | Meaning in this lab |
|---|---|---|
ttft_ms |
Lower is better | Time to the first token measurement point |
prompt_tokens_per_s |
Higher is better | Prompt prefill throughput |
decode_tokens_per_s |
Higher is better | Generated-token throughput |
peak_wired_mb |
Lower is better | Peak wired memory |
Every observation must carry the same metric set, and every fixture needs at least three observations. The comparator uses the median. Three synthetic samples do not establish statistical significance; they simply make the aggregation rule visible.
Step 2: Run the Bounded Walkthrough
Run:
python run_lab.py
The runner prints one JSON object with 20 named conditions. A successful run ends with:
{
"ok": true,
"passed": 20,
"total": 20
}
The full output includes every condition rather than only the totals. That matters when the lab fails because you can see which boundary changed. The runner creates JSON, TSV, and Markdown projections inside a temporary directory, checks their agreement, and removes the directory before reporting cleanup_complete.
The runner does not hide a failed assertion behind a zero exit status. If any condition is false, it exits nonzero.
Step 3: Compare an Optimization Candidate
Run the comparator directly:
python benchmark_compare.py fixtures/baseline.json fixtures/candidate-optimization.json
The candidate uses the same artifact, workload, sampling, machine, context, cache state, profiling state, concurrency, and batch size as the baseline. Its source fingerprint differs, so the comparator labels the experiment optimization.
The report shows baseline median, candidate median, absolute delta, percentage delta, expected direction, and outcome for each metric. Notice that a negative delta can be an improvement for latency or memory, while a positive delta can be an improvement for throughput. Direction belongs to the metric definition; it should not be guessed by a report viewer.
Try the other formats:
python benchmark_compare.py fixtures/baseline.json fixtures/candidate-optimization.json --format json
python benchmark_compare.py fixtures/baseline.json fixtures/candidate-optimization.json --format tsv
All three projections come from the same validated comparison object. The tool does not parse its own Markdown or recalculate a TSV independently.
Step 4: Turn It into a Repeatability Check
Copy candidate-optimization.json to a temporary file and replace its source_fingerprint with the baseline source fingerprint. Leave the other identity fields alone, then compare it again.
The classification remains comparable, but the experiment kind changes to repeatability. That distinction prevents an unchanged program being advertised as a code optimization merely because normal run-to-run variation moved a median.
Do not edit the artifact or workload fingerprint for this step. Those changes describe different content, not a repeat of the same experiment.
Step 5: Watch an Incompatible Run Stop
Now run:
python benchmark_compare.py fixtures/baseline.json fixtures/candidate-ineligible.json
The command exits with status 2 and reports ineligible. The candidate is cold while the baseline is warm. Its numbers may be interesting in a cold-start experiment, but they cannot be mixed into this warm comparison.
The same stop occurs when artifact, workload, sampling, machine, memory, profiling, concurrency, batch, or context identity differs. Matching profiled runs also stop because profiling is diagnostic and does not belong in a throughput baseline. The report names the reason and emits no metric rows. A rejection is not a failed benchmark run. It is the comparator doing its job before a misleading conclusion escapes.
Step 6: Prove That Correctness Blocks Speed
Copy the optimization fixture, set correctness.passed to false, and compare it with the baseline. You can also leave passed true and change the normalized output hash.
Either change produces correctness_failed, with no performance metrics in the result. The tool does not print faster numbers with a warning beneath them. I made that a terminal state because warnings are easy to quote around.
This rule is intentionally stricter than many exploratory notebooks. If you are evaluating an intentional quality change, define a different experiment and quality contract. Do not silently reuse a deterministic regression contract that the candidate no longer satisfies.
Step 7: Break the Schema on Purpose
Add an unexpected field such as hostname to the identity object and run the comparator again. It fails schema validation. Then remove the field and change memory_gb from 32 to true. That also fails.
Python treats bool as a subclass of int, so a careless numeric validator will accept true as memory capacity. The lab rejects it explicitly. It also rejects missing fields, uppercase or malformed fingerprints, empty observations, inconsistent metric columns, negative values, and non-finite values such as infinity or NaN.
Failing closed here is not pedantry. Quietly accepting a field that one version ignores and another interprets is how evidence formats become ambiguous.
Step 8: Run the Regression Tests
Run:
python -m unittest -v test_lab.py
python -B -m py_compile benchmark_compare.py run_lab.py test_lab.py
The nine tests cover optimization and repeatability classification, cross-engine labels under a matched contract, identity mismatches, exclusion of matching profiled runs, correctness drift, exact schema behavior, median and direction-aware outcomes, a zero baseline percentage, deterministic projections, and absence of prompt or output fields.
The cross-engine-label test deserves a note. The engine string may differ because the lab is engine-neutral. That does not relax artifact, workload, sampling, machine, or execution-state compatibility. In a real cross-format comparison, you would still need a documented equivalence contract if exact artifact identity cannot be preserved.
Step 9: Confirm the Privacy Boundary
Search the package:
rg -n 'hostname|ip_address|api_key|bearer|customer|conversation|output_text' .
The test source contains some forbidden names as negative cases, but the fixtures and reports contain no real host, account, address, private path, prompt, completion, token, credential, or model-generated artifact. Fingerprints and IDs are visibly invented. They do not authenticate a real model or source revision.
The metric name prompt_tokens_per_s is safe because it describes a numeric measurement, not prompt content. A privacy rule that merely searches for the substring prompt would confuse a metric with the material the metric describes.
Step 10: Clean Up
run_lab.py cleans its generated reports automatically. If you created modified fixture copies during the walkthrough, remove only those copies. Do not remove the eight companion files.
Confirm that no interpreter cache was retained:
find . -type d -name __pycache__ -print
find . -type f -name '*.pyc' -print
The intended result is no output. The lab starts no service, loads no model, opens no network connection, creates no persistent cache, and touches no operator configuration.
Current State
The companion is an original standard-library teaching implementation with three invented fixtures, 20 bounded acceptance conditions, and nine unit tests. It validates an exact closed schema, distinguishes optimization from repeatability, rejects incompatible and profiled runs, requires matching normalized correctness, aggregates medians, calculates direction-aware deltas, and emits deterministic bounded projections.
Its synthetic values test comparator behavior only. They are not Apple Silicon measurements, MLXForge evidence, engine recommendations, or hardware guidance.
Next Work
The natural extension is a small adapter that maps a privacy-reviewed export from an inference engine into this schema. That adapter should remain outside the comparator and should never scrape a private run directory indiscriminately. It needs explicit ownership of redaction, field mapping, units, token accounting, and artifact-equivalence rules.
Real benchmark publication would add frozen source identity, retained correctness evidence, repeated observations with spread, contention notes, and a reviewed disclosure package. The comparator should stay boring. Its job is to stop bad comparisons early and make the surviving arithmetic easy to inspect.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.