| """Trusted benchmark reproduction (decision 45; Phase 3 exit gate). |
| |
| Each benchmark builds its reference system, runs the registered solver through the normal |
| SimulationService (so it produces immutable run records), and compares with the registered |
| reference value and tolerance from configs/solver_registry.yaml. Results are reported as-is: |
| a failed benchmark is reported as failed. |
| """ |
| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| from ase.cluster import Icosahedron |
|
|
| from core.representation import MatterStructure |
| from discovery.generators import known_family |
|
|
|
|
| def _system(bid: str) -> tuple[MatterStructure, str, dict]: |
| if bid == "emt_cu_lattice_constant": |
| ms = known_family("fcc", element="Cu", a=3.6, cubic=True); ms.environment["lattice_type"] = "fcc" |
| return ms, "lattice_constant", {"n_points": 11, "span": 0.05} |
| if bid == "lj13_global_minimum": |
| ms = MatterStructure.from_ase(Icosahedron("Ar", noshells=2, latticeconstant=1.55), boundary_profile="cluster") |
| return ms, "relaxed_energy", {"fmax": 1e-5, "steps": 5000} |
| if bid == "lj38_global_minimum": |
| from ase.cluster import Octahedron |
| atoms = Octahedron("Ar", length=4, cutoff=1, latticeconstant=1.55) |
| ms = MatterStructure.from_ase(atoms, boundary_profile="cluster") |
| return ms, "relaxed_energy", {"fmax": 1e-5, "steps": 5000} |
| if bid == "h2_rhf_sto3g": |
| from ase.units import Bohr |
| ms = MatterStructure(["H", "H"], [[0, 0, 0], [0, 0, 1.4 * Bohr]]) |
| return ms, "energy", {"basis": "sto-3g"} |
| raise KeyError(bid) |
|
|
|
|
| def run_benchmark(app, bid: str, project_id: str) -> dict[str, Any]: |
| bm = app.registry.benchmarks[bid] |
| ms, observable, params = _system(bid) |
| cand = app.candidates.register(ms, project_id, generation_method=f"benchmark:{bid}", screen=True) |
| rec = app.simulation.submit_run(cand.candidate_id, bm["solver"], observable, params, priority="validation", |
| router_rationale=f"trusted benchmark {bid}") |
| ref = bm["reference"] |
| key = {"lattice_constant": "lattice_constant", "relaxed_energy": "relaxed_energy", "energy": "energy"}[observable] |
| if bid == "h2_rhf_sto3g": |
| key = "energy_hartree" |
| val = (rec.observables.get(key) or {}).get("value") |
| passed = None; dev = None |
| if rec.exit_status == "SUCCESS" and val is not None: |
| dev = val - ref["value"] |
| if "tolerance_relative" in bm: |
| passed = abs(dev / ref["value"]) <= bm["tolerance_relative"] |
| else: |
| passed = abs(dev) <= bm["tolerance_absolute"] |
| else: |
| passed = False |
| result = {"benchmark": bid, "solver": bm["solver"], "run_id": rec.run_id, "candidate_id": cand.candidate_id, |
| "computed": val, "unit": (rec.observables.get(key) or {}).get("unit"), "reference": ref, |
| "deviation": dev, "relative_deviation": (dev / ref["value"]) if dev is not None else None, |
| "tolerance": bm.get("tolerance_relative", bm.get("tolerance_absolute")), |
| "tolerance_kind": "relative" if "tolerance_relative" in bm else "absolute", |
| "passed": bool(passed), "exit_status": rec.exit_status, "convergence": rec.convergence_status} |
| app.db.audit("validation_agent", "benchmark_run", bid, result) |
| return result |
|
|
|
|
| def run_all(app, project_id: str, ids: list[str] | None = None) -> list[dict]: |
| ids = ids or list(app.registry.benchmarks) |
| out = [] |
| for b in ids: |
| try: |
| out.append(run_benchmark(app, b, project_id)) |
| except Exception as e: |
| out.append({"benchmark": b, "passed": False, "error": repr(e)}) |
| return out |
|
|
