"""Evidence-bound, immutable review continuations of completed recipe calculations."""
from __future__ import annotations
import hashlib
import json
import shutil
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from vibesketch.document.campaign import (
CampaignRefused,
Compiled,
_canonical,
_record,
_vibe_queue_digest,
_vq,
_write_json,
discard_output,
inside,
source_digest,
)
from vibesketch.document.model import CampaignRecord, Document
from vibesketch.stages import VALIDATOR
from vibesketch.stages.validator import assess, reviewable_reference, sha256
_REPLAY = """import json
import shutil
from pathlib import Path
from run_stage import check_start, sha256
root = Path.cwd()
record = json.loads((root / "replay.json").read_text())
config = json.loads((root / "stage.json").read_text())
check_start(config, root)
if sha256(root / "start/geometry.json") != record["geometry_sha256"]:
raise ValueError("reused result has a different parent geometry")
if sha256(root / "retained-result.qvf") != record["result_sha256"]:
raise ValueError("retained computation result has changed")
shutil.copyfile(root / "retained-result.qvf", root / "result.qvf")
(root / "runner.log").write_text("Reused verified computation; see replay.json.\\n")
"""
[docs]
def review_campaign(
document: Document,
recipe_id: str,
campaign_id: str,
directory: str,
*,
node: str,
reviewer: str,
reason: str,
criteria: dict[str, Any],
warnings: list[str],
evidence: list[str],
) -> Compiled:
"""Compile a separate continuation for one rejected terminal DLPNO check.
All computation artifacts must already be verified. This operation reuses
those bytes, never runs chemistry or edits the old journal, and records
the original attempts. Only the restricted-reference external-instability
review scope is supported. Other terminal reference failures stay rejected
until separately reviewed; all other scientific checks must have succeeded.
"""
return _continue_campaign(
document,
recipe_id,
campaign_id,
directory,
node=node,
reviewer=reviewer,
reason=reason,
criteria=criteria,
warnings=warnings,
evidence=evidence,
)
[docs]
def revalidate_campaign(
document: Document,
recipe_id: str,
campaign_id: str,
directory: str,
*,
node: str,
reviewer: str,
reason: str,
) -> Compiled:
"""Revalidate an unchanged DLPNO result rejected by the older energy reader.
The current strict assessment must pass without a scientific waiver. A new
campaign replays verified calculation bytes and retains the original failure.
Other reference failures require their own subsequent explicit reviews.
"""
return _continue_campaign(
document,
recipe_id,
campaign_id,
directory,
node=node,
reviewer=reviewer,
reason=reason,
criteria={},
warnings=[],
evidence=[],
revalidate=True,
)
def _continue_campaign(
document: Document,
recipe_id: str,
campaign_id: str,
directory: str,
*,
node: str,
reviewer: str,
reason: str,
criteria: dict[str, Any],
warnings: list[str],
evidence: list[str],
revalidate: bool = False,
) -> Compiled:
vq = _vq()
recipe, record = _record(document, recipe_id, campaign_id)
root = Path(record.directory)
target = Path(directory).expanduser().resolve()
if target.exists():
raise CampaignRefused(["a review continuation needs a new directory"])
old = 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(old)
if digest != record.campaign_sha256 or journal.campaign_sha256 != digest:
raise CampaignRefused(
["the original campaign or journal does not match its recorded digest"]
)
if any(item["problem"] for item in vq.check_sources(old, root).values()):
raise CampaignRefused(["the original campaign sources have changed"])
role = record.nodes.get(node, {})
if role.get("role") != "check" or recipe.stage_by_id(role["stage"]).kind != "dlpno":
raise CampaignRefused(["only a terminal DLPNO check permits this review scope"])
if any(node in n.edges() for n in old.nodes):
raise CampaignRefused(["the reviewed check must be terminal"])
attempts = {n.id: [a for a in journal.attempts if a.node == n.id] for n in old.nodes}
if not attempts[node] or attempts[node][-1].state != "failed":
raise CampaignRefused(["review requires an existing failed validation attempt"])
failed_checks = set()
for n in old.nodes:
if attempts[n.id] and attempts[n.id][-1].state == "succeeded":
continue
other_role = record.nodes.get(n.id, {})
if (
attempts[n.id]
and attempts[n.id][-1].state == "failed"
and other_role.get("role") == "check"
and recipe.stage_by_id(other_role["stage"]).kind == "dlpno"
and not any(n.id in consumer.edges() for consumer in old.nodes)
):
failed_checks.add(n.id)
else:
raise CampaignRefused([f"{n.id}: finish the original campaign before review"])
for n in old.nodes:
attempt = attempts[n.id][-1]
for item in n.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 binding.producer_attempt != producer.attempt
or binding.manifest_sha256 != output.manifest_sha256
):
raise CampaignRefused([f"{n.id}: stale or mismatched parent artifact binding"])
if (
not reviewer.strip()
or not reason.strip()
or (not warnings and not revalidate)
or not all(isinstance(w, str) and w.strip() for w in warnings)
):
raise CampaignRefused(["reviewer, reason and explicit warnings are required"])
evidence_paths = [Path(p).expanduser().resolve() for p in evidence]
if (not evidence_paths and not revalidate) or any(not p.is_file() for p in evidence_paths):
raise CampaignRefused(["review evidence must name retained files"])
def artifact(node_id: str, output: str) -> tuple[Path, dict[str, Any]]:
attempt = attempts[node_id][-1]
item = attempt.artifacts.get(output)
base = root / "run/artifacts" / node_id / str(attempt.attempt)
if item is None or item.status != "verified":
raise CampaignRefused([f"{node_id}: {output} has no verified artifact"])
manifest = vq.load_manifest(base / f"{output}.manifest.json")
path = base / 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 CampaignRefused([f"{node_id}: artifact integrity or attempt mismatch"])
return path, {
"attempt": attempt.model_dump(mode="json"),
"manifest": vq.manifest_payload(manifest),
}
# Check every retained artifact before creating the continuation.
retained: dict[str, tuple[Path, dict[str, Any]]] = {}
for n in old.nodes:
for output in n.outputs:
if n.id not in failed_checks:
checked = artifact(n.id, output.name)
if record.nodes[n.id]["role"] == "run" and output.name == "result":
retained[n.id] = checked
run_id = node.removesuffix("-check") + "-run"
result = retained[run_id][0]
check_node = old.node(node)
check_dir = inside(root, check_node.sources.path, "check sources")
check = json.loads((check_dir / "check.json").read_text())
raw = assess(check, result)
from vq.campaign_controller import CliQueue, FetchError
rejected: dict[str, bytes] = {}
for failed_id in sorted(failed_checks):
failed_job = attempts[failed_id][-1].job
if failed_job is None:
raise CampaignRefused(["the failed check has no recorded queue job"])
failed_dir = inside(root, old.node(failed_id).sources.path, "check sources")
failed_check = json.loads((failed_dir / "check.json").read_text())
failed_result = retained[failed_id.removesuffix("-check") + "-run"][0]
failed_raw = assess(failed_check, failed_result)
upgrading = revalidate and failed_id == node
if not upgrading and not reviewable_reference(failed_check, failed_result, failed_raw):
raise CampaignRefused([f"{failed_id}: failure is outside restricted-reference review"])
try:
with tempfile.TemporaryDirectory(prefix="vibesketch-review-") as temporary:
fetched = CliQueue().fetch(failed_job, "accepted", Path(temporary))
failed_verdict_path = fetched / "verdict.json"
if failed_verdict_path.is_symlink():
raise CampaignRefused(["the original verdict must be a regular file"])
rejected[failed_id] = failed_verdict_path.read_bytes()
failed_verdict = json.loads(rejected[failed_id])
except (OSError, ValueError, FetchError) as error:
raise CampaignRefused([f"cannot retain the original failed verdict: {error}"]) from None
if not upgrading and failed_verdict != failed_raw:
raise CampaignRefused(["the original failed verdict differs from raw revalidation"])
failed_verdict_bytes = rejected[node]
decision = {
"schema": "vibe-sketch.operator-review/1",
"scope": "restricted-reference-external-instability",
"reviewer": reviewer,
"reason": reason,
"criteria": criteria,
"warnings": warnings,
"created": datetime.now(UTC).isoformat(),
"result_sha256": sha256(result),
"check_sha256": _canonical(check),
"raw_assessment_sha256": _canonical(raw),
"original_verdict_sha256": hashlib.sha256(failed_verdict_bytes).hexdigest(),
"original_campaign_sha256": digest,
"original_journal_sha256": hashlib.sha256(journal_bytes).hexdigest(),
"failed_check_attempt": attempts[node][-1].model_dump(mode="json"),
"evidence": [
{"path": f"review-evidence/{i}", "sha256": sha256(p)}
for i, p in enumerate(evidence_paths)
],
}
if revalidate:
decision = {
"schema": "vibe-sketch.validator-revalidation/1",
"scope": "structured-correlated-energy-reader",
"reviewer": reviewer,
"reason": reason,
"created": datetime.now(UTC).isoformat(),
"result_sha256": sha256(result),
"check_sha256": _canonical(check),
"original_validator_sha256": sha256(check_dir / "validate_stage.py"),
"validator_sha256": sha256(VALIDATOR),
"original_verdict_sha256": hashlib.sha256(failed_verdict_bytes).hexdigest(),
"original_verdict": json.loads(failed_verdict_bytes),
"original_campaign_file_sha256": hashlib.sha256(
(json.dumps(old.model_dump(mode="json"), indent=2, sort_keys=True) + "\n").encode()
).hexdigest(),
"original_campaign_sha256": digest,
"original_journal_sha256": hashlib.sha256(journal_bytes).hexdigest(),
"failed_check_attempt": attempts[node][-1].model_dump(mode="json"),
"warning": "Revalidated unchanged calculation with the structured correlated-energy "
"reader; the original validator rejection remains in the archive.",
}
verdict = assess(
check, result, None if revalidate else decision, decision if revalidate else None
)
if not verdict["accepted"]:
raise CampaignRefused(verdict["reasons"])
identifier = f"{recipe.id}-c{len(recipe.campaigns) + 1}"
payload = old.model_dump(mode="json", exclude_none=True)
payload["campaign_id"] = identifier
try:
target.mkdir()
lineage = {
"campaign_sha256": digest,
"journal": vq.journal_payload(journal),
"revalidation" if revalidate else "review": decision,
"raw_assessment": raw,
"unreviewed_checks": sorted(failed_checks - {node}),
}
_write_json(target / "lineage.json", lineage)
for item in payload["nodes"]:
original = inside(root, item["sources"]["path"], "node sources")
dest = inside(target, item["sources"]["path"], "node sources")
shutil.copytree(original, dest)
if source_digest(dest) != item["sources"]["sha256"]:
raise CampaignRefused(["original sources changed during review compilation"])
if record.nodes[item["id"]]["role"] == "run":
path, origin = retained[item["id"]]
stage_check_id = item["id"].removesuffix("-run") + "-check"
stage_check = json.loads(
(
inside(root, old.node(stage_check_id).sources.path, "check sources")
/ "check.json"
).read_text()
)
original_verdict = assess(stage_check, path)
start = original_verdict.get("start") or {}
if not start.get("geometry_sha256"):
raise CampaignRefused(["the retained result lacks its starting geometry hash"])
shutil.copyfile(path, dest / "retained-result.qvf")
_write_json(
dest / "replay.json",
{
"original_campaign_sha256": digest,
"original_execution": {
key: item.get(key)
for key in (
"host",
"program",
"command",
"cpus",
"mem_mb",
"wall_time_seconds",
)
},
"source": origin,
"original_sources_sha256": item["sources"]["sha256"],
"stage_input_sha256": sha256(dest / "stage.qvf"),
"geometry_sha256": start["geometry_sha256"],
"parent_result_sha256": start.get("result_sha256"),
"result_sha256": sha256(path),
},
)
(dest / "replay.py").write_text(_REPLAY)
item["cpus"] = 1
item.pop("mem_mb", None)
item.pop("wall_time_seconds", None)
check_budget = old.node(stage_check_id).wall_time_seconds
if check_budget is not None:
item["wall_time_seconds"] = check_budget
item["kind"] = "other"
item["command"] = [item["command"][0], "replay.py"]
else:
# Older campaigns omitted the managed program on check nodes.
producer = old.node(item["id"].removesuffix("-check") + "-run")
if producer.program is not None and "program" not in item:
item["program"] = producer.program
shutil.copyfile(dest / "validate_stage.py", dest / "original-validator.py")
shutil.copyfile(VALIDATOR, dest / "validate_stage.py")
if item["id"] in rejected:
# Keep each original failure through subsequent review continuations.
history = dest / "rejected-attempts" / digest
history.mkdir(parents=True)
(history / "verdict.json").write_bytes(rejected[item["id"]])
_write_json(
history / "attempt.json", attempts[item["id"]][-1].model_dump(mode="json")
)
if item["id"] == node:
_write_json(
dest
/ ("validator-revalidation.json" if revalidate else "operator-review.json"),
decision,
)
_write_json(dest / "raw-assessment.json", raw)
(dest / "original-verdict.json").write_bytes(failed_verdict_bytes)
_write_json(dest / "original-lineage.json", lineage)
(dest / "original-journal.json").write_bytes(journal_bytes)
_write_json(dest / "original-campaign.json", old.model_dump(mode="json"))
if revalidate:
shutil.copyfile(
check_dir / "validate_stage.py", dest / "revalidation-original-validator.py"
)
(dest / "revalidation-original-verdict.json").write_bytes(failed_verdict_bytes)
(dest / "review-evidence").mkdir(exist_ok=True)
for i, path in enumerate(evidence_paths):
copied = dest / "review-evidence" / str(i)
shutil.copyfile(path, copied)
if sha256(copied) != decision["evidence"][i]["sha256"]:
raise CampaignRefused(["review evidence changed while retaining it"])
item["sources"]["sha256"] = source_digest(dest)
if (root / "run/journal.json").read_bytes() != journal_bytes:
raise CampaignRefused(["the original journal changed during review compilation"])
_write_json(target / "campaign.json", payload)
new_digest, notes = _vibe_queue_digest(target / "campaign.json")
except (OSError, ValueError):
discard_output(target, False)
raise
new_record = CampaignRecord(
id=identifier,
directory=str(target),
campaign_sha256=new_digest,
created=datetime.now(UTC),
nodes=record.nodes,
model_license_acknowledged=record.model_license_acknowledged,
)
updated = document.model_copy(
update={
"recipes": [
r.model_copy(update={"campaigns": [*r.campaigns, new_record]})
if r.id == recipe_id
else r
for r in document.recipes
]
}
)
return Compiled(Document.model_validate(updated.to_json_data()), new_record, notes)