#!/usr/bin/env python3
"""Zero-network synthetic compatibility smoke for a supplied runtime candidate."""

from __future__ import annotations

import json
import socket
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import NoReturn, final
from zoneinfo import ZoneInfo


NOW = 4_000_000_000
SESSION = "f1e2d3c4b5a697887766554433221100"
OWNER = "opaque-owner-v1"
CHAT = "opaque-chat-v1"
TOPIC = "opaque-topic-v1"
MESSAGE = "opaque-message-v1"
BODYWEIGHT = "61.4"
CALORIES = "2460"


@final
class _SocketGuard:
    """Fail the smoke on any attempted network socket construction."""

    _original: object
    network_events: int

    def __init__(self) -> None:
        self._original = socket.socket
        self.network_events = 0

    def __enter__(self) -> _SocketGuard:
        setattr(socket, "socket", self._blocked_socket)
        return self

    def __exit__(
        self,
        exception_type: object,
        exception: object,
        traceback: object,
    ) -> bool:
        _ = (exception_type, exception, traceback)
        setattr(socket, "socket", self._original)
        return False

    def _blocked_socket(self, *args: object, **kwargs: object) -> NoReturn:
        _ = (args, kwargs)
        self.network_events += 1
        raise AssertionError("network access is forbidden in the compatibility smoke")


def _parse_args(argv: list[str]) -> tuple[Path, bool]:
    expected_root: Path | None = None
    require_wheels = False
    arguments = iter(argv)
    for argument in arguments:
        if argument == "--expected-runtime-root":
            try:
                expected_root = Path(next(arguments))
            except StopIteration as exc:
                raise RuntimeError("--expected-runtime-root requires a path") from exc
        elif argument == "--require-wheel-runtime":
            require_wheels = True
        else:
            raise RuntimeError(f"unsupported argument: {argument}")
    if expected_root is None:
        raise RuntimeError("--expected-runtime-root is required")
    return expected_root, require_wheels


def _is_within(path: Path, root: Path) -> bool:
    try:
        _ = path.resolve().relative_to(root)
    except ValueError:
        return False
    return True


def _load_runtime(expected_root: Path, *, require_wheels: bool) -> dict[str, str]:
    """Import the exact runtime and reject source-tree or stale-wheel leakage."""
    from importlib import metadata

    runtime_text = str(expected_root)
    if runtime_text not in sys.path:
        sys.path.insert(0, runtime_text)
    profile_source = expected_root / "dualcoach" / "profile"
    if (profile_source / "checkin_cli").is_dir():
        source_text = str(profile_source)
        if source_text not in sys.path:
            sys.path.insert(0, source_text)

    import checkin_cli
    import gateway
    from checkin_cli import wizard as checkin_wizard
    from gateway.platforms import physique_checkin

    modules = {
        "gateway": Path(gateway.__file__).resolve(),
        "gateway.platforms.physique_checkin": Path(physique_checkin.__file__).resolve(),
        "checkin_cli": Path(checkin_cli.__file__).resolve(),
        "checkin_cli.wizard": Path(checkin_wizard.__file__).resolve(),
    }
    for name, path in modules.items():
        if not _is_within(path, expected_root):
            raise RuntimeError(f"{name} imported outside the supplied runtime: {path}")
    if require_wheels:
        for distribution_name in ("hermes-agent", "physique-checkin-cli"):
            distribution_root = Path(str(
                metadata.distribution(distribution_name).locate_file("")
            )).resolve()
            if not _is_within(distribution_root, expected_root):
                message = (
                    f"{distribution_name} distribution is outside the supplied runtime: "
                    f"{distribution_root}"
                )
                raise RuntimeError(message)
    return {name: str(path) for name, path in modules.items()}


