Hands-On: Tune One llama.cpp Variable at a Time
This is Hands-On 11A in the Local-First Agent Operations series. It accompanies Part 11, Where Local Inference Performance Actually Comes From, where I separated operator controls from model-compatible acceleration and architecture decisions. Here I want to make the experimental discipline runnable without asking you to download my model, reproduce my machine, or trust a benchmark number I have not published.
The lab is deliberately model-free. It uses invented records to answer a less glamorous but more important question: would this pair of runs be eligible for a performance comparison at all?
What We Are Testing
The example starts with an invented runtime where speculative generation is disabled. Its candidate enables an abstract method named candidate_method. The name is intentionally dull. It does not imply that your llama.cpp build, your model, or any particular artifact supports the method.
The candidate is allowed to change three related pieces of configuration: the method name, an optional draft-artifact identity, and the method’s arguments. Together they form one structured intervention. The target model, workload, tokenizer, template, sampling, cache, context, concurrency, batch, thread count, machine class, binary, source, and unrelated arguments remain frozen.
That distinction matters because a real separate-draft experiment cannot change only a single scalar flag. It needs a draft artifact and method arguments too. Those are dependent parts of one intervention, not three excuses to change the rest of the system.
The metric projection is the last step. Most mistakes should stop the experiment before arithmetic begins.
Get Oriented
The companion contains ten small files, including four fixtures:
| File | What it does |
|---|---|
experiment.schema.json |
Documents the closed experiment record |
command-templates.txt |
Puts the baseline and candidate commands beside each other |
experiment_lab.py |
Validates records, classifies eligibility, invokes metric projection, and renders reports |
run_lab.py |
Runs the bounded invented walkthrough |
test_experiment_lab.py |
Exercises schema, identity, correctness, privacy, adapter, and cleanup behavior |
fixtures/*.json |
Supplies eligible, ineligible, and active-path-unproven examples |
The companion expects Hands-On 10A beside it because I did not want to invent another median calculator and another definition of metric direction. Part 11 owns experiment eligibility and runtime-configuration classification. Part 10A supplies only its reviewed median and direction-aware projection.
Download the complete Hands-On 11A lab. The archive contains the ten lab files under post-11a/ and the unchanged comparator under post-10a/. Keep those directories beside each other. You do not need to download Part 10A separately.
Every source file is available here without leaving the article:
README.md markdown View source
# Hands-On 11A Companion: One Inference Intervention at a Time
This model-neutral Python 3.10+ lab validates an invented runtime-configuration experiment without downloading or running a model. It requires greedy sampling for exact normalized-output equality, proves that only one structured intervention changed, requires method-specific active-path evidence, and reuses the Hands-On 10A comparator only for median and direction-aware metric projection.
All model, machine, fingerprint, marker, and metric values are invented. They are not llama.cpp, MLX, Apple Silicon, or MLXForge benchmark results.
The teaching schema accepts only `none` and `candidate_method` as method names. Both validators enforce that vocabulary before a comparison can produce a report, including a rejected comparison. Adding a real method requires a separately reviewed schema change; raw runtime strings do not belong in this field.
## Files
| File | Purpose |
| --- | --- |
| `experiment.schema.json` | Human-readable closed JSON Schema for the teaching record |
| `command-templates.txt` | Baseline and candidate templates with one visible structured intervention |
| `experiment_lab.py` | Exact validator, eligibility classifier, versioned Part 10A projection adapter, and report rendering |
| `run_lab.py` | Bounded walkthrough using four invented fixtures and temporary reports |
| `test_experiment_lab.py` | Focused schema, identity, intervention, correctness, privacy, adapter, and cleanup tests |
| `fixtures/*.json` | Invented eligible, ineligible, and active-path-unproven records |
The companion expects the reviewed Hands-On 10A directory beside this directory. It imports `../post-10a/benchmark_compare.py` at runtime and does not alter that accepted schema or its source-fingerprint classification.
## Run the Lab
```bash
python -B run_lab.py
python -B -m unittest -v test_experiment_lab.py
python -B -c 'from pathlib import Path; [compile(p.read_text(), str(p), "exec") for p in Path(".").glob("*.py")]'
```
The lab creates reports only inside a temporary directory and verifies that the directory is removed. It starts no service, opens no network connection, creates no cache, and modifies no operator configuration.
## What the Contract Enforces
The target artifact, source, binary, tokenizer, template, workload, greedy sampling, cache, context, concurrency, batch, threads, machine class, warm/cold state, profiling state, and unrelated arguments remain frozen. The runtime configuration has its own canonical fingerprint. The reviewed `speculative_method` intervention may change only its method, optional draft-artifact identity, and method arguments.
The candidate must confirm the expected active-path marker and record a positive invented acceptance counter. Markers are short sanitized symbolic identifiers, not raw log text, paths, model names, or command lines. Missing, inactive, contradictory, unknown, or unsafe marker evidence emits no metric projection. Correctness failure or exact normalized-output drift emits `correctness_failed`. Other identity drift emits `ineligible`.
The Part 11 validator owns the `runtime_configuration_optimization` classification. The Part 10A adapter supplies only the already-reviewed median and direction-aware metric projection. It never changes the source fingerprint to manufacture an optimization classification.
## Current State
This is original model-free teaching code with invented fixtures. The walkthrough checks 15 conditions, and the test suite contains 19 tests. It does not establish compatibility or performance for a real model or runtime.
## Next Work
A separately reviewed live adapter could capture records from an operator-owned model and runtime. Any stochastic sampling mode needs a task-specific correctness contract and is intentionally rejected here.
command-templates.txt text View source
# Illustrative templates only. These commands are not executed by the lab.
# Replace every angle-bracket placeholder in an operator-owned, separately reviewed adapter.
# Baseline: speculative method disabled
<llama-server> \
--model <authorized-target-artifact> \
--temp 0 \
--ctx-size <frozen-context> \
--parallel <frozen-slots> \
--cache-type-k <frozen-kv-format> \
--cache-type-v <frozen-kv-format> \
--spec-type none
# Candidate: the same frozen command with one structured intervention
<llama-server> \
--model <same-authorized-target-artifact> \
--temp 0 \
--ctx-size <same-frozen-context> \
--parallel <same-frozen-slots> \
--cache-type-k <same-frozen-kv-format> \
--cache-type-v <same-frozen-kv-format> \
--spec-type <candidate_method> \
--spec-draft-model <optional-reviewed-draft-artifact> \
--spec-draft-n-max <reviewed-draft-count>
# The intervention owns only spec-type, optional draft identity, and method arguments.
# Any other difference makes the experiment ineligible.
experiment.schema.json json View source
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://example.invalid/local-first-agent-operations/post-11a/experiment.schema.json",
"title": "Model-neutral one-intervention inference experiment",
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "identity", "intervention", "active_path", "correctness", "observations"],
"properties": {
"schema_version": {"const": 1},
"identity": {
"type": "object",
"additionalProperties": false,
"required": ["run_id", "engine", "source_fingerprint", "binary_fingerprint", "target_artifact_fingerprint", "tokenizer_fingerprint", "template_fingerprint", "workload_fingerprint", "sampling_fingerprint", "extra_arguments_fingerprint", "runtime_configuration_fingerprint", "runtime_configuration", "sampling", "cache", "machine_class", "memory_gb", "warm_cold", "profiled", "concurrency", "batch_size", "context_tokens", "threads"],
"properties": {
"run_id": {"type": "string", "minLength": 1},
"engine": {"type": "string", "minLength": 1},
"source_fingerprint": {"$ref": "#/$defs/fingerprint"},
"binary_fingerprint": {"$ref": "#/$defs/fingerprint"},
"target_artifact_fingerprint": {"$ref": "#/$defs/fingerprint"},
"tokenizer_fingerprint": {"$ref": "#/$defs/fingerprint"},
"template_fingerprint": {"$ref": "#/$defs/fingerprint"},
"workload_fingerprint": {"$ref": "#/$defs/fingerprint"},
"sampling_fingerprint": {"$ref": "#/$defs/fingerprint"},
"extra_arguments_fingerprint": {"$ref": "#/$defs/fingerprint"},
"runtime_configuration_fingerprint": {"$ref": "#/$defs/fingerprint"},
"runtime_configuration": {"$ref": "#/$defs/runtimeConfiguration"},
"sampling": {"$ref": "#/$defs/sampling"},
"cache": {"$ref": "#/$defs/cache"},
"machine_class": {"type": "string", "minLength": 1},
"memory_gb": {"type": "integer", "minimum": 1},
"warm_cold": {"enum": ["warm", "cold"]},
"profiled": {"type": "boolean"},
"concurrency": {"type": "integer", "minimum": 1},
"batch_size": {"type": "integer", "minimum": 1},
"context_tokens": {"type": "integer", "minimum": 1},
"threads": {"type": "integer", "minimum": 1}
}
},
"intervention": {"$ref": "#/$defs/intervention"},
"active_path": {"$ref": "#/$defs/activePath"},
"correctness": {"$ref": "#/$defs/correctness"},
"observations": {
"type": "array",
"minItems": 3,
"items": {"$ref": "#/$defs/observation"}
}
},
"$defs": {
"fingerprint": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
"runtimeConfiguration": {
"type": "object",
"additionalProperties": false,
"required": ["speculative_method", "draft_artifact_fingerprint", "method_arguments"],
"properties": {
"speculative_method": {"type": "string", "enum": ["none", "candidate_method"]},
"draft_artifact_fingerprint": {"anyOf": [{"$ref": "#/$defs/fingerprint"}, {"type": "null"}]},
"method_arguments": {
"type": "object",
"additionalProperties": false,
"required": ["draft_tokens"],
"properties": {"draft_tokens": {"type": "integer", "minimum": 0}}
}
}
},
"sampling": {
"type": "object",
"additionalProperties": false,
"required": ["mode", "seed"],
"properties": {"mode": {"const": "greedy"}, "seed": {"type": "null"}}
},
"cache": {
"type": "object",
"additionalProperties": false,
"required": ["kv_format", "prompt_cache"],
"properties": {"kv_format": {"type": "string", "minLength": 1}, "prompt_cache": {"type": "boolean"}}
},
"intervention": {
"type": "object",
"additionalProperties": false,
"required": ["kind", "phase", "reviewed_dependent_paths"],
"properties": {
"kind": {"const": "speculative_method"},
"phase": {"const": "decode"},
"reviewed_dependent_paths": {
"const": ["runtime_configuration.speculative_method", "runtime_configuration.draft_artifact_fingerprint", "runtime_configuration.method_arguments"]
}
}
},
"activePath": {
"type": "object",
"additionalProperties": false,
"required": ["status", "expected_marker", "observed_marker", "counter_name", "counter_value"],
"properties": {
"status": {"enum": ["confirmed", "inactive", "contradictory", "unknown"]},
"expected_marker": {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,63}$"},
"observed_marker": {"type": "string", "pattern": "^(|[a-z][a-z0-9_]{0,63})$"},
"counter_name": {"const": "accepted_draft_tokens"},
"counter_value": {"type": "integer", "minimum": 0}
}
},
"correctness": {
"type": "object",
"additionalProperties": false,
"required": ["passed", "normalized_output_hash"],
"properties": {"passed": {"type": "boolean"}, "normalized_output_hash": {"$ref": "#/$defs/fingerprint"}}
},
"observation": {
"type": "object",
"additionalProperties": false,
"required": ["ttft_ms", "prompt_tokens_per_s", "decode_tokens_per_s", "peak_wired_mb"],
"properties": {
"ttft_ms": {"type": "number", "minimum": 0},
"prompt_tokens_per_s": {"type": "number", "minimum": 0},
"decode_tokens_per_s": {"type": "number", "minimum": 0},
"peak_wired_mb": {"type": "number", "minimum": 0}
}
}
}
}
experiment_lab.py python View source
#!/usr/bin/env python3
"""Validate and compare invented one-intervention inference records."""
from __future__ import annotations
import importlib.util
import json
import math
import re
import sys
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path
from types import ModuleType
from typing import Any
HEX64 = re.compile(r"^[0-9a-f]{64}$")
MARKER_ID = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
TOP_KEYS = {"schema_version", "identity", "intervention", "active_path", "correctness", "observations"}
IDENTITY_KEYS = {
"run_id", "engine", "source_fingerprint", "binary_fingerprint",
"target_artifact_fingerprint", "tokenizer_fingerprint", "template_fingerprint",
"workload_fingerprint", "sampling_fingerprint", "extra_arguments_fingerprint",
"runtime_configuration_fingerprint", "runtime_configuration", "sampling", "cache",
"machine_class", "memory_gb", "warm_cold", "profiled", "concurrency",
"batch_size", "context_tokens", "threads",
}
FINGERPRINT_FIELDS = {
"source_fingerprint", "binary_fingerprint", "target_artifact_fingerprint",
"tokenizer_fingerprint", "template_fingerprint", "workload_fingerprint",
"sampling_fingerprint", "extra_arguments_fingerprint", "runtime_configuration_fingerprint",
}
FROZEN_IDENTITY_FIELDS = IDENTITY_KEYS - {"run_id", "runtime_configuration_fingerprint", "runtime_configuration"}
INTERVENTION_KEYS = {"kind", "phase", "reviewed_dependent_paths"}
REVIEWED_PATHS = [
"runtime_configuration.speculative_method",
"runtime_configuration.draft_artifact_fingerprint",
"runtime_configuration.method_arguments",
]
ACTIVE_PATH_KEYS = {"status", "expected_marker", "observed_marker", "counter_name", "counter_value"}
CORRECTNESS_KEYS = {"passed", "normalized_output_hash"}
METRICS = {"ttft_ms", "prompt_tokens_per_s", "decode_tokens_per_s", "peak_wired_mb"}
class SchemaError(ValueError):
"""Raised when a record violates the closed teaching schema."""
@dataclass(frozen=True)
class ExperimentComparison:
"""Terminal classification and optional Part 10A metric projection."""
classification: str
experiment_kind: str | None
reasons: tuple[str, ...]
baseline_configuration_fingerprint: str
candidate_configuration_fingerprint: str
intervention: dict[str, Any]
intervention_change: dict[str, Any]
active_path: dict[str, Any]
projection_contract: str | None
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),
"baseline_configuration_fingerprint": self.baseline_configuration_fingerprint,
"candidate_configuration_fingerprint": self.candidate_configuration_fingerprint,
"intervention": self.intervention,
"intervention_change": self.intervention_change,
"active_path": self.active_path,
"projection_contract": self.projection_contract,
"metrics": list(self.metrics),
}
def _exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None:
actual = set(value)
if actual != expected:
raise SchemaError(f"{label} keys differ; missing={sorted(expected - actual)}, unknown={sorted(actual - expected)}")
def _integer(value: Any, label: str, *, minimum: int = 0) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
raise SchemaError(f"{label} must be an integer >= {minimum}")
return value
def _fingerprint(value: Any, label: str) -> str:
if not isinstance(value, str) or not HEX64.fullmatch(value):
raise SchemaError(f"{label} must be lowercase 64-hex")
return value
def configuration_fingerprint(configuration: dict[str, Any]) -> str:
"""Return the canonical SHA-256 fingerprint for a runtime configuration."""
encoded = json.dumps(configuration, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
return sha256(encoded).hexdigest()
def validate_record(value: Any) -> dict[str, Any]:
"""Return a validated experiment record or raise SchemaError."""
if not isinstance(value, dict):
raise SchemaError("record must be an object")
_exact_keys(value, TOP_KEYS, "record")
if type(value["schema_version"]) is not int or value["schema_version"] != 1:
raise SchemaError("schema_version must be integer 1")
identity = value["identity"]
intervention = value["intervention"]
active_path = value["active_path"]
correctness = value["correctness"]
observations = value["observations"]
for item, label in ((identity, "identity"), (intervention, "intervention"), (active_path, "active_path"), (correctness, "correctness")):
if not isinstance(item, dict):
raise SchemaError(f"{label} must be an object")
_exact_keys(identity, IDENTITY_KEYS, "identity")
_exact_keys(intervention, INTERVENTION_KEYS, "intervention")
_exact_keys(active_path, ACTIVE_PATH_KEYS, "active_path")
_exact_keys(correctness, CORRECTNESS_KEYS, "correctness")
for field in FINGERPRINT_FIELDS:
_fingerprint(identity[field], f"identity.{field}")
for field in ("run_id", "engine", "machine_class"):
if not isinstance(identity[field], str) or not identity[field].strip():
raise SchemaError(f"identity.{field} must be a nonempty string")
for field in ("memory_gb", "concurrency", "batch_size", "context_tokens", "threads"):
_integer(identity[field], f"identity.{field}", minimum=1)
if type(identity["profiled"]) is not bool:
raise SchemaError("identity.profiled must be Boolean")
if identity["warm_cold"] not in {"warm", "cold"}:
raise SchemaError("identity.warm_cold must be warm or cold")
sampling = identity["sampling"]
cache = identity["cache"]
configuration = identity["runtime_configuration"]
for item, label in ((sampling, "identity.sampling"), (cache, "identity.cache"), (configuration, "identity.runtime_configuration")):
if not isinstance(item, dict):
raise SchemaError(f"{label} must be an object")
_exact_keys(sampling, {"mode", "seed"}, "identity.sampling")
if sampling != {"mode": "greedy", "seed": None}:
raise SchemaError("identity.sampling must be greedy with a null seed")
_exact_keys(cache, {"kv_format", "prompt_cache"}, "identity.cache")
if not isinstance(cache["kv_format"], str) or not cache["kv_format"].strip() or type(cache["prompt_cache"]) is not bool:
raise SchemaError("identity.cache has invalid values")
_exact_keys(configuration, {"speculative_method", "draft_artifact_fingerprint", "method_arguments"}, "identity.runtime_configuration")
if not isinstance(configuration["speculative_method"], str) or configuration["speculative_method"] not in {"none", "candidate_method"}:
raise SchemaError("runtime speculative_method must be none or candidate_method")
draft_fingerprint = configuration["draft_artifact_fingerprint"]
if draft_fingerprint is not None:
_fingerprint(draft_fingerprint, "identity.runtime_configuration.draft_artifact_fingerprint")
arguments = configuration["method_arguments"]
if not isinstance(arguments, dict):
raise SchemaError("method_arguments must be an object")
_exact_keys(arguments, {"draft_tokens"}, "identity.runtime_configuration.method_arguments")
_integer(arguments["draft_tokens"], "method_arguments.draft_tokens")
if configuration_fingerprint(configuration) != identity["runtime_configuration_fingerprint"]:
raise SchemaError("runtime_configuration_fingerprint does not match canonical configuration")
if intervention != {"kind": "speculative_method", "phase": "decode", "reviewed_dependent_paths": REVIEWED_PATHS}:
raise SchemaError("intervention must use the reviewed speculative_method contract")
if active_path["status"] not in {"confirmed", "inactive", "contradictory", "unknown"}:
raise SchemaError("active_path.status is invalid")
for field in ("expected_marker", "observed_marker", "counter_name"):
if not isinstance(active_path[field], str):
raise SchemaError(f"active_path.{field} must be a string")
if not MARKER_ID.fullmatch(active_path["expected_marker"]):
raise SchemaError("active_path.expected_marker must be a sanitized symbolic identifier")
if active_path["observed_marker"] and not MARKER_ID.fullmatch(active_path["observed_marker"]):
raise SchemaError("active_path.observed_marker must be empty or a sanitized symbolic identifier")
if active_path["counter_name"] != "accepted_draft_tokens":
raise SchemaError("active_path marker or counter contract is invalid")
_integer(active_path["counter_value"], "active_path.counter_value")
if type(correctness["passed"]) is not bool:
raise SchemaError("correctness.passed must be Boolean")
_fingerprint(correctness["normalized_output_hash"], "correctness.normalized_output_hash")
if not isinstance(observations, list) or len(observations) < 3:
raise SchemaError("observations must contain at least three rows")
for index, row in enumerate(observations):
if not isinstance(row, dict):
raise SchemaError(f"observations[{index}] must be an object")
_exact_keys(row, METRICS, f"observations[{index}]")
for name, metric in row.items():
if isinstance(metric, bool) or not isinstance(metric, (int, float)) or not math.isfinite(metric) or metric < 0:
raise SchemaError(f"observations[{index}].{name} must be finite and nonnegative")
return value
def load_record(path: str | Path) -> dict[str, Any]:
"""Load and validate a JSON experiment record."""
with Path(path).open(encoding="utf-8") as handle:
return validate_record(json.load(handle))
def _terminal(baseline: dict[str, Any], candidate: dict[str, Any], classification: str, reasons: tuple[str, ...]) -> ExperimentComparison:
return ExperimentComparison(
classification=classification,
experiment_kind=None,
reasons=reasons,
baseline_configuration_fingerprint=baseline["identity"]["runtime_configuration_fingerprint"],
candidate_configuration_fingerprint=candidate["identity"]["runtime_configuration_fingerprint"],
intervention=candidate["intervention"],
intervention_change=_intervention_change(baseline, candidate),
active_path=candidate["active_path"],
projection_contract=None,
metrics=(),
)
def _intervention_change(baseline: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]:
"""Return the sanitized reviewed runtime-configuration change."""
return {
"baseline": baseline["identity"]["runtime_configuration"],
"candidate": candidate["identity"]["runtime_configuration"],
}
def compare_records(baseline: dict[str, Any], candidate: dict[str, Any], *, part10a_path: Path | None = None) -> ExperimentComparison:
"""Classify one runtime intervention and project eligible metrics."""
baseline = validate_record(baseline)
candidate = validate_record(candidate)
mismatches = tuple(sorted(field for field in FROZEN_IDENTITY_FIELDS if baseline["identity"][field] != candidate["identity"][field]))
if mismatches:
return _terminal(baseline, candidate, "ineligible", mismatches)
if baseline["intervention"] != candidate["intervention"]:
return _terminal(baseline, candidate, "ineligible", ("intervention_contract_mismatch",))
if baseline["identity"]["profiled"]:
return _terminal(baseline, candidate, "ineligible", ("profiled_runs_excluded",))
baseline_config = baseline["identity"]["runtime_configuration"]
candidate_config = candidate["identity"]["runtime_configuration"]
if baseline_config["speculative_method"] != "none" or candidate_config["speculative_method"] != "candidate_method":
return _terminal(baseline, candidate, "ineligible", ("abstract_method_transition_required",))
if baseline["identity"]["runtime_configuration_fingerprint"] == candidate["identity"]["runtime_configuration_fingerprint"]:
return _terminal(baseline, candidate, "ineligible", ("configuration_unchanged",))
if baseline_config["draft_artifact_fingerprint"] is not None or baseline_config["method_arguments"]["draft_tokens"] != 0:
return _terminal(baseline, candidate, "ineligible", ("baseline_intervention_not_disabled",))
active = candidate["active_path"]
if active["status"] != "confirmed" or active["expected_marker"] != active["observed_marker"] or active["counter_value"] <= 0:
return _terminal(baseline, candidate, "active_path_unproven", ("candidate_active_path_not_confirmed",))
baseline_active = baseline["active_path"]
if baseline_active["status"] != "confirmed" or baseline_active["expected_marker"] != baseline_active["observed_marker"] or baseline_active["counter_value"] != 0:
return _terminal(baseline, candidate, "active_path_unproven", ("baseline_disabled_path_not_confirmed",))
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 _terminal(baseline, candidate, "correctness_failed", tuple(correctness_reasons))
metrics = _part10a_projection(baseline, candidate, part10a_path=part10a_path)
return ExperimentComparison(
classification="comparable",
experiment_kind="runtime_configuration_optimization",
reasons=(),
baseline_configuration_fingerprint=baseline["identity"]["runtime_configuration_fingerprint"],
candidate_configuration_fingerprint=candidate["identity"]["runtime_configuration_fingerprint"],
intervention=candidate["intervention"],
intervention_change=_intervention_change(baseline, candidate),
active_path=candidate["active_path"],
projection_contract="part-10a-v1-metric-projection",
metrics=metrics,
)
def _load_part10a(path: Path) -> ModuleType:
spec = importlib.util.spec_from_file_location("post10a_benchmark_compare", path)
if spec is None or spec.loader is None:
raise RuntimeError(f"cannot load Part 10A comparator from {path}")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def _part10a_projection(baseline: dict[str, Any], candidate: dict[str, Any], *, part10a_path: Path | None) -> tuple[dict[str, Any], ...]:
comparator_path = part10a_path or Path(__file__).resolve().parents[1] / "post-10a" / "benchmark_compare.py"
module = _load_part10a(comparator_path)
def transform(record: dict[str, Any]) -> dict[str, Any]:
identity = record["identity"]
return {
"schema_version": 1,
"identity": {
"run_id": identity["run_id"],
"engine": identity["engine"],
"source_fingerprint": identity["source_fingerprint"],
"artifact_fingerprint": identity["target_artifact_fingerprint"],
"workload_fingerprint": identity["workload_fingerprint"],
"machine_class": identity["machine_class"],
"memory_gb": identity["memory_gb"],
"warm_cold": identity["warm_cold"],
"profiled": identity["profiled"],
"concurrency": identity["concurrency"],
"batch_size": identity["batch_size"],
"context_tokens": identity["context_tokens"],
"sampling_fingerprint": identity["sampling_fingerprint"],
},
"correctness": record["correctness"],
"observations": record["observations"],
}
result = module.compare_bundles(transform(baseline), transform(candidate))
if result.classification != "comparable":
raise RuntimeError(f"Part 10A metric projection rejected an eligible Part 11 pair: {result.reasons}")
return tuple(result.metrics)
def render_json(result: ExperimentComparison) -> str:
"""Render one stable JSON report."""
return json.dumps(result.as_dict(), indent=2, sort_keys=True) + "\n"
def render_markdown(result: ExperimentComparison) -> str:
"""Render one concise Markdown report."""
dependent_paths = ", ".join(result.intervention.get("reviewed_dependent_paths", []))
baseline_change = result.intervention_change.get("baseline", {})
candidate_change = result.intervention_change.get("candidate", {})
baseline_arguments = json.dumps(baseline_change.get("method_arguments", {}), sort_keys=True, separators=(",", ":"))
candidate_arguments = json.dumps(candidate_change.get("method_arguments", {}), sort_keys=True, separators=(",", ":"))
lines = [
f"Classification: `{result.classification}`",
f"Experiment: `{result.experiment_kind or 'none'}`",
f"Projection: `{result.projection_contract or 'none'}`",
f"Reasons: `{', '.join(result.reasons) if result.reasons else 'none'}`",
f"Baseline configuration: `{result.baseline_configuration_fingerprint}`",
f"Candidate configuration: `{result.candidate_configuration_fingerprint}`",
f"Intervention kind: `{result.intervention.get('kind', 'unknown')}`",
f"Intervention phase: `{result.intervention.get('phase', 'unknown')}`",
f"Reviewed dependent paths: `{dependent_paths}`",
f"Baseline method: `{baseline_change.get('speculative_method', 'unknown')}`",
f"Baseline draft identity: `{baseline_change.get('draft_artifact_fingerprint') or 'none'}`",
f"Baseline method arguments: `{baseline_arguments}`",
f"Candidate method: `{candidate_change.get('speculative_method', 'unknown')}`",
f"Candidate draft identity: `{candidate_change.get('draft_artifact_fingerprint') or 'none'}`",
f"Candidate method arguments: `{candidate_arguments}`",
f"Active-path status: `{result.active_path.get('status', 'unknown')}`",
f"Expected marker ID: `{result.active_path.get('expected_marker', '')}`",
f"Observed marker ID: `{result.active_path.get('observed_marker', '')}`",
f"Counter: `{result.active_path.get('counter_name', '')}`",
f"Counter value: `{result.active_path.get('counter_value', '')}`",
]
if result.metrics:
lines.extend(("", "| Metric | Unit | Direction | Baseline median | Candidate median | Outcome |", "| --- | --- | --- | ---: | ---: | --- |"))
for row in result.metrics:
lines.append(f"| {row['metric']} | {row['unit']} | {row['direction']} | {row['baseline_median']:.3f} | {row['candidate_median']:.3f} | {row['outcome']} |")
return "\n".join(lines) + "\n"
run_lab.py python View source
#!/usr/bin/env python3
"""Run the bounded, invented Hands-On 11A experiment lab."""
from __future__ import annotations
import copy
import json
import tempfile
from pathlib import Path
from experiment_lab import SchemaError, compare_records, configuration_fingerprint, load_record, render_json, render_markdown, validate_record
ROOT = Path(__file__).resolve().parent
FIXTURES = ROOT / "fixtures"
def main() -> int:
baseline = load_record(FIXTURES / "baseline.json")
candidate = load_record(FIXTURES / "candidate.json")
ineligible = load_record(FIXTURES / "candidate-ineligible.json")
unproven = load_record(FIXTURES / "candidate-unproven.json")
accepted = compare_records(baseline, candidate)
rejected = compare_records(baseline, ineligible)
path_unknown = compare_records(baseline, unproven)
correctness = copy.deepcopy(candidate)
correctness["correctness"]["passed"] = False
correctness_result = compare_records(baseline, correctness)
stochastic = copy.deepcopy(candidate)
stochastic["identity"]["sampling"] = {"mode": "stochastic", "seed": 7}
try:
validate_record(stochastic)
stochastic_rejected = False
except SchemaError:
stochastic_rejected = True
with tempfile.TemporaryDirectory(prefix="post-11a-lab-") as temporary:
report_dir = Path(temporary)
json_report = render_json(accepted)
markdown_report = render_markdown(accepted)
(report_dir / "comparison.json").write_text(json_report, encoding="utf-8")
(report_dir / "comparison.md").write_text(markdown_report, encoding="utf-8")
json_payload = json.loads(json_report)
parity_values = (
accepted.classification,
accepted.experiment_kind,
accepted.projection_contract,
accepted.intervention["kind"],
accepted.intervention["phase"],
*accepted.intervention["reviewed_dependent_paths"],
accepted.intervention_change["baseline"]["speculative_method"],
accepted.intervention_change["candidate"]["speculative_method"],
accepted.intervention_change["candidate"]["draft_artifact_fingerprint"],
accepted.active_path["status"],
accepted.active_path["expected_marker"],
accepted.active_path["observed_marker"],
accepted.active_path["counter_name"],
str(accepted.active_path["counter_value"]),
)
report_agreement = json_payload == accepted.as_dict() and all(str(value) in markdown_report for value in parity_values)
for side, record in (("baseline", baseline), ("candidate", candidate)):
configuration = record["identity"]["runtime_configuration"]
fingerprint = record["identity"]["runtime_configuration_fingerprint"]
expected_lines = (
f"{side.title()} configuration: `{fingerprint}`",
f"{side.title()} method: `{configuration['speculative_method']}`",
f"{side.title()} draft identity: `{configuration['draft_artifact_fingerprint'] or 'none'}`",
f"{side.title()} method arguments: `{json.dumps(configuration['method_arguments'], sort_keys=True, separators=(',', ':'))}`",
)
report_agreement = (
report_agreement
and json_payload["intervention_change"][side] == configuration
and json_payload[f"{side}_configuration_fingerprint"] == fingerprint
and all(line in markdown_report.splitlines() for line in expected_lines)
)
command_templates = (ROOT / "command-templates.txt").read_text(encoding="utf-8")
schema = json.loads((ROOT / "experiment.schema.json").read_text(encoding="utf-8"))
command_contract_matches = (
"--seed" not in command_templates
and command_templates.count("--temp 0") == 2
and schema["$defs"]["sampling"]["properties"]["mode"]["const"] == "greedy"
and schema["$defs"]["sampling"]["properties"]["seed"]["type"] == "null"
)
unsafe_marker = copy.deepcopy(candidate)
unsafe_marker["active_path"]["observed_marker"] = "../private/runtime.log"
try:
validate_record(unsafe_marker)
unsafe_marker_rejected = False
except SchemaError:
unsafe_marker_rejected = True
unsafe_method = copy.deepcopy(candidate)
identity = unsafe_method["identity"]
identity["runtime_configuration"]["speculative_method"] = "../private/runtime.log"
identity["runtime_configuration_fingerprint"] = configuration_fingerprint(identity["runtime_configuration"])
try:
compare_records(baseline, unsafe_method)
unsafe_method_rejected = False
except SchemaError:
unsafe_method_rejected = True
checks = {
"eligible_one_intervention": accepted.classification == "comparable",
"runtime_optimization_classified_by_part11": accepted.experiment_kind == "runtime_configuration_optimization",
"part10a_projection_versioned": accepted.projection_contract == "part-10a-v1-metric-projection",
"configuration_fingerprints_preserved": bool(accepted.baseline_configuration_fingerprint and accepted.candidate_configuration_fingerprint),
"active_path_preserved": accepted.active_path["status"] == "confirmed" and accepted.active_path["counter_value"] > 0,
"unrelated_cache_change_rejected": rejected.classification == "ineligible" and "cache" in rejected.reasons,
"unknown_active_path_rejected": path_unknown.classification == "active_path_unproven" and not path_unknown.metrics,
"correctness_blocks_projection": correctness_result.classification == "correctness_failed" and not correctness_result.metrics,
"stochastic_sampling_rejected": stochastic_rejected,
"reports_agree": report_agreement,
"command_template_matches_greedy_schema": command_contract_matches,
"unsafe_marker_rejected": unsafe_marker_rejected,
"unsafe_method_rejected_before_report": unsafe_method_rejected,
"privacy_fields_absent": _privacy_check(baseline, candidate, ineligible, unproven, accepted.as_dict()),
"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 _privacy_check(*values: object) -> bool:
text = json.dumps(values, sort_keys=True).lower()
forbidden = ('"prompt"', '"completion"', '"output_text"', '"hostname"', '"user"', '"ip_address"', '"model_path"', '"credential"')
return not any(term in text for term in forbidden)
if __name__ == "__main__":
raise SystemExit(main())
test_experiment_lab.py python View source
#!/usr/bin/env python3
"""Tests for the model-neutral Hands-On 11A companion."""
from __future__ import annotations
import copy
import json
import tempfile
import unittest
from pathlib import Path
from experiment_lab import SchemaError, compare_records, configuration_fingerprint, load_record, render_json, render_markdown, validate_record
ROOT = Path(__file__).resolve().parent
FIXTURES = ROOT / "fixtures"
class ExperimentLabTests(unittest.TestCase):
def setUp(self) -> None:
self.baseline = load_record(FIXTURES / "baseline.json")
self.candidate = load_record(FIXTURES / "candidate.json")
def test_eligible_structured_intervention(self) -> None:
result = compare_records(self.baseline, self.candidate)
self.assertEqual(result.classification, "comparable")
self.assertEqual(result.experiment_kind, "runtime_configuration_optimization")
self.assertEqual(result.projection_contract, "part-10a-v1-metric-projection")
self.assertTrue(result.metrics)
def test_stochastic_sampling_rejected_even_with_seed(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["identity"]["sampling"] = {"mode": "stochastic", "seed": 7}
with self.assertRaises(SchemaError):
validate_record(changed)
def test_unknown_field_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["identity"]["hostname"] = "not accepted"
with self.assertRaises(SchemaError):
validate_record(changed)
def test_missing_identity_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
del changed["identity"]["binary_fingerprint"]
with self.assertRaises(SchemaError):
validate_record(changed)
def test_boolean_numeric_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["identity"]["threads"] = True
with self.assertRaises(SchemaError):
validate_record(changed)
def test_canonical_configuration_fingerprint_required(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["identity"]["runtime_configuration"]["method_arguments"]["draft_tokens"] = 5
with self.assertRaises(SchemaError):
validate_record(changed)
changed["identity"]["runtime_configuration_fingerprint"] = configuration_fingerprint(changed["identity"]["runtime_configuration"])
validate_record(changed)
def test_unrelated_change_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["identity"]["cache"]["kv_format"] = "invented-q8"
result = compare_records(self.baseline, changed)
self.assertEqual(result.classification, "ineligible")
self.assertIn("cache", result.reasons)
def test_reviewed_dependent_fields_accepted(self) -> None:
config = self.candidate["identity"]["runtime_configuration"]
self.assertEqual(config["speculative_method"], "candidate_method")
self.assertIsNotNone(config["draft_artifact_fingerprint"])
self.assertGreater(config["method_arguments"]["draft_tokens"], 0)
self.assertEqual(compare_records(self.baseline, self.candidate).classification, "comparable")
def test_warm_cold_mismatch_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["identity"]["warm_cold"] = "cold"
result = compare_records(self.baseline, changed)
self.assertEqual(result.classification, "ineligible")
self.assertIn("warm_cold", result.reasons)
def test_profiled_pair_rejected(self) -> None:
baseline = copy.deepcopy(self.baseline)
candidate = copy.deepcopy(self.candidate)
baseline["identity"]["profiled"] = True
candidate["identity"]["profiled"] = True
result = compare_records(baseline, candidate)
self.assertEqual(result.classification, "ineligible")
self.assertIn("profiled_runs_excluded", result.reasons)
def test_active_path_unknown_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["active_path"]["status"] = "unknown"
changed["active_path"]["observed_marker"] = ""
result = compare_records(self.baseline, changed)
self.assertEqual(result.classification, "active_path_unproven")
self.assertFalse(result.metrics)
def test_correctness_failure_blocks_metrics(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["correctness"]["passed"] = False
result = compare_records(self.baseline, changed)
self.assertEqual(result.classification, "correctness_failed")
self.assertFalse(result.metrics)
def test_output_drift_blocks_metrics(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["correctness"]["normalized_output_hash"] = "f" * 64
result = compare_records(self.baseline, changed)
self.assertEqual(result.classification, "correctness_failed")
self.assertIn("normalized_output_hash_mismatch", result.reasons)
def test_report_preserves_contract_evidence(self) -> None:
result = compare_records(self.baseline, self.candidate)
json_report = render_json(result)
markdown_report = render_markdown(result)
payload = json.loads(json_report)
self.assertEqual(payload, result.as_dict())
self.assertIn(result.baseline_configuration_fingerprint, json_report)
self.assertIn(result.candidate_configuration_fingerprint, markdown_report)
expected_values = (
result.classification,
result.experiment_kind,
result.projection_contract,
result.intervention["kind"],
result.intervention["phase"],
*result.intervention["reviewed_dependent_paths"],
result.intervention_change["baseline"]["speculative_method"],
result.intervention_change["candidate"]["speculative_method"],
result.intervention_change["candidate"]["draft_artifact_fingerprint"],
result.active_path["status"],
result.active_path["expected_marker"],
result.active_path["observed_marker"],
result.active_path["counter_name"],
str(result.active_path["counter_value"]),
)
for value in expected_values:
self.assertIn(str(value), json_report)
self.assertIn(str(value), markdown_report)
for side, record in (("baseline", self.baseline), ("candidate", self.candidate)):
configuration = record["identity"]["runtime_configuration"]
fingerprint = record["identity"]["runtime_configuration_fingerprint"]
self.assertEqual(payload["intervention_change"][side], configuration)
self.assertEqual(payload[f"{side}_configuration_fingerprint"], fingerprint)
expected_lines = (
f"{side.title()} configuration: `{fingerprint}`",
f"{side.title()} method: `{configuration['speculative_method']}`",
f"{side.title()} draft identity: `{configuration['draft_artifact_fingerprint'] or 'none'}`",
f"{side.title()} method arguments: `{json.dumps(configuration['method_arguments'], sort_keys=True, separators=(',', ':'))}`",
)
for line in expected_lines:
self.assertIn(line, markdown_report.splitlines())
def test_unsafe_method_rejected_before_report(self) -> None:
for side in ("baseline", "candidate"):
with self.subTest(side=side):
baseline = copy.deepcopy(self.baseline)
candidate = copy.deepcopy(self.candidate)
identity = (baseline if side == "baseline" else candidate)["identity"]
identity["runtime_configuration"]["speculative_method"] = "../private/runtime.log"
identity["runtime_configuration_fingerprint"] = configuration_fingerprint(identity["runtime_configuration"])
with self.assertRaisesRegex(SchemaError, "speculative_method must be none or candidate_method"):
compare_records(baseline, candidate)
def test_method_vocabulary_matches_schema(self) -> None:
schema = json.loads((ROOT / "experiment.schema.json").read_text(encoding="utf-8"))
methods = schema["$defs"]["runtimeConfiguration"]["properties"]["speculative_method"]["enum"]
self.assertEqual(methods, ["none", "candidate_method"])
for method in (*methods, "unreviewed_method", "", None, []):
with self.subTest(method=method):
changed = copy.deepcopy(self.candidate)
identity = changed["identity"]
identity["runtime_configuration"]["speculative_method"] = method
identity["runtime_configuration_fingerprint"] = configuration_fingerprint(identity["runtime_configuration"])
if method in methods:
validate_record(changed)
else:
with self.assertRaises(SchemaError):
validate_record(changed)
def test_unsafe_marker_rejected(self) -> None:
changed = copy.deepcopy(self.candidate)
changed["active_path"]["observed_marker"] = "../private/runtime.log"
with self.assertRaises(SchemaError):
validate_record(changed)
def test_command_template_matches_greedy_schema(self) -> None:
commands = (ROOT / "command-templates.txt").read_text(encoding="utf-8")
schema = json.loads((ROOT / "experiment.schema.json").read_text(encoding="utf-8"))
self.assertNotIn("--seed", commands)
self.assertEqual(commands.count("--temp 0"), 2)
sampling = schema["$defs"]["sampling"]["properties"]
self.assertEqual(sampling["mode"]["const"], "greedy")
self.assertEqual(sampling["seed"]["type"], "null")
def test_temporary_report_cleanup(self) -> None:
result = compare_records(self.baseline, self.candidate)
with tempfile.TemporaryDirectory(prefix="post-11a-test-") as temporary:
directory = Path(temporary)
(directory / "report.json").write_text(render_json(result), encoding="utf-8")
self.assertTrue((directory / "report.json").is_file())
self.assertFalse(directory.exists())
if __name__ == "__main__":
unittest.main()
baseline.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "invented-baseline",
"engine": "invented-engine",
"source_fingerprint": "1111111111111111111111111111111111111111111111111111111111111111",
"binary_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"target_artifact_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"tokenizer_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444",
"template_fingerprint": "5555555555555555555555555555555555555555555555555555555555555555",
"workload_fingerprint": "6666666666666666666666666666666666666666666666666666666666666666",
"sampling_fingerprint": "7777777777777777777777777777777777777777777777777777777777777777",
"extra_arguments_fingerprint": "8888888888888888888888888888888888888888888888888888888888888888",
"runtime_configuration_fingerprint": "df0c8acfff17073fc2c8af4c11937ee0796fb13bf599e00bc2d92d4b0bbe1890",
"runtime_configuration": {
"speculative_method": "none",
"draft_artifact_fingerprint": null,
"method_arguments": {"draft_tokens": 0}
},
"sampling": {"mode": "greedy", "seed": null},
"cache": {"kv_format": "invented-default", "prompt_cache": false},
"machine_class": "invented-machine-class",
"memory_gb": 64,
"warm_cold": "warm",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"threads": 8
},
"intervention": {
"kind": "speculative_method",
"phase": "decode",
"reviewed_dependent_paths": [
"runtime_configuration.speculative_method",
"runtime_configuration.draft_artifact_fingerprint",
"runtime_configuration.method_arguments"
]
},
"active_path": {
"status": "confirmed",
"expected_marker": "speculation_disabled",
"observed_marker": "speculation_disabled",
"counter_name": "accepted_draft_tokens",
"counter_value": 0
},
"correctness": {
"passed": true,
"normalized_output_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
},
"observations": [
{"ttft_ms": 210.0, "prompt_tokens_per_s": 100.0, "decode_tokens_per_s": 20.0, "peak_wired_mb": 12000.0},
{"ttft_ms": 205.0, "prompt_tokens_per_s": 102.0, "decode_tokens_per_s": 20.5, "peak_wired_mb": 12050.0},
{"ttft_ms": 215.0, "prompt_tokens_per_s": 99.0, "decode_tokens_per_s": 19.8, "peak_wired_mb": 11980.0}
]
}
candidate.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "invented-candidate",
"engine": "invented-engine",
"source_fingerprint": "1111111111111111111111111111111111111111111111111111111111111111",
"binary_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"target_artifact_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"tokenizer_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444",
"template_fingerprint": "5555555555555555555555555555555555555555555555555555555555555555",
"workload_fingerprint": "6666666666666666666666666666666666666666666666666666666666666666",
"sampling_fingerprint": "7777777777777777777777777777777777777777777777777777777777777777",
"extra_arguments_fingerprint": "8888888888888888888888888888888888888888888888888888888888888888",
"runtime_configuration_fingerprint": "a378f7e9372233fe10c6f3f117638f7677ed53764f1e5a1b3ab8b9fd8cb0a209",
"runtime_configuration": {
"speculative_method": "candidate_method",
"draft_artifact_fingerprint": "9999999999999999999999999999999999999999999999999999999999999999",
"method_arguments": {"draft_tokens": 4}
},
"sampling": {"mode": "greedy", "seed": null},
"cache": {"kv_format": "invented-default", "prompt_cache": false},
"machine_class": "invented-machine-class",
"memory_gb": 64,
"warm_cold": "warm",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"threads": 8
},
"intervention": {
"kind": "speculative_method",
"phase": "decode",
"reviewed_dependent_paths": [
"runtime_configuration.speculative_method",
"runtime_configuration.draft_artifact_fingerprint",
"runtime_configuration.method_arguments"
]
},
"active_path": {
"status": "confirmed",
"expected_marker": "candidate_method_selected",
"observed_marker": "candidate_method_selected",
"counter_name": "accepted_draft_tokens",
"counter_value": 12
},
"correctness": {
"passed": true,
"normalized_output_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
},
"observations": [
{"ttft_ms": 212.0, "prompt_tokens_per_s": 99.5, "decode_tokens_per_s": 24.0, "peak_wired_mb": 12400.0},
{"ttft_ms": 208.0, "prompt_tokens_per_s": 101.0, "decode_tokens_per_s": 24.5, "peak_wired_mb": 12450.0},
{"ttft_ms": 214.0, "prompt_tokens_per_s": 98.5, "decode_tokens_per_s": 23.8, "peak_wired_mb": 12380.0}
]
}
candidate-ineligible.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "invented-ineligible",
"engine": "invented-engine",
"source_fingerprint": "1111111111111111111111111111111111111111111111111111111111111111",
"binary_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"target_artifact_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"tokenizer_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444",
"template_fingerprint": "5555555555555555555555555555555555555555555555555555555555555555",
"workload_fingerprint": "6666666666666666666666666666666666666666666666666666666666666666",
"sampling_fingerprint": "7777777777777777777777777777777777777777777777777777777777777777",
"extra_arguments_fingerprint": "8888888888888888888888888888888888888888888888888888888888888888",
"runtime_configuration_fingerprint": "a378f7e9372233fe10c6f3f117638f7677ed53764f1e5a1b3ab8b9fd8cb0a209",
"runtime_configuration": {
"speculative_method": "candidate_method",
"draft_artifact_fingerprint": "9999999999999999999999999999999999999999999999999999999999999999",
"method_arguments": {"draft_tokens": 4}
},
"sampling": {"mode": "greedy", "seed": null},
"cache": {"kv_format": "invented-q8", "prompt_cache": false},
"machine_class": "invented-machine-class",
"memory_gb": 64,
"warm_cold": "warm",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"threads": 8
},
"intervention": {
"kind": "speculative_method",
"phase": "decode",
"reviewed_dependent_paths": [
"runtime_configuration.speculative_method",
"runtime_configuration.draft_artifact_fingerprint",
"runtime_configuration.method_arguments"
]
},
"active_path": {
"status": "confirmed",
"expected_marker": "candidate_method_selected",
"observed_marker": "candidate_method_selected",
"counter_name": "accepted_draft_tokens",
"counter_value": 12
},
"correctness": {
"passed": true,
"normalized_output_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
},
"observations": [
{"ttft_ms": 212.0, "prompt_tokens_per_s": 99.5, "decode_tokens_per_s": 24.0, "peak_wired_mb": 11000.0},
{"ttft_ms": 208.0, "prompt_tokens_per_s": 101.0, "decode_tokens_per_s": 24.5, "peak_wired_mb": 11050.0},
{"ttft_ms": 214.0, "prompt_tokens_per_s": 98.5, "decode_tokens_per_s": 23.8, "peak_wired_mb": 10980.0}
]
}
candidate-unproven.json json View source
{
"schema_version": 1,
"identity": {
"run_id": "invented-unproven",
"engine": "invented-engine",
"source_fingerprint": "1111111111111111111111111111111111111111111111111111111111111111",
"binary_fingerprint": "2222222222222222222222222222222222222222222222222222222222222222",
"target_artifact_fingerprint": "3333333333333333333333333333333333333333333333333333333333333333",
"tokenizer_fingerprint": "4444444444444444444444444444444444444444444444444444444444444444",
"template_fingerprint": "5555555555555555555555555555555555555555555555555555555555555555",
"workload_fingerprint": "6666666666666666666666666666666666666666666666666666666666666666",
"sampling_fingerprint": "7777777777777777777777777777777777777777777777777777777777777777",
"extra_arguments_fingerprint": "8888888888888888888888888888888888888888888888888888888888888888",
"runtime_configuration_fingerprint": "a378f7e9372233fe10c6f3f117638f7677ed53764f1e5a1b3ab8b9fd8cb0a209",
"runtime_configuration": {
"speculative_method": "candidate_method",
"draft_artifact_fingerprint": "9999999999999999999999999999999999999999999999999999999999999999",
"method_arguments": {"draft_tokens": 4}
},
"sampling": {"mode": "greedy", "seed": null},
"cache": {"kv_format": "invented-default", "prompt_cache": false},
"machine_class": "invented-machine-class",
"memory_gb": 64,
"warm_cold": "warm",
"profiled": false,
"concurrency": 1,
"batch_size": 1,
"context_tokens": 4096,
"threads": 8
},
"intervention": {
"kind": "speculative_method",
"phase": "decode",
"reviewed_dependent_paths": [
"runtime_configuration.speculative_method",
"runtime_configuration.draft_artifact_fingerprint",
"runtime_configuration.method_arguments"
]
},
"active_path": {
"status": "unknown",
"expected_marker": "candidate_method_selected",
"observed_marker": "",
"counter_name": "accepted_draft_tokens",
"counter_value": 0
},
"correctness": {
"passed": true,
"normalized_output_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
},
"observations": [
{"ttft_ms": 212.0, "prompt_tokens_per_s": 99.5, "decode_tokens_per_s": 24.0, "peak_wired_mb": 12400.0},
{"ttft_ms": 208.0, "prompt_tokens_per_s": 101.0, "decode_tokens_per_s": 24.5, "peak_wired_mb": 12450.0},
{"ttft_ms": 214.0, "prompt_tokens_per_s": 98.5, "decode_tokens_per_s": 23.8, "peak_wired_mb": 12380.0}
]
}
Part 10A comparator dependency 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())
Extract the archive in a working directory, then enter the lab:
unzip hands-on-11a-one-inference-intervention.zip
cd post-11a
python --version
Python 3.10 or newer is sufficient; there are no third-party dependencies. From that directory, run:
python -B run_lab.py
python -B -m unittest -v test_experiment_lab.py
python -B -c 'from pathlib import Path; [compile(p.read_text(), str(p), "exec") for p in Path(".").glob("*.py")]'
The walkthrough should report fifteen passing conditions. The test suite should report nineteen passing tests. The compile command should return without output. None of those results says that the invented candidate is faster on real hardware. They say the teaching contract behaves as designed.
Step 1: Begin With a Closed Record
Each record has six top-level sections: schema version, identity, intervention, active-path evidence, correctness, and repeated observations. Missing and unknown fields fail validation. A Boolean cannot sneak into an integer field simply because Python happens to treat bool as a subclass of int.
The closed shape also creates a privacy boundary. There is no field for a prompt, completion, username, hostname, IP address, model path, credential, or conversation. If somebody adds one, the validator rejects the record rather than carrying it into a report.
Open the baseline and candidate fixtures side by side. The long repeated hexadecimal strings are obviously invented fingerprints. They authenticate nothing. Their purpose is to show where a real capture process would place content identities without publishing the content itself.
Step 2: Make the Runtime Configuration Prove Its Own Identity
The runtime configuration is a small canonical object:
{
"speculative_method": "candidate_method",
"draft_artifact_fingerprint": "9999999999999999999999999999999999999999999999999999999999999999",
"method_arguments": {
"draft_tokens": 4
}
}
The validator serializes that object with sorted keys and fixed separators, calculates its SHA-256 fingerprint, and compares the result with runtime_configuration_fingerprint. Change draft_tokens without updating the fingerprint and the record fails immediately.
This fingerprint is separate from source identity. That is important. Turning a runtime method off or on does not rewrite the llama.cpp source tree. Pretending the source changed would make the record say something untrue merely to fit another comparator’s classification rules.
Step 3: Require Greedy Generation for Exact Equality
This first lab requires greedy generation and a null seed. A fixed seed is not enough to make stochastic sampling identical across CPU implementations, backends, floating-point paths, and runtime revisions. If I expect exact normalized-output hashes to match, the generation contract needs to be deterministic enough for that expectation to make sense.
Try changing the candidate sampling section to this:
{
"mode": "stochastic",
"seed": 7
}
The validator rejects it before comparison. A future live adapter may support stochastic evaluation, but it will need a task-specific correctness contract reviewed on its own merits. It cannot quietly reuse exact output equality and call the result deterministic.
Step 4: Freeze Everything Outside the Intervention
The pair comparison freezes every identity field except the run ID and the runtime configuration. If the candidate changes from warm to cold, switches the KV format, uses another workload, alters concurrency, changes context, or runs under profiling, it is ineligible.
The candidate-ineligible.json fixture demonstrates this by changing the KV format while also enabling the speculative method. The lab returns ineligible and names cache as the reason. It emits no metric projection.
This is the part of benchmarking that saves me from my own enthusiasm. If I turn on speculation, reduce the cache, shorten the context, and change parallel slots in one run, I may get a better number. I will not know which change produced it, and I may not have performed the same work.
Step 5: Prove the Intended Path Was Active
Successful startup is not active-path evidence. The candidate declares an expected marker, records the observed marker, and retains a method-specific counter. These markers are sanitized symbolic identifiers such as candidate_method_selected, not raw log lines, paths, model names, or commands. The validator limits them to short lowercase letters, numbers, and underscores. A live adapter may derive an identifier from private evidence, but the private evidence does not belong in the portable record. In the invented fixture, the expected and observed markers match and the accepted-draft-token counter is positive.
The candidate-unproven.json fixture leaves the observed marker empty, sets the path state to unknown, and records no accepted tokens. Its terminal classification is active_path_unproven. Again, there is no speed comparison.
Different mechanisms will expose different evidence. A prompt-cache experiment wants an eligible identity and a confirmed hit. A KV-cache-format experiment wants the observed effective format. A speculative method wants a selected-path marker and acceptance evidence. The schema should not pretend those counters are interchangeable, but every intervention still needs a positive answer to the same question: did the runtime actually exercise what I intended to test?
Step 6: Put Correctness Before Projection
The baseline and candidate both have to pass correctness, and their normalized-output hashes have to match under the greedy contract. Change the candidate hash or set passed to false and the result becomes correctness_failed. Metrics remain empty.
Only after schema, identity, intervention, active-path, and correctness checks pass does the versioned adapter call the Hands-On 10A comparator. Both transformed records retain the same source fingerprint, so Part 10A sees them as repeatability-shaped inputs. Part 11 does not misuse that classification. It takes only the median and direction-aware metric rows, then emits its own explicit runtime_configuration_optimization classification with both configuration fingerprints, the intervention contract, and active-path evidence preserved.
That separation is a little more work than changing a source hash. It is also honest.
Step 7: Read the Invented Report Carefully
The eligible fixture produces JSON and Markdown projections in a temporary directory. Both reports contain the terminal classification, experiment kind, projection contract, both configuration fingerprints, intervention kind, phase, reviewed dependent paths, sanitized baseline and candidate configuration values, complete active-path marker and counter evidence, and median metric rows.
The invented report shows improved decode throughput alongside regressions in the other three metrics. Those numbers demonstrate report behavior, not a real performance result. They say nothing about speculative decoding, llama.cpp, MLX, Apple Silicon, or any model.
The walkthrough prints its checks and deletes its temporary reports. To read the Markdown report itself, run this from post-11a/; it prints the report without writing a file:
python -B -c 'from experiment_lab import load_record, compare_records, render_markdown; print(render_markdown(compare_records(load_record("fixtures/baseline.json"), load_record("fixtures/candidate.json"))))'
When run_lab.py leaves its temporary-directory block, it verifies that the report directory no longer exists. These commands do not leave an interpreter cache, start a server, call a provider, or modify operator configuration.
Mapping the Lab to a Real Experiment
The command templates show how a later operator-owned adapter might express the baseline and candidate. Every path and value remains a placeholder. Before using a real model, I would need to qualify the artifact and license, capture the effective command rather than the intended command, define the runtime markers, retain machine and contention state, and decide which normalized correctness contract applies to the workload.
I would also keep the experiments separate. Embedded MTP off versus on is one pair. Prompt-cache cold versus warm is another. A KV-cache representation change is another. N-gram speculation, context length, and concurrency each get their own pair. The one-intervention rule is not there to make benchmarking tedious. It is what gives the result a cause I can defend.
Current State
The model-free companion contains a closed schema, canonical runtime-configuration fingerprints, greedy exact-output correctness, structured intervention dependencies, sanitized active-path identifiers, terminal fail-closed classifications, a versioned Part 10A projection adapter, four invented fixtures, a fifteen-condition walkthrough, and nineteen tests. It runs with the Python standard library and performs no model, service, network, or persistent-cache operation.
Next Work
The next useful addition would be a separately reviewed adapter for an operator-owned runtime. That means choosing and qualifying a model, capturing what actually ran, and deciding which correctness checks the workload needs. Until then, this lab deliberately stops at invented records. It gives me a place to test the rules before I trust them with a real performance claim.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.