#!/usr/bin/env python3
"""Offline, read-only validation for the OMNI deployment contracts.

This script parses local JSON/YAML and checks deployment invariants. It never
opens a socket, starts a process, imports a model runtime, or writes a file.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any

try:
    import yaml
except ImportError as exc:  # pragma: no cover - explicit operator guidance
    raise SystemExit("PyYAML is required for offline YAML validation") from exc


EXPECTED_PORTS = {
    8006,
    8007,
    22400,
    22500,
    8100,
    8101,
    8102,
    8103,
    8110,
    8111,
    8642,
    8765,
    8766,
    30010,
    30011,
    30020,
}

MODEL_SERVICES = {
    "minicpm-worker",
    "asr",
    "qwen",
    "h3-controller",
    "h3-worker",
    "hunyuan",
}


class Checks:
    def __init__(self) -> None:
        self.passed: list[str] = []
        self.failed: list[str] = []

    def require(self, condition: bool, message: str) -> None:
        (self.passed if condition else self.failed).append(message)

    def equal(self, actual: Any, expected: Any, message: str) -> None:
        self.require(actual == expected, f"{message} (got {actual!r}, expected {expected!r})")


def load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        value = json.load(handle)
    if not isinstance(value, dict):
        raise ValueError(f"{path} must contain a JSON object")
    return value


def load_yaml(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        value = yaml.safe_load(handle)
    if not isinstance(value, dict):
        raise ValueError(f"{path} must contain a YAML mapping")
    return value


def environment_map(service: dict[str, Any]) -> dict[str, str]:
    value = service.get("environment", {})
    if isinstance(value, dict):
        return {str(key): str(item) for key, item in value.items()}
    if isinstance(value, list):
        result: dict[str, str] = {}
        for item in value:
            key, _, val = str(item).partition("=")
            result[key] = val
        return result
    return {}


def has_read_only_model_mount(service: dict[str, Any]) -> bool:
    for volume in service.get("volumes", []) or []:
        if isinstance(volume, dict):
            target = str(volume.get("target", ""))
            if target.startswith("/models/") and volume.get("read_only") is True:
                return True
        elif isinstance(volume, str):
            fields = volume.rsplit(":", 2)
            if "/models/" in volume and fields[-1] == "ro":
                return True
    return False


def validate_schemas(root: Path, checks: Checks) -> None:
    contract_dir = root / "contracts"
    names = {
        "event-envelope.schema.json",
        "route.schema.json",
        "media-job.schema.json",
    }
    checks.equal(
        {path.name for path in contract_dir.glob("*.schema.json")},
        names,
        "the contract directory contains the three expected schemas",
    )
    for name in sorted(names):
        schema = load_json(contract_dir / name)
        checks.equal(
            schema.get("$schema"),
            "https://json-schema.org/draft/2020-12/schema",
            f"{name} uses JSON Schema 2020-12",
        )
        checks.require(bool(schema.get("$id")), f"{name} has a stable $id")
        checks.require(bool(schema.get("title")), f"{name} has a title")

    event = load_json(contract_dir / "event-envelope.schema.json")
    required = set(event.get("required", []))
    checks.require(
        {
            "protocol_version",
            "type",
            "event_id",
            "seq",
            "timestamp",
            "session_id",
            "turn_id",
            "operation_id",
            "trace_id",
            "source",
            "payload",
        }
        <= required,
        "event envelope requires correlation, ordering and payload fields",
    )
    checks.equal(
        event.get("properties", {}).get("protocol_version", {}).get("const"),
        "1.0",
        "event protocol version is pinned",
    )


def validate_pipeline(root: Path, checks: Checks) -> dict[str, Any]:
    pipeline = load_yaml(root / "config" / "pipeline.example.yaml")
    deployment = pipeline.get("deployment", {})
    checks.equal(deployment.get("only_public_port"), 8006, "only public application port is 8006")
    checks.equal(deployment.get("auto_start_models"), False, "model auto-start is disabled")
    checks.equal(deployment.get("auto_download_models"), False, "model auto-download is disabled")
    checks.equal(deployment.get("local_files_only"), True, "deployment is local-files-only")

    network_services = pipeline.get("network", {}).get("services", {})
    ports = [definition.get("port") for definition in network_services.values()]
    checks.equal(set(ports), EXPECTED_PORTS, "port plan matches the target topology")
    checks.equal(len(ports), len(set(ports)), "every topology port is unique")
    public_ports = {
        definition.get("port")
        for definition in network_services.values()
        if definition.get("public") is True
    }
    checks.equal(public_ports, {8006}, "only 8006 is marked public")

    hardware = pipeline.get("hardware", {})
    profiles = hardware.get("profiles", {})
    mi300x = profiles.get("mi300x_single", {})
    checks.equal(mi300x.get("status"), "recommended", "single MI300X is the recommended profile")
    checks.equal(mi300x.get("device_count"), 1, "MI300X profile describes one device")
    checks.equal(mi300x.get("per_device_memory_gib"), 192, "MI300X memory is reported per device")
    checks.equal(mi300x.get("preferred_precision"), "bf16", "MI300X prefers BF16")
    checks.equal(
        mi300x.get("placement", {}).get("do_not_overcommit_from_nominal_capacity"),
        True,
        "MI300X colocation requires a live memory probe",
    )
    mi50 = profiles.get("mi50_12x16_legacy", {})
    checks.equal(mi50.get("status"), "legacy_unsupported_current_rocm", "MI50 profile is marked legacy")
    checks.equal(mi50.get("device_count"), 12, "MI50 profile describes twelve devices")
    checks.equal(mi50.get("per_device_memory_gib"), 16, "MI50 memory is reported per device")
    checks.equal(
        mi50.get("memory_is_single_device_address_space"),
        False,
        "MI50 devices are not presented as one address space",
    )
    checks.equal(
        mi50.get("never_report_summed_memory_as_one_pool"),
        True,
        "MI50 memory is never falsely aggregated",
    )
    checks.require("total_memory_gib" not in mi50, "MI50 profile contains no misleading total-memory field")
    checks.equal(mi50.get("automatic_selection"), False, "MI50 legacy profile is never auto-selected")

    conversation = pipeline.get("conversation", {})
    checks.equal(conversation.get("native_duplex_path"), True, "MiniCPM keeps its native duplex path")
    checks.equal(
        conversation.get("speculative_decoding", {}).get("enabled"),
        False,
        "MiniCPM speculative decoding is disabled",
    )
    checks.equal(
        conversation.get("context", {}).get("strategy"),
        "sliding",
        "MiniCPM uses context sliding",
    )

    qwen = pipeline.get("qwen_inference", {})
    launcher = qwen.get("launcher", {})
    checks.equal(launcher.get("module"), "qwen_runtime.launcher", "Qwen uses the safe runtime launcher")
    checks.equal(launcher.get("shell"), False, "Qwen launcher never uses a shell")
    checks.equal(launcher.get("never_download"), True, "Qwen launcher never downloads")
    checks.equal(
        launcher.get("require_exact_target_model_id"),
        "Qwen/Qwen3.8-27B",
        "Qwen launcher pins the exact target model ID",
    )
    checks.equal(
        launcher.get("bind", {}).get("lan_publication"),
        False,
        "Qwen launcher contract forbids LAN publication",
    )
    identity = qwen.get("exact_identity", {})
    checks.equal(identity.get("require_local_path"), True, "Qwen requires a local path")
    checks.equal(identity.get("require_revision_pin"), True, "Qwen requires a pinned revision")
    modes = qwen.get("modes", {})
    dflash = modes.get("dflash2_exact", {})
    checks.equal(dflash.get("exact_target_required"), True, "DFlash2 requires an exact target")
    checks.equal(dflash.get("capability_gated"), True, "DFlash2 is capability-gated")
    checks.equal(dflash.get("default_enabled"), False, "DFlash2 is off by default")
    checks.equal(dflash.get("lossless_verification"), True, "DFlash2 uses target verification")
    checks.equal(
        dflash.get("require_pinned_40_char_commit"),
        True,
        "DFlash2 requires a pinned vLLM commit",
    )
    checks.equal(
        dflash.get("enable_only_after_parity_test"),
        True,
        "DFlash2 requires an AR parity test",
    )
    dflash_amd = dflash.get("amd", {})
    checks.equal(dflash_amd.get("capability_gated"), True, "DFlash2 on AMD is capability-gated")
    checks.equal(dflash_amd.get("canary_only"), True, "DFlash2 on AMD is canary-only")
    checks.equal(
        dflash_amd.get("automatic_selection"),
        False,
        "DFlash2 is never auto-selected on AMD",
    )
    checks.equal(
        dflash_amd.get("failure_fallback"),
        ["native_mtp", "autoregressive"],
        "DFlash2 on AMD falls back through MTP to AR",
    )
    mtp = modes.get("native_mtp", {})
    checks.equal(mtp.get("require_native_model_mtp_head"), True, "MTP requires a native model head")
    checks.equal(mtp.get("num_speculative_tokens"), 1, "MTP starts with one speculative token")
    checks.equal(
        qwen.get("mode_selection", {}).get("high_concurrency_mode"),
        "autoregressive",
        "high-concurrency Qwen mode is AR",
    )
    checks.equal(
        qwen.get("mode_selection", {}).get("hardware_preferences", {}).get("amd_mi300x"),
        ["native_mtp", "autoregressive"],
        "MI300X prefers native MTP and then AR",
    )

    hermes = pipeline.get("hermes", {})
    checks.equal(
        hermes.get("runtime"),
        "native_windows_api",
        "Hermes is configured as a native Windows API",
    )
    checks.equal(hermes.get("api_base"), "http://127.0.0.1:8642", "Hermes uses loopback port 8642")
    checks.equal(hermes.get("bind_loopback_only"), True, "Hermes is loopback-only")
    hermes_model = hermes.get("official_model_config", {})
    development_model = hermes_model.get("development", {})
    checks.equal(development_model.get("provider"), "openai-codex", "Hermes uses ChatGPT/Codex auth in development")
    checks.equal(
        development_model.get("auth"),
        "chatgpt_device_code_managed_by_hermes",
        "Hermes owns the development device-code login",
    )
    production_model = hermes_model.get("production", {})
    checks.equal(production_model.get("provider"), "openai-api", "Hermes uses the OpenAI API provider in production")
    checks.equal(
        production_model.get("base_url"),
        "https://api.openai.com/v1",
        "Hermes uses the official OpenAI API in production",
    )
    checks.equal(production_model.get("api_key_env"), "OPENAI_API_KEY", "Hermes reads the production key from its secret environment")
    hermes_api = hermes.get("api_server", {})
    checks.equal(hermes_api.get("host"), "127.0.0.1", "Hermes API server remains loopback-only")
    checks.equal(hermes_api.get("bearer_key_env"), "API_SERVER_KEY", "Hermes API uses its separate bearer key")
    toolsets = {item.get("name") for item in hermes.get("toolsets", [])}
    checks.require(
        {"desktop", "browser", "filesystem", "shell_and_code", "internet_search"} <= toolsets,
        "Hermes declares the complete toolset inventory",
    )
    hermes_toolsets = {
        item.get("name"): item for item in hermes.get("toolsets", [])
        if isinstance(item, dict)
    }
    for toolset_name in ("desktop", "browser", "filesystem"):
        checks.equal(
            hermes_toolsets.get(toolset_name, {}).get("transport"),
            "paired_sidecar_mcp",
            f"Hermes {toolset_name} requires the paired sidecar bridge",
        )
    search_tool = next(
        (item for item in hermes.get("toolsets", []) if item.get("name") == "internet_search"),
        {},
    )
    checks.equal(search_tool.get("transport"), "hermes_core", "internet search is a Hermes core tool")
    checks.equal(search_tool.get("tool_name"), "web_search", "Hermes web_search is explicitly named")
    checks.require(
        "no_model_download" in search_tool.get("restrictions", []),
        "internet search cannot download models",
    )

    remote = pipeline.get("remote_compute", {})
    checks.equal(
        remote.get("local_voice_and_video_are_authoritative"),
        True,
        "remote compute keeps voice and video local",
    )
    direct_local = remote.get("direct_local_hermes_provider", {})
    checks.equal(
        set(direct_local.get("capabilities", [])),
        {"hermes.run", "code", "tools", "internet.search"},
        "direct local Hermes advertises only safe capabilities",
    )
    checks.equal(
        set(direct_local.get("permissions", [])),
        {"tool.execute", "network.search"},
        "direct local Hermes has only safe direct permissions",
    )
    forbidden_local = set(direct_local.get("forbidden_without_paired_sidecar", []))
    checks.require(
        {"desktop.observe", "desktop.control", "files.read", "files.write", "webcam.read", "context.vision"}
        <= forbidden_local,
        "privileged local resources require a paired sidecar",
    )
    discovery = remote.get("discovery", {})
    checks.equal(discovery.get("method"), "mdns", "remote nodes use mDNS discovery")
    checks.equal(discovery.get("discovery_is_never_trust"), True, "mDNS discovery grants no trust")
    checks.equal(
        set(discovery.get("txt_allowlist", [])),
        {"protocol_version", "node_id", "pairing_state", "api_port"},
        "mDNS TXT contains only non-trust discovery hints",
    )
    pairing = remote.get("pairing", {})
    checks.equal(pairing.get("single_use"), True, "node pairing is single-use")
    checks.equal(
        pairing.get("close_pairing_window_after_success"),
        True,
        "pairing closes immediately after success",
    )
    listener = remote.get("listener", {})
    checks.equal(listener.get("default_host"), "127.0.0.1", "remote listener defaults to loopback")
    rejected_binds = set(listener.get("reject_unspecified_bind", []))
    checks.require({"0.0.0.0", "::"} <= rejected_binds, "unauthenticated wildcard listeners are rejected")
    checks.equal(
        listener.get("require_mtls_before_accepting_jobs"),
        True,
        "remote listener requires mTLS before jobs",
    )
    security = remote.get("security", {})
    checks.equal(security.get("transport"), "mtls_tls_1_3", "remote transport is mTLS 1.3")
    checks.equal(
        security.get("require_certificate_and_node_allowlist"),
        True,
        "remote nodes require certificate and allowlist",
    )
    grants = remote.get("capability_grants", {})
    checks.equal(grants.get("default"), "deny", "remote capabilities default to deny")
    checks.equal(
        grants.get("require_both_certificate_role_and_local_grant"),
        True,
        "remote capability requires certificate role and local grant",
    )
    available_grants = grants.get("available", {})
    checks.require(
        {
            "qwen.inference",
            "hermes.computer_use",
            "screen.capture",
            "files.read",
            "files.write",
            "webcam.capture",
            "internet.search",
        }
        <= set(available_grants),
        "paired nodes expose the complete grant-controlled feature set",
    )
    webcam_grant = available_grants.get("webcam.capture", {})
    checks.equal(
        webcam_grant.get("interactive_confirmation"),
        "always_per_session",
        "remote webcam requires confirmation every session",
    )
    checks.equal(
        webcam_grant.get("never_activate_on_unpaired_node"),
        True,
        "an unpaired node can never activate its webcam",
    )
    advertisement = remote.get("advertisement", {})
    checks.equal(advertisement.get("fetch_only_after_mtls"), True, "capabilities are fetched after mTLS")
    checks.equal(advertisement.get("signed"), True, "capability advertisements are signed")
    checks.equal(
        advertisement.get("accelerator_memory", {}).get("report_per_device_only"),
        True,
        "remote nodes report accelerator memory per device",
    )
    selection = remote.get("selection", {})
    checks.equal(selection.get("prefer_profile"), "mi300x_single", "remote selection prefers MI300X")
    checks.equal(
        selection.get("ignore_self_reported_latency_without_local_probe"),
        True,
        "remote selection uses locally verified latency",
    )
    sessions = remote.get("sessions", {})
    checks.equal(sessions.get("idempotency_key"), "operation_id", "remote operations are idempotent")
    checks.equal(sessions.get("cancel_requires_ack"), True, "remote cancel requires acknowledgement")
    checks.equal(
        sessions.get("failover", {}).get("unknown_commit_state"),
        "reconcile_required",
        "uncertain side effects are reconciled rather than replayed",
    )
    checks.equal(
        sessions.get("failover", {}).get("screen_files_webcam"),
        "no_cross_node_failover_without_new_grant",
        "screen, files and webcam never fail over without a new grant",
    )
    targeting = remote.get("targeting", {})
    checks.require(
        {"auto", "paired_node_id"} <= set(targeting.get("modes", [])),
        "remote targeting supports auto and an explicit paired node",
    )
    checks.equal(targeting.get("reject_unpaired_target"), True, "unpaired explicit targets are rejected")
    payload = remote.get("payload_policy", {})
    forbidden = set(payload.get("forbidden", []))
    checks.require(
        {"raw_microphone_audio", "raw_camera_video", "raw_desktop_capture"} <= forbidden,
        "remote payload policy forbids raw voice, video and desktop capture",
    )
    mock = remote.get("offline_mock", {})
    checks.equal(mock.get("never_open_socket"), True, "remote-node mock is offline-only")
    indicators = remote.get("indicators", {})
    checks.equal(
        indicators.get("remote_camera", {}).get("visible_before_first_frame"),
        True,
        "remote camera indicator appears before its first frame",
    )
    checks.equal(
        indicators.get("remote_camera", {}).get("may_not_be_suppressed_by_remote_node"),
        True,
        "remote node cannot suppress its camera indicator",
    )
    emergency = remote.get("emergency_stop", {})
    checks.equal(emergency.get("local_always_available"), True, "remote emergency stop is always local")
    checks.require(
        "stop_remote_camera_before_next_frame" in emergency.get("stop_actions", []),
        "emergency stop halts the remote camera immediately",
    )
    audit = remote.get("audit", {})
    checks.equal(audit.get("enabled"), True, "remote-node audit trail is enabled")
    checks.require(
        {"raw_audio", "raw_video", "raw_screen_frame"} <= set(audit.get("never_store", [])),
        "audit trail never stores raw media",
    )

    media = pipeline.get("media", {})
    checks.equal(
        media.get("load_cold_models_on_authenticated_job"), True,
        "authenticated media jobs may load cold local weights",
    )
    checks.equal(media.get("never_auto_download"), True, "media auto-download is disabled")
    video = media.get("video", {})
    checks.equal(video.get("precision", {}).get("preferred"), "bf16", "H3 defaults to BF16")
    checks.equal(
        video.get("precision", {}).get("fp16", {}).get("capability_gated"),
        True,
        "H3 FP16 is capability-gated",
    )
    checks.equal(video.get("lifecycle", {}).get("min_loaded"), 0, "H3 starts as a cold pool")
    image = media.get("image", {})
    checks.equal(image.get("lifecycle", {}).get("pool"), "cold", "Hunyuan uses a cold pool")
    checks.equal(image.get("lifecycle", {}).get("min_loaded"), 0, "Hunyuan keeps zero warm models")

    guards = pipeline.get("offline_guards", {}).get("environment", {})
    for key in ("HF_HUB_OFFLINE", "TRANSFORMERS_OFFLINE", "HF_DATASETS_OFFLINE"):
        checks.equal(str(guards.get(key)), "1", f"offline guard {key} is enabled")
    return pipeline


def validate_compose(root: Path, checks: Checks) -> dict[str, Any]:
    compose_path = root / "docker-compose.pipeline.yml"
    compose_text = compose_path.read_text(encoding="utf-8")
    compose = load_yaml(compose_path)
    services = compose.get("services", {})
    checks.require(bool(services), "compose contains services")
    checks.require(
        "router" not in services,
        "compose does not start a nonexistent learned-router placeholder",
    )
    checks.require("build:" not in compose_text, "compose never builds an image implicitly")
    checks.require(not re.search(r":(?:latest|main)(?:[\"'}\s]|$)", compose_text), "compose has no floating image tags")
    hardware_profiles = compose.get("x-hardware-profiles", {})
    checks.equal(
        hardware_profiles.get("mi300x_single", {}).get("status"),
        "recommended",
        "compose metadata recommends one MI300X",
    )
    compose_mi50 = hardware_profiles.get("mi50_12x16_legacy", {})
    checks.equal(
        compose_mi50.get("memory_is_not_one_aggregate_pool"),
        True,
        "compose metadata does not aggregate MI50 memory",
    )
    checks.equal(compose_mi50.get("automatic_selection"), False, "compose never auto-selects MI50")

    publishers = {name for name, service in services.items() if service.get("ports")}
    checks.equal(
        publishers,
        {"gateway", "qwen", "conductor"},
        "only gateway and private host-loopback AI bridges publish ports",
    )
    gateway_ports = services.get("gateway", {}).get("ports", [])
    checks.require(
        len(gateway_ports) == 1 and str(gateway_ports[0]).endswith(":8006"),
        "gateway publishes only container port 8006",
    )
    checks.require(
        str(gateway_ports[0]).startswith("127.0.0.1:"),
        "gateway host mapping is loopback-bound",
    )
    conductor_ports = services.get("conductor", {}).get("ports", [])
    checks.require(
        len(conductor_ports) == 1 and str(conductor_ports[0]).startswith("127.0.0.1:"),
        "model-tool conductor bridge is loopback-bound",
    )
    qwen_ports = services.get("qwen", {}).get("ports", [])
    checks.require(
        len(qwen_ports) == 1 and str(qwen_ports[0]).endswith(":8103"),
        "Qwen publishes only container port 8103 for host Hermes",
    )
    checks.require(
        str(qwen_ports[0]).startswith("127.0.0.1:"),
        "Qwen host mapping is private loopback, never LAN-bound",
    )

    for name, service in services.items():
        checks.require(bool(service.get("profiles")), f"compose service {name} is behind a profile")
        checks.equal(service.get("pull_policy"), "never", f"compose service {name} never pulls")
        checks.require(str(service.get("image", "")).startswith("${"), f"compose service {name} uses an image variable")

    for name in sorted(MODEL_SERVICES):
        service = services.get(name, {})
        checks.require(bool(service), f"model service {name} exists")
        env = environment_map(service)
        for key in ("HF_HUB_OFFLINE", "TRANSFORMERS_OFFLINE", "HF_DATASETS_OFFLINE"):
            checks.equal(env.get(key), "1", f"{name} enables {key}")
        checks.require(has_read_only_model_mount(service), f"{name} has a read-only local model mount")
        if name == "qwen":
            checks.require(bool(service.get("ports")), "qwen has the loopback-only Hermes bridge")
        else:
            checks.require(not service.get("ports"), f"{name} publishes no host port")

    asr_env = environment_map(services.get("asr", {}))
    checks.equal(asr_env.get("OMNI_ASR_PROVIDER"), "${OMNI_ASR_PROVIDER:-transformers_local}", "ASR defaults to the local in-process provider")
    checks.equal(asr_env.get("OMNI_ASR_MODEL"), "${OMNI_ASR_MODEL:-/models/asr}", "ASR uses the read-only mounted model path")
    checks.equal(asr_env.get("OMNI_ASR_DEVICE"), "${OMNI_ASR_DEVICE:-cuda:0}", "ASR defaults to the first ROCm-visible torch device")
    checks.equal(asr_env.get("OMNI_ASR_ENDPOINT"), "${OMNI_ASR_ENDPOINT:-}", "local ASR config has no network endpoint")
    manager_env = environment_map(services.get("manager", {}))
    checks.equal(manager_env.get("OMNI_MANAGER_LAUNCH_ENABLED"), "0", "safe compose disables manager launch")
    manager_command = " ".join(str(item) for item in services.get("manager", {}).get("command", []))
    media_command = " ".join(str(item) for item in services.get("media", {}).get("command", []))
    checks.require("python -m omni_manager" in manager_command, "compose starts the implemented manager module")
    checks.require("omni_manager.server" not in manager_command, "compose has no stale manager module path")
    checks.require("python -m omni_media" in media_command, "compose starts the implemented media module")
    checks.require("omni_media.server" not in media_command, "compose has no stale media module path")
    checks.require((root / "omni_manager" / "__main__.py").is_file(), "manager module entry point exists")
    checks.require((root / "omni_media" / "__main__.py").is_file(), "media module entry point exists")
    checks.equal(manager_env.get("OMNI_MANAGER_ALLOW_CONTAINER_BIND"), "1", "manager container bind is explicit")
    checks.equal(manager_env.get("OMNI_MANAGER_ALLOW_CONTAINER_ENDPOINTS"), "1", "manager local container endpoints are explicit")
    media_env = environment_map(services.get("media", {}))
    checks.equal(media_env.get("OMNI_MEDIA_AUTO_START"), "0", "safe compose disables media auto-start")
    checks.equal(media_env.get("OMNI_MEDIA_ALLOW_CONTAINER_BIND"), "1", "media container bind is explicit")
    checks.equal(media_env.get("OMNI_MEDIA_ALLOW_CONTAINER_ENDPOINTS"), "1", "media local container endpoints are explicit")
    conductor_env = environment_map(services.get("conductor", {}))
    gateway_env = environment_map(services.get("gateway", {}))
    required_conductor_token = (
        "${OMNI_CONDUCTOR_TOKEN:?set OMNI_CONDUCTOR_TOKEN to a strong local secret}"
    )
    checks.equal(
        conductor_env.get("OMNI_CONDUCTOR_TOKEN"),
        required_conductor_token,
        "conductor refuses a container start without an explicit token",
    )
    checks.equal(
        gateway_env.get("OMNI_CONDUCTOR_TOKEN"),
        required_conductor_token,
        "gateway receives the same required conductor token",
    )
    checks.equal(
        conductor_env.get("OMNI_HERMES_URL"),
        "http://host.docker.internal:8642",
        "compose reaches the host Hermes API through the explicit host bridge",
    )
    checks.equal(
        conductor_env.get("HERMES_API_KEY"),
        "${HERMES_API_KEY:-}",
        "compose injects the separate Hermes API bearer key",
    )
    checks.equal(
        conductor_env.get("OMNI_REMOTE_NODES_ENABLED"),
        "${OMNI_REMOTE_NODES_ENABLED:-0}",
        "compose disables remote providers by default",
    )
    checks.equal(
        conductor_env.get("OMNI_MDNS_DISCOVERY_ENABLED"),
        "${OMNI_MDNS_DISCOVERY_ENABLED:-0}",
        "compose disables mDNS discovery by default",
    )
    checks.equal(conductor_env.get("OMNI_REMOTE_LISTENER_ENABLED"), "0", "compose remote listener defaults off")
    checks.equal(conductor_env.get("OMNI_NODE_LISTEN_HOST"), "127.0.0.1", "compose node listener is loopback")
    checks.equal(conductor_env.get("OMNI_MTLS_REQUIRED"), "1", "compose requires mTLS for nodes")
    checks.equal(conductor_env.get("OMNI_REMOTE_CAPABILITIES_DEFAULT"), "deny", "compose denies remote capabilities by default")
    checks.equal(conductor_env.get("OMNI_REMOTE_WEBCAM_DEFAULT"), "deny", "compose denies remote webcam by default")
    checks.equal(
        conductor_env.get("OMNI_REMOTE_EMERGENCY_STOP_ENABLED"),
        "1",
        "compose enables local remote-node emergency stop",
    )
    minicpm_env = environment_map(services.get("minicpm-worker", {}))
    checks.equal(minicpm_env.get("ENABLE_SPECULATIVE_DECODING"), "0", "compose disables MiniCPM speculation")
    qwen_service = services.get("qwen", {})
    checks.equal(
        qwen_service.get("command"),
        ["python", "-m", "qwen_runtime.launcher"],
        "compose routes Qwen through the safe local launcher without a shell",
    )
    qwen_env = environment_map(qwen_service)
    checks.equal(qwen_env.get("INFERENCE_MODE"), "${QWEN_INFERENCE_MODE:-auto}", "Qwen launcher defaults to auto MTP1-or-AR selection")
    checks.equal(qwen_env.get("QWEN_DFLASH2_CANARY"), "${QWEN_DFLASH2_CANARY:-0}", "DFlash2 canary is disabled by default")
    checks.equal(qwen_env.get("NATIVE_MTP_TOKENS"), "${QWEN_NATIVE_MTP_TOKENS:-1}", "Qwen native MTP is fixed to one token by default")
    checks.equal(qwen_env.get("QWEN_CONTAINER_INTERNAL_ONLY"), "1", "Qwen wildcard bind is marked container-internal")
    checks.equal(qwen_env.get("QWEN_BIND_HOST"), "0.0.0.0", "Qwen listens on its isolated container network")
    checks.equal(qwen_env.get("QWEN_CHECKPOINT_MANIFEST_PATH"), "/models/qwen/.omni-checkpoint.json", "Qwen uses a local exact-checkpoint attestation")
    checks.equal(qwen_env.get("QWEN_CHECKPOINT_MANIFEST_SHA256"), "${QWEN_CHECKPOINT_MANIFEST_SHA256}", "Qwen externally pins its checkpoint attestation")
    checks.equal(qwen_env.get("PYTHONPATH"), "/opt/omni", "Qwen can import the read-only OMNI launcher")
    qwen_volumes = qwen_service.get("volumes", []) or []
    checks.require(
        any(
            isinstance(volume, dict)
            and volume.get("target") == "/opt/omni/qwen_runtime"
            and volume.get("read_only") is True
            for volume in qwen_volumes
        ),
        "Qwen launcher source is mounted read-only",
    )
    h3_env = environment_map(services.get("h3-worker", {}))
    checks.equal(h3_env.get("H3_DTYPE"), "${H3_DTYPE:-bfloat16}", "compose H3 uses the full BF16 FL2VA pipeline")
    checks.equal(h3_env.get("H3_MODEL_VARIANT"), "${H3_MODEL_VARIANT:-fl2va}", "compose selects the duplicate-free H3 FL2VA subtree")
    checks.equal(h3_env.get("H3_REQUIRE_PRECISION_PROBE"), "1", "compose requires H3 precision probe")
    hunyuan_env = environment_map(services.get("hunyuan", {}))
    checks.equal(hunyuan_env.get("HUNYUAN_MIN_LOADED"), "0", "compose Hunyuan pool is cold")
    checks.equal(hunyuan_env.get("HUNYUAN_QUANTIZATION"), "${HUNYUAN_QUANTIZATION:-nf4-v2}", "compose selects the pinned Hunyuan NF4 v2 runtime")
    checks.equal(hunyuan_env.get("HUNYUAN_EXTERNAL_SAFETY_CHECKER"), "${HUNYUAN_EXTERNAL_SAFETY_CHECKER:-0}", "external image safety checker is disabled locally")
    gateway_dockerfile = (root / "docker" / "Dockerfile.gateway").read_text(encoding="utf-8")
    dockerignore = (root / ".dockerignore").read_text(encoding="utf-8")
    gateway_source = (root / "gateway.py").read_text(encoding="utf-8")
    checks.require("AS mobile-builder" not in gateway_dockerfile, "gateway does not build the retired mobile UI")
    checks.require("COPY frontend/mobile" not in gateway_dockerfile, "gateway image excludes retired mobile sources")
    checks.require(
        "frontend/mobile/" in dockerignore.splitlines(),
        "gateway build context excludes retired mobile sources",
    )
    checks.require(
        '@app.api_route("/mobile/{asset_path:path}"' not in gateway_source,
        "gateway has no retired mobile asset route",
    )
    checks.require(
        '_RETIRED_UI_STATIC_PREFIXES = ("/static/mobile",)' in gateway_source,
        "gateway blocks stale retired mobile assets",
    )
    return compose


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--root",
        type=Path,
        default=Path(__file__).resolve().parents[1],
        help="OMNI root (defaults to the parent of scripts/)",
    )
    parser.add_argument("--json", action="store_true", help="emit a compact JSON result")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    root = args.root.resolve()
    checks = Checks()
    try:
        validate_schemas(root, checks)
        validate_pipeline(root, checks)
        validate_compose(root, checks)
    except (OSError, ValueError, json.JSONDecodeError, yaml.YAMLError) as exc:
        checks.failed.append(f"parse error: {exc}")

    result = {
        "ok": not checks.failed,
        "root": str(root),
        "passed": len(checks.passed),
        "failed": checks.failed,
    }
    if args.json:
        print(json.dumps(result, ensure_ascii=False, sort_keys=True))
    else:
        print(f"OMNI pipeline checks: {len(checks.passed)} passed, {len(checks.failed)} failed")
        for message in checks.failed:
            print(f"FAIL: {message}")
    return 0 if not checks.failed else 1


if __name__ == "__main__":
    sys.exit(main())