#!/usr/bin/env python3
"""Independent, read-only verification of the Task 26 isolated runtime."""

from __future__ import annotations

import base64
import csv
import hashlib
import importlib
import importlib.metadata
import json
import os
import stat
import subprocess
import sys
import zipfile
from collections.abc import Callable, Mapping, Sequence
from pathlib import Path
from typing import Protocol, cast
from urllib.parse import unquote, urlparse

RUNTIME = Path("/home/cube/.hermes/profiles/dualcoachtest/.strict-runtime/dac4e812")
PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest")
ARTIFACTS = Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-strict-final-candidate-successor-v3-st_01a00f35/artifacts")
WHEELS = {
    "hermes-agent": (ARTIFACTS / "hermes_agent-0.17.0-py3-none-any.whl", "ee787fb63fd8ce61d35eb2d7a43c11d7e7adef268c1d09a0a40c1d881b8ea742"),
    "physique-checkin-cli": (ARTIFACTS / "physique_checkin_cli-0.1.0-py3-none-any.whl", "fb48a1931828ee60ab3812faf795550bf4e672a54c56f78cd10dcf674c45560d"),
}

class _TelegramInputs(Protocol):
    secret: str


class _PreflightInputs(Protocol):
    telegram: _TelegramInputs
    redacted_identity: Mapping[str, object]


class _MembershipInventory(Protocol):
    rows: Sequence[object]


class _ControllerState(Protocol):
    value: str


class _Capability(Protocol):
    state: _ControllerState


MODULES = (
    "gateway.config",
    "gateway.platforms.dualcoach_admin",
    "gateway.platforms.telegram_staff_membership_gate",
    "gateway.platforms.dualcoach_tasks21_25_controller",
    "checkin_cli",
    "checkin_cli.customer_admin",
)


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def profile_inventory() -> str:
    rows: list[tuple[object, ...]] = []
    for root, directories, files in os.walk(PROFILE, followlinks=False):
        names = sorted(directories + files)
        for name in names:
            path = Path(root) / name
            info = path.lstat()
            relative = path.relative_to(PROFILE).as_posix()
            mode = stat.S_IMODE(info.st_mode)
            if path.is_symlink():
                rows.append((relative, "link", mode, info.st_uid, info.st_gid, os.readlink(path)))
            elif path.is_file():
                rows.append((relative, "file", mode, info.st_uid, info.st_gid, info.st_size, info.st_mtime_ns, sha256(path)))
            else:
                rows.append((relative, "dir", mode, info.st_uid, info.st_gid, info.st_mtime_ns))
    return hashlib.sha256(json.dumps(rows, separators=(",", ":"), ensure_ascii=True).encode()).hexdigest()


def verify_wheel(name: str, wheel: Path, expected_sha: str, site: Path) -> dict[str, object]:
    assert sha256(wheel) == expected_sha
    distribution = importlib.metadata.distribution(name)
    assert Path(str(distribution.locate_file(""))).resolve() == site.resolve()
    direct_url_file = next(item for item in (distribution.files or ()) if item.name == "direct_url.json")
    direct_url_path = Path(str(distribution.locate_file(direct_url_file)))
    loaded_direct_url = cast(object, json.loads(direct_url_path.read_text(encoding="utf-8")))
    assert isinstance(loaded_direct_url, dict)
    direct_url = cast(dict[str, object], loaded_direct_url)
    url = direct_url.get("url")
    assert isinstance(url, str)
    origin = Path(unquote(urlparse(url).path))
    assert origin == wheel
    checked = 0
    mismatches: list[str] = []
    with zipfile.ZipFile(wheel) as archive:
        record_name = next(item for item in archive.namelist() if item.endswith(".dist-info/RECORD"))
        for member, encoded_hash, size in csv.reader(archive.read(record_name).decode().splitlines()):
            if not encoded_hash:
                continue
            algorithm, value = encoded_hash.split("=", 1)
            assert algorithm == "sha256"
            data_marker = ".data/data/"
            target = site / member
            if data_marker in member:
                target = site.parents[2] / member.split(data_marker, 1)[1]
            expected = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)).hex()
            if not target.is_file() or target.stat().st_size != int(size) or sha256(target) != expected:
                mismatches.append(member)
            checked += 1
    assert not mismatches, mismatches
    return {"version": distribution.version, "origin": str(origin), "record_files_checked": checked, "mismatches": 0}


def main() -> int:
    assert sys.executable == str(RUNTIME / "venv/bin/python")
    assert stat.S_IMODE(RUNTIME.stat().st_mode) == 0o700
    assert stat.S_IMODE((PROFILE / "config.yaml").stat().st_mode) == 0o600
    assert sha256(PROFILE / "config.yaml") == "f93106b16643227e2ef9dec67a5bbd497e1d353e779da62287898a087071af87"
    site = RUNTIME / "venv/lib/python3.12/site-packages"
    assert os.environ.get("PYTHONPATH") is None
    assert all("/home/cube/projects/richard/hermes-agent" not in entry for entry in sys.path)

    origins: dict[str, str] = {}
    for module_name in MODULES:
        module_file = importlib.import_module(module_name).__file__
        assert module_file is not None
        module_path = Path(module_file).resolve()
        assert module_path.is_relative_to(site)
        origins[module_name] = str(module_path)

    gateway_config = importlib.import_module("gateway.config")
    controller = importlib.import_module("gateway.platforms.dualcoach_tasks21_25_controller")
    membership_gate = importlib.import_module("gateway.platforms.telegram_staff_membership_gate")
    load_preflight = cast(Callable[[Path], _PreflightInputs], getattr(gateway_config, "load_gateway_preflight_inputs"))
    build_membership = cast(Callable[[object, object], _MembershipInventory], getattr(membership_gate, "build_staff_chat_inventory"))
    issue_capability = cast(Callable[[Path], _Capability], getattr(controller, "issue_delivery_capability"))

    before = profile_inventory()
    inputs = load_preflight(PROFILE)
    after = profile_inventory()
    assert before == after
    identity = dict(inputs.redacted_identity)
    assert inputs.telegram.secret not in repr(inputs)
    assert inputs.telegram.secret not in json.dumps(identity, sort_keys=True)

    membership = build_membership(
        {"owner": {"user_id": "12", "chat_id": "12", "topic_id": "0"}, "customers": []},
        {"platforms": {"telegram": {"extra": {"nutrition_coaching": {"operator_card_route": {"user_id": "12", "chat_id": "12", "topic_id": "0"}}}}}},
    )
    capability = issue_capability(Path("/nonexistent/isolated-runtime-launch.json"))
    assert membership.rows and capability.state.value == "unarmed"

    service_rows = subprocess.run(
        ["systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", "--property=ActiveState,SubState,MainPID"],
        check=True, capture_output=True, text=True,
    ).stdout.splitlines()
    service = dict(row.split("=", 1) for row in service_rows)
    assert service == {"ActiveState": "inactive", "SubState": "dead", "MainPID": "0"}, service

    result = {
        "status": "PASS",
        "runtime": str(RUNTIME),
        "python": sys.executable,
        "distributions": {name: verify_wheel(name, *values, site) for name, values in WHEELS.items()},
        "module_origins": origins,
        "profile_inventory_mutations": 0,
        "redacted_identity": identity,
        "token_disclosed": False,
        "membership_driver_rows": len(membership.rows),
        "capability_driver": "UNARMED_NO_INPUT_NO_MUTATION",
        "service": "inactive/dead/MainPID0",
        "network_actions": 0,
        "customer_actions": 0,
        "service_actions": 0,
    }
    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())