"""Check four vLLM/PyTorch binary roles against Eggcracker's exact release pins.""" from __future__ import annotations import argparse import json import os import stat import sys from collections.abc import Callable, Sequence from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from lumi_eggcracker import __version__, elfmarkers from lumi_eggcracker.elfmarkers import ( PYTORCH_ATEN_EVIDENCE_ID, PYTORCH_BRIDGE_EVIDENCE_ID, VLLM_EXTENSION_EVIDENCE_ID, VLLM_PYTHON_EVIDENCE_ID, RuntimeEvidence, ) SCHEMA = "lumi-eggcracker.vllm-compatibility.v1" PROFILE = "content.safetensors-vllm" DescriptorInspector = Callable[[int], RuntimeEvidence | None] def _absolute_path(parser: argparse.ArgumentParser, value: str) -> Path: path = Path(value) if not path.is_absolute(): parser.error("vLLM and PyTorch binary paths must be absolute") return path def _inspect_regular_path( path: Path, inspector: DescriptorInspector ) -> RuntimeEvidence | None: try: metadata = path.lstat() except OSError: return None if not stat.S_ISREG(metadata.st_mode): return None flags = ( os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) ) try: descriptor = os.open(path, flags) except OSError: return None try: if not stat.S_ISREG(os.fstat(descriptor).st_mode): return None return inspector(descriptor) except OSError: return None finally: os.close(descriptor) def compatibility( pytorch_bridge: Path, pytorch_aten: Path, vllm_python: Path, vllm_extension: Path, ) -> dict[str, object]: evidence = { "pytorch_bridge_identity": _inspect_regular_path( pytorch_bridge, elfmarkers._inspect_pytorch_descriptor ), "pytorch_aten_identity": _inspect_regular_path( pytorch_aten, elfmarkers._inspect_pytorch_descriptor ), "vllm_python_identity": _inspect_regular_path( vllm_python, elfmarkers._inspect_vllm_python_descriptor ), "vllm_extension_identity": _inspect_regular_path( vllm_extension, elfmarkers._inspect_vllm_extension_descriptor ), } expected = { "pytorch_bridge_identity": PYTORCH_BRIDGE_EVIDENCE_ID, "pytorch_aten_identity": PYTORCH_ATEN_EVIDENCE_ID, "vllm_python_identity": VLLM_PYTHON_EVIDENCE_ID, "vllm_extension_identity": VLLM_EXTENSION_EVIDENCE_ID, } checks = { role: ( "SUPPORTED" if value is not None and value.evidence_id == expected[role] else "UNSUPPORTED" ) for role, value in evidence.items() } supported = all(value == "SUPPORTED" for value in checks.values()) return { "checks": checks, "limitations": [ "BINARY_IDENTITY_ONLY", "SAFETENSORS_MODEL_NOT_INSPECTED", "LIVE_TOPOLOGY_NOT_PROVEN", "EXECUTION_CONTEXT_NOT_QUALIFIED", "COMPATIBILITY_CHECK_NOT_CONTAINMENT_EVIDENCE", "COMPATIBILITY_CHECK_NOT_ADOPTION_EVIDENCE", ], "result": "SUPPORTED" if supported else "UNSUPPORTED", "schema_version": SCHEMA, "target_profile": PROFILE, "version": __version__, } def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser( description=( "Check exact vLLM and PyTorch binary identities without installing or starting " "Eggcracker." ) ) value.add_argument( "--pytorch-bridge", required=True, help="absolute path to the PyTorch bridge ELF" ) value.add_argument( "--pytorch-aten", required=True, help="absolute path to the PyTorch ATen ELF" ) value.add_argument( "--vllm-python", required=True, help="absolute path to the vLLM CPython ELF" ) value.add_argument( "--vllm-extension", required=True, help="absolute path to the vLLM extension ELF" ) return value def main(argv: Sequence[str] | None = None) -> int: argument_parser = parser() args = argument_parser.parse_args(argv) result = compatibility( _absolute_path(argument_parser, args.pytorch_bridge), _absolute_path(argument_parser, args.pytorch_aten), _absolute_path(argument_parser, args.vllm_python), _absolute_path(argument_parser, args.vllm_extension), ) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if result["result"] == "SUPPORTED" else 2 if __name__ == "__main__": raise SystemExit(main())