| """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 | |
| 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) | |