"""Explicit, hash-pinned direct-API chain evidence, never invented campaign runs.
The input ZIP retains original producer reports, terminal receipts and sidecars.
Its descriptor only maps those files to document species; it cannot override
producer verdicts. A reference QVF and a correlated event have separate roles.
"""
from __future__ import annotations
import hashlib
import io
import json
import math
import zipfile
from collections import Counter
from pathlib import PurePosixPath
from typing import Any
from vibesketch.document.model import Document
from vibesketch.document.results import ResultRefused, _finite, species_digest
from vibesketch.stages.validator import reference_reasons
from vibesketch.units import HARTREE_TO_KCAL_MOL
MAX_EVIDENCE_BYTES = 64 * 1024 * 1024
STAGES = ("mace", "pbe-opt", "pbe-hessian", "dlpno")
def _sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _require(condition: bool, message: str) -> None:
if not condition:
raise ResultRefused(message)
def _json(data: bytes) -> Any:
def reject(value: str) -> Any:
raise ValueError(f"nonfinite JSON: {value}")
return json.loads(data, parse_constant=reject)
def _zip(data: bytes) -> dict[str, bytes]:
_require(0 < len(data) <= MAX_EVIDENCE_BYTES, "external evidence exceeds byte limit")
with zipfile.ZipFile(io.BytesIO(data)) as archive:
names = archive.namelist()
_require(
len(names) <= 2048 and len(names) == len(set(names)), "duplicate/excess archive members"
)
_require(
sum(i.file_size for i in archive.infolist()) <= MAX_EVIDENCE_BYTES,
"expanded evidence exceeds byte limit",
)
for name in names:
path = PurePosixPath(name)
_require(
not path.is_absolute() and ".." not in path.parts and "\\" not in name,
"unsafe evidence member",
)
return {name: archive.read(name) for name in names}
def _xyz(data: bytes) -> tuple[list[str], list[list[float]]]:
lines = data.decode("utf-8").splitlines()
count = int(lines[0])
rows = [line.split() for line in lines[2:] if line.strip()]
_require(
0 < count <= 5000 and len(rows) == count and all(len(r) == 4 for r in rows),
"invalid external geometry",
)
positions = [[float(v) for v in r[1:]] for r in rows]
_require(all(math.isfinite(v) for row in positions for v in row), "nonfinite geometry")
return [r[0] for r in rows], positions
def _qvf(data: bytes) -> dict[str, Any]:
members = _zip(data)
manifest: dict[str, Any] = _json(members["manifest.json"])
for section in manifest.get("sections", []):
for item in section.get("members", {}).values():
if isinstance(item, dict) and "path" in item:
_require(
_sha(members[item["path"]]) == item.get("sha256"), "QVF member checksum differs"
)
return manifest
def _source(
document: Document, sid: str, selection: dict[str, Any], files: dict[str, bytes], contract: str
) -> dict[str, Any]:
species = document.species_by_id(sid)
_require(
selection["species_sha256"] == species_digest(species),
f"{sid}: external evidence is stale for this species/geometry",
)
_require(
bool(species.conformers) and species.multiplicity == 1,
"external restricted chain requires a singlet with explicit geometry",
)
journal = _json(files[selection["journal"]])
preparation_bytes = files[selection["preparation"]]
preparation = _json(preparation_bytes)
_require(preparation.get("accepted") is True, "preparation was not accepted")
parent_report = _sha(preparation_bytes)
parent_geometry = preparation["geometry_sha256"]
lineage = []
final: dict[str, Any] = {}
_require(set(selection["stages"]) == set(STAGES), "select the complete external chain")
for stage in STAGES:
item = selection["stages"][stage]
prefix = item["directory"].rstrip("/") + "/"
def get(name: str, directory: str = prefix) -> bytes:
return files[directory + name]
report_bytes = get("acceptance.json")
report = _json(report_bytes)
terminal = _json(get("_vq/terminal-diagnosis.json"))
jobs = [j for j in journal["jobs"] if j.get("id") == item["job_id"]]
_require(
len(jobs) == 1
and jobs[0].get("stage") == stage
and jobs[0].get("state") == "completed"
and jobs[0].get("accepted") is True,
"selected original job is not a completed accepted stage",
)
_require(
terminal.get("jobid") == item["job_id"]
and type(terminal.get("exit_code")) is int
and terminal["exit_code"] == 0
and get("_vq/exit-code").strip() == b"0",
"external terminal identity or exit status differs",
)
_require(
report.get("stage") == stage
and report.get("accepted") is True
and bool(report.get("checks"))
and all(v is True for v in report["checks"].values()),
"external scientific stage was not accepted",
)
_require(
report.get("charge") == species.charge and report.get("multiplicity") == 1,
"external charge/spin differs from species",
)
_require(
report.get("parent_report_sha256") == parent_report
and report.get("parent_geometry_sha256") == parent_geometry
and _sha(get("input.xyz")) == parent_geometry,
"external stage has a stale or substituted parent",
)
for name, digest in report["files_sha256"].items():
_require(_sha(get(name)) == digest, "producer output checksum differs")
symbols, xyz = _xyz(get("input.xyz"))
_require(symbols == species.conformers[0].symbols, "external atom order differs")
if stage == "mace":
_require(
all(
math.dist(a, b) <= 1e-8
for a, b in zip(xyz, species.conformers[0].xyz_angstrom, strict=True)
),
"external starting coordinates differ from document",
)
qvf_name = {
"mace": "mace.qvf",
"pbe-opt": "pbe-opt.qvf",
"pbe-hessian": "pbe-hessian.qvf",
"dlpno": "dlpno.qvf",
}[stage]
manifest = _qvf(get(qvf_name))
qvf_files = _zip(get(qvf_name))
specs = [s for s in manifest["sections"] if s.get("kind") == "job.spec"]
_require(len(specs) == 1, "external reference needs one producer job specification")
spec = _json(qvf_files[specs[0]["members"]["spec"]["path"]])
expected_method = {"mace": "mace", "pbe-opt": "rks", "pbe-hessian": "rks", "dlpno": "rhf"}[
stage
]
_require(spec.get("method") == expected_method, "external reference method differs")
if stage != "mace":
expected_basis = "cc-pvdz" if stage == "dlpno" else "def2-svp"
_require(
str(spec.get("basis", "")).lower() == expected_basis,
"external reference basis differs from the supported chain",
)
if stage in ("mace", "pbe-opt"):
force = _finite(report["max_force_ev_angstrom"], "maximum force")
threshold = _finite(report["force_threshold_ev_angstrom"], "force threshold")
_require(
0 <= force <= threshold <= (0.005 if stage == "mace" else 0.001),
"external optimization fails its force bound",
)
if stage.startswith("pbe"):
_require(
spec.get("functional", "").lower() == "pbe"
and report.get("method") == "PBE/def2-SVP"
and report.get("dispersion") is None
and report.get("density_fit") is False
and report.get("cosx") is False,
"external PBE surface differs",
)
if stage == "pbe-hessian":
frequencies = report["frequencies_cm1"]
half = report["half_step_frequencies_cm1"]
modes = sorted(_finite(v, "frequency") for v in frequencies if abs(v) > 1)
half_modes = sorted(_finite(v, "frequency") for v in half if abs(v) > 1)
_require(
len(frequencies) == len(half) == 3 * len(symbols)
and all(math.isfinite(v) for v in frequencies + half)
and len(modes) == len(half_modes) == 3 * len(symbols) - 6
and all(v > 0 for v in modes + half_modes)
and max(abs(a - b) for a, b in zip(modes, half_modes, strict=True)) <= 5,
"external Hessian minimum/stencil check failed",
)
reference = report.get("producer_verdict")
if stage != "mace":
_require(
manifest.get("provenance", {}).get("verdict") == reference,
"reference QVF and accepted verdict disagree",
)
check = {
"kind": "dlpno" if stage == "dlpno" else "dft",
"multiplicity": 1,
"checks": {"reference_contract": contract} if stage == "dlpno" else {},
}
_require(
not reference_reasons(check, reference),
"external reference violates the selected stability contract",
)
output_name = stage + ".xyz"
_require(_sha(get(output_name)) == report["geometry_sha256"], "final geometry differs")
if stage in ("pbe-hessian", "dlpno"):
final_symbols, final_xyz = _xyz(get(output_name))
_require(
final_symbols == symbols
and all(math.dist(a, b) <= 1e-9 for a, b in zip(xyz, final_xyz, strict=True)),
"single point/Hessian changed accepted geometry",
)
lineage.append(
{
"stage": stage,
"original_job": item["job_id"],
"terminal_sha256": _sha(get("_vq/terminal-diagnosis.json")),
"report_sha256": _sha(report_bytes),
"input_sha256": parent_geometry,
"parent_report_sha256": parent_report,
"geometry_sha256": report["geometry_sha256"],
"qvf_sha256": _sha(get(qvf_name)),
}
)
parent_report, parent_geometry = _sha(report_bytes), report["geometry_sha256"]
if stage == "dlpno":
events = [_json(line) for line in get("dlpno.scf.jsonl").splitlines() if line.strip()]
done = [e for e in events if e.get("event") == "dlpno_ccsd_done"]
ended = [e for e in events if e.get("event") == "job_end"]
_require(
len(done) == len(ended) == 1, "external correlated log is ambiguous/incomplete"
)
event, end = done[0], ended[0]
energy = _finite(event["e_total"], "correlated total")
_require(
event.get("converged") is True
and end.get("converged") is True
and event.get("variant") == "dlpno-ccsd(t)"
and end.get("verdict") == reference
and energy == end.get("energy") == report.get("energy_total_hartree"),
"external correlated completion does not match acceptance",
)
_require(
abs(
energy
- sum(
_finite(report[k], k)
for k in (
"rhf_energy_hartree",
"ccsd_correlation_hartree",
"triples_hartree",
)
)
)
< 1e-9,
"correlated energy components disagree",
)
_require(
event.get("n_frozen") == report.get("frozen_core_orbitals")
and event.get("triples_mode") == "t1",
"correlated method metadata differs",
)
_require(
report.get("method") == "DLPNO-CCSD(T)/cc-pVDZ"
and report.get("pno") == "tight"
and event.get("threshold_preset") == "TightPNO",
"external correlated method/preset differs",
)
options = _json(get("requested-cc-options.json"))
_require(
options.get("compute_triples") is True
and options.get("triples_mode") == "t1"
and bool(event["threshold_applied"])
and all(options.get(k) == v for k, v in event["threshold_applied"].items()),
"actual correlated thresholds differ from requested options",
)
version = report["runtime"]["vibeqc"]
records = [s for s in manifest["sections"] if s.get("kind") == "run.record"]
_require(
len(records) == 1 and records[0].get("program_version") == version,
"external runtime version differs from reference QVF",
)
final = {
"species": sid,
"run": item["job_id"],
"program": "vibe-qc",
"program_version": version,
"computation": {"kind": "external-import"},
"reference_verdict": reference,
"lineage": lineage,
"original_journal_sha256": _sha(files[selection["journal"]]),
"reference_qvf_sha256": _sha(get(qvf_name)),
"correlated_log_sha256": _sha(get("dlpno.scf.jsonl")),
"correlated_event": event,
"requested_options": options,
"result": {"energy": {"value": energy, "units": "hartree"}},
"settings": {
"method": event["variant"],
"basis": "cc-pvdz",
"task": "sp",
"reference_contract": contract,
"triples_algorithm": event["triples_algorithm"],
},
}
return final
[docs]
def external_chain_report(
document: Document, data: bytes, expected_sha256: str, reference_contract: str
) -> list[dict[str, Any]]:
"""Validate an explicitly pinned external archive, without creating Queue runs.
This is integrity and cross-record consistency, not an independent rerun or
a signature authenticating an arbitrary producer. Select a trusted archive.
"""
try:
_require(reference_contract == "real_restricted_cc", "select real_restricted_cc explicitly")
_require(_sha(data) == expected_sha256, "external evidence digest differs")
files = _zip(data)
descriptor = _json(files["external-chain.json"])
_require(
descriptor.get("schema") == "vibe-sketch.external-chain/1",
"unsupported evidence schema",
)
hashes = descriptor["files"]
_require(
set(files) == set(hashes) | {"external-chain.json"}, "unlisted/missing evidence files"
)
for path, digest in hashes.items():
_require(_sha(files[path]) == digest, "external evidence member checksum differs")
historical = []
for item in descriptor.get("prior_evidence", []):
_require(
item["role"]
in {"original-rejection", "original-campaign", "independent-reference"},
"unknown historical evidence role",
)
historical.append(
{"role": item["role"], "path": item["path"], "sha256": _sha(files[item["path"]])}
)
chosen = {
sid: _source(document, sid, entry, files, reference_contract)
for sid, entry in descriptor["species"].items()
}
_require(bool(chosen) and bool(document.reactions), "select an external reaction chain")
rows = []
for reaction in document.reactions:
_require(
reaction.arrow not in {"resonance", "retrosynthetic"},
"external reports require a forward reaction",
)
terms = []
balance: Counter[str] = Counter()
protocols = set()
for ids, sign in ((reaction.reactants, -1), (reaction.products, 1)):
for sid in ids:
source = chosen[sid]
species = document.species_by_id(sid)
balance.update(
{k: sign * v for k, v in Counter(species.conformers[0].symbols).items()}
)
balance.update({"charge": sign * species.charge})
terms.append({**source, "coefficient": sign})
protocols.add(
json.dumps(
[
source["program_version"],
source["settings"],
source["requested_options"],
],
sort_keys=True,
)
)
_require(
all(v == 0 for v in balance.values()),
"external reaction is not atom/charge balanced",
)
_require(len(protocols) == 1, "external protocols/software differ")
delta = math.fsum(t["coefficient"] * t["result"]["energy"]["value"] for t in terms)
_require(math.isfinite(delta), "nonfinite reaction energy")
rows.append(
{
"reaction": reaction.id,
"program": "vibe-qc",
"specification": terms[0]["settings"],
"temperature_k": None,
"pressure_atm": None,
"sources": terms,
"energies": {
"energy": {
"value": delta,
"units": "hartree",
"kcal_mol": delta * HARTREE_TO_KCAL_MOL,
}
},
"external_evidence_sha256": expected_sha256,
"prior_evidence": historical,
"warnings": [
"Imported direct-API chain; not a Sketch campaign or new calculation.",
"Electronic endpoint energies only; complex stability "
"and correlated accuracy are not certified.",
],
}
)
return rows
except ResultRefused:
raise
except (
ValueError,
KeyError,
TypeError,
OSError,
IndexError,
AttributeError,
OverflowError,
zipfile.BadZipFile,
) as error:
raise ResultRefused(f"invalid external chain evidence: {error}") from None
[docs]
def external_chain_export(
document: Document,
data: bytes,
expected_sha256: str,
reference_contract: str,
format: str = "json",
) -> bytes:
from vibesketch.document.tables import _table_rows
rows = external_chain_report(document, data, expected_sha256, reference_contract)
if format in {"csv", "markdown", "latex"}:
return _table_rows(rows, format) # type: ignore[arg-type]
if format != "json":
raise ResultRefused("external reports support json, csv, markdown or latex")
return (json.dumps(rows, indent=2, allow_nan=False) + "\n").encode()