"""Fail-closed, local-files-only Qwen3.8-27B vLLM launcher. The launcher does not import vLLM (which can initialize an accelerator). It inspects package metadata and local source files, verifies a pinned checkpoint attestation, and then selects AR, native one-token MTP, or an explicitly gated DFlash2 canary. It never downloads a model and never invokes a shell. """ from __future__ import annotations import argparse import hashlib import importlib.metadata import json import os import re import shutil import subprocess import sys from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping, Sequence TARGET_MODEL_ID = "Qwen/Qwen3.8-27B" TARGET_ARCHITECTURE = "Qwen3_5ForConditionalGeneration" DFLASH2_MODEL_ID = "incoai/Qwen3.8-27B-DFlash2" DFLASH2_ARCHITECTURE = "DFlash2DraftModel" PINNED_REVISION_RE = re.compile(r"^[0-9a-fA-F]{40}$") SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$") GPU_ARCH_RE = re.compile(r"\bgfx[0-9a-f]+\b", re.IGNORECASE) MAX_JSON_BYTES = 16 * 1024 * 1024 class LauncherError(RuntimeError): """Raised when a runtime invariant cannot be established safely.""" def _is_true(value: str | None) -> bool: return str(value or "").strip().lower() in {"1", "true", "yes", "on"} def _require_pinned_revision(value: str, label: str) -> str: value = value.strip() if not PINNED_REVISION_RE.fullmatch(value): raise LauncherError(f"{label} must be an immutable 40-character commit SHA") return value.lower() def _require_sha256(value: str, label: str) -> str: value = value.strip() if not SHA256_RE.fullmatch(value): raise LauncherError(f"{label} must be a 64-character SHA-256 digest") return value.lower() def _read_json(path: Path, *, label: str) -> dict[str, Any]: try: stat = path.stat() except OSError as exc: raise LauncherError(f"{label} is not readable: {path}") from exc if not path.is_file() or stat.st_size <= 0 or stat.st_size > MAX_JSON_BYTES: raise LauncherError(f"{label} must be a non-empty JSON file below {MAX_JSON_BYTES} bytes") try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise LauncherError(f"{label} is not valid UTF-8 JSON: {path}") from exc if not isinstance(value, dict): raise LauncherError(f"{label} must contain a JSON object") return value def _sha256_file(path: Path) -> str: digest = hashlib.sha256() try: with path.open("rb") as handle: for chunk in iter(lambda: handle.read(4 * 1024 * 1024), b""): digest.update(chunk) except OSError as exc: raise LauncherError(f"cannot hash local file: {path}") from exc return digest.hexdigest() def _local_root(path: Path, *, label: str) -> Path: raw = str(path) if "://" in raw or raw.startswith(("hf:", "s3:", "gs:")): raise LauncherError(f"{label} must be a local directory, not a URL") if not path.is_absolute(): raise LauncherError(f"{label} must be an absolute local path") try: resolved = path.resolve(strict=True) except OSError as exc: raise LauncherError(f"{label} does not exist: {path}") from exc if not resolved.is_dir(): raise LauncherError(f"{label} is not a directory: {resolved}") return resolved def _child_file(root: Path, relative: str, *, label: str) -> Path: relative_path = Path(relative) if relative_path.is_absolute() or ".." in relative_path.parts: raise LauncherError(f"{label} must be a safe path relative to the checkpoint") try: candidate = (root / relative_path).resolve(strict=True) except OSError as exc: raise LauncherError(f"{label} is missing: {relative}") from exc try: candidate.relative_to(root) except ValueError as exc: raise LauncherError(f"{label} escapes the checkpoint root: {relative}") from exc if not candidate.is_file() or candidate.stat().st_size <= 0: raise LauncherError(f"{label} is not a non-empty local file: {relative}") return candidate @dataclass(frozen=True) class CheckpointInspection: root: Path model_id: str revision: str config_sha256: str weight_index_sha256: str shard_count: int tensor_count: int native_mtp_head: bool mtp_reason: str max_model_len: int def public_dict(self) -> dict[str, Any]: return { "root": str(self.root), "model_id": self.model_id, "revision": self.revision, "config_sha256": self.config_sha256, "weight_index_sha256": self.weight_index_sha256, "shard_count": self.shard_count, "tensor_count": self.tensor_count, "native_mtp_head": self.native_mtp_head, "mtp_reason": self.mtp_reason, "max_model_len": self.max_model_len, } def build_checkpoint_attestation( model_path: Path, *, model_id: str, revision: str, ) -> dict[str, Any]: """Build a small local metadata attestation without loading weight data. This is an operator convenience, not a download provenance oracle. The caller must already have obtained and verified the immutable revision. """ root = _local_root(model_path, label="Qwen model path") if model_id != TARGET_MODEL_ID: raise LauncherError(f"MODEL_ID must be exactly {TARGET_MODEL_ID!r}") revision = _require_pinned_revision(revision, "MODEL_REVISION") config_path = _child_file(root, "config.json", label="target config") config = _read_json(config_path, label="target config") text_config = config.get("text_config") if config.get("architectures") != [TARGET_ARCHITECTURE] or config.get("model_type") != "qwen3_5": raise LauncherError("checkpoint is not the exact Qwen3.8-27B architecture") if not isinstance(text_config, dict): raise LauncherError("target config has no text_config object") exact_shape = { "model_type": "qwen3_5_text", "hidden_size": 5120, "num_hidden_layers": 64, "vocab_size": 248320, } for key, expected in exact_shape.items(): if text_config.get(key) != expected: raise LauncherError(f"target text_config.{key} is not the Qwen3.8-27B value {expected!r}") index_path = _child_file(root, "model.safetensors.index.json", label="target weight index") index = _read_json(index_path, label="target weight index") weight_map = index.get("weight_map") if not isinstance(weight_map, dict) or not weight_map: raise LauncherError("target weight index has no weight_map") shard_names = {str(value) for value in weight_map.values()} for shard in sorted(shard_names): _child_file(root, shard, label="target weight shard") return { "schema_version": "1", "model_id": model_id, "revision": revision, "config_sha256": _sha256_file(config_path), "weight_index_path": "model.safetensors.index.json", "weight_index_sha256": _sha256_file(index_path), "expected_shard_count": len(shard_names), } def inspect_target_checkpoint( model_path: Path, *, model_id: str, revision: str, manifest_path: Path | None = None, expected_manifest_sha256: str | None = None, ) -> CheckpointInspection: """Validate the exact local target without loading model weights.""" root = _local_root(model_path, label="Qwen model path") if model_id != TARGET_MODEL_ID: raise LauncherError(f"MODEL_ID must be exactly {TARGET_MODEL_ID!r}") revision = _require_pinned_revision(revision, "MODEL_REVISION") config_path = _child_file(root, "config.json", label="target config") config = _read_json(config_path, label="target config") architectures = config.get("architectures") text_config = config.get("text_config") if architectures != [TARGET_ARCHITECTURE] or config.get("model_type") != "qwen3_5": raise LauncherError("checkpoint is not the exact Qwen3.8-27B architecture") if not isinstance(text_config, dict): raise LauncherError("target config has no text_config object") exact_shape = { "model_type": "qwen3_5_text", "hidden_size": 5120, "num_hidden_layers": 64, "vocab_size": 248320, } for key, expected in exact_shape.items(): if text_config.get(key) != expected: raise LauncherError(f"target text_config.{key} is not the Qwen3.8-27B value {expected!r}") max_model_len = text_config.get("max_position_embeddings") if not isinstance(max_model_len, int) or max_model_len <= 0: raise LauncherError("target max_position_embeddings is invalid") manifest_path = manifest_path or root / ".omni-checkpoint.json" try: manifest_resolved = manifest_path.resolve(strict=True) except OSError as exc: raise LauncherError( "exact checkpoint attestation is missing; expected .omni-checkpoint.json" ) from exc manifest = _read_json(manifest_resolved, label="checkpoint attestation") if expected_manifest_sha256 is not None: expected_manifest_sha = _require_sha256( expected_manifest_sha256, "QWEN_CHECKPOINT_MANIFEST_SHA256" ) if _sha256_file(manifest_resolved) != expected_manifest_sha: raise LauncherError("checkpoint attestation SHA-256 does not match its external pin") if manifest.get("schema_version") != "1": raise LauncherError("checkpoint attestation schema_version must be '1'") if manifest.get("model_id") != model_id: raise LauncherError("checkpoint attestation model_id does not match MODEL_ID") manifest_revision = _require_pinned_revision( str(manifest.get("revision", "")), "checkpoint attestation revision" ) if manifest_revision != revision: raise LauncherError("checkpoint attestation revision does not match MODEL_REVISION") config_sha = _sha256_file(config_path) expected_config_sha = _require_sha256( str(manifest.get("config_sha256", "")), "checkpoint config_sha256" ) if config_sha != expected_config_sha: raise LauncherError("checkpoint config.json SHA-256 does not match its attestation") index_relative = str(manifest.get("weight_index_path", "model.safetensors.index.json")) index_path = _child_file(root, index_relative, label="target weight index") index_sha = _sha256_file(index_path) expected_index_sha = _require_sha256( str(manifest.get("weight_index_sha256", "")), "checkpoint weight_index_sha256" ) if index_sha != expected_index_sha: raise LauncherError("checkpoint weight index SHA-256 does not match its attestation") index = _read_json(index_path, label="target weight index") weight_map = index.get("weight_map") if not isinstance(weight_map, dict) or not weight_map: raise LauncherError("target weight index has no weight_map") shard_names = {str(value) for value in weight_map.values()} for shard in sorted(shard_names): _child_file(root, shard, label="target weight shard") expected_shards = manifest.get("expected_shard_count") if not isinstance(expected_shards, int) or expected_shards <= 0: raise LauncherError("checkpoint attestation expected_shard_count must be positive") if len(shard_names) != expected_shards: raise LauncherError("checkpoint shard count does not match its attestation") tensor_names = {str(name) for name in weight_map} mtp_layers = int(text_config.get("mtp_num_hidden_layers", 0) or 0) required_mtp_groups = ( any(name.startswith("mtp.layers.0.") for name in tensor_names), any(name.startswith("mtp.fc.") for name in tensor_names), any(name.startswith("mtp.norm.") for name in tensor_names), ) native_mtp = mtp_layers == 1 and all(required_mtp_groups) if mtp_layers != 1: mtp_reason = "text_config.mtp_num_hidden_layers is not exactly 1" elif not all(required_mtp_groups): mtp_reason = "weight index does not contain the complete mtp.layers/fc/norm groups" else: mtp_reason = "one native MTP layer and its indexed weight groups are present" return CheckpointInspection( root=root, model_id=model_id, revision=revision, config_sha256=config_sha, weight_index_sha256=index_sha, shard_count=len(shard_names), tensor_count=len(weight_map), native_mtp_head=native_mtp, mtp_reason=mtp_reason, max_model_len=max_model_len, ) @dataclass(frozen=True) class VllmCapabilities: installed: bool version: str = "" source_commit: str = "" package_root: Path | None = None qwen35_ar: bool = False native_mtp: bool = False dflash2: bool = False reasons: tuple[str, ...] = () def public_dict(self) -> dict[str, Any]: return { "installed": self.installed, "version": self.version, "source_commit": self.source_commit or None, "package_root": str(self.package_root) if self.package_root else None, "qwen35_ar": self.qwen35_ar, "native_mtp": self.native_mtp, "dflash2": self.dflash2, "reasons": list(self.reasons), } def _read_small_source(path: Path) -> str: try: if not path.is_file() or path.stat().st_size > 4 * 1024 * 1024: return "" return path.read_text(encoding="utf-8", errors="replace") except OSError: return "" def _version_tuple(value: str) -> tuple[int, int, int]: match = re.match(r"^(\d+)\.(\d+)\.(\d+)", value) if not match: return (0, 0, 0) return tuple(int(part) for part in match.groups()) def _git_commit_from_distribution(distribution: importlib.metadata.Distribution) -> str: try: direct_url_text = distribution.read_text("direct_url.json") if direct_url_text: direct_url = json.loads(direct_url_text) commit = str(direct_url.get("vcs_info", {}).get("commit_id", "")) if PINNED_REVISION_RE.fullmatch(commit): return commit.lower() except (OSError, json.JSONDecodeError, AttributeError, TypeError): pass return "" def probe_installed_vllm() -> VllmCapabilities: """Inspect installed vLLM metadata/source without importing it.""" try: distribution = importlib.metadata.distribution("vllm") except importlib.metadata.PackageNotFoundError: return VllmCapabilities(installed=False, reasons=("vLLM distribution is not installed",)) version = distribution.version try: root = Path(distribution.locate_file("vllm")).resolve(strict=True) except OSError: return VllmCapabilities( installed=True, version=version, reasons=("installed vLLM package root is unreadable",) ) commit = _git_commit_from_distribution(distribution) api_server = root / "entrypoints" / "openai" / "api_server.py" qwen_target = root / "model_executor" / "models" / "qwen3_5.py" qwen_mtp = root / "model_executor" / "models" / "qwen3_5_mtp.py" speculative = root / "config" / "speculative.py" registry = root / "model_executor" / "models" / "registry.py" dflash_model = root / "model_executor" / "models" / "qwen3_dflash2.py" dflash_worker = root / "v1" / "worker" / "gpu" / "spec_decode" / "dflash2" / "speculator.py" ar_ok = bool(_read_small_source(api_server)) and "Qwen3_5" in _read_small_source(qwen_target) mtp_source = _read_small_source(qwen_mtp) speculative_source = _read_small_source(speculative).lower() mtp_ok = ( ar_ok and _version_tuple(version) >= (0, 18, 0) and "qwen3_5" in mtp_source.lower() and "mtp" in mtp_source.lower() and "mtp" in speculative_source and "num_speculative_tokens" in speculative_source ) registry_source = _read_small_source(registry) dflash_ok = ( ar_ok and "DFlash2DraftModel" in registry_source and "DFlash2" in _read_small_source(dflash_model) and "DFlash2" in _read_small_source(dflash_worker) ) reasons: list[str] = [] if not ar_ok: reasons.append("Qwen3.5 target model or OpenAI server source is absent") if not mtp_ok: reasons.append("native Qwen3.5 MTP source/config support was not proven") if not dflash_ok: reasons.append("DFlash2 model/worker/registry support was not proven") if not commit: reasons.append("installed vLLM source commit is not available from wheel metadata") return VllmCapabilities( installed=True, version=version, source_commit=commit, package_root=root, qwen35_ar=ar_ok, native_mtp=mtp_ok, dflash2=dflash_ok, reasons=tuple(reasons), ) @dataclass(frozen=True) class LauncherSettings: model_path: Path model_id: str model_revision: str checkpoint_manifest: Path checkpoint_manifest_sha256: str = "" requested_mode: str = "auto" mtp_tokens: int = 1 bind_host: str = "127.0.0.1" port: int = 8103 served_model_name: str = "qwen3.8-27b" api_key: str = "omni-local-only" max_model_len: int = 65536 gpu_memory_utilization: float = 0.60 dtype: str = "float16" required_gpu_arch: str = "" hardware_profile: str = "mi300x_single" dflash2_canary: bool = False dflash2_path: Path | None = None dflash2_revision: str = "" dflash2_manifest_sha256: str = "" dflash2_vllm_commit: str = "" dflash2_gate_path: Path | None = None dflash2_gate_sha256: str = "" dflash2_min_p95_improvement_percent: float = 5.0 @classmethod def from_env(cls, environment: Mapping[str, str] | None = None) -> "LauncherSettings": env = dict(os.environ if environment is None else environment) model_path = Path(env.get("MODEL_PATH", "/models/qwen")) manifest = Path(env.get("QWEN_CHECKPOINT_MANIFEST_PATH", str(model_path / ".omni-checkpoint.json"))) mode = env.get("INFERENCE_MODE", "auto").strip().lower() aliases = {"autoregressive": "ar", "native_mtp": "mtp", "dflash2": "dflash2_canary"} mode = aliases.get(mode, mode) if mode not in {"auto", "ar", "mtp", "dflash2_canary"}: raise LauncherError("INFERENCE_MODE must be auto, autoregressive, native_mtp, or dflash2_canary") try: port = int(env.get("QWEN_PORT", "8103")) mtp_tokens = int(env.get("NATIVE_MTP_TOKENS", "1")) max_model_len = int(env.get("QWEN_MAX_MODEL_LEN", "65536")) gpu_util = float(env.get("QWEN_GPU_MEMORY_UTILIZATION", "0.60")) min_improvement = float(env.get("QWEN_DFLASH2_MIN_P95_IMPROVEMENT_PERCENT", "5.0")) except ValueError as exc: raise LauncherError("numeric Qwen launcher settings are invalid") from exc if not 1 <= port <= 65535: raise LauncherError("QWEN_PORT is outside 1..65535") if mtp_tokens != 1: raise LauncherError("native MTP is deliberately fixed to exactly one speculative token") if not 1024 <= max_model_len <= 262144: raise LauncherError("QWEN_MAX_MODEL_LEN is outside 1024..262144") if not 0.20 <= gpu_util <= 0.90: raise LauncherError("QWEN_GPU_MEMORY_UTILIZATION is outside 0.20..0.90") if not 0.0 < min_improvement <= 100.0: raise LauncherError("DFlash2 minimum p95 improvement must be in (0, 100]") dtype = env.get("QWEN_DTYPE", "float16").strip().lower() dtype = {"fp16": "float16", "half": "float16"}.get(dtype, dtype) if dtype != "float16": raise LauncherError("QWEN_DTYPE is fixed to float16 for the Nelloz Qwen profile") bind_host = env.get("QWEN_BIND_HOST", "127.0.0.1").strip() if bind_host == "0.0.0.0": if not _is_true(env.get("QWEN_CONTAINER_INTERNAL_ONLY")): raise LauncherError("wildcard Qwen bind requires QWEN_CONTAINER_INTERNAL_ONLY=1") elif bind_host not in {"127.0.0.1", "::1", "localhost"}: raise LauncherError("Qwen may bind only to loopback or an explicitly internal container network") dflash_path_raw = env.get("DFLASH2_SPECULATOR_PATH", "").strip() gate_path_raw = env.get("QWEN_DFLASH2_GATE_PATH", "").strip() return cls( model_path=model_path, model_id=env.get("MODEL_ID", "").strip(), model_revision=env.get("MODEL_REVISION", "").strip(), checkpoint_manifest=manifest, checkpoint_manifest_sha256=env.get("QWEN_CHECKPOINT_MANIFEST_SHA256", "").strip(), requested_mode=mode, mtp_tokens=mtp_tokens, bind_host=bind_host, port=port, served_model_name=env.get("QWEN_SERVED_MODEL_NAME", "qwen3.8-27b").strip(), api_key=env.get("QWEN_API_KEY", "omni-local-only"), max_model_len=max_model_len, gpu_memory_utilization=gpu_util, dtype=dtype, required_gpu_arch=env.get("QWEN_REQUIRE_GPU_ARCH", "").strip().lower(), hardware_profile=env.get("OMNI_HARDWARE_PROFILE", "mi300x_single").strip(), dflash2_canary=_is_true(env.get("QWEN_DFLASH2_CANARY")), dflash2_path=Path(dflash_path_raw) if dflash_path_raw else None, dflash2_revision=env.get("DFLASH2_SPECULATOR_REVISION", "").strip(), dflash2_manifest_sha256=env.get("DFLASH2_MANIFEST_SHA256", "").strip(), dflash2_vllm_commit=env.get("DFLASH2_VLLM_COMMIT", "").strip(), dflash2_gate_path=Path(gate_path_raw) if gate_path_raw else None, dflash2_gate_sha256=env.get("QWEN_DFLASH2_GATE_SHA256", "").strip(), dflash2_min_p95_improvement_percent=min_improvement, ) @dataclass(frozen=True) class Decision: mode: str reasons: tuple[str, ...] = () speculative_config: dict[str, Any] | None = None def public_dict(self) -> dict[str, Any]: return { "mode": self.mode, "reasons": list(self.reasons), "speculative_config": self.speculative_config, } def detect_rocm_architectures() -> tuple[str, ...]: """Read ROCm's local hardware report; never imports torch or opens a socket.""" command = shutil.which("rocminfo") if not command and Path("/opt/rocm/bin/rocminfo").is_file(): command = "/opt/rocm/bin/rocminfo" if not command: return () try: completed = subprocess.run( [command], check=False, capture_output=True, text=True, timeout=8, env={**os.environ, "ROCR_VISIBLE_DEVICES": os.environ.get("ROCR_VISIBLE_DEVICES", "0")}, ) except (OSError, subprocess.SubprocessError): return () combined = (completed.stdout + "\n" + completed.stderr)[:8_000_000] return tuple(sorted({match.lower() for match in GPU_ARCH_RE.findall(combined)})) def _verify_speculator(settings: LauncherSettings, target: CheckpointInspection) -> Path: if settings.dflash2_path is None: raise LauncherError("DFLASH2_SPECULATOR_PATH is required") root = _local_root(settings.dflash2_path, label="DFlash2 speculator path") revision = _require_pinned_revision(settings.dflash2_revision, "DFlash2 speculator revision") config_path = _child_file(root, "config.json", label="DFlash2 config") config = _read_json(config_path, label="DFlash2 config") if config.get("architectures") != [DFLASH2_ARCHITECTURE]: raise LauncherError("DFlash2 config does not declare DFlash2DraftModel") if ( config.get("hidden_size") != 5120 or config.get("num_target_layers") != 64 or config.get("num_hidden_layers") != 5 or config.get("vocab_size") != 248320 ): raise LauncherError("DFlash2 config is not shaped for the exact Qwen3.8-27B target") dflash_config = config.get("dflash_config") if not isinstance(dflash_config, dict) or { "block_size": dflash_config.get("block_size"), "conv_group_size": dflash_config.get("conv_group_size"), "conv_kernel_size": dflash_config.get("conv_kernel_size"), "selector_rank": dflash_config.get("selector_rank"), "selector_top_k": dflash_config.get("selector_top_k"), "target_layer_ids": dflash_config.get("target_layer_ids"), } != { "block_size": 8, "conv_group_size": 16, "conv_kernel_size": 2, "selector_rank": 256, "selector_top_k": 16, "target_layer_ids": [5, 19, 33, 47, 61], }: raise LauncherError("DFlash2 config does not match the exact published drafter shape") manifest_path = root / ".omni-speculator.json" expected_manifest_sha = _require_sha256( settings.dflash2_manifest_sha256, "DFLASH2_MANIFEST_SHA256" ) if _sha256_file(manifest_path) != expected_manifest_sha: raise LauncherError("DFlash2 attestation file SHA-256 does not match the pinned value") manifest = _read_json(manifest_path, label="DFlash2 attestation") if manifest.get("schema_version") != "1" or manifest.get("model_id") != DFLASH2_MODEL_ID: raise LauncherError("DFlash2 attestation has the wrong schema or exact model_id") if _require_pinned_revision(str(manifest.get("revision", "")), "DFlash2 attestation revision") != revision: raise LauncherError("DFlash2 attestation revision does not match its pin") target_block = manifest.get("target") if not isinstance(target_block, dict) or target_block != { "model_id": target.model_id, "revision": target.revision, }: raise LauncherError("DFlash2 attestation is not bound to the exact target checkpoint") config_sha = _require_sha256(str(manifest.get("config_sha256", "")), "DFlash2 config_sha256") if _sha256_file(config_path) != config_sha: raise LauncherError("DFlash2 config SHA-256 does not match its attestation") weight_files = manifest.get("weight_files") if not isinstance(weight_files, list) or len(weight_files) != 1: raise LauncherError("DFlash2 attestation must list its one exact weight file") for item in weight_files: if not isinstance(item, dict): raise LauncherError("DFlash2 weight_files entries must be objects") if item.get("path") != "model.safetensors": raise LauncherError("DFlash2 weight file must be exactly model.safetensors") weight_path = _child_file(root, str(item.get("path", "")), label="DFlash2 weight file") expected_size = item.get("size") if not isinstance(expected_size, int) or weight_path.stat().st_size != expected_size: raise LauncherError("DFlash2 weight file size does not match its attestation") expected_sha = _require_sha256(str(item.get("sha256", "")), "DFlash2 weight SHA-256") if _sha256_file(weight_path) != expected_sha: raise LauncherError("DFlash2 weight SHA-256 does not match its attestation") return root def _verify_dflash2_gate( settings: LauncherSettings, target: CheckpointInspection, vllm: VllmCapabilities, hardware_arches: Sequence[str], ) -> dict[str, Any]: if not settings.dflash2_canary: raise LauncherError("QWEN_DFLASH2_CANARY is not explicitly enabled") if settings.hardware_profile != "mi300x_single" or "gfx942" not in hardware_arches: raise LauncherError("DFlash2 canary gate was not measured on the active MI300X/gfx942") if not vllm.dflash2: raise LauncherError("installed vLLM does not prove DFlash2 model and V2 worker support") required_commit = _require_pinned_revision(settings.dflash2_vllm_commit, "DFLASH2_VLLM_COMMIT") if not vllm.source_commit or vllm.source_commit != required_commit: raise LauncherError("installed vLLM commit does not equal the DFlash2 commit pin") speculator_root = _verify_speculator(settings, target) if settings.dflash2_gate_path is None: raise LauncherError("QWEN_DFLASH2_GATE_PATH is required") expected_gate_sha = _require_sha256(settings.dflash2_gate_sha256, "QWEN_DFLASH2_GATE_SHA256") if _sha256_file(settings.dflash2_gate_path) != expected_gate_sha: raise LauncherError("DFlash2 parity/benchmark gate SHA-256 does not match its pin") gate = _read_json(settings.dflash2_gate_path, label="DFlash2 parity/benchmark gate") exact_binding = { "target_model_id": target.model_id, "target_revision": target.revision, "speculator_model_id": DFLASH2_MODEL_ID, "speculator_revision": settings.dflash2_revision.lower(), "vllm_commit": required_commit, "hardware_profile": "mi300x_single", "gpu_arch": "gfx942", } if gate.get("schema_version") != "1": raise LauncherError("DFlash2 gate schema_version must be '1'") for key, expected in exact_binding.items(): actual = str(gate.get(key, "")).lower() if key.endswith(("revision", "commit")) else gate.get(key) if actual != expected: raise LauncherError(f"DFlash2 gate {key} is not bound to the active runtime") parity = gate.get("parity") if not isinstance(parity, dict) or parity.get("passed") is not True: raise LauncherError("DFlash2 AR parity gate has not passed") if parity.get("covers") != ["greedy", "sampling", "tool_json", "stop_and_cancel"]: raise LauncherError("DFlash2 parity gate does not cover all required quality cases") _require_sha256(str(parity.get("corpus_sha256", "")), "DFlash2 parity corpus_sha256") _require_sha256(str(parity.get("result_sha256", "")), "DFlash2 parity result_sha256") benchmark = gate.get("benchmark") if not isinstance(benchmark, dict) or benchmark.get("passed") is not True: raise LauncherError("DFlash2 target-hardware benchmark gate has not passed") concurrency = benchmark.get("concurrency") if not isinstance(concurrency, list) or not {1, 2}.issubset(set(concurrency)): raise LauncherError("DFlash2 benchmark must cover concurrency 1 and 2") samples = benchmark.get("samples_per_mode") ar_p95 = benchmark.get("ar_p95_ms") draft_p95 = benchmark.get("dflash2_p95_ms") if not isinstance(samples, int) or samples < 100: raise LauncherError("DFlash2 benchmark needs at least 100 samples per mode") if not isinstance(ar_p95, (int, float)) or not isinstance(draft_p95, (int, float)): raise LauncherError("DFlash2 benchmark p95 values are invalid") if ar_p95 <= 0 or draft_p95 <= 0 or draft_p95 >= ar_p95: raise LauncherError("DFlash2 did not beat AR p95 latency") improvement = (float(ar_p95) - float(draft_p95)) / float(ar_p95) * 100.0 if improvement < settings.dflash2_min_p95_improvement_percent: raise LauncherError("DFlash2 p95 improvement is below the configured canary threshold") return { "method": "dflash", "model": str(speculator_root), "num_speculative_tokens": 7, } def select_mode( target: CheckpointInspection, vllm: VllmCapabilities, settings: LauncherSettings, *, hardware_arches: Sequence[str] = (), ) -> Decision: """Choose a lossless mode; failed speculation checks always fall back.""" if not vllm.installed or not vllm.qwen35_ar: raise LauncherError("installed vLLM cannot serve the exact Qwen3.8 target even in AR mode") normalized_arches = tuple(str(item).lower() for item in hardware_arches) if settings.required_gpu_arch and settings.required_gpu_arch not in normalized_arches: raise LauncherError( f"required GPU architecture {settings.required_gpu_arch!r} was not found locally" ) reasons: list[str] = [] mtp_ready = target.native_mtp_head and vllm.native_mtp and settings.mtp_tokens == 1 if not target.native_mtp_head: reasons.append(f"native MTP disabled: {target.mtp_reason}") if not vllm.native_mtp: reasons.append("native MTP disabled: installed vLLM capability probe failed") if settings.requested_mode == "ar": return Decision(mode="autoregressive", reasons=("AR was explicitly requested",)) if settings.requested_mode == "dflash2_canary": try: speculative = _verify_dflash2_gate(settings, target, vllm, normalized_arches) except LauncherError as exc: reasons.append(f"DFlash2 canary disabled: {exc}") else: return Decision( mode="dflash2_canary", reasons=("explicit exact DFlash2 parity and target-hardware benchmark gates passed",), speculative_config=speculative, ) if mtp_ready: reasons.append("exact native MTP head and installed vLLM support were proven") return Decision( mode="native_mtp", reasons=tuple(reasons), speculative_config={"method": "mtp", "num_speculative_tokens": 1}, ) reasons.append("using the quality-baseline autoregressive fallback") return Decision(mode="autoregressive", reasons=tuple(reasons)) def build_server_argv( settings: LauncherSettings, target: CheckpointInspection, decision: Decision, ) -> list[str]: if settings.max_model_len > target.max_model_len: raise LauncherError("QWEN_MAX_MODEL_LEN exceeds the checkpoint's declared limit") if not settings.served_model_name or len(settings.served_model_name) > 128: raise LauncherError("QWEN_SERVED_MODEL_NAME is invalid") if not settings.api_key or len(settings.api_key) > 512: raise LauncherError("QWEN_API_KEY must be a non-empty local API key") argv = [ sys.executable, "-m", "vllm.entrypoints.openai.api_server", "--host", settings.bind_host, "--port", str(settings.port), "--model", str(target.root), "--served-model-name", settings.served_model_name, "--dtype", settings.dtype, "--tensor-parallel-size", "1", "--gpu-memory-utilization", f"{settings.gpu_memory_utilization:.3f}", "--max-model-len", str(settings.max_model_len), "--enable-prefix-caching", "--enable-chunked-prefill", "--api-key", settings.api_key, ] if decision.speculative_config is not None: argv.extend( [ "--speculative-config", json.dumps(decision.speculative_config, separators=(",", ":"), sort_keys=True), ] ) return argv def _redacted_argv(argv: Sequence[str]) -> list[str]: result = list(argv) for index, value in enumerate(result[:-1]): if value == "--api-key": result[index + 1] = "" return result def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--inspect-only", action="store_true", help="validate and print the decision without starting vLLM or loading a model", ) return parser.parse_args(argv) def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) for key in ("HF_HUB_OFFLINE", "TRANSFORMERS_OFFLINE", "HF_DATASETS_OFFLINE"): os.environ[key] = "1" os.environ["VLLM_USE_MODELSCOPE"] = "0" os.environ["VLLM_NO_USAGE_STATS"] = "1" os.environ["DO_NOT_TRACK"] = "1" try: settings = LauncherSettings.from_env() target = inspect_target_checkpoint( settings.model_path, model_id=settings.model_id, revision=settings.model_revision, manifest_path=settings.checkpoint_manifest, expected_manifest_sha256=settings.checkpoint_manifest_sha256, ) vllm = probe_installed_vllm() hardware = detect_rocm_architectures() if settings.required_gpu_arch or settings.requested_mode == "dflash2_canary" else () decision = select_mode(target, vllm, settings, hardware_arches=hardware) server_argv = build_server_argv(settings, target, decision) except LauncherError as exc: print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False), file=sys.stderr) return 2 report = { "ok": True, "offline_only": True, "will_start_model": not args.inspect_only, "checkpoint": target.public_dict(), "vllm": vllm.public_dict(), "hardware_arches": list(hardware), "decision": decision.public_dict(), "server_argv": _redacted_argv(server_argv), } print(json.dumps(report, ensure_ascii=False, sort_keys=True), flush=True) if args.inspect_only: return 0 os.execv(sys.executable, server_argv) return 127 if __name__ == "__main__": raise SystemExit(main())