#!/usr/bin/env python3
"""Offline installed-wheel QA for r59's Monday no-task sidecar regression."""

from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[1]
RUN_ROOT = ROOT / "workspace" / "run-root-verified"
RESULT_PATH = ROOT / "r59-weekly-wheel-qa-result.json"
SOURCE_ROOT = Path(
    "/home/cube/projects/richard/traning coach/.omo/evidence/"
    "nutricoach-v150-combined/task-v15r59-candidate/snapshot/source"
)

checks = 0
network_attempts: list[str] = []


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


def result(status: str, error: str | None = None) -> None:
    payload: dict[str, object] = {
        "status": status,
        "explicit_assertions": checks,
        "scenario": "authorized_monday_no_task_no_sidecar_weekly_tick",
        "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
    from checkin_cli.weekly_operations import CustomerKey
    from checkin_cli.weekly_operations_layout import customer_data_name
    from checkin_cli.weekly_operations_store import WeeklyOperationsStore
    from tests.gateway._nutrition_weekly_dispatcher_cases import (
        AuthorizedTelegramHost,
        dispatcher_fixture,
    )

    expected_site = Path(sys.prefix) / "lib" / "python3.12" / "site-packages"
    check(str(Path(checkin_cli.__file__).resolve()).startswith(str(expected_site)), "checkin_cli loaded from installed wheel")
    check(str(Path(gateway.__file__).resolve()).startswith(str(expected_site)), "gateway loaded from installed wheel")

    fixture = dispatcher_fixture(RUN_ROOT, "cutoff")
    sidecar = (
        fixture.reminder.store.authority.observed_path
        / customer_data_name(fixture.reminder.store.customer_identity_digest)
    )
    topic59_ledger = RUN_ROOT / "data" / "weekly-operations-topic59.jsonl"
    check(not sidecar.exists(), "precondition: no weekly sidecar")

    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
    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
    now = datetime(2026, 8, 17, 23, tzinfo=ZoneInfo("Asia/Seoul"))
    try:
        fence = checkin_cli.initialize_schedule_delivery_fence(RUN_ROOT)
        check(fence.state == "ready", "schedule fence is ready")

        first = await fixture.host.run_authorized(now)
        check(first.success is True, "authorized Monday no-task tick succeeds")
        check(due_calls == 1, "tick obtains an empty due-task set")
        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"", "new sidecar has an empty durable history")
        check(stat.S_IMODE(initial.st_mode) == 0o600, "sidecar mode is exactly 0600")
        check(initial.st_nlink == 1, "sidecar link count is exactly one")
        check(fixture.provider.calls == [], "no customer or provider message is sent")
        check(not topic59_ledger.exists(), "no Topic-59 card is projected without a task row")

        check(len(captured_requests) == 1, "Monday weekly-summary request is generated")
        request = captured_requests[0]
        summary = request.bound_summary.summary
        check(summary.starts_on.isoformat() == "2026-08-10", "summary starts on the prior Monday")
        check(summary.ends_on.isoformat() == "2026-08-16", "summary ends on the prior Sunday")
        check((summary.submitted_count, summary.late_count, summary.missed_count) == (0, 0, 0), "empty sidecar produces zero weekly statuses")
        check((summary.completed_days, summary.calendar_days) == (0, 7), "summary retains the seven-day denominator")
        check(summary.adherence_percent == 0.0, "empty week adherence is zero")
        check(fixture.coordinator.owner_routes == [("owner", "owner-dm", "owner")], "summary is persisted to the owner route")
        check(tuple(fixture.coordinator.drafts) == ("weekly-1",), "one weekly owner draft is created")
        body = fixture.coordinator.drafts["weekly-1"].text
        check("기간: 2026-08-10~2026-08-16" in body, "owner draft contains the summary week")
        check("완료: 0/7" in body, "owner draft contains the empty-week completion count")

        reopened = WeeklyOperationsStore.for_authority(
            fixture.reminder.store.authority,
            CustomerKey(fixture.reminder.runtime.spec.customer_key),
        )
        check(reopened.read() == (), "fresh store instance reads the initialized empty sidecar")
        before_bytes = sidecar.read_bytes()
        before_identity = (initial.st_dev, initial.st_ino)

        restarted = AuthorizedTelegramHost.create(
            fixture.coordinator, fixture.provider, fixture.host.config
        )
        second = await restarted.run_authorized(now)
        check(second.success is True, "idempotent authorized rerun succeeds")
        check(due_calls == 2, "rerun again obtains no due tasks")
        check(len(captured_requests) == 2, "rerun validates the same Monday summary")
        final = sidecar.stat(follow_symlinks=False)
        check(sidecar.read_bytes() == before_bytes, "rerun leaves sidecar bytes unchanged")
        check((final.st_dev, final.st_ino) == before_identity, "rerun preserves sidecar identity")
        check(stat.S_IMODE(final.st_mode) == 0o600, "rerun preserves sidecar mode")
        check(final.st_nlink == 1, "rerun preserves one sidecar link")
        check(fixture.provider.calls == [], "rerun sends no provider message")
        check(fixture.coordinator.owner_routes == [("owner", "owner-dm", "owner")], "rerun does not duplicate owner persistence")
        check(tuple(fixture.coordinator.drafts) == ("weekly-1",), "rerun reuses the single bound owner draft")
        check(network_attempts == [], "no socket connection was attempted")
    finally:
        checkin_cli.build_due_customer_tasks = original_due


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:
        result("FAIL", f"{type(error).__name__}: {error}")
        raise
    finally:
        socket.socket.connect = original_connect
    result("PASS")
    return 0


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