def _run(module_paths: dict[str, str]) -> dict[str, bool | int | str | dict[str, str]]:
    from pydantic import ValidationError

    from checkin_cli.wizard import WizardService
    from checkin_cli.wizard_models import (
        WizardContext,
        WizardFlow,
        WizardSession,
        WizardStatus,
    )
    from gateway.platforms.physique_checkin import CallbackData, PhysiqueCheckinBridge
    from gateway.platforms.physique_checkin_bindings import BindingStoreCorruption
    from gateway.platforms.physique_checkin_config import PhysiqueCheckinConfig

    def kst_day() -> str:
        return datetime.now(ZoneInfo("Asia/Seoul")).date().isoformat()

    def config() -> PhysiqueCheckinConfig:
        return PhysiqueCheckinConfig(OWNER, CHAT, TOPIC, 3_600, False, False)

    def legacy_session(
        *,
        step: str = "calories",
        version: int = 1,
        answers: dict[str, str] | None = None,
    ) -> dict[str, object]:
        session = WizardSession(
            session_id=SESSION,
            flow=WizardFlow.NUTRITION,
            owner_id=OWNER,
            customer_key=None,
            topic_id=TOPIC,
            kst_day=kst_day(),
            version=version,
            step=step,
            answers=answers or {"bodyweight": BODYWEIGHT},
            state_schema_version=1,
        )
        return session.model_dump(
            mode="json",
            exclude={"unknown_steps", "step_history"},
        )

    def legacy_binding(
        *,
        step: str = "calories",
        version: int = 1,
        expires_at: int = NOW + 3_600,
    ) -> dict[str, object]:
        return {
            "version": 1,
            "active_session_id": SESSION,
            "bindings": [{
                "session_id": SESSION,
                "owner_id": OWNER,
                "chat_id": CHAT,
                "topic_id": TOPIC,
                "step": step,
                "version": version,
                "message_id": MESSAGE,
                "expires_at": expires_at,
            }],
        }

    def write_legacy_state(
        root: Path,
        *,
        session: dict[str, object] | None = None,
        binding: dict[str, object] | None = None,
    ) -> tuple[Path, Path]:
        wizard_root = root / "wizard"
        drafts = wizard_root / "drafts"
        _ = drafts.mkdir(parents=True, mode=0o700)
        draft_path = drafts / f"{SESSION}.json"
        _ = draft_path.write_text(
            json.dumps(
                legacy_session() if session is None else session,
                separators=(",", ":"),
                sort_keys=True,
            ),
            encoding="utf-8",
        )
        _ = draft_path.chmod(0o600)
        binding_path = root / "bindings.json"
        _ = binding_path.write_text(
            json.dumps(
                legacy_binding() if binding is None else binding,
                separators=(",", ":"),
                sort_keys=True,
            ),
            encoding="utf-8",
        )
        _ = binding_path.chmod(0o600)
        return wizard_root, binding_path

    def load_session(wizard_root: Path) -> WizardSession:
        return WizardSession.model_validate_json(
            (wizard_root / "drafts" / f"{SESSION}.json").read_text(encoding="utf-8")
        )

    def bridge_for(root: Path) -> tuple[PhysiqueCheckinBridge, WizardService, Path, Path]:
        wizard_root, binding_path = write_legacy_state(root)
        service = WizardService.for_standalone(wizard_root)
        bridge = PhysiqueCheckinBridge(config(), service=service, binding_path=binding_path)
        return bridge, service, wizard_root, binding_path

    def callback(bridge: PhysiqueCheckinBridge, action: str) -> str:
        prompt = bridge.active_prompt()
        if prompt is None:
            raise AssertionError("active prompt is unavailable")
        for row in prompt.button_rows:
            for _label, data in row:
                parsed = CallbackData.parse(data)
                if parsed is not None and parsed.action == action:
                    return data
        raise AssertionError(f"missing {action!r} callback")

    with tempfile.TemporaryDirectory(prefix="nutricoach-v150-legacy-") as temporary:
        root = Path(temporary)

        wizard_root, binding_path = write_legacy_state(root / "baseline")
        draft_path = wizard_root / "drafts" / f"{SESSION}.json"
        draft_before = draft_path.read_bytes()
        binding_before = binding_path.read_bytes()
        baseline = PhysiqueCheckinBridge(
            config(),
            service=WizardService.for_standalone(wizard_root),
            binding_path=binding_path,
        )
        cursor = baseline.active_cursor_identity(now_epoch=NOW)
        snapshot = baseline.active_checkin_snapshot()
        if (
            cursor is None
            or (cursor[0].session_id, cursor[0].step, cursor[0].version)
            != (SESSION, "calories", 1)
            or baseline.active_prompt_message_id() != MESSAGE
            or snapshot is None
            or snapshot.get("answers") != {"bodyweight": BODYWEIGHT}
            or load_session(wizard_root).unknown_steps != ()
            or draft_path.read_bytes() != draft_before
            or binding_path.read_bytes() != binding_before
        ):
            raise AssertionError("v1 baseline identity was not preserved")
        repeated = PhysiqueCheckinBridge(
            config(),
            service=WizardService.for_standalone(wizard_root),
            binding_path=binding_path,
        )
        startup_mutations = int(draft_path.read_bytes() != draft_before) + int(
            binding_path.read_bytes() != binding_before
        )
        if (
            baseline.cursor_identity(SESSION, now_epoch=NOW) is None
            or repeated.cursor_identity(SESSION, now_epoch=NOW) is None
        ):
            raise AssertionError("future v1 binding is not addressable")
        if (wizard_root / "events.jsonl").exists():
            raise AssertionError("startup created an event")

        known_bridge, known_service, known_root, _known_binding = bridge_for(root / "known")
        known_prompt = known_bridge.active_prompt()
        if known_prompt is None or not known_prompt.text.startswith("진행 2/12"):
            raise AssertionError("legacy calories prompt did not preserve progress")
        known = known_bridge.handle_text(CALORIES, OWNER, CHAT, TOPIC, now_epoch=NOW)
        known_cursor = known_bridge.active_cursor_identity(now_epoch=NOW)
        known_session = load_session(known_root)
        stale = known_service.answer(
            WizardContext(OWNER, TOPIC), SESSION, 1, "unknown",
        )
        if (
            known is None
            or not known.accepted
            or known_cursor is None
            or (known_cursor[0].step, known_cursor[0].version) != ("macros", 2)
            or known.prompt is None
            or "진행 3/12" not in known.prompt.text
            or known_session.answers.get("calories") != CALORIES
            or known_session.unknown_steps != ()
            or stale.status is not WizardStatus.REJECTED
        ):
            raise AssertionError("known v1 calories did not advance exactly once")

        unknown_bridge, _unknown_service, unknown_root, _unknown_binding = bridge_for(
            root / "unknown"
        )
        unknown_callback = callback(unknown_bridge, "u")
        unknown = unknown_bridge.handle_callback(
            unknown_callback, OWNER, CHAT, TOPIC, MESSAGE, now_epoch=NOW,
        )
        unknown_cursor = unknown_bridge.active_cursor_identity(now_epoch=NOW)
        unknown_session = load_session(unknown_root)
        if (
            not unknown.accepted
            or unknown_cursor is None
            or (unknown_cursor[0].step, unknown_cursor[0].version) != ("macros", 2)
            or unknown.prompt is None
            or "진행 3/12" not in unknown.prompt.text
            or "calories" in unknown_session.answers
            or unknown_session.unknown_steps != ("calories",)
        ):
            raise AssertionError("explicit unknown v1 calories did not advance exactly once")
        duplicate_before = (
            (unknown_root / "drafts" / f"{SESSION}.json").read_bytes(),
            unknown_bridge.active_cursor_identity(now_epoch=NOW),
        )
        duplicate = PhysiqueCheckinBridge(
            config(),
            service=WizardService.for_standalone(unknown_root),
            binding_path=root / "unknown" / "bindings.json",
        ).handle_callback(
            unknown_callback,
            OWNER,
            CHAT,
            TOPIC,
            MESSAGE,
            now_epoch=NOW,
        )
        duplicate_after = (
            (unknown_root / "drafts" / f"{SESSION}.json").read_bytes(),
            unknown_bridge.active_cursor_identity(now_epoch=NOW),
        )
        if duplicate.accepted or duplicate_after != duplicate_before:
            raise AssertionError("duplicate callback advanced a v1-origin session")
        duplicate_advances = 0

        expired_root, expired_binding = write_legacy_state(
            root / "expired",
            binding=legacy_binding(expires_at=1),
        )
        expired_draft = expired_root / "drafts" / f"{SESSION}.json"
        expired_draft_before = expired_draft.read_bytes()
        expired_binding_before = expired_binding.read_bytes()
        expired_service = WizardService.for_standalone(expired_root)
        expired_bridge = PhysiqueCheckinBridge(
            config(), service=expired_service, binding_path=expired_binding,
        )
        if (
            expired_bridge.active_cursor_identity() is not None
            or expired_bridge.active_prompt() is not None
            or expired_draft.read_bytes() != expired_draft_before
            or expired_binding.read_bytes() != expired_binding_before
            or (expired_root / "events.jsonl").exists()
        ):
            raise AssertionError("expired binding mutated or resumed automatically")
        explicit = expired_bridge.open_launcher(
            "nutrition_daily", message_id="opaque-explicit-resume", now_epoch=NOW,
        )
        if explicit.callback_data is None:
            raise AssertionError("explicit resume did not create a launcher")
        resumed = expired_bridge.handle_callback(
            explicit.callback_data,
            OWNER,
            CHAT,
            TOPIC,
            "opaque-explicit-resume",
            now_epoch=NOW,
        )
        if not resumed.accepted:
            raise AssertionError("expired binding did not require a working explicit resume")
        expired_sends = 0

        corrupt_path = root / "corrupt" / "bindings.json"
        _ = corrupt_path.parent.mkdir(mode=0o700)
        _ = corrupt_path.write_text("{", encoding="utf-8")
        _ = corrupt_path.chmod(0o600)
        try:
            _ = PhysiqueCheckinBridge(
                config(),
                service=WizardService.for_standalone(root / "corrupt" / "wizard"),
                binding_path=corrupt_path,
            )
        except BindingStoreCorruption:
            pass
        else:
            raise AssertionError("corrupt binding was accepted")

        contradictory = legacy_session(
            step="macros",
            version=2,
            answers={"bodyweight": BODYWEIGHT, "calories": CALORIES},
        )
        contradictory["state_schema_version"] = 2
        contradictory["unknown_steps"] = ["calories"]
        contradictory["step_history"] = ["bodyweight", "calories"]
        try:
            _ = WizardSession.model_validate(contradictory)
        except ValidationError:
            pass
        else:
            raise AssertionError("contradictory candidate was accepted")

        mismatched_bridge, _mismatched_service, mismatched_root, mismatched_binding = bridge_for(
            root / "mismatched"
        )
        _ = mismatched_binding.write_text(
            json.dumps(legacy_binding(step="macros", version=2), separators=(",", ":")),
            encoding="utf-8",
        )
        _ = mismatched_binding.chmod(0o600)
        mismatched_bridge = PhysiqueCheckinBridge(
            config(),
            service=WizardService.for_standalone(mismatched_root),
            binding_path=mismatched_binding,
        )
        mismatched_before = (
            mismatched_root / "drafts" / f"{SESSION}.json"
        ).read_bytes()
        mismatched = mismatched_bridge.handle_text(
            "180 120 50", OWNER, CHAT, TOPIC, now_epoch=NOW,
        )
        if (
            mismatched is None
            or mismatched.accepted
            or (mismatched_root / "drafts" / f"{SESSION}.json").read_bytes()
            != mismatched_before
        ):
            raise AssertionError("mismatched candidate did not fail closed")

        legacy_free_text = {
            "bodyweight": BODYWEIGHT,
            "calories": CALORIES,
            "macros": "180 120 50",
            "meals": "unknown",
        }
        free_root, free_binding = write_legacy_state(
            root / "free-text",
            session=legacy_session(step="water", version=4, answers=legacy_free_text),
            binding=legacy_binding(step="water", version=4),
        )
        free_bridge = PhysiqueCheckinBridge(
            config(),
            service=WizardService.for_standalone(free_root),
            binding_path=free_binding,
        )
        free_reply = free_bridge.handle_text("2.0", OWNER, CHAT, TOPIC, now_epoch=NOW)
        free_session = load_session(free_root)
        if (
            free_reply is None
            or not free_reply.accepted
            or free_session.answers.get("meals") != "unknown"
            or free_session.unknown_steps != ()
        ):
            raise AssertionError("legacy free text was reinterpreted as explicit unknown")

    return {
        "duplicate_advances": duplicate_advances,
        "expired_sends": expired_sends,
        "known_step": "macros",
        "known_version": 2,
        "module_paths": module_paths,
        "preserved_v1_identity": True,
        "startup_mutations": startup_mutations,
        "telegram_sends": 0,
        "unknown_step": "macros",
        "unknown_version": 2,
    }


def _run_without_network(
    module_paths: dict[str, str],
) -> dict[str, bool | int | str | dict[str, str]]:
    with _SocketGuard() as sockets:
        result = _run(module_paths)
        result["network_events"] = sockets.network_events
        result["status"] = "PASS"
        return result
    raise AssertionError("socket guard unexpectedly suppressed the smoke result")


def main() -> int:
    runtime_root, require_wheels = _parse_args(sys.argv[1:])
    expected_root = runtime_root.resolve()
    if not expected_root.is_dir():
        raise RuntimeError(f"expected runtime root does not exist: {expected_root}")
    module_paths = _load_runtime(expected_root, require_wheels=require_wheels)
    result = _run_without_network(module_paths)
    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
    return 0


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