#!/usr/bin/env python3
"""Offline installed-wheel r61 deterministic prior-session weekly QA."""

from __future__ import annotations

import asyncio
import json
import os
import socket
import stat
import sys
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parents[1]
RUN_ROOT = ROOT / "workspace" / "run-root-verified"
RESULT_PATH = ROOT / "r61-weekly-wheel-qa-result.json"
SOURCE_ROOT = Path(
    "/home/cube/projects/richard/traning coach/.omo/evidence/"
    "nutricoach-v150-combined/task-v15r61-candidate/snapshot/source"
)
KST = ZoneInfo("Asia/Seoul")
CHECKS = 0
NETWORK_ATTEMPTS: list[str] = []


def check(condition: object, label: str) -> None:
    global CHECKS
    CHECKS += 1
    if not condition:
        raise AssertionError(label)


def write_result(status: str, error: str | None = None) -> None:
    payload: dict[str, object] = {
        "status": status,
        "explicit_assertions": CHECKS,
        "scenario": "authorized_monday_no_task_prior_day_equal_mtime_session_tie",
        "network": "socket connect blocked; fake in-process provider only",
        "run_root": str(RUN_ROOT),
        "source_support": str(SOURCE_ROOT),
    }
    if error is not None:
        payload["error"] = error
    RESULT_PATH.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")


async def exercise() -> None:
    import checkin_cli
    import gateway
    import gateway.platforms.nutrition_coaching as nutrition_coaching
    import gateway.platforms.physique_checkin as physique_checkin
    from checkin_cli.weekly_operations_layout import customer_data_name
    from checkin_cli.wizard_models import WizardBranch, WizardFlow, WizardSession
    from checkin_cli.wizard_storage import WizardStorage
    from gateway.platforms.physique_checkin import (
        PhysiqueCheckinBridge,
        PhysiqueCheckinConfig,
    )
    from tests.gateway._nutrition_weekly_dispatcher_cases import (
        AuthorizedTelegramHost,
        dispatcher_fixture,
    )

    expected_site = Path(sys.prefix) / "lib" / "python3.12" / "site-packages"
    for module, label in (
        (checkin_cli, "checkin_cli"),
        (gateway, "gateway"),
        (nutrition_coaching, "nutrition_coaching"),
        (physique_checkin, "physique_checkin"),
    ):
        check(
            str(Path(module.__file__).resolve()).startswith(str(expected_site)),
            f"{label} loaded from installed wheel",
        )

    # Use real installed storage and bridge code, with a fixed Monday only for
    # deterministic local clock semantics.  No current-day final is written.
    fixed_now = datetime(2026, 8, 17, 23, tzinfo=KST)
    original_datetime = physique_checkin.datetime

    class FixedDatetime(datetime):
        @classmethod
        def now(cls, tz: ZoneInfo | None = None) -> datetime:
            return fixed_now if tz is not None else fixed_now.replace(tzinfo=None)

    physique_checkin.datetime = FixedDatetime
    try:
        storage = WizardStorage(RUN_ROOT / "prior-session" / "wizard")
        service = SimpleNamespace(_storage=storage)
        bridge = PhysiqueCheckinBridge(
            PhysiqueCheckinConfig("user-1", "customer-chat", "41", 600, False, False),
            service=service,
            binding_path=RUN_ROOT / "prior-session" / "bindings.json",
        )
        low_id = "a" * 32
        high_id = "b" * 32

        def finalized(session_id: str, event_id: str) -> WizardSession:
            return WizardSession(
                session_id=session_id,
                flow=WizardFlow.NUTRITION,
                owner_id="user-1",
                customer_key="client_001",
                topic_id="41",
                kst_day="2026-08-16",
                version=12,
                step="summary",
                answers={"calories": "2300", "bodyweight": "71.4"},
                branch=WizardBranch.NORMAL,
                finalized_event_id=event_id,
            )

        with storage.locked():
            storage.save(finalized(low_id, "event-low"))
            storage.save(finalized(high_id, "event-high"))
        drafts = RUN_ROOT / "prior-session" / "wizard" / "drafts"
        tie_ns = 1_770_000_000_000_000_000
        for session_id in (low_id, high_id):
            os.utime(drafts / f"{session_id}.json", ns=(tie_ns, tie_ns))

        current = storage.find_latest_finalized("user-1", "41", "2026-08-17")
        check(current is None, "no current-day finalized check-in exists")
        selected = bridge.latest_available_finalized_coaching_snapshot()
        check(isinstance(selected, dict), "prior-day safe snapshot is available")
        check(selected.get("session_id") == high_id, "equal-mtime prior sessions use lexically greatest session id")
        check(selected.get("kst_day") == "2026-08-16", "selected snapshot is the prior day")
        check(selected.get("answers") == {"calories": "2300", "bodyweight": "71.4"}, "selected snapshot is safe finalized content")

        selection_coordinator = SimpleNamespace(
            _by_key={"client_001": SimpleNamespace(bridge=bridge)}
        )
        chosen_session = nutrition_coaching.NutritionCoachingCoordinator.weekly_session_id(
            selection_coordinator, "client_001"
        )
        check(chosen_session == high_id, "installed coordinator selects the deterministic prior session")

        weekly_root = RUN_ROOT / "weekly"
        weekly_root.mkdir(parents=True, exist_ok=True)
        fixture = dispatcher_fixture(weekly_root, "cutoff")
        sidecar = (
            fixture.reminder.store.authority.observed_path
            / customer_data_name(fixture.reminder.store.customer_identity_digest)
        )
        check(not sidecar.exists(), "precondition: weekly sidecar is absent")
        due_calls = 0

        def no_due(*_args: object, **_kwargs: object) -> tuple[()]:
            nonlocal due_calls
            due_calls += 1
            return ()

        original_due = checkin_cli.build_due_customer_tasks
        checkin_cli.build_due_customer_tasks = no_due
        persisted_sessions: list[str | None] = []
        original_persist = fixture.coordinator.persist_grounded_weekly_review

        def persist_with_installed_selection(customer_key: str, owner_key: tuple[str, str, str], source: object):
            session = nutrition_coaching.NutritionCoachingCoordinator.weekly_session_id(
                selection_coordinator, customer_key
            )
            persisted_sessions.append(session)
            return original_persist(customer_key, owner_key, source)

        fixture.coordinator.persist_grounded_weekly_review = persist_with_installed_selection
        captured_requests: list[object] = []
        original_create = fixture.coordinator.create_grounded_weekly_owner_draft

        def capture_create(request: object, model: object):
            captured_requests.append(request)
            return original_create(request, model)

        fixture.coordinator.create_grounded_weekly_owner_draft = capture_create
        try:
            fence = checkin_cli.initialize_schedule_delivery_fence(fixture.root)
            check(fence.state == "ready", "delivery fence is ready")
            first = await fixture.host.run_authorized(fixed_now)
            check(first.success is True, "authorized Monday no-task tick succeeds")
            check(due_calls == 1, "first tick receives no due tasks")
            initial = sidecar.stat(follow_symlinks=False)
            check(stat.S_ISREG(initial.st_mode), "sidecar is a regular file")
            check(not stat.S_ISLNK(initial.st_mode), "sidecar is not a symlink")
            check(sidecar.read_bytes() == b"", "sidecar initializes with an empty durable history")
            check(stat.S_IMODE(initial.st_mode) == 0o600, "sidecar mode is exactly 0600")
            check(initial.st_nlink == 1, "sidecar has exactly one link")
            check(fixture.provider.calls == [], "no provider/customer message is sent")
            check(len(captured_requests) == 1, "Monday weekly owner-draft request is generated")
            check(persisted_sessions == [high_id], "owner draft persistence uses the selected prior-day session")
            summary = captured_requests[0].bound_summary.summary
            check(summary.starts_on.isoformat() == "2026-08-10", "summary begins on the prior Monday")
            check(summary.ends_on.isoformat() == "2026-08-16", "summary ends on the prior Sunday")
            check((summary.completed_days, summary.calendar_days) == (0, 7), "summary retains the empty seven-day denominator")
            check(tuple(fixture.coordinator.drafts) == ("weekly-1",), "weekly owner draft is persisted")
            check(fixture.coordinator.owner_routes == [("owner", "owner-dm", "owner")], "owner draft uses only the owner route")
            before_bytes = sidecar.read_bytes()
            before_identity = (initial.st_dev, initial.st_ino)

            restarted = AuthorizedTelegramHost.create(
                fixture.coordinator, fixture.provider, fixture.host.config
            )
            replay = await restarted.run_authorized(fixed_now)
            check(replay.success is True, "idempotent replay succeeds")
            check(due_calls == 2, "replay again receives no due tasks")
            check(len(captured_requests) == 2, "replay validates the weekly request")
            final = sidecar.stat(follow_symlinks=False)
            check(sidecar.read_bytes() == before_bytes, "replay leaves sidecar bytes unchanged")
            check((final.st_dev, final.st_ino) == before_identity, "replay preserves sidecar identity")
            check(stat.S_IMODE(final.st_mode) == 0o600, "replay preserves sidecar mode")
            check(final.st_nlink == 1, "replay preserves sidecar link count")
            check(persisted_sessions == [high_id], "replay reuses the bound weekly owner draft")
            check(tuple(fixture.coordinator.drafts) == ("weekly-1",), "exactly one owner draft remains")
            check(fixture.coordinator.owner_routes == [("owner", "owner-dm", "owner")], "replay does not duplicate owner draft persistence")
            check(fixture.provider.calls == [], "replay sends no provider/customer message")
            check(NETWORK_ATTEMPTS == [], "no socket connection was attempted")
        finally:
            checkin_cli.build_due_customer_tasks = original_due
    finally:
        physique_checkin.datetime = original_datetime


def main() -> int:
    RUN_ROOT.mkdir(parents=True, exist_ok=True)
    original_connect = socket.socket.connect

    def block_connect(_socket: socket.socket, address: object) -> None:
        NETWORK_ATTEMPTS.append(repr(address))
        raise AssertionError(f"network connection blocked: {address!r}")

    socket.socket.connect = block_connect
    try:
        asyncio.run(exercise())
    except Exception as error:
        write_result("FAIL", f"{type(error).__name__}: {error}")
        raise
    finally:
        socket.socket.connect = original_connect
    write_result("PASS")
    return 0


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