"""Electronic reaction reports from accepted campaign artifacts, without invented runs."""
from __future__ import annotations
import hashlib
import json
import math
import zipfile
from collections import Counter
from pathlib import Path
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from vibesketch.document.campaign import _record, _vq, inside
from vibesketch.document.model import Document
from vibesketch.document.results import ResultRefused, _finite, species_digest
from vibesketch.stages.validator import _verified_member, assess, sha256
from vibesketch.units import HARTREE_TO_KCAL_MOL
[docs]
class CampaignSelection(BaseModel):
"""Explicit immutable campaign and correlated stage; no implicit latest campaign."""
model_config = ConfigDict(extra="forbid", strict=True)
recipe: str
campaign: str
stage: str
reactions: list[str] | None = None
coefficients: dict[str, dict[str, dict[str, float]]] = Field(default_factory=dict)
CampaignFormat = Literal["json", "csv", "markdown", "latex", "svg", "qvf"]
def _json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def _job_and_document(path: Path) -> tuple[dict[str, Any], Document]:
with zipfile.ZipFile(path) as archive:
manifest = json.loads(archive.read("manifest.json"))
jobs = [s for s in manifest["sections"] if s.get("kind") == "job.spec"]
documents = [s for s in manifest["sections"] if s.get("id") == "vibesketch_document"]
if len(jobs) != 1 or len(documents) != 1:
raise ResultRefused("campaign result lacks a unique job and compiled document")
spec = json.loads(_verified_member(archive, jobs[0]["members"]["spec"]))
document = Document.model_validate_json(
_verified_member(archive, documents[0]["members"]["document"])
)
return spec, document
def _accepted_sources(document: Document, selection: CampaignSelection) -> dict[str, Any]:
vq = _vq()
recipe, record = _record(document, selection.recipe, selection.campaign)
stage = recipe.stage_by_id(selection.stage)
if stage.kind != "dlpno":
raise ResultRefused("campaign reaction reports require a correlated single-point stage")
root = Path(record.directory)
campaign = vq.load_campaign(root / "campaign.json")
journal_bytes = (root / "run/journal.json").read_bytes()
journal = vq.Journal.model_validate_json(journal_bytes)
digest = vq.campaign_digest(campaign)
if (
digest != record.campaign_sha256
or journal.campaign_sha256 != digest
or campaign.campaign_id != record.id
):
raise ResultRefused("campaign or journal differs from the document's recorded digest")
expected_nodes = {
f"{sid}-{step.id}-{role}": {"species": sid, "stage": step.id, "role": role}
for sid in recipe.species
for step in recipe.stages
if step.kind != "prepare"
for role in ("run", "check")
}
if record.nodes != expected_nodes or {node.id for node in campaign.nodes} != set(
expected_nodes
):
raise ResultRefused("campaign nodes do not match the complete recipe")
for node in campaign.nodes:
inside(root, node.sources.path, "campaign sources")
if any(item["problem"] for item in vq.check_sources(campaign, root).values()):
raise ResultRefused("campaign sources differ from their pins")
attempts = {n.id: [a for a in journal.attempts if a.node == n.id] for n in campaign.nodes}
for node, history in attempts.items():
if (
not history
or [a.attempt for a in history] != sorted({a.attempt for a in history})
or history[-1].state != "succeeded"
):
raise ResultRefused(f"{node}: finish all campaign checks before reporting energies")
for node in campaign.nodes:
attempt = attempts[node.id][-1]
for item in node.inputs:
binding = attempt.inputs.get(item.name)
producer = attempts[item.producer][-1]
output = producer.artifacts.get(item.output)
if (
binding is None
or output is None
or output.status != "verified"
or binding.producer_attempt != producer.attempt
or binding.manifest_sha256 != output.manifest_sha256
):
raise ResultRefused(f"{node.id}: stale accepted-parent artifact binding")
def artifact(node_id: str, output: str) -> tuple[Path, dict[str, Any]]:
attempt = attempts[node_id][-1]
item = attempt.artifacts.get(output)
base = inside(root, f"run/artifacts/{node_id}/{attempt.attempt}", "artifact")
if item is None or item.status != "verified":
raise ResultRefused(f"{node_id}: missing verified {output} artifact")
manifest_path = inside(
root,
str(base.relative_to(root.resolve()) / f"{output}.manifest.json"),
"artifact manifest",
)
manifest = vq.load_manifest(manifest_path)
path = inside(root, str(base.relative_to(root.resolve()) / output), "artifact output")
if (
vq.manifest_digest(manifest) != item.manifest_sha256
or manifest.node != node_id
or manifest.attempt != attempt.attempt
or manifest.output != output
or vq.verify_artifact(path, manifest)["status"] != "verified"
):
raise ResultRefused(f"{node_id}: artifact integrity or attempt mismatch")
return path, {
"node": node_id,
"attempt": attempt.attempt,
"manifest_sha256": item.manifest_sha256,
"queue_job_id": attempt.job.job_id if attempt.job is not None else None,
}
# A succeeded process alone is not scientific acceptance, including parents.
for node in campaign.nodes:
if record.nodes.get(node.id, {}).get("role") == "check":
accepted, _ = artifact(node.id, "accepted")
verdict = _json(accepted / "verdict.json")
if verdict.get("accepted") is not True:
raise ResultRefused(f"{node.id}: the stage is not scientifically accepted")
if sha256(accepted / "result.qvf") != verdict.get("result_sha256"):
raise ResultRefused(f"{node.id}: accepted verdict names a different result")
chosen = {}
for sid in recipe.species:
run_id = f"{sid}-{stage.id}-run"
check_id = f"{sid}-{stage.id}-check"
if run_id not in attempts or check_id not in attempts:
raise ResultRefused(f"{sid}: campaign does not contain the selected stage")
run_node, check_node = campaign.node(run_id), campaign.node(check_id)
if any(check_id in n.edges() for n in campaign.nodes):
raise ResultRefused("reaction reports require a terminal correlated stage")
run_sources = inside(root, run_node.sources.path, "run sources")
check_sources = inside(root, check_node.sources.path, "check sources")
spec, compiled = _job_and_document(run_sources / "stage.qvf")
compiled_recipe = next(r for r in compiled.recipes if r.id == recipe.id)
if compiled_recipe.stages != recipe.stages or compiled_recipe.species != recipe.species:
raise ResultRefused("the recipe has changed since campaign compilation")
if species_digest(compiled.species_by_id(sid)) != species_digest(
document.species_by_id(sid)
):
raise ResultRefused(f"{sid}: species or input geometry changed since compilation")
accepted, origin = artifact(check_id, "accepted")
produced, computation = artifact(run_id, "result")
result = accepted / "result.qvf"
if sha256(result) != sha256(produced):
raise ResultRefused(f"{sid}: accepted result differs from its computation")
replay_origin = None
replay_path = run_sources / "replay.json"
computation["kind"] = "calculation"
if replay_path.exists():
replay = _json(replay_path)
if replay.get("result_sha256") != sha256(result):
raise ResultRefused(f"{sid}: replay identifies a different retained calculation")
prior = replay["source"]
prior_manifest = vq.ArtifactManifest.model_validate(prior["manifest"])
prior_attempt = prior["attempt"]
prior_digest = vq.manifest_digest(prior_manifest)
if (
prior_attempt["artifacts"]["result"]["manifest_sha256"] != prior_digest
or prior_attempt["state"] != "succeeded"
or prior_manifest.node != run_id
or prior_manifest.attempt != prior_attempt["attempt"]
or prior_manifest.output != "result"
or prior_manifest.kind != "file"
or prior_manifest.sha256 != sha256(result)
):
raise ResultRefused(f"{sid}: replay origin has a conflicting artifact identity")
computation["kind"] = "replay"
replay_origin = {
"campaign_sha256": replay["original_campaign_sha256"],
"sources_sha256": replay["original_sources_sha256"],
"result_sha256": replay["result_sha256"],
"producer": {
"node": prior_manifest.node,
"attempt": prior_manifest.attempt,
"manifest_sha256": prior_digest,
"queue_job_id": (prior_attempt.get("job") or {}).get("job_id"),
},
"scope": "immediate artifact origin; earlier lineage is in the execution archive",
}
actual_spec, _ = _job_and_document(result)
if actual_spec != spec:
raise ResultRefused(f"{sid}: executed job differs from its compiled input")
check = _json(check_sources / "check.json")
if (check.get("species"), check.get("stage"), check.get("kind")) != (
sid,
stage.id,
"dlpno",
):
raise ResultRefused(f"{sid}: mismatched selected stage check")
verdict = _json(accepted / "verdict.json")
review = verdict.get("operator_review")
if review is not None:
if _json(check_sources / "operator-review.json") != review:
raise ResultRefused("accepted review differs from its pinned decision")
if sha256(check_sources / "original-verdict.json") != review["original_verdict_sha256"]:
raise ResultRefused("original rejected verdict differs from the review binding")
for item in review["evidence"]:
if sha256(inside(check_sources, item["path"], "review evidence")) != item["sha256"]:
raise ResultRefused("review evidence differs from its recorded hash")
revalidation = verdict.get("validator_revalidation")
if revalidation is not None:
if _json(check_sources / "validator-revalidation.json") != revalidation:
raise ResultRefused("accepted revalidation differs from its pinned decision")
for name, key in (
("revalidation-original-validator.py", "original_validator_sha256"),
("revalidation-original-verdict.json", "original_verdict_sha256"),
("original-campaign.json", "original_campaign_file_sha256"),
("original-journal.json", "original_journal_sha256"),
("validate_stage.py", "validator_sha256"),
):
if (
sha256(inside(check_sources, name, "revalidation evidence"))
!= revalidation[key]
):
raise ResultRefused("revalidation evidence differs from its recorded hash")
if (
_json(check_sources / "revalidation-original-verdict.json")
!= revalidation["original_verdict"]
):
raise ResultRefused("revalidation original verdict differs from its binding")
fresh = assess(check, result, review, revalidation)
if fresh != verdict or fresh.get("accepted") is not True:
raise ResultRefused(f"{sid}: accepted verdict does not match current revalidation")
energy = fresh.get("energy") or {}
source = energy.get("source") or {}
event = source.get("event") or {}
# Root totals alone do not establish the actual correlated algorithm.
required = (
"engine",
"source_kind",
"aux_basis",
"triples_mode",
"triples_algorithm",
"frozen_core_convention",
"threshold_preset",
"threshold_requested",
"threshold_applied",
)
if not source or any(event.get(key) is None for key in required):
raise ResultRefused(f"{sid}: missing actual correlated algorithm/threshold evidence")
if any(not isinstance(event[key], str) or not event[key].strip() for key in required[:-2]):
raise ResultRefused(f"{sid}: correlated algorithm labels must be explicit")
for key in ("threshold_requested", "threshold_applied"):
values = event[key]
if (
not isinstance(values, dict)
or not values
or any(_finite(value, "correlated threshold") < 0 for value in values.values())
):
raise ResultRefused(f"{sid}: missing finite correlated thresholds")
if any(
event["threshold_applied"].get(key) != value
for key, value in event["threshold_requested"].items()
):
raise ResultRefused(f"{sid}: applied correlated thresholds differ from the request")
if event.get("threshold_unsupported") != [] or event.get("threshold_inactive") != []:
raise ResultRefused(
f"{sid}: correlated result has unsupported or unverified thresholds"
)
total = _finite(energy.get("total_hartree"), "correlated total")
if energy.get("method") != spec.get("method"):
raise ResultRefused(f"{sid}: accepted energy has a different method")
parent_inputs = [item for item in run_node.inputs if item.name == "start"]
if len(parent_inputs) != 1:
raise ResultRefused(f"{sid}: correlated stage lacks its accepted geometry parent")
parent_input = parent_inputs[0]
parent, parent_origin = artifact(parent_input.producer, parent_input.output)
start = fresh.get("start") or {}
if start.get("geometry_sha256") != sha256(parent / "geometry.json") or start.get(
"result_sha256"
) != sha256(parent / "result.qvf"):
raise ResultRefused(f"{sid}: result does not identify its accepted parent geometry")
program, version = fresh.get("program"), fresh.get("program_version")
if (
not isinstance(program, str)
or not program
or not isinstance(version, str)
or not version
):
raise ResultRefused(f"{sid}: missing calculation program/version")
settings = {
k: v for k, v in spec.items() if k not in {"structure_ref", "charge", "multiplicity"}
}
chosen[sid] = {
"program": program,
"program_version": version,
"settings": settings,
"actual_algorithm": {key: event[key] for key in required},
"species": sid,
"run": computation["queue_job_id"] or run_id,
"job": source["job_ref"],
"campaign_sha256": digest,
"journal_sha256": hashlib.sha256(journal_bytes).hexdigest(),
"computation": computation,
"replay_origin": replay_origin,
"acceptance": origin,
"accepted_parent": {**parent_origin, **start},
"acceptance_kind": fresh.get("acceptance_kind", "automatic"),
"operator_review": review,
"validator_revalidation": revalidation,
"warnings": fresh.get("review") or [],
"result": {
"source_sha256": sha256(result),
"energy": {
"value": total,
"units": "hartree",
"source": source,
},
},
}
return chosen
[docs]
def campaign_energies(document: Document, selection: CampaignSelection) -> list[dict[str, Any]]:
"""Verified electronic products minus reactants, preserving real campaign lineage."""
try:
chosen = _accepted_sources(document, selection)
return _aggregate(document, selection, chosen)
except (OSError, ValueError, KeyError, TypeError, StopIteration, zipfile.BadZipFile) as error:
if isinstance(error, ResultRefused):
raise
raise ResultRefused(f"cannot report campaign energies: {error}") from None
def _aggregate(
document: Document, selection: CampaignSelection, chosen: dict[str, Any]
) -> list[dict[str, Any]]:
reactions = {r.id: r for r in document.reactions}
wanted = selection.reactions if selection.reactions is not None else list(reactions)
if not wanted or len(set(wanted)) != len(wanted) or not set(wanted) <= reactions.keys():
raise ResultRefused("select known, distinct reactions")
if not set(selection.coefficients) <= set(wanted):
raise ResultRefused("coefficients name an unselected reaction")
rows = []
for identifier in wanted:
reaction = reactions[identifier]
if reaction.arrow == "resonance":
raise ResultRefused("a resonance arrow has no reaction energy")
reactants, products = reaction.reactants, reaction.products
if reaction.arrow == "retrosynthetic":
reactants, products = products, reactants
coefficients = selection.coefficients.get(
identifier,
{
"reactants": dict.fromkeys(reactants, 1.0),
"products": dict.fromkeys(products, 1.0),
},
)
if (
set(coefficients) != {"reactants", "products"}
or set(coefficients["reactants"]) != set(reactants)
or set(coefficients["products"]) != set(products)
):
raise ResultRefused("coefficients must name every species on both chemical sides")
terms = []
balance: dict[str, float] = {}
protocols = set()
for side, sign in (("reactants", -1), ("products", 1)):
for sid, coefficient in coefficients[side].items():
factor = _finite(coefficient, "coefficient")
if factor <= 0 or sid not in chosen:
raise ResultRefused(f"{sid}: need a positive coefficient and accepted stage")
source = chosen[sid]
species = document.species_by_id(sid)
for element, count in Counter(species.conformers[0].symbols).items():
balance[element] = balance.get(element, 0.0) + sign * factor * count
balance["charge"] = balance.get("charge", 0.0) + sign * factor * species.charge
protocols.add(
json.dumps(
{
key: source[key]
for key in (
"program",
"program_version",
"settings",
"actual_algorithm",
)
},
sort_keys=True,
allow_nan=False,
)
)
terms.append({**source, "coefficient": sign * factor})
if any(not math.isfinite(value) or abs(value) > 1e-8 for value in balance.values()):
raise ResultRefused(
"reaction atoms/charge are not balanced; supply complete coefficients"
)
if len(protocols) != 1:
raise ResultRefused(
"selected campaign results have incompatible settings/software/algorithms"
)
try:
delta = math.fsum(t["coefficient"] * t["result"]["energy"]["value"] for t in terms)
converted = delta * HARTREE_TO_KCAL_MOL
except (OverflowError, ValueError):
raise ResultRefused("reaction energy overflow") from None
if not math.isfinite(delta) or not math.isfinite(converted):
raise ResultRefused("reaction energy overflow")
spec = {k: v for k, v in terms[0]["settings"].items() if k not in {"tasks", "job_type"}}
spec.update(task="sp", triples_algorithm=terms[0]["actual_algorithm"]["triples_algorithm"])
rows.append(
{
"reaction": identifier,
"recipe": selection.recipe,
"campaign": selection.campaign,
"stage": selection.stage,
"coefficients": coefficients,
"specification": spec,
"program": terms[0]["program"],
"energies": {
"energy": {
"value": delta,
"units": "hartree",
"kcal_mol": converted,
}
},
"temperature_k": None,
"pressure_atm": None,
"sources": terms,
"warnings": sorted({warning for t in terms for warning in t["warnings"]}),
}
)
return rows
[docs]
def campaign_energy_export(
document: Document, selection: CampaignSelection, format: CampaignFormat = "json"
) -> bytes:
"""Report, table, SVG or QVF from the same verified campaign selection."""
from vibesketch.document.qvf import _to_qvf_rows
from vibesketch.document.results import _figure_rows
from vibesketch.document.tables import _table_rows
rows = campaign_energies(document, selection)
if format in {"csv", "markdown", "latex"}:
return _table_rows(rows, format)
if format == "svg":
return _figure_rows(document, rows)
if format == "qvf":
return _to_qvf_rows(document, rows=rows)
if format != "json":
raise ResultRefused("unsupported campaign report format")
return (json.dumps(rows, indent=2, allow_nan=False) + "\n").encode()