lmxopcua/scripts/validation/opcua-lifecycle-memory/launch_saturation.py

#!/usr/bin/env python3
"""Launch only a previously reviewed immutable candidate saturation bundle. Never an acceptance run."""
import argparse
import hashlib
import json
from pathlib import Path
import subprocess
import time
import uuid
import xml.etree.ElementTree as ET


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def require(condition, message):
    if not condition:
        raise ValueError(message)


def verify(root, expected):
    receipt_path = root / "receipt.json"
    if digest(receipt_path) != expected:
        raise ValueError("Receipt hash differs from the reviewed artifact")
    receipt = json.loads(receipt_path.read_text())
    for name, value in receipt["files"].items():
        if digest(root / name) != value:
            raise ValueError("Immutable bundle file changed: " + name)
    require(receipt["protocol"] == "sdk2-saturation-v1" and receipt["acceptance"] is False, "Invalid diagnostic protocol")
    require((receipt["canonicalCycles"], receipt["completedRestarts"], receipt["settlingCycles"], receipt["maximumPhaseSeconds"]) == (384, 96, 1, 1800), "Changed prospective limits")
    return receipt


def trx(path, minimum):
    document = ET.parse(path)
    counters = document.find(".//{*}Counters")
    if counters is None:
        raise ValueError("Missing TRX counters")
    executed = int(counters.attrib.get("executed", "0"))
    skipped = max(int(counters.attrib.get("notExecuted", "0")),
                  sum(result.get("outcome") == "NotExecuted" for result in document.findall("./{*}Results/{*}UnitTestResult")))
    if executed < minimum or skipped:
        raise ValueError("Required diagnostic tests were skipped or not executed")
    return {"executed": executed, "skipped": skipped}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--bundle", type=Path, required=True)
    parser.add_argument("--receipt-sha256", required=True, help="Hash reviewed by root and Astra before launch")
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--verify-only", action="store_true")
    options = parser.parse_args()
    root = options.bundle.resolve(strict=True)
    receipt = verify(root, options.receipt_sha256)
    if options.verify_only:
        print("Immutable bundle verified; no experiment launched")
        return 0
    output = options.output.resolve()
    output.mkdir(parents=True, exist_ok=False)
    run_id = str(uuid.uuid4())
    container = "opcua-saturation-" + uuid.uuid4().hex[:12]
    command = ["docker", "run", "--rm", "--name", container, "--network", "none",
               "-e", "OTOPCUA_SOAK_BUILD_SHA=" + receipt["sourceRevision"],
               "-e", "OTOPCUA_SDK2_SATURATION_RUN_ID=" + run_id,
               "-v", str(root / "app") + ":/app:ro", "-v", str(root / "observer") + ":/observer:ro",
               "-v", str(root / "tools") + ":/tools:ro", "-v", str(output) + ":/task",
               receipt["image"], "bash", "/tools/container_saturation.sh"]
    (output / "launch.json").write_text(json.dumps({"command": command, "receiptSha256": options.receipt_sha256,
                                                   "sourceRevision": receipt["sourceRevision"], "acceptance": False}, indent=2) + "\n")
    start = time.monotonic()
    try:
        with (output / "container.log").open("w") as log:
            process = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, timeout=2040)
        status = process.returncode
    except subprocess.TimeoutExpired:
        status = 124
    finally:
        subprocess.run(["docker", "stop", "--time", "10", container], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20)
    (output / "container-exit-status.txt").write_text(str(status) + "\n")
    results = {"acceptance": False, "containerExitStatus": status, "wallSeconds": time.monotonic() - start,
               "sourceRevision": receipt["sourceRevision"], "receiptSha256": options.receipt_sha256}
    try:
        results["descriptorTests"] = trx(output / "results/descriptor-controls.trx", 2)
        results["experimentTests"] = trx(output / "results/saturation.trx", 1)
        rows = [json.loads(line) for line in (output / "resources.jsonl").read_text().splitlines()]
        header = rows[0]
        require(header["mode"] == "saturation" and header["requestedSeconds"] == 1800 and header["build"] == receipt["sourceRevision"], "Wrong executed source or mode")
        require(not any(row["kind"] == "complete" for row in rows), "Saturation must never emit an ordinary completion result")
        outcome = [row for row in rows if row["kind"] == "saturation-result"]
        require(len(outcome) == 1, "Missing unique diagnostic result")
        outcome = outcome[0]
        require((outcome["canonicalCycles"], outcome["settlingCycles"], outcome["restarts"]) == (384, 1, 96), "Incomplete fixed work")
        require(outcome["phaseSeconds"] <= 1800 and outcome["shutdownOwnershipVerified"], "Cutoff or cleanup proof failed")
        require(outcome["acceptance"] is False, "This is not an acceptance run")
        results["diagnosticOutcome"] = outcome
        observer_status = int((output / "observer-exit-status.txt").read_text())
        test_status = int((output / "test-exit-status.txt").read_text())
        require(observer_status == 0, "Observer failed")
        require((test_status != 0) == (outcome["coldBudgetRejected"] or not outcome["readiness"]), "Cold-cap exit was not preserved")
        results["protocolCompleted"] = True
    except Exception as error:
        results["protocolCompleted"] = False
        results["validationErrorType"] = type(error).__name__
        status = status or 1
    verify(root, options.receipt_sha256)
    (output / "run-receipt.json").write_text(json.dumps(results, indent=2) + "\n")
    print(json.dumps(results, indent=2))
    return status


if __name__ == "__main__":
    raise SystemExit(main())