"""Compare saved MLX scores with CUDA using identical, frozen temperatures. No fitting, parameter selection, or changes to inference weights take place. Partial panels are diagnostic only and can never qualify a release. """ import argparse import json from collections import defaultdict from pathlib import Path import numpy as np from solomon_mlx._vendor.semantics import listed_probs, p_yes from solomon_mlx.api import TASKS from solomon_mlx.artifacts import digest, runtime_identity, sha256 from solomon_mlx.evaluation import compare_rows, load_panel, read_cuda_scores def decision_probabilities(row, temperature): """Parity includes every branch, even when its gold label is not a listed option.""" if row["task"] in ("boolean", "entity", "multilabel"): p = p_yes(row["letter_logits"], temperature) return np.array([1 - p, p]) width = row["n"] - 2 if row["head_key"].endswith("choiceR") else row["n"] return listed_probs(row["letter_logits"], width, temperature) def compare(panel, scores, cuda_directory, reference, output, *, allow_partial=False): panel, scores, output = Path(panel), Path(scores), Path(output) if output.exists(): raise FileExistsError("Parity reports are immutable") jobs, manifest = load_panel(panel) identity = json.loads((scores / "identity.json").read_text()) model_binding = json.loads(Path("models/quality/binding.json").read_text()) if identity["runtime"] != runtime_identity(model_binding): raise ValueError("Scores belong to another MLX runtime") if identity["panel_sha256"] != manifest["jobs_sha256"]: raise ValueError("Scores belong to another panel") groups = defaultdict(list) for job in jobs: groups[job["document_key"]].append(job) rows, files = [], {} for key, group in groups.items(): path = scores / (key + ".json") if not path.exists() and allow_partial: continue record = json.loads(path.read_text()) body = {k: v for k, v in record.items() if k != "sha256"} if ( record["sha256"] != digest(body) or record["identity"] != digest(identity) or [r["id"] for r in record["rows"]] != [r["id"] for r in group] ): raise ValueError("Corrupt or mismatched score document") rows.extend(record["rows"]) files[path.name] = sha256(path) complete = len(files) == len(groups) if not allow_partial: marker = json.loads((scores / "complete.json").read_text()) if marker != { "identity": digest(identity), "documents": len(groups), "branches": len(jobs), "files": files, }: raise ValueError("Incomplete or mismatched completion manifest") ref = json.loads(Path(reference).read_text()) cuda = read_cuda_scores(cuda_directory, panel, ref["identity"]) selected = {r["id"] for r in rows} cuda = [r for r in cuda if r["id"] in selected] binding_path = Path("evaluations/cuda-acceptance/input/serving-binding.json") source_manifest = json.loads((binding_path.parent / "manifest.json").read_text()) if sha256(binding_path) != source_manifest["files"][binding_path.name]: raise ValueError("CUDA acceptance binding checksum mismatch") binding = json.loads(binding_path.read_text()) for key in ( "adapter_sha256", "trained_heads_sha256", "model_sha256", "numerics", "placement", "arithmetic", ): if binding["runtime"][key] != ref["identity"][key]: raise ValueError("CUDA temperatures belong to another reference") temperatures = {task: binding["temperatures"]["models"][task]["temperature"] for task in TASKS} comparisons = {} cuda_by_id = {r["id"]: r for r in cuda} for name, temps in (("temperature_one", dict.fromkeys(TASKS, 1.0)), ("cuda_serving", temperatures)): result = compare_rows(rows, cuda, temperatures=temps, reference_temperatures=temps) result.pop("quality_gate_passed") result["accuracy_units"] = result["units"] result["accuracy_questions"] = result["questions"] worst, questions = [], defaultdict(list) for row in rows: other = {**row, "letter_logits": cuda_by_id[row["id"]]["letter_logits"]} p = decision_probabilities(row, temps[row["task"]]) q = decision_probabilities(other, temps[row["task"]]) if not np.isfinite(p).all() or not np.isfinite(q).all(): raise ValueError("Nonfinite parity probability") agrees = int(np.argmax(p)) == int(np.argmax(q)) questions[row["question_id"]].append(agrees) worst.append( { "id": row["id"], "task": row["task"], "max_probability_drift": float(np.max(np.abs(p - q))), "decision_agrees": agrees, } ) result.update( units=len(rows), questions=len(questions), unit_decision_agreement=float(np.mean([r["decision_agrees"] for r in worst])), question_decision_agreement=float(np.mean([all(v) for v in questions.values()])), max_probability_drift=max(r["max_probability_drift"] for r in worst), mean_probability_drift=float(np.mean([r["max_probability_drift"] for r in worst])), ) result["agreement_gate_passed"] = ( result["unit_decision_agreement"] >= 0.999 and result["question_decision_agreement"] >= 0.999 ) result["largest_probability_differences"] = sorted( worst, key=lambda r: r["max_probability_drift"], reverse=True )[:10] comparisons[name] = result report = { "scope": "complete text parity panel" if complete else "partial text parity diagnostic", "complete": complete, "documents": len(files), "total_documents": len(groups), "branches": len(rows), "total_branches": len(jobs), "tasks": sorted({r["task"] for r in rows}), "runtime": identity["runtime"], "cuda_runtime": ref["identity"], "panel_sha256": identity["panel_sha256"], "score_files_sha256": digest(files), "cuda_serving_binding_sha256": sha256(binding_path), "temperature_fitting_performed": False, "temperature_policy": "identical settings on both backends; not MLX calibration", "comparisons": comparisons, "parity_gate_passed": complete and all(r["agreement_gate_passed"] for r in comparisons.values()), "bitwise_equality_claimed": False, "image_qualification": False, } output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2)) print( json.dumps( {k: report[k] for k in ("scope", "documents", "branches", "tasks", "parity_gate_passed")}, indent=2, ) ) return report if __name__ == "__main__": parser = argparse.ArgumentParser() for field in ("panel", "scores", "cuda-directory", "output"): parser.add_argument("--" + field, required=True) parser.add_argument("--reference", default="evaluations/bf16-reference-1789901869/report.json") parser.add_argument("--allow-partial", action="store_true") args = parser.parse_args() compare( args.panel, args.scores, args.cuda_directory, args.reference, args.output, allow_partial=args.allow_partial, )