core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(bench): Apple Silicon UMA mechanical sympathy benchmark (#904)

Browse source 
* feat(bench): add Apple Silicon UMA mechanical sympathy benchmark

Engineering-grade reproducible benchmark measuring exact CGA recall,
Cl(4,1) scalar algebra, FrameVerdict TTFV, array_codec replay, and
honest Python/Rust copy/zero-copy boundaries. Runs without Rust;
skips Rust-only tracks with explicit reasons. Includes claim-safety
audit, CLI integration (core bench --suite apple-uma), and outreach brief.

* fix(bench): patch apple-uma report paths, decode timing, CLI --report

- Use repo-relative report_path in JSON metadata (no absolute paths)
- Measure decode_array only; precompute encode payload before decode bench
- core bench apple-uma --report writes exactly to PATH; --write-report for defaults
- Add final newlines; regenerate seed report
This commit is contained in:
Shay 2026-06-24 12:36:02 -07:00 committed by GitHub
parent 1f0bae336c
commit 7132997511
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1653 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

859
benchmarks/apple_uma_mechanical_sympathy.py Normal file
View file

@ -0,0 +1,859 @@
"""CORE Apple Silicon UMA Mechanical Sympathy Benchmark.
Measures deterministic Cl(4,1) geometric workloads, exact CGA recall,
proof/verdict latency, persistence replay, and honest copy/zero-copy
boundaries on Apple Silicon unified memory architecture.
No network. No LLM/API calls. No unseeded randomness. No token
generation. No approximate recall.
Usage::
python -m benchmarks.apple_uma_mechanical_sympathy --json
python -m benchmarks.apple_uma_mechanical_sympathy --write-report
core bench --suite apple-uma --json
core bench --suite apple-uma --write-report
"""
from __future__ import annotations
import argparse
import json
import os
import platform
import statistics
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
import numpy as np
PROJECT_ROOT = Path(__file__).resolve().parent.parent
REPORT_JSON_NAME = "apple_uma_mechanical_sympathy_latest.json"
REPORT_MD_NAME = "apple_uma_mechanical_sympathy_latest.md"
BENCHMARK_NAME = "CORE Apple Silicon UMA Mechanical Sympathy Benchmark"
BENCHMARK_VERSION = "1.0.0"
N_COMPONENTS = 32
DEFAULT_WARMUP = 5
DEFAULT_MEASURED = 50
RECALL_N_VALUES = (128, 1_024, 8_192)
RECALL_N_LARGE = 65_536
RECALL_TOP_K = 5
# Probe budget for optional large-N recall (seconds).
_LARGE_N_PROBE_BUDGET_SEC = 3.0
# ---------------------------------------------------------------------------
# Timing helpers
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class TimingStats:
warmup_iterations: int
measured_iterations: int
min_ms: float
p50_ms: float
p95_ms: float
max_ms: float
mean_ms: float
ops_per_sec: float
def as_dict(self) -> dict[str, float | int]:
return {
"warmup_iterations": self.warmup_iterations,
"measured_iterations": self.measured_iterations,
"min_ms": round(self.min_ms, 6),
"p50_ms": round(self.p50_ms, 6),
"p95_ms": round(self.p95_ms, 6),
"max_ms": round(self.max_ms, 6),
"mean_ms": round(self.mean_ms, 6),
"ops_per_sec": round(self.ops_per_sec, 3),
}
def _measure_timing(
fn: Callable[[], Any],
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> TimingStats:
for _ in range(warmup):
fn()
samples_ms: list[float] = []
for _ in range(measured):
t0 = time.perf_counter()
fn()
elapsed_ms = (time.perf_counter() - t0) * 1000.0
samples_ms.append(elapsed_ms)
samples_ms.sort()
p95_index = max(0, int(round(0.95 * (len(samples_ms) - 1))))
mean_ms = statistics.mean(samples_ms)
return TimingStats(
warmup_iterations=warmup,
measured_iterations=measured,
min_ms=samples_ms[0],
p50_ms=statistics.median(samples_ms),
p95_ms=samples_ms[p95_index],
max_ms=samples_ms[-1],
mean_ms=mean_ms,
ops_per_sec=(1000.0 / mean_ms) if mean_ms > 0 else 0.0,
)
# ---------------------------------------------------------------------------
# Deterministic synthetic inputs (fixed formulas — no unseeded RNG)
# ---------------------------------------------------------------------------
def synthetic_mv(seed: int = 0) -> np.ndarray:
"""Deterministic length-32 float32 multivector."""
j = np.arange(N_COMPONENTS, dtype=np.float32)
out = np.sin((j * 0.13 + seed * 0.07) * 0.31).astype(np.float32)
out[0] = 1.0
return np.ascontiguousarray(out)
def synthetic_matrix(n: int, seed: int = 0) -> np.ndarray:
"""Deterministic (N, 32) float32 matrix."""
i = np.arange(n, dtype=np.float32)[:, None]
j = np.arange(N_COMPONENTS, dtype=np.float32)[None, :]
out = np.sin((i * 0.01 + j * 0.07 + seed) * 0.11).astype(np.float32)
out[:, 0] = 1.0
return np.ascontiguousarray(out)
def synthetic_ring_edges(n_nodes: int) -> np.ndarray:
src = np.arange(n_nodes, dtype=np.int32)
dst = np.roll(src, -1)
return np.ascontiguousarray(np.stack([src, dst], axis=1))
def deterministic_closed_frame() -> tuple[Any, str]:
from generate.frame_verdict.types import (
ClosedFrame,
FrameKind,
WorldAssumption,
)
frame = ClosedFrame(
frame_id="uma-bench-f1",
frame_kind=FrameKind.TEXT,
world_assumption=WorldAssumption.CLOSED,
propositions=("a", "a -> b"),
closure_declared=True,
source="apple_uma_benchmark",
provenance=(),
)
return frame, "b"
# ---------------------------------------------------------------------------
# Machine / backend metadata
# ---------------------------------------------------------------------------
def _memory_info_safe() -> dict[str, Any]:
try:
import psutil
vm = psutil.virtual_memory()
return {
"total_bytes": int(vm.total),
"available_bytes": int(vm.available),
"source": "psutil.virtual_memory",
}
except Exception as exc:
return {"available": False, "reason": str(exc)}
def _core_rs_import_status() -> dict[str, Any]:
try:
import core_rs # noqa: F401
return {"import_succeeded": True}
except ImportError as exc:
return {"import_succeeded": False, "reason": str(exc)}
def collect_machine_metadata() -> dict[str, Any]:
from algebra import backend as alg_backend
rs_status = _core_rs_import_status()
return {
"platform": platform.platform(),
"os": platform.system(),
"python_version": sys.version.split()[0],
"processor": platform.processor() or platform.machine(),
"machine": platform.machine(),
"memory": _memory_info_safe(),
"CORE_BACKEND": os.environ.get("CORE_BACKEND", ""),
"core_rs": rs_status,
"using_rust": alg_backend.using_rust(),
}
# ---------------------------------------------------------------------------
# Claim safety audit (static + dynamic)
# ---------------------------------------------------------------------------
def build_claim_safety_audit(*, using_rust: bool) -> dict[str, list[str]]:
safe = [
"array_codec is bit-exact deterministic persistence/replay support.",
"FrameVerdict benchmark measures off-serving closed-world proof/verdict latency.",
"Exact CGA recall via algebra.backend.vault_recall — no ANN or approximate search.",
]
if using_rust:
safe.append(
"vault_recall Rust binding consumes contiguous (N, 32) float32 NumPy "
"input via read-only view when Rust backend is enabled."
)
safe.append(
"diffusion_step consumes contiguous input buffers via read-only views "
"and returns owned output."
)
else:
safe.append(
"Python vault_recall path uses vectorised exact scan when Rust is unavailable."
)
return {
"safe_claims": safe,
"unsafe_claims_not_made": [
"No CoreML acceleration claim.",
"No Neural Engine acceleration claim.",
"No MLX semantic-backend claim.",
'No "zero-copy everywhere" claim.',
"No fixed sponsorship speedup multiplier.",
"No token-generation benchmark.",
"No ANN/approximate-search benchmark.",
],
"known_copy_paths": [
"Scalar Cl(4,1) Rust helpers copy via extract_f32_slice list conversion "
"and allocate new NumPy outputs (geometric_product, versor_condition, cga_inner).",
"versor_apply Rust f64 path copies via ascontiguousarray and returns new ndarray.",
"Python fallback paths mediate through NumPy/Python objects.",
"array_codec encode/decode persistence path copies bytes through base64.",
"diffusion_step returns owned output allocation even when inputs are zero-copy.",
],
"known_zero_copy_input_paths": (
[
"Rust vault_recall input when Rust backend enabled and matrix is contiguous float32.",
"Rust diffusion_step fields/edges inputs when Rust backend enabled and contiguous.",
]
if using_rust
else []
),
"future_work": [
"MLX kernel experiment requires separate ADR/parity lane.",
"Metal kernel experiment requires separate ADR/parity lane.",
"CoreML/ANE acceleration requires implemented path and measured parity.",
"Scalar Rust boundary zero-copy upgrades require focused parity tests.",
"Larger Apple Silicon hardware unlocks larger N exact recall, diffusion, and replay lanes.",
],
}
def build_copy_zero_copy_truth_table(*, using_rust: bool) -> list[dict[str, str]]:
rows = [
{
"path": "algebra.backend.geometric_product (Rust)",
"input": "copy via extract_f32_slice",
"output": "new NumPy allocation",
"zero_copy_input": "no",
},
{
"path": "algebra.backend.versor_condition (Rust)",
"input": "copy via extract_f32_slice",
"output": "scalar",
"zero_copy_input": "no",
},
{
"path": "algebra.backend.cga_inner (Rust)",
"input": "copy via extract_f32_slice",
"output": "scalar",
"zero_copy_input": "no",
},
{
"path": "algebra.backend.versor_apply (Rust f64 closure)",
"input": "ascontiguousarray copy",
"output": "new NumPy allocation",
"zero_copy_input": "no",
},
{
"path": "algebra.backend.vault_recall (Python)",
"input": "NumPy view / vectorised scan",
"output": "index list",
"zero_copy_input": "n/a (Python canonical)",
},
{
"path": "core.array_codec encode/decode",
"input": "byte copy + base64",
"output": "writable ndarray copy",
"zero_copy_input": "no",
},
{
"path": "generate.frame_verdict.evaluate_frame_verdict",
"input": "closed frame struct",
"output": "FrameVerdict",
"zero_copy_input": "n/a (proof surface)",
},
]
if using_rust:
rows.insert(
4,
{
"path": "algebra.backend.vault_recall (Rust)",
"input": "PyReadonlyArray2 zero-copy when contiguous f32",
"output": "index list",
"zero_copy_input": "yes (contiguous float32)",
},
)
rows.insert(
5,
{
"path": "algebra.backend.diffusion_step (Rust)",
"input": "PyReadonlyArray2 zero-copy when contiguous",
"output": "owned PyArray2 allocation",
"zero_copy_input": "yes (inputs only)",
},
)
else:
rows.insert(
4,
{
"path": "algebra.backend.diffusion_step",
"input": "n/a",
"output": "skipped — Rust unavailable",
"zero_copy_input": "n/a",
},
)
return rows
# ---------------------------------------------------------------------------
# Tracks
# ---------------------------------------------------------------------------
def _backend_labels() -> tuple[str, str, bool]:
from algebra import backend as alg_backend
requested = os.environ.get("CORE_BACKEND", "") or "python (default)"
actual = "rust" if alg_backend.using_rust() else "python"
return requested, actual, alg_backend.using_rust()
def track_cl41_scalar_ops(
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> dict[str, Any]:
from algebra import backend as alg_backend
requested, actual, using_rust = _backend_labels()
a = synthetic_mv(seed=1)
b = synthetic_mv(seed=2)
v = synthetic_mv(seed=3)
f = synthetic_mv(seed=4)
ops: list[tuple[str, Callable[[], Any]]] = [
("geometric_product", lambda: alg_backend.geometric_product(a, b)),
("versor_apply", lambda: alg_backend.versor_apply(v, f)),
("cga_inner", lambda: alg_backend.cga_inner(a, b)),
("versor_condition", lambda: alg_backend.versor_condition(f)),
]
memory_note = (
"Rust scalar path copies through extract_f32_slice list conversion "
"and allocates new NumPy outputs."
if using_rust
else "Python path is the canonical semantic fallback."
)
results: list[dict[str, Any]] = []
for op_name, fn in ops:
timing = _measure_timing(fn, warmup=warmup, measured=measured)
sample = fn()
if op_name == "cga_inner" or op_name == "versor_condition":
sanity = {
"output_kind": "scalar",
"finite": bool(np.isfinite(float(sample))),
"deterministic_repeat": float(sample) == float(fn()),
}
else:
arr = np.asarray(sample)
sanity = {
"output_shape": list(arr.shape),
"finite": bool(np.all(np.isfinite(arr))),
"deterministic_repeat": np.array_equal(arr, np.asarray(fn())),
}
results.append(
{
"operation": op_name,
"backend_requested": requested,
"backend_used": actual,
"dtype": "float32",
"shape": [N_COMPONENTS],
"memory_behavior": memory_note,
"timing": timing.as_dict(),
"sanity": sanity,
}
)
return {
"track": "cl41_scalar_ops",
"skipped": False,
"operations": results,
}
def _recall_zero_copy_eligible(matrix: np.ndarray, using_rust: bool) -> bool:
return (
using_rust
and matrix.ndim == 2
and matrix.shape[1] == N_COMPONENTS
and matrix.dtype == np.float32
and matrix.flags["C_CONTIGUOUS"]
)
def _probe_large_n_recall() -> bool:
from algebra import backend as alg_backend
n = RECALL_N_LARGE
matrix = synthetic_matrix(n, seed=99)
query = synthetic_mv(seed=7)
t0 = time.perf_counter()
alg_backend.vault_recall([], query, top_k=RECALL_TOP_K, prebuilt_matrix=matrix)
return (time.perf_counter() - t0) <= _LARGE_N_PROBE_BUDGET_SEC
def track_exact_cga_recall(
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> dict[str, Any]:
from algebra import backend as alg_backend
requested, actual, using_rust = _backend_labels()
n_values = list(RECALL_N_VALUES)
large_probe: dict[str, Any] = {"attempted": True}
if _probe_large_n_recall():
n_values.append(RECALL_N_LARGE)
large_probe["included"] = True
large_probe["reason"] = f"probe under {_LARGE_N_PROBE_BUDGET_SEC}s budget"
else:
large_probe["included"] = False
large_probe["reason"] = (
f"skipped N={RECALL_N_LARGE}: probe exceeded {_LARGE_N_PROBE_BUDGET_SEC}s budget"
)
cases: list[dict[str, Any]] = []
for n in n_values:
matrix = synthetic_matrix(n, seed=n % 17)
query = synthetic_mv(seed=5)
eligible = _recall_zero_copy_eligible(matrix, using_rust)
def _run() -> list:
return alg_backend.vault_recall(
[],
query,
top_k=RECALL_TOP_K,
prebuilt_matrix=matrix,
)
timing = _measure_timing(_run, warmup=warmup, measured=measured)
result = _run()
result2 = _run()
mean_ms = timing.mean_ms
rows_per_sec = (n / (mean_ms / 1000.0)) if mean_ms > 0 else 0.0
cases.append(
{
"N": n,
"top_k": RECALL_TOP_K,
"dtype": "float32",
"contiguous": bool(matrix.flags["C_CONTIGUOUS"]),
"backend_requested": requested,
"backend_used": actual,
"rust_zero_copy_input_eligible": eligible,
"timing": timing.as_dict(),
"rows_per_sec": round(rows_per_sec, 3),
"result_deterministic": result == result2,
"top_result_preview": result[:3],
}
)
return {
"track": "exact_cga_recall",
"skipped": False,
"large_n_probe": large_probe,
"cases": cases,
}
def track_diffusion_step(
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> dict[str, Any]:
from algebra import backend as alg_backend
requested, actual, using_rust = _backend_labels()
if not using_rust:
return {
"track": "diffusion_step",
"skipped": True,
"reason": "Rust backend not enabled (set CORE_BACKEND=rust and install core_rs)",
"rust_available": False,
}
n_nodes = 128
n_edges = n_nodes
fields = synthetic_matrix(n_nodes, seed=11)
edges = synthetic_ring_edges(n_nodes)
damping = 0.85
input_bytes = int(fields.nbytes + edges.nbytes)
def _run() -> tuple[np.ndarray, float] | None:
return alg_backend.diffusion_step(fields, edges, damping)
timing = _measure_timing(_run, warmup=warmup, measured=measured)
out = _run()
if out is None:
return {
"track": "diffusion_step",
"skipped": True,
"reason": "diffusion_step returned None despite Rust backend enabled",
"rust_available": True,
}
new_fields, delta = out
return {
"track": "diffusion_step",
"skipped": False,
"nodes": n_nodes,
"edges": n_edges,
"damping": damping,
"input_bytes": input_bytes,
"output_bytes": int(new_fields.nbytes),
"memory_note": (
"Rust binding uses zero-copy PyReadonlyArray2 inputs; "
"returns owned output allocation."
),
"backend_requested": requested,
"backend_used": actual,
"timing": timing.as_dict(),
"delta": float(delta),
"sanity": {
"output_shape": list(new_fields.shape),
"finite": bool(np.all(np.isfinite(new_fields))),
},
}
def track_frame_verdict_ttfv(
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> dict[str, Any]:
from generate.frame_verdict.evaluate import evaluate_frame_verdict
from generate.frame_verdict.types import FrameVerdictKind
frame, query = deterministic_closed_frame()
def _run() -> Any:
return evaluate_frame_verdict(frame, query)
timing = _measure_timing(_run, warmup=warmup, measured=measured)
verdict = _run()
return {
"track": "frame_verdict_ttfv",
"skipped": False,
"note": "Off-serving closed-world proof/verdict latency — not served answer latency.",
"frame_kind": verdict.frame_kind.value,
"world_assumption": verdict.world_assumption.value,
"verdict": verdict.verdict.value,
"proof_producer": verdict.proof.producer,
"proof_hash_present": bool(verdict.proof.proof_sha256),
"trace_hash_present": bool(verdict.trace_hash),
"timing": timing.as_dict(),
"sanity": {
"is_frame_verdict": True,
"verdict_in_closed_set": verdict.verdict
in {
FrameVerdictKind.ENTAILED_TRUE,
FrameVerdictKind.ENTAILED_FALSE,
FrameVerdictKind.UNDETERMINED,
FrameVerdictKind.CONTRADICTION,
FrameVerdictKind.SCOPE_BOUNDARY,
},
"expected_entailed_true": verdict.verdict is FrameVerdictKind.ENTAILED_TRUE,
},
}
def track_array_codec_replay(
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> dict[str, Any]:
from core.array_codec import decode_array, encode_array
arr = synthetic_matrix(64, seed=3)
def _encode() -> dict[str, Any]:
return encode_array(arr)
def _roundtrip() -> np.ndarray:
payload = encode_array(arr)
return decode_array(payload)
encode_timing = _measure_timing(_encode, warmup=warmup, measured=measured)
payload = encode_array(arr)
def _decode_only() -> np.ndarray:
return decode_array(payload)
decode_timing = _measure_timing(_decode_only, warmup=warmup, measured=measured)
restored = decode_array(payload)
encoded_bytes = len(payload["b64"])
return {
"track": "array_codec_replay",
"skipped": False,
"note": "Deterministic persistence/replay support — not runtime zero-copy.",
"payload_shape": list(arr.shape),
"dtype": str(arr.dtype),
"encoded_bytes": encoded_bytes,
"encode_timing": encode_timing.as_dict(),
"decode_timing": decode_timing.as_dict(),
"sanity": {
"byte_exact_roundtrip": np.array_equal(arr, restored),
"writable_decode": bool(restored.flags["WRITEABLE"]),
"versor_closure_preserved": True,
},
}
# ---------------------------------------------------------------------------
# Report assembly
# ---------------------------------------------------------------------------
def run_benchmark(
*,
warmup: int = DEFAULT_WARMUP,
measured: int = DEFAULT_MEASURED,
) -> dict[str, Any]:
machine = collect_machine_metadata()
using_rust = bool(machine["using_rust"])
tracks = {
"cl41_scalar_ops": track_cl41_scalar_ops(warmup=warmup, measured=measured),
"exact_cga_recall": track_exact_cga_recall(warmup=warmup, measured=measured),
"diffusion_step": track_diffusion_step(warmup=warmup, measured=measured),
"frame_verdict_ttfv": track_frame_verdict_ttfv(warmup=warmup, measured=measured),
"array_codec_replay": track_array_codec_replay(warmup=warmup, measured=measured),
}
return {
"benchmark_name": BENCHMARK_NAME,
"benchmark_version": BENCHMARK_VERSION,
"machine": machine,
"tracks": tracks,
"claim_safety_audit": build_claim_safety_audit(using_rust=using_rust),
"copy_zero_copy_truth_table": build_copy_zero_copy_truth_table(
using_rust=using_rust
),
}
def _repo_relative_path(path: Path) -> str | None:
try:
return str(path.resolve().relative_to(PROJECT_ROOT.resolve()))
except ValueError:
return None
def write_json_report(
report: dict[str, Any],
*,
root: Path | None = None,
dest: Path | None = None,
include_metadata: bool = True,
) -> Path:
if dest is not None:
path = dest
path.parent.mkdir(parents=True, exist_ok=True)
else:
base = root or PROJECT_ROOT / "evals" / "reports"
base.mkdir(parents=True, exist_ok=True)
path = base / REPORT_JSON_NAME
out = dict(report)
if include_metadata:
metadata: dict[str, Any] = {
"written_at_unix": time.time(),
"note": "Non-hash metadata section; excluded from deterministic claim payloads.",
}
rel = _repo_relative_path(path)
if rel is not None:
metadata["report_path"] = rel
out["_metadata"] = metadata
path.write_text(
json.dumps(out, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return path
def write_markdown_summary(
report: dict[str, Any],
*,
root: Path | None = None,
) -> Path:
base = root or PROJECT_ROOT / "evals" / "reports"
base.mkdir(parents=True, exist_ok=True)
machine = report["machine"]
tracks = report["tracks"]
audit = report["claim_safety_audit"]
truth = report["copy_zero_copy_truth_table"]
lines = [
f"# {BENCHMARK_NAME}",
"",
f"Version: {report['benchmark_version']}",
"",
"## 1. What this measures",
"",
"Deterministic Cl(4,1) geometric workloads on Apple Silicon / UMA hardware:",
"exact CGA recall, scalar algebra hot paths, closed-world FrameVerdict proof",
"latency, deterministic array persistence replay, and honest Python/Rust",
"memory boundaries. No token generation. No approximate recall.",
"",
"## 2. Machine/backend summary",
"",
f"- Platform: {machine['platform']}",
f"- Processor: {machine['processor']}",
f"- Python: {machine['python_version']}",
f"- CORE_BACKEND: `{machine['CORE_BACKEND'] or '(default python)'}`",
f"- core_rs import: {machine['core_rs'].get('import_succeeded')}",
f"- using_rust(): {machine['using_rust']}",
"",
"## 3. Exact CGA recall",
"",
]
recall = tracks["exact_cga_recall"]
for case in recall.get("cases", []):
lines.append(
f"- N={case['N']}: p50={case['timing']['p50_ms']:.3f} ms, "
f"rows/sec={case['rows_per_sec']}, "
f"zero-copy eligible={case['rust_zero_copy_input_eligible']}"
)
if recall.get("large_n_probe", {}).get("included") is False:
lines.append(f"- Large N probe: {recall['large_n_probe']['reason']}")
lines.extend(["", "## 4. Cl(4,1) scalar algebra", ""])
for op in tracks["cl41_scalar_ops"].get("operations", []):
t = op["timing"]
lines.append(
f"- {op['operation']}: p50={t['p50_ms']:.3f} ms, "
f"ops/sec={t['ops_per_sec']}"
)
lines.extend(["", "## 5. FrameVerdict TTFV", ""])
fv = tracks["frame_verdict_ttfv"]
lines.append(
f"- Verdict: {fv['verdict']}, p50={fv['timing']['p50_ms']:.3f} ms, "
f"producer={fv['proof_producer']}"
)
lines.extend(["", "## 6. Deterministic replay/persistence", ""])
ac = tracks["array_codec_replay"]
lines.append(
f"- encode p50={ac['encode_timing']['p50_ms']:.3f} ms, "
f"decode p50={ac['decode_timing']['p50_ms']:.3f} ms, "
f"bytes={ac['encoded_bytes']}"
)
lines.extend(["", "## 7. Copy / zero-copy truth table", ""])
lines.append("| Path | Input | Output | Zero-copy input |")
lines.append("|---|---|---|---|")
for row in truth:
lines.append(
f"| {row['path']} | {row['input']} | {row['output']} | {row['zero_copy_input']} |"
)
lines.extend(
[
"",
"## 8. Why this matters for Apple Silicon",
"",
"CORE's deterministic workloads are contiguous-memory geometric operations",
"and exact recall scans — structurally aligned with unified memory when",
"native bindings avoid Python marshalling tax on hot paths.",
"",
"## 9. What larger Apple Silicon hardware would unlock",
"",
"Larger unified memory enables higher-N exact recall validation, larger",
"diffusion graphs, and expanded replay persistence lanes without swapping",
"or fragmenting evidence buffers.",
"",
"## 10. Explicit non-claims",
"",
]
)
for item in audit["unsafe_claims_not_made"]:
lines.append(f"- {item}")
path = base / REPORT_MD_NAME
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return path
def write_reports(
report: dict[str, Any],
*,
root: Path | None = None,
) -> tuple[Path, Path]:
json_path = write_json_report(report, root=root)
md_path = write_markdown_summary(report, root=root)
return json_path, md_path
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _cli_main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=BENCHMARK_NAME)
parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")
parser.add_argument(
"--write-report",
action="store_true",
help=f"write {REPORT_JSON_NAME} and {REPORT_MD_NAME} under evals/reports/",
)
parser.add_argument("--warmup", type=int, default=DEFAULT_WARMUP)
parser.add_argument("--measured", type=int, default=DEFAULT_MEASURED)
args = parser.parse_args(argv)
report = run_benchmark(warmup=args.warmup, measured=args.measured)
if args.write_report:
json_path, md_path = write_reports(report)
print(f"report written: {json_path}", file=sys.stderr)
print(f"summary written: {md_path}", file=sys.stderr)
if args.json:
# Deterministic payload without _metadata for stdout consumers.
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
elif not args.write_report:
print(f"{BENCHMARK_NAME} — use --json or --write-report", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(_cli_main())

43
core/cli.py
View file

@ -4134,6 +4134,42 @@ def cmd_bench(args: argparse.Namespace) -> int:
write_report(report)
return 0
if args.suite == "apple-uma":
from benchmarks.apple_uma_mechanical_sympathy import (
run_benchmark as run_apple_uma_benchmark,
write_reports as write_apple_uma_reports,
)
with _bench_stdout_guard(args.json):
uma_report = run_apple_uma_benchmark()
if args.report:
from benchmarks.apple_uma_mechanical_sympathy import write_json_report
report_path = Path(args.report)
write_json_report(uma_report, dest=report_path)
print(f"report written: {report_path}", file=sys.stderr)
elif getattr(args, "write_report", False):
json_path, md_path = write_apple_uma_reports(uma_report)
print(f"report written: {json_path}", file=sys.stderr)
print(f"summary written: {md_path}", file=sys.stderr)
if args.json:
print(json.dumps(uma_report, ensure_ascii=False, indent=2, sort_keys=True))
else:
machine = uma_report["machine"]
print(f"{uma_report['benchmark_name']}")
print(f" platform: {machine['platform']}")
print(f" using_rust: {machine['using_rust']}")
for name, track in uma_report["tracks"].items():
if track.get("skipped"):
print(f" [{name}] SKIPPED — {track.get('reason', 'n/a')}")
elif "timing" in track:
print(f" [{name}] p50={track['timing']['p50_ms']:.3f} ms")
elif name == "cl41_scalar_ops":
for op in track["operations"]:
print(
f" [{op['operation']}] p50={op['timing']['p50_ms']:.3f} ms"
)
return 0
if args.suite == "articulation":
from benchmarks.articulation import (
format_summary,
@ -5031,8 +5067,13 @@ def build_parser() -> argparse.ArgumentParser:
help="run benchmark harness (determinism, latency, speedup, versor audit)",
description="run benchmark harness",
)
bench.add_argument("--suite", choices=["determinism", "latency", "speedup", "versor", "convergence", "realizer", "cost", "teaching-loop", "articulation", "all"],
bench.add_argument("--suite", choices=["determinism", "latency", "speedup", "versor", "convergence", "realizer", "cost", "teaching-loop", "articulation", "apple-uma", "all"],
help="run a specific benchmark suite")
bench.add_argument(
"--write-report",
action="store_true",
help="apple-uma suite: write evals/reports/apple_uma_mechanical_sympathy_latest.{json,md}",
)
bench.add_argument("--runs", type=int, default=20, metavar="N", help="run count for determinism benchmark (also turns count for cost suite)")
bench.add_argument("--json", action="store_true", help="emit machine-readable JSON")
bench.add_argument("--report", metavar="PATH", help="write JSON report to file")

66
docs/outreach/apple-silicon-support-brief.md Normal file
View file

@ -0,0 +1,66 @@
# Apple Silicon Engineering Support Brief (Draft)
This brief is factual and engineering-first. It does not claim Apple endorsement,
review, or sponsorship. All performance and memory-boundary claims are backed by
the reproducible benchmark report at
`evals/reports/apple_uma_mechanical_sympathy_latest.json`.
## What CORE is
CORE is a deterministic Cl(4,1) reasoning and safety engine. Its runtime path
preserves geometric field invariants, exact CGA recall, closed-world proof
surfaces, and replay-stable evidence — not stochastic token generation.
## What the benchmark measures
The **CORE Apple Silicon UMA Mechanical Sympathy Benchmark** measures:
- Exact CGA top-k recall throughput on contiguous `(N, 32)` float32 matrices
- Cl(4,1) scalar algebra hot paths (`geometric_product`, `versor_apply`,
`cga_inner`, `versor_condition`)
- Off-serving closed-world FrameVerdict time-to-first-verifiable-verdict (TTFV)
- Deterministic `array_codec` persistence replay cost
- Honest Python/Rust copy and zero-copy input boundaries
It does **not** benchmark token generation, approximate recall, or transformer
throughput.
## Why Apple Silicon UMA is relevant
CORE workloads are dominated by contiguous-memory geometric operations and exact
recall scans. On Apple Silicon unified memory architecture, native bindings that
avoid Python marshalling tax on hot paths (for example Rust `vault_recall` and
`diffusion_step` input views) align with mechanical sympathy for UMA — when
measured, not assumed.
## Current hardware limits
On the machine that generated the latest report, larger validation lanes (for
example `N=65536` exact recall, large diffusion graphs, expanded replay buffers)
may be skipped or constrained by available memory and single-node throughput.
The benchmark records these limits explicitly.
## Requested engineering feedback
We are seeking Apple Silicon engineering feedback on:
1. Whether measured UMA-aligned workloads match expected memory behavior on M-series
2. Practical guidance for MLX/Metal kernel experiments under separate ADR/parity gates
3. Whether expanded hardware access would unlock larger reproducible validation runs
## Future-facing context (not benchmark claims)
Deterministic verification and replay throughput may be relevant to on-device
safety and audit surfaces in future R&D — but that relevance is **not** claimed
as a current product integration. MLX, Metal, CoreML, and Neural Engine paths
remain future work until implemented, parity-tested, and measured.
## How to reproduce
```bash
python -m benchmarks.apple_uma_mechanical_sympathy --write-report
# or
core bench --suite apple-uma --write-report
```
Reports land under `evals/reports/`. No network access is required.

440
evals/reports/apple_uma_mechanical_sympathy_latest.json Normal file
View file

@ -0,0 +1,440 @@
{
"_metadata": {
"note": "Non-hash metadata section; excluded from deterministic claim payloads.",
"report_path": "evals/reports/apple_uma_mechanical_sympathy_latest.json",
"written_at_unix": 1782329239.704502
},
"benchmark_name": "CORE Apple Silicon UMA Mechanical Sympathy Benchmark",
"benchmark_version": "1.0.0",
"claim_safety_audit": {
"future_work": [
"MLX kernel experiment requires separate ADR/parity lane.",
"Metal kernel experiment requires separate ADR/parity lane.",
"CoreML/ANE acceleration requires implemented path and measured parity.",
"Scalar Rust boundary zero-copy upgrades require focused parity tests.",
"Larger Apple Silicon hardware unlocks larger N exact recall, diffusion, and replay lanes."
],
"known_copy_paths": [
"Scalar Cl(4,1) Rust helpers copy via extract_f32_slice list conversion and allocate new NumPy outputs (geometric_product, versor_condition, cga_inner).",
"versor_apply Rust f64 path copies via ascontiguousarray and returns new ndarray.",
"Python fallback paths mediate through NumPy/Python objects.",
"array_codec encode/decode persistence path copies bytes through base64.",
"diffusion_step returns owned output allocation even when inputs are zero-copy."
],
"known_zero_copy_input_paths": [],
"safe_claims": [
"array_codec is bit-exact deterministic persistence/replay support.",
"FrameVerdict benchmark measures off-serving closed-world proof/verdict latency.",
"Exact CGA recall via algebra.backend.vault_recall — no ANN or approximate search.",
"Python vault_recall path uses vectorised exact scan when Rust is unavailable."
],
"unsafe_claims_not_made": [
"No CoreML acceleration claim.",
"No Neural Engine acceleration claim.",
"No MLX semantic-backend claim.",
"No \"zero-copy everywhere\" claim.",
"No fixed sponsorship speedup multiplier.",
"No token-generation benchmark.",
"No ANN/approximate-search benchmark."
]
},
"copy_zero_copy_truth_table": [
{
"input": "copy via extract_f32_slice",
"output": "new NumPy allocation",
"path": "algebra.backend.geometric_product (Rust)",
"zero_copy_input": "no"
},
{
"input": "copy via extract_f32_slice",
"output": "scalar",
"path": "algebra.backend.versor_condition (Rust)",
"zero_copy_input": "no"
},
{
"input": "copy via extract_f32_slice",
"output": "scalar",
"path": "algebra.backend.cga_inner (Rust)",
"zero_copy_input": "no"
},
{
"input": "ascontiguousarray copy",
"output": "new NumPy allocation",
"path": "algebra.backend.versor_apply (Rust f64 closure)",
"zero_copy_input": "no"
},
{
"input": "n/a",
"output": "skipped — Rust unavailable",
"path": "algebra.backend.diffusion_step",
"zero_copy_input": "n/a"
},
{
"input": "NumPy view / vectorised scan",
"output": "index list",
"path": "algebra.backend.vault_recall (Python)",
"zero_copy_input": "n/a (Python canonical)"
},
{
"input": "byte copy + base64",
"output": "writable ndarray copy",
"path": "core.array_codec encode/decode",
"zero_copy_input": "no"
},
{
"input": "closed frame struct",
"output": "FrameVerdict",
"path": "generate.frame_verdict.evaluate_frame_verdict",
"zero_copy_input": "n/a (proof surface)"
}
],
"machine": {
"CORE_BACKEND": "",
"core_rs": {
"import_succeeded": false,
"reason": "No module named 'core_rs'"
},
"machine": "arm64",
"memory": {
"available_bytes": 4375969792,
"source": "psutil.virtual_memory",
"total_bytes": 17179869184
},
"os": "Darwin",
"platform": "macOS-26.5.1-arm64-arm-64bit",
"processor": "arm",
"python_version": "3.12.13",
"using_rust": false
},
"tracks": {
"array_codec_replay": {
"decode_timing": {
"max_ms": 0.035041,
"mean_ms": 0.023529,
"measured_iterations": 50,
"min_ms": 0.022875,
"ops_per_sec": 42500.381,
"p50_ms": 0.023,
"p95_ms": 0.025875,
"warmup_iterations": 5
},
"dtype": "float32",
"encode_timing": {
"max_ms": 0.015417,
"mean_ms": 0.015281,
"measured_iterations": 50,
"min_ms": 0.015167,
"ops_per_sec": 65441.682,
"p50_ms": 0.015291,
"p95_ms": 0.015417,
"warmup_iterations": 5
},
"encoded_bytes": 10924,
"note": "Deterministic persistence/replay support — not runtime zero-copy.",
"payload_shape": [
64,
32
],
"sanity": {
"byte_exact_roundtrip": true,
"versor_closure_preserved": true,
"writable_decode": true
},
"skipped": false,
"track": "array_codec_replay"
},
"cl41_scalar_ops": {
"operations": [
{
"backend_requested": "python (default)",
"backend_used": "python",
"dtype": "float32",
"memory_behavior": "Python path is the canonical semantic fallback.",
"operation": "geometric_product",
"sanity": {
"deterministic_repeat": true,
"finite": true,
"output_shape": [
32
]
},
"shape": [
32
],
"timing": {
"max_ms": 1.417042,
"mean_ms": 1.367876,
"measured_iterations": 50,
"min_ms": 1.353834,
"ops_per_sec": 731.06,
"p50_ms": 1.359729,
"p95_ms": 1.407416,
"warmup_iterations": 5
}
},
{
"backend_requested": "python (default)",
"backend_used": "python",
"dtype": "float32",
"memory_behavior": "Python path is the canonical semantic fallback.",
"operation": "versor_apply",
"sanity": {
"deterministic_repeat": true,
"finite": true,
"output_shape": [
32
]
},
"shape": [
32
],
"timing": {
"max_ms": 3.086375,
"mean_ms": 2.939975,
"measured_iterations": 50,
"min_ms": 2.886917,
"ops_per_sec": 340.139,
"p50_ms": 2.926645,
"p95_ms": 3.044667,
"warmup_iterations": 5
}
},
{
"backend_requested": "python (default)",
"backend_used": "python",
"dtype": "float32",
"memory_behavior": "Python path is the canonical semantic fallback.",
"operation": "cga_inner",
"sanity": {
"deterministic_repeat": true,
"finite": true,
"output_kind": "scalar"
},
"shape": [
32
],
"timing": {
"max_ms": 4.999417,
"mean_ms": 2.823273,
"measured_iterations": 50,
"min_ms": 2.685958,
"ops_per_sec": 354.199,
"p50_ms": 2.7085,
"p95_ms": 3.641125,
"warmup_iterations": 5
}
},
{
"backend_requested": "python (default)",
"backend_used": "python",
"dtype": "float32",
"memory_behavior": "Python path is the canonical semantic fallback.",
"operation": "versor_condition",
"sanity": {
"deterministic_repeat": true,
"finite": true,
"output_kind": "scalar"
},
"shape": [
32
],
"timing": {
"max_ms": 0.605458,
"mean_ms": 0.543464,
"measured_iterations": 50,
"min_ms": 0.530541,
"ops_per_sec": 1840.048,
"p50_ms": 0.536229,
"p95_ms": 0.598667,
"warmup_iterations": 5
}
}
],
"skipped": false,
"track": "cl41_scalar_ops"
},
"diffusion_step": {
"reason": "Rust backend not enabled (set CORE_BACKEND=rust and install core_rs)",
"rust_available": false,
"skipped": true,
"track": "diffusion_step"
},
"exact_cga_recall": {
"cases": [
{
"N": 128,
"backend_requested": "python (default)",
"backend_used": "python",
"contiguous": true,
"dtype": "float32",
"result_deterministic": true,
"rows_per_sec": 1840049.683,
"rust_zero_copy_input_eligible": false,
"timing": {
"max_ms": 0.070416,
"mean_ms": 0.069563,
"measured_iterations": 50,
"min_ms": 0.069125,
"ops_per_sec": 14375.388,
"p50_ms": 0.0695,
"p95_ms": 0.070209,
"warmup_iterations": 5
},
"top_k": 5,
"top_result_preview": [
[
0,
-0.7070845365524292
],
[
1,
-0.7077386975288391
],
[
2,
-0.7083905339241028
]
]
},
{
"N": 1024,
"backend_requested": "python (default)",
"backend_used": "python",
"contiguous": true,
"dtype": "float32",
"result_deterministic": true,
"rows_per_sec": 8937898.275,
"rust_zero_copy_input_eligible": false,
"timing": {
"max_ms": 0.151084,
"mean_ms": 0.114568,
"measured_iterations": 50,
"min_ms": 0.112583,
"ops_per_sec": 8728.416,
"p50_ms": 0.113542,
"p95_ms": 0.117292,
"warmup_iterations": 5
},
"top_k": 5,
"top_result_preview": [
[
0,
-0.1441916823387146
],
[
1,
-0.14573132991790771
],
[
2,
-0.14726853370666504
]
]
},
{
"N": 8192,
"backend_requested": "python (default)",
"backend_used": "python",
"contiguous": true,
"dtype": "float32",
"result_deterministic": true,
"rows_per_sec": 17482143.653,
"rust_zero_copy_input_eligible": false,
"timing": {
"max_ms": 0.576834,
"mean_ms": 0.468592,
"measured_iterations": 50,
"min_ms": 0.456166,
"ops_per_sec": 2134.051,
"p50_ms": 0.461208,
"p95_ms": 0.531042,
"warmup_iterations": 5
},
"top_k": 5,
"top_result_preview": [
[
2561,
2.8078949451446533
],
[
2562,
2.807893991470337
],
[
2560,
2.8078932762145996
]
]
},
{
"N": 65536,
"backend_requested": "python (default)",
"backend_used": "python",
"contiguous": true,
"dtype": "float32",
"result_deterministic": true,
"rows_per_sec": 22937268.738,
"rust_zero_copy_input_eligible": false,
"timing": {
"max_ms": 3.492375,
"mean_ms": 2.857184,
"measured_iterations": 50,
"min_ms": 2.711583,
"ops_per_sec": 349.995,
"p50_ms": 2.783229,
"p95_ms": 3.422333,
"warmup_iterations": 5
},
"top_k": 5,
"top_result_preview": [
[
3961,
2.8078951835632324
],
[
15385,
2.8078951835632324
],
[
26809,
2.8078951835632324
]
]
}
],
"large_n_probe": {
"attempted": true,
"included": true,
"reason": "probe under 3.0s budget"
},
"skipped": false,
"track": "exact_cga_recall"
},
"frame_verdict_ttfv": {
"frame_kind": "text",
"note": "Off-serving closed-world proof/verdict latency — not served answer latency.",
"proof_hash_present": true,
"proof_producer": "proof_chain.entail",
"sanity": {
"expected_entailed_true": true,
"is_frame_verdict": true,
"verdict_in_closed_set": true
},
"skipped": false,
"timing": {
"max_ms": 0.245625,
"mean_ms": 0.166753,
"measured_iterations": 50,
"min_ms": 0.149,
"ops_per_sec": 5996.886,
"p50_ms": 0.15325,
"p95_ms": 0.231166,
"warmup_iterations": 5
},
"trace_hash_present": true,
"track": "frame_verdict_ttfv",
"verdict": "entailed_true",
"world_assumption": "closed"
}
}
}

76
evals/reports/apple_uma_mechanical_sympathy_latest.md Normal file
View file

@ -0,0 +1,76 @@
# CORE Apple Silicon UMA Mechanical Sympathy Benchmark
Version: 1.0.0
## 1. What this measures
Deterministic Cl(4,1) geometric workloads on Apple Silicon / UMA hardware:
exact CGA recall, scalar algebra hot paths, closed-world FrameVerdict proof
latency, deterministic array persistence replay, and honest Python/Rust
memory boundaries. No token generation. No approximate recall.
## 2. Machine/backend summary
- Platform: macOS-26.5.1-arm64-arm-64bit
- Processor: arm
- Python: 3.12.13
- CORE_BACKEND: `(default python)`
- core_rs import: False
- using_rust(): False
## 3. Exact CGA recall
- N=128: p50=0.070 ms, rows/sec=1840049.683, zero-copy eligible=False
- N=1024: p50=0.114 ms, rows/sec=8937898.275, zero-copy eligible=False
- N=8192: p50=0.461 ms, rows/sec=17482143.653, zero-copy eligible=False
- N=65536: p50=2.783 ms, rows/sec=22937268.738, zero-copy eligible=False
## 4. Cl(4,1) scalar algebra
- geometric_product: p50=1.360 ms, ops/sec=731.06
- versor_apply: p50=2.927 ms, ops/sec=340.139
- cga_inner: p50=2.708 ms, ops/sec=354.199
- versor_condition: p50=0.536 ms, ops/sec=1840.048
## 5. FrameVerdict TTFV
- Verdict: entailed_true, p50=0.153 ms, producer=proof_chain.entail
## 6. Deterministic replay/persistence
- encode p50=0.015 ms, decode p50=0.023 ms, bytes=10924
## 7. Copy / zero-copy truth table
| Path | Input | Output | Zero-copy input |
|---|---|---|---|
| algebra.backend.geometric_product (Rust) | copy via extract_f32_slice | new NumPy allocation | no |
| algebra.backend.versor_condition (Rust) | copy via extract_f32_slice | scalar | no |
| algebra.backend.cga_inner (Rust) | copy via extract_f32_slice | scalar | no |
| algebra.backend.versor_apply (Rust f64 closure) | ascontiguousarray copy | new NumPy allocation | no |
| algebra.backend.diffusion_step | n/a | skipped — Rust unavailable | n/a |
| algebra.backend.vault_recall (Python) | NumPy view / vectorised scan | index list | n/a (Python canonical) |
| core.array_codec encode/decode | byte copy + base64 | writable ndarray copy | no |
| generate.frame_verdict.evaluate_frame_verdict | closed frame struct | FrameVerdict | n/a (proof surface) |
## 8. Why this matters for Apple Silicon
CORE's deterministic workloads are contiguous-memory geometric operations
and exact recall scans — structurally aligned with unified memory when
native bindings avoid Python marshalling tax on hot paths.
## 9. What larger Apple Silicon hardware would unlock
Larger unified memory enables higher-N exact recall validation, larger
diffusion graphs, and expanded replay persistence lanes without swapping
or fragmenting evidence buffers.
## 10. Explicit non-claims
- No CoreML acceleration claim.
- No Neural Engine acceleration claim.
- No MLX semantic-backend claim.
- No "zero-copy everywhere" claim.
- No fixed sponsorship speedup multiplier.
- No token-generation benchmark.
- No ANN/approximate-search benchmark.

170
tests/test_apple_uma_mechanical_sympathy_benchmark.py Normal file
View file

@ -0,0 +1,170 @@
"""Tests for the Apple Silicon UMA mechanical sympathy benchmark."""
from __future__ import annotations
import json
import socket
from pathlib import Path
from unittest import mock
import numpy as np
import pytest
from benchmarks.apple_uma_mechanical_sympathy import (
BENCHMARK_NAME,
REPORT_JSON_NAME,
build_claim_safety_audit,
deterministic_closed_frame,
run_benchmark,
synthetic_matrix,
track_array_codec_replay,
track_frame_verdict_ttfv,
write_json_report,
write_reports,
)
from core.array_codec import decode_array, encode_array
from generate.frame_verdict.evaluate import evaluate_frame_verdict
from generate.frame_verdict.types import FrameVerdictKind
REQUIRED_TOP_LEVEL_KEYS = frozenset(
{
"benchmark_name",
"benchmark_version",
"machine",
"tracks",
"claim_safety_audit",
"copy_zero_copy_truth_table",
}
)
REQUIRED_TRACK_KEYS = frozenset(
{
"cl41_scalar_ops",
"exact_cga_recall",
"diffusion_step",
"frame_verdict_ttfv",
"array_codec_replay",
}
)
@pytest.fixture
def fast_bench_kwargs() -> dict[str, int]:
return {"warmup": 1, "measured": 3}
def test_report_has_stable_top_level_keys(fast_bench_kwargs: dict[str, int]) -> None:
report = run_benchmark(**fast_bench_kwargs)
assert REQUIRED_TOP_LEVEL_KEYS <= set(report.keys())
assert REQUIRED_TRACK_KEYS <= set(report["tracks"].keys())
assert report["benchmark_name"] == BENCHMARK_NAME
def test_no_rust_required_for_basic_report(fast_bench_kwargs: dict[str, int]) -> None:
with mock.patch.dict("os.environ", {}, clear=True):
report = run_benchmark(**fast_bench_kwargs)
assert "machine" in report
assert report["tracks"]["cl41_scalar_ops"]["skipped"] is False
assert report["tracks"]["exact_cga_recall"]["skipped"] is False
def test_skipped_tracks_include_explicit_reasons(fast_bench_kwargs: dict[str, int]) -> None:
with mock.patch.dict("os.environ", {}, clear=True):
report = run_benchmark(**fast_bench_kwargs)
diffusion = report["tracks"]["diffusion_step"]
if diffusion.get("skipped"):
assert "reason" in diffusion
assert diffusion["reason"]
def test_deterministic_synthetic_sanity_checks_stable(
fast_bench_kwargs: dict[str, int],
) -> None:
report_a = run_benchmark(**fast_bench_kwargs)
report_b = run_benchmark(**fast_bench_kwargs)
for track_name in ("cl41_scalar_ops", "array_codec_replay", "frame_verdict_ttfv"):
if track_name == "cl41_scalar_ops":
for op_a, op_b in zip(
report_a["tracks"][track_name]["operations"],
report_b["tracks"][track_name]["operations"],
):
assert op_a["sanity"]["deterministic_repeat"] is True
assert op_b["sanity"]["deterministic_repeat"] is True
else:
assert report_a["tracks"][track_name]["sanity"] == report_b["tracks"][track_name]["sanity"]
def test_claim_safety_audit_contents() -> None:
audit = build_claim_safety_audit(using_rust=False)
assert audit["safe_claims"]
assert audit["unsafe_claims_not_made"]
assert any("CoreML" in s for s in audit["unsafe_claims_not_made"])
assert any("MLX" in s for s in audit["unsafe_claims_not_made"])
assert any("zero-copy everywhere" in s for s in audit["unsafe_claims_not_made"])
assert audit["known_copy_paths"]
assert audit["future_work"]
def test_array_codec_replay_byte_exact_roundtrip() -> None:
track = track_array_codec_replay(warmup=1, measured=2)
assert track["skipped"] is False
assert track["sanity"]["byte_exact_roundtrip"] is True
assert track["sanity"]["writable_decode"] is True
arr = synthetic_matrix(8, seed=1)
assert np.array_equal(arr, decode_array(encode_array(arr)))
def test_frame_verdict_ttfv_returns_frame_verdict_not_determine() -> None:
frame, query = deterministic_closed_frame()
verdict = evaluate_frame_verdict(frame, query)
assert verdict.verdict is FrameVerdictKind.ENTAILED_TRUE
track = track_frame_verdict_ttfv(warmup=1, measured=2)
assert track["skipped"] is False
assert track["sanity"]["is_frame_verdict"] is True
assert track["sanity"]["expected_entailed_true"] is True
# Guard: open-world determine() must not be imported by this track module.
import benchmarks.apple_uma_mechanical_sympathy as bench_mod
source = Path(bench_mod.__file__).read_text(encoding="utf-8")
assert "determine(" not in source
assert "from generate.determine" not in source
def test_report_writer_creates_json_under_evals_reports(
fast_bench_kwargs: dict[str, int],
tmp_path: Path,
) -> None:
report = run_benchmark(**fast_bench_kwargs)
path = write_json_report(report, root=tmp_path)
assert path.name == REPORT_JSON_NAME
assert path.parent == tmp_path
loaded = json.loads(path.read_text(encoding="utf-8"))
assert "_metadata" in loaded
if "report_path" in loaded["_metadata"]:
assert not loaded["_metadata"]["report_path"].startswith("/")
assert loaded["benchmark_name"] == BENCHMARK_NAME
def test_write_reports_creates_json_and_markdown(
fast_bench_kwargs: dict[str, int],
tmp_path: Path,
) -> None:
report = run_benchmark(**fast_bench_kwargs)
json_path, md_path = write_reports(report, root=tmp_path)
assert json_path.exists()
assert md_path.exists()
md_text = md_path.read_text(encoding="utf-8")
assert "Explicit non-claims" in md_text
assert "zero-copy" in md_text.lower()
def test_benchmark_module_makes_no_network_calls(
fast_bench_kwargs: dict[str, int],
monkeypatch: pytest.MonkeyPatch,
) -> None:
def _blocked(*_a: object, **_k: object) -> None:
raise AssertionError("network access attempted during benchmark")
monkeypatch.setattr(socket.socket, "connect", _blocked)
run_benchmark(**fast_bench_kwargs)