"""Load the export's own verified runtime; never import an experiment package.""" from dataclasses import dataclass, field from pathlib import Path, PurePosixPath import copy import hashlib import importlib import importlib.util import json import re import sys import uuid from ._compat import RUNTIME_SHA256 def sha256(path): digest = hashlib.sha256() with Path(path).open("rb") as stream: for chunk in iter(lambda: stream.read(1 << 20), b""): digest.update(chunk) return digest.hexdigest() def verify_files(directory, expected_manifest_sha256): """Stdlib pre-import verification, including every weight and runtime file.""" root = Path(directory).resolve(strict=True) if not re.fullmatch(r"[0-9a-f]{64}", expected_manifest_sha256 or ""): raise ValueError("An explicit trusted MANIFEST SHA256 is required") if sha256(root / "MANIFEST.json") != expected_manifest_sha256: raise ValueError("Native manifest identity mismatch") manifest = json.loads((root / "MANIFEST.json").read_text()) files = manifest.get("files") if manifest.get("schema") != "decision.files.v1" or not isinstance(files, dict): raise ValueError("Unsupported native manifest") for name, ref in files.items(): if not isinstance(name, str) or not isinstance(ref, dict): raise ValueError("Invalid manifest entry") rel = PurePosixPath(name) if (rel.is_absolute() or ".." in rel.parts or rel.as_posix() != name or "\\" in name or not rel.parts): raise ValueError("Unsafe manifest path") path = root / name if (not path.is_file() or path.is_symlink() or set(ref) != {"bytes", "sha256"} or path.stat().st_size != ref["bytes"] or sha256(path) != ref["sha256"]): raise ValueError("Native file mismatch: " + name) actual = set() for path in root.rglob("*"): if path.is_symlink(): raise ValueError("Materialized native required; symlinks are not supported") rel = path.relative_to(root) if path.is_file() and rel.as_posix() != "MANIFEST.json": if "__pycache__" in rel.parts and path.suffix == ".pyc": continue actual.add(rel.as_posix()) if actual != set(files): raise ValueError("Native file roster mismatch") runtime = {name: ref["sha256"] for name, ref in files.items() if name.endswith(".py")} if runtime != RUNTIME_SHA256: raise ValueError("Unsupported runtime revision; architecture name alone is insufficient") return root, manifest @dataclass class Native: directory: Path manifest_sha256: str model: object collator: object config: dict model_api: object contract: object artifacts: object training_state: dict | None = None last_training_provenance: dict | None = None package_name: str = field(default="", repr=False) def _amd_device(device): import torch selected = torch.device(device) if selected.type != "cuda" or torch.version.hip is None or not torch.cuda.is_available(): raise ValueError("This runtime requires a real ROCm CUDA device; no CPU fallback") if selected.index is None: selected = torch.device("cuda", torch.cuda.current_device()) return selected def load_native(directory, *, expected_manifest_sha256, device="cuda:0"): """Load any compatible weight export with the pinned public runtime revision. Verification precedes executing bundled Python. This function loads a model; importing decision_runtime or running its stdlib checks does not. """ root, manifest = verify_files(directory, expected_manifest_sha256) selected = _amd_device(device) namespace = "_decision_native_" + uuid.uuid4().hex spec = importlib.util.spec_from_file_location(namespace, root / "__init__.py", submodule_search_locations=[str(root)]) package = importlib.util.module_from_spec(spec) sys.modules[namespace] = package try: spec.loader.exec_module(package) artifacts = importlib.import_module(namespace + ".artifacts") contract = importlib.import_module(namespace + ".contract") api = importlib.import_module(namespace + ".model") declared = contract.validate_config(json.loads((root / "decision_config.json").read_text())) if (declared["arm"], declared["training_arm"]) != ("all22", "S22"): raise ValueError("Only the complete three-path all22/S22 architecture is supported") model, collator, cfg = artifacts.load_export(root, device=str(selected)) if (cfg["arm"], cfg["training_arm"]) != ("all22", "S22"): raise ValueError("Only the complete three-path all22/S22 architecture is supported") if type(model) is not api.DecisionModel or artifacts.verify_native(root) != (manifest, cfg): raise ValueError("Loaded class or native identity mismatch") return Native(root, expected_manifest_sha256, model, collator, cfg, api, contract, artifacts, package_name=namespace) except BaseException: for name in tuple(sys.modules): if name == namespace or name.startswith(namespace + "."): del sys.modules[name] raise def _native_mode(native): if native.training_state is not None: raise ValueError("Restore native policy before native inference or export") native.model.verify_inventory(check_values=False) def predict(native, records, *, batch_size=8): """Unchanged bundled inference, FP32; no training graph is used here.""" _native_mode(native) device = next(native.model.parameters()).device _amd_device(device) return native.model_api.predict(native.model, native.collator, records, batch_size=batch_size, device=str(device), precision="fp32") def export_native(native, directory, *, provenance): """Use the original strict exporter after explicit policy restoration.""" _native_mode(native) if native.last_training_provenance is None or not isinstance(provenance, dict): raise ValueError("Restored training provenance and an explicit user provenance dict required") if Path(directory).exists(): raise ValueError("Export destination must be fresh") details = {"parent_native_manifest_sha256": native.manifest_sha256, "parent_provenance": copy.deepcopy(native.config.get("provenance", {})), "training_adapter": copy.deepcopy(native.last_training_provenance), "user": copy.deepcopy(provenance)} json.dumps(details, allow_nan=False) return native.artifacts.export_model(native.model, native.collator.tokenizer, directory, native.config, details)