#!/usr/bin/env python3
"""Offline, installed-wheel invite-to-first-check-in QA.

This runner imports only the candidate wheels installed in the sibling venv. It
uses synthetic local Telegram identities and a fake in-process delivery method;
no provider client, production root, or real customer is used.
"""

from __future__ import annotations

import asyncio
import json
import os
import socket
import stat
import subprocess
import sys
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import parse_qs, urlparse

from checkin_cli.nutrition_onboarding import (
    QUESTION_FIELDS,
    MessageEvidence,
    NutritionOnboardingService,
    OnboardingAuthority,
    canonical_digest,
    example_answer,
)
from checkin_cli.wizard import WizardService
from checkin_cli.wizard_models import WizardContext, WizardStatus
from gateway.platforms.dualcoach_customer_invite import CustomerInviteError
from gateway.platforms.telegram import TelegramAdapter
from gateway.platforms.telegram_activation_notice import ActivationNoticeStore
from gateway.platforms.telegram_customer_bootstrap import (
    BootstrapError,
    BootstrapState,
    ConsentHandoff,
    CustomerDraft,
    GenerationConflict,
    RecoverySlot,
    RoomBootstrapStore,
    consent_handoff_digest,
    room_bootstrap_state_dir,
)
from gateway.platforms.telegram_staff_membership_gate import (
    MembershipGateError,
    MembershipJournal,
)

QA_ROOT = Path(__file__).resolve().parents[1]
RUN_ROOT = QA_ROOT / "workspace" / "run-root"
RESULT_PATH = QA_ROOT / "installed-wheel-qa-result.json"
CHECKS = 0
SCENARIOS: dict[str, str] = {}


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


def expect(exc_type: type[BaseException], label: str, action) -> BaseException:
    global CHECKS
    try:
        action()
    except exc_type as exc:
        CHECKS += 1
        return exc
    except Exception as exc:  # pragma: no cover - evidence must name unexpected errors
        raise AssertionError(f"{label}: expected {exc_type.__name__}, got {type(exc).__name__}: {exc}") from exc
    raise AssertionError(f"{label}: expected {exc_type.__name__}")


def private_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True, mode=0o700)
    path.chmod(0o700)


def private_json(path: Path, payload: object) -> None:
    private_dir(path.parent)
    path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
    path.chmod(0o600)


def token_from_link(link: str) -> str:
    value = parse_qs(urlparse(link).query).get("start", [])
    check(len(value) == 1 and value[0].startswith("rc1_"), "invite has one opaque rc1 token")
    return value[0]


def draft_payload(customer_key: str = "client_001", customer_user_id: str = "100") -> dict[str, object]:
    return {
        "customer_key": customer_key,
        "display_name": "Synthetic Customer",
        "starts_on": "2026-09-01",
        "daily_time": "08:00",
        "weekly_weekday": 0,
        "monthly_day": 1,
        "calories_kcal": 2200,
        "protein_g": 160,
        "meals": ["synthetic meal"],
        "customer_user_id": customer_user_id,
    }


def make_invite_profile(root: Path, *, capacity: int = 1, enabled_customers: list[dict[str, object]] | None = None) -> Path:
    private_dir(root)
    private_dir(root / "data")
    private_dir(root / "customers")
    private_json(
        root / "config.yaml",
        {
            "platforms": {
                "telegram": {
                    "extra": {
                        "production_preflight": {"expected_bot_username": "test_bot"}
                    }
                }
            }
        },
    )
    private_json(
        root / "customers" / "registry.json",
        {
            "owner": {"user_id": "900"},
            "customers": enabled_customers or [],
            "admission_policy": {"max_enabled_customers": capacity},
        },
    )
    draft = root / "draft.json"
    private_json(draft, draft_payload())
    return draft


def invoke(*args: str) -> subprocess.CompletedProcess[str]:
    env = {
        "PATH": os.environ["PATH"],
        "HOME": str(QA_ROOT / "home"),
        "PIP_NO_INDEX": "1",
        "NO_PROXY": "*",
        "PYTHONDONTWRITEBYTECODE": "1",
    }
    return subprocess.run(
        [sys.executable, *args],
        cwd=QA_ROOT / "workspace",
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )


def await_consent(store: RoomBootstrapStore, token: str) -> object:
    claimed = store.claim_rehearsal_customer_invite(
        token, user_id="100", chat_id="100", message_id="11"
    )
    return store.transition(
        claimed.session_id,
        expected_generation=claimed.generation,
        target=BootstrapState.AWAITING_CONSENT,
    )


def test_invite_claim_consent_activation() -> None:
    root = RUN_ROOT / "invite-golden"
    draft = make_invite_profile(root)
    issued = invoke(
        "-m", "gateway.platforms.dualcoach_admin", "customer", "invite",
        "--profile-root", str(root), "--draft", str(draft), "--json",
    )
    check(issued.returncode == 0, f"installed invite CLI succeeds: {issued.stderr}")
    invite = json.loads(issued.stdout)
    check(invite["customer_key"] == "client_001", "invite CLI reports customer key")
    check(invite["generation"] == 1, "invite CLI starts generation one")
    token = token_from_link(invite["customer_link"])
    store = RoomBootstrapStore(room_bootstrap_state_dir(root))
    expect(
        BootstrapError,
        "wrong claimant is denied before mutation",
        lambda: store.claim_rehearsal_customer_invite(
            token, user_id="101", chat_id="101", message_id="10"
        ),
    )
    claimed = store.claim_rehearsal_customer_invite(
        token, user_id="100", chat_id="100", message_id="11"
    )
    check(claimed.state is BootstrapState.REGISTERING, "intended private-DM claimant is accepted")
    expect(
        BootstrapError,
        "single-use invite duplicate claim is denied",
        lambda: store.claim_rehearsal_customer_invite(
            token, user_id="100", chat_id="100", message_id="12"
        ),
    )
    awaiting = store.transition(
        claimed.session_id,
        expected_generation=claimed.generation,
        target=BootstrapState.AWAITING_CONSENT,
    )
    check(awaiting.state is BootstrapState.AWAITING_CONSENT, "claim advances only to consent")
    reserved = store.reserve_consent_publication(
        awaiting.session_id, expected_generation=awaiting.generation
    )
    attempt = store.reserve_recovery_attempt(
        reserved.session_id,
        slot=RecoverySlot.CONSENT_CARD,
        chat_id="100",
        expected_generation=reserved.generation,
    )
    bound = store.bind_recovery_receipt(
        reserved.session_id,
        slot=RecoverySlot.CONSENT_CARD,
        chat_id="100",
        attempt_generation=attempt.generation,
        receipt_id="501",
    )
    handoff = ConsentHandoff(
        update_id=5010,
        actor_id=100,
        chat_id=100,
        topic_id=0,
        message_id=501,
        callback_data="cc1:local:grant",
        bootstrap_generation=bound.generation,
        recorded_at=datetime.now(timezone.utc),
        provenance_digest="",
    )
    # The installed store requires the exact canonical callback; use its public
    # callback builder through the package import rather than a fabricated wire value.
    from gateway.platforms.telegram_customer_bootstrap import customer_consent_callback
    handoff = replace(
        handoff,
        callback_data=customer_consent_callback("client_001", "privacy-v1", "g"),
        provenance_digest="",
    )
    handoff = replace(handoff, provenance_digest=consent_handoff_digest(bound, handoff))
    handed = store.bind_consent_handoff(
        bound.session_id, expected_generation=bound.generation, handoff=handoff
    )
    check(handed.consent_handoff == handoff, "authenticated consent handoff is persisted")
    restarted = RoomBootstrapStore(room_bootstrap_state_dir(root))
    after_restart = restarted.get(handed.session_id)
    check(after_restart.consent_handoff == handoff, "consent handoff survives store restart")
    waiting_activation = restarted.reconcile_committed_consent(
        after_restart.session_id,
        expected_generation=after_restart.generation,
        publication_attempt=1,
        consent_card_message_id="501",
    )
    check(waiting_activation.state is BootstrapState.AWAITING_ACTIVATION, "consent reconciles to activation")
    active = restarted.activate_bound_customer(
        waiting_activation.session_id,
        expected_generation=waiting_activation.generation,
        customer_key="client_001",
        customer_user_id="100",
        owner_id="900",
    )
    check(active.state is BootstrapState.ACTIVE, "bound customer activation succeeds once")
    replay = restarted.activate_bound_customer(
        active.session_id,
        expected_generation=active.generation,
        customer_key="client_001",
        customer_user_id="100",
        owner_id="900",
    )
    check(replay == active, "activation replay is idempotent")
    expect(
        GenerationConflict,
        "stale activation generation is denied",
        lambda: restarted.activate_bound_customer(
            active.session_id,
            expected_generation=waiting_activation.generation - 1,
            customer_key="client_001",
            customer_user_id="100",
            owner_id="900",
        ),
    )
    SCENARIOS["invite_claim_consent_activation"] = "PASS"


def test_expiry_uncertain_capacity_symlink_permissions() -> None:
    now = [datetime(2026, 9, 1, tzinfo=timezone.utc)]
    expiry_store = RoomBootstrapStore(RUN_ROOT / "expiry" / "state", now=lambda: now[0])
    draft = CustomerDraft(**draft_payload())
    prepared = expiry_store.prepare_rehearsal_customer_invite(
        draft, bot_username="test_bot", owner_id="900"
    )
    expiry_token = token_from_link(prepared.customer_link)
    now[0] += timedelta(hours=24, seconds=1)
    expect(
        BootstrapError,
        "expired invite cannot be claimed",
        lambda: expiry_store.rehearsal_customer_invite_session(expiry_token),
    )
    expired = expiry_store.get(prepared.session.session_id)
    check(expired.state is BootstrapState.EXPIRED, "expired prepared invite is terminal")

    uncertain = RoomBootstrapStore(RUN_ROOT / "uncertain" / "state")
    prepared_uncertain = uncertain.prepare_rehearsal_customer_invite(
        draft, bot_username="test_bot", owner_id="900"
    )
    awaiting = await_consent(uncertain, token_from_link(prepared_uncertain.customer_link))
    first_publication = uncertain.reserve_consent_publication(
        awaiting.session_id, expected_generation=awaiting.generation
    )
    first_attempt = uncertain.reserve_recovery_attempt(
        first_publication.session_id,
        slot=RecoverySlot.CONSENT_CARD,
        chat_id="100",
        expected_generation=first_publication.generation,
    )
    recovered_once = uncertain.recover_uncertain_consent_publication(
        first_publication.session_id,
        expected_generation=uncertain.get(first_publication.session_id).generation,
    )
    check(
        recovered_once.state is BootstrapState.AWAITING_CONSENT
        and recovered_once.failure_code == "consent_publication_superseded",
        "first uncertain consent publication is superseded",
    )
    second_publication = uncertain.reserve_consent_publication(
        recovered_once.session_id, expected_generation=recovered_once.generation
    )
    second_attempt = uncertain.reserve_recovery_attempt(
        second_publication.session_id,
        slot=RecoverySlot.CONSENT_CARD,
        chat_id="100",
        expected_generation=second_publication.generation,
    )
    exhausted = uncertain.recover_uncertain_consent_publication(
        second_publication.session_id,
        expected_generation=uncertain.get(second_publication.session_id).generation,
    )
    check(
        exhausted.state is BootstrapState.FAILED
        and exhausted.failure_code == "consent_publication_uncertain_exhausted",
        "second uncertain consent publication fails closed",
    )

    capacity_root = RUN_ROOT / "capacity"
    capacity_draft = make_invite_profile(
        capacity_root,
        capacity=1,
        enabled_customers=[{"customer_key": "other", "enabled": True}],
    )
    capacity_draft.write_text(json.dumps(draft_payload("client_002")), encoding="utf-8")
    capacity_draft.chmod(0o600)
    capacity = invoke(
        "-m", "gateway.platforms.dualcoach_admin", "customer", "invite",
        "--profile-root", str(capacity_root), "--draft", str(capacity_draft), "--json",
    )
    check(capacity.returncode != 0 and "capacity" in capacity.stderr, "full capacity denies invite CLI")

    permission_root = RUN_ROOT / "permission"
    permission_draft = make_invite_profile(permission_root)
    permission_draft.chmod(0o644)
    permission = invoke(
        "-m", "gateway.platforms.dualcoach_admin", "customer", "invite",
        "--profile-root", str(permission_root), "--draft", str(permission_draft), "--json",
    )
    check(permission.returncode != 0 and "customer draft" in permission.stderr, "non-private draft denies invite CLI")

    outside = RUN_ROOT / "outside"
    private_dir(outside)
    symlink_root = RUN_ROOT / "symlink-profile"
    private_dir(symlink_root)
    (symlink_root / "data").symlink_to(outside, target_is_directory=True)
    expect(
        ValueError,
        "onboarding data symlink is denied",
        lambda: NutritionOnboardingService(
            profile_root=symlink_root,
            customer_key="client_001",
            enforce_current_authority=False,
        ),
    )
    check(not (outside / "customers").exists(), "symlink denial leaves external location untouched")
    SCENARIOS["expiry_uncertain_capacity_symlink_permissions"] = "PASS"


def customer_authority() -> OnboardingAuthority:
    return OnboardingAuthority(
        customer_key="client_001",
        customer_user_id=100,
        customer_chat_id=-100,
        customer_topic_id=20,
        owner_user_id=900,
        owner_chat_id=-100,
        owner_topic_id=22,
        consent_notice_version="privacy-v1",
        consent_granted=True,
        customer_enabled=False,
    )


def evidence(*, actor: int = 100, topic: int = 20, message: int, update: int) -> MessageEvidence:
    return MessageEvidence(
        actor_user_id=actor,
        chat_id=-100,
        topic_id=topic,
        message_id=message,
        update_id=update,
    )


def test_22_answers_restart_revise_attest_review() -> None:
    root = RUN_ROOT / "onboarding"
    private_dir(root)
    auth = customer_authority()
    service = NutritionOnboardingService(
        profile_root=root,
        customer_key="client_001",
        enforce_current_authority=False,
    )
    denied_auth = auth.model_copy(update={"consent_granted": False})
    expect(
        ValueError,
        "absent privacy consent blocks onboarding",
        lambda: service.start_or_resume(
            authority=denied_auth, evidence=evidence(message=1, update=1)
        ),
    )
    started = service.start_or_resume(authority=auth, evidence=evidence(message=1, update=1))
    check(started.answer_count == 0 and started.next_field == QUESTION_FIELDS[0], "onboarding starts at first answer")
    expect(
        ValueError,
        "wrong customer route is denied",
        lambda: service.submit_answer(
            field=QUESTION_FIELDS[0], value=example_answer(QUESTION_FIELDS[0]),
            authority=auth, evidence=evidence(actor=101, message=2, update=2),
        ),
    )
    first = service.submit_answer(
        field=QUESTION_FIELDS[0], value=example_answer(QUESTION_FIELDS[0]),
        authority=auth, evidence=evidence(message=2, update=2),
    )
    check(first.answer_count == 1, "first onboarding answer persists")
    restarted = NutritionOnboardingService(
        profile_root=root,
        customer_key="client_001",
        enforce_current_authority=False,
    )
    resumed = restarted.start_or_resume(authority=auth, evidence=evidence(message=3, update=3))
    check(resumed.answer_count == 1 and resumed.next_field == QUESTION_FIELDS[1], "onboarding restart resumes redacted cursor")
    expect(
        ValueError,
        "out-of-order answer is denied",
        lambda: restarted.submit_answer(
            field=QUESTION_FIELDS[2], value=example_answer(QUESTION_FIELDS[2]),
            authority=auth, evidence=evidence(message=4, update=4),
        ),
    )
    for index, field in enumerate(QUESTION_FIELDS[1:], start=1):
        status = restarted.submit_answer(
            field=field,
            value=example_answer(field),
            authority=auth,
            evidence=evidence(message=index + 4, update=index + 4),
        )
        check(status.answer_count == index + 1, f"all-22 answer {index + 1}: {field}")
    check(status.state.value == "customer_attestation", "22 answers reach customer attestation")
    answers = restarted.reconciliation_answers(authority=auth)
    digest = canonical_digest(answers)
    revised = restarted.revise_reconciliation_answer(
        field="meal_count",
        value=4,
        expected_answers_digest=digest,
        authority=auth,
        evidence=evidence(message=50, update=50),
    )
    check(revised.state.value == "customer_attestation", "customer reconciliation revision returns to attestation")
    attested = restarted.attest_baseline(
        authority=auth, evidence=evidence(message=51, update=51)
    )
    check(attested.state.value == "owner_review", "customer attestation transfers only to owner review")
    approved = restarted.review_as_owner(
        decision="approved",
        authority=auth,
        evidence=evidence(actor=900, topic=22, message=52, update=52),
    )
    check(approved.state.value == "finalizing", "owner review approval reaches finalization")
    SCENARIOS["onboarding_22_restart_revise_attest_owner_review"] = "PASS"


def answer_wizard(service: WizardService, context: WizardContext, result, action: str, value: str | None, label: str):
    updated = service.answer(context, result.session_id, result.version, action, value)
    check(updated.status is WizardStatus.ADVANCED, label)
    return updated


def test_navigation_pause_resume_save_edit_dedup_safety() -> None:
    root = RUN_ROOT / "first-checkin"
    context = WizardContext(owner_id="owner", topic_id="topic")
    service = WizardService.for_standalone(root)
    result = service.start_nutrition(context, "2026-09-01")
    check(result.status is WizardStatus.ADVANCED and result.step == "bodyweight", "first check-in opens nutrition wizard")
    result = answer_wizard(service, context, result, "value", "70", "first check-in bodyweight")
    result = service.answer(context, result.session_id, result.version, "previous")
    check(result.status is WizardStatus.ADVANCED and result.step == "bodyweight", "wizard navigation returns to prior question")
    result = answer_wizard(service, context, result, "value", "71", "wizard replaces navigated answer")
    deferred = service.answer(context, result.session_id, result.version, "defer")
    check(deferred.status is WizardStatus.DEFERRED, "wizard pause defers durable draft")
    resumed_service = WizardService.for_standalone(root)
    result = resumed_service.start_nutrition(context, "2026-09-01")
    check(
        result.status is WizardStatus.ADVANCED and result.session_id == deferred.session_id and result.step == "calories",
        "wizard resume restores paused draft",
    )
    for action, value, label in (
        ("value", "2200", "calories"),
        ("value", "250 160 60", "macros"),
        ("value", "synthetic meals", "meals"),
        ("value", "2", "water"),
        ("value", "7", "sleep duration"),
        ("select", "4", "sleep quality"),
        ("select", "normal", "digestion"),
        ("select", "4", "condition"),
        ("value", "appetite 3 stress 2", "appetite stress"),
        ("select", "none", "training summary"),
        ("select", "skip", "optional note"),
    ):
        result = answer_wizard(resumed_service, context, result, action, value, f"first check-in {label}")
    check(result.step == "summary", "completed first check-in reaches summary")
    saved = resumed_service.answer(context, result.session_id, result.version, "save")
    check(saved.status is WizardStatus.SAVED, "first check-in saves canonical event")
    event = resumed_service.finalized_event(saved.session_id)
    check(event is not None and event.event_type.value == "nutrition_checkin", "saved first check-in has nutrition event")
    replay = resumed_service.start_nutrition(context, "2026-09-01")
    check(replay.status is WizardStatus.SAVED and replay.session_id == saved.session_id, "saved first check-in replay is deduplicated")
    correction = resumed_service.start_nutrition_correction(context, "2026-09-01")
    check(correction.status is WizardStatus.ADVANCED and correction.step == "summary", "first check-in correction opens at edit summary")
    correction = resumed_service.answer(correction_context := context, correction.session_id, correction.version, "edit", "calories")
    check(correction.status is WizardStatus.ADVANCED and correction.step == "calories", "first check-in edit selects calories")
    correction = answer_wizard(resumed_service, context, correction, "value", "2300", "first check-in edit replaces calories")
    edited = resumed_service.answer(context, correction.session_id, correction.version, "save")
    check(edited.status is WizardStatus.SAVED, "edited first check-in saves correction")
    edited_event = resumed_service.finalized_event(edited.session_id)
    check(edited_event is not None and edited_event.supersedes == event.event_id, "edited check-in supersedes original exactly")

    cli_home = RUN_ROOT / "record-cli"
    first = invoke(
        "-m", "checkin_cli.cli", "--home", str(cli_home), "record",
        "--message-id", "local-1", "--received-at", "2026-09-01T08:00:00+09:00",
        "--text", "체중: 70\n칼로리: 2200\n수면: 7\n운동: 휴식",
    )
    check(first.returncode == 0, f"installed first-check-in CLI save: {first.stderr}")
    first_payload = json.loads(first.stdout)
    check(first_payload["outcome"] == "recorded", "CLI first check-in records once")
    duplicate = invoke(
        "-m", "checkin_cli.cli", "--home", str(cli_home), "record",
        "--message-id", "local-1", "--received-at", "2026-09-01T08:00:00+09:00",
        "--text", "체중: 70\n칼로리: 2200\n수면: 7\n운동: 휴식",
    )
    duplicate_payload = json.loads(duplicate.stdout)
    check(duplicate.returncode == 0 and duplicate_payload["outcome"] == "duplicate", "CLI exact first-check-in duplicate is idempotent")

    safety = WizardService.for_standalone(RUN_ROOT / "safety")
    safety_result = safety.start_morning(context, "2026-09-02")
    for action, value, label in (
        ("value", "70", "safety bodyweight"),
        ("value", "7", "safety sleep duration"),
        ("select", "4", "safety sleep quality"),
        ("select", "4", "safety condition"),
    ):
        safety_result = answer_wizard(safety, context, safety_result, action, value, label)
    stopped = safety.answer(safety_result and context, safety_result.session_id, safety_result.version, "value", "chest pain")
    check(stopped.status is WizardStatus.SAFETY_STOP and stopped.step == "safety_ack", "safety signal stops check-in")
    audited = safety.answer(context, stopped.session_id, stopped.version, "acknowledge")
    check(audited.status is WizardStatus.SAVED, "safety stop requires and accepts audit acknowledgement")
    SCENARIOS["navigation_pause_resume_first_checkin_save_edit_dedup_safety"] = "PASS"


def test_membership_and_notice_defer_drain() -> None:
    journal = MembershipJournal(RUN_ROOT / "membership" / "events.jsonl")
    first = journal.append_transition(
        update_id=1,
        chat_id=-100,
        customer_user_id=100,
        old_status="member",
        new_status="left",
        event_date_utc="2026-09-01",
        subscription_epoch_id="local-epoch",
    )
    replay = journal.append_transition(
        update_id=1,
        chat_id=-100,
        customer_user_id=100,
        old_status="member",
        new_status="left",
        event_date_utc="2026-09-01",
        subscription_epoch_id="local-epoch",
    )
    check(replay == first and len(journal.verify()) == 1, "membership evidence journal is idempotent and hash-valid")
    expect(
        MembershipGateError,
        "conflicting duplicate membership evidence is denied",
        lambda: journal.append_transition(
            update_id=1,
            chat_id=-101,
            customer_user_id=100,
            old_status="member",
            new_status="left",
            event_date_utc="2026-09-01",
            subscription_epoch_id="local-epoch",
        ),
    )

    profile = RUN_ROOT / "notice"
    notice_store = ActivationNoticeStore(profile)
    receipt = notice_store.reserve(
        customer_key="client_001",
        starts_on="2026-09-03",
        daily_time="08:00",
        destination={"user_id": "100", "chat_id": "100", "topic_id": "0"},
        authority_digest="a" * 64,
    )
    adapter = object.__new__(TelegramAdapter)
    calls: list[dict[str, str]] = []

    async def fake_send_nutrition_topic(**kwargs):
        calls.append(kwargs)
        return SimpleNamespace(message_id=700)

    adapter._send_nutrition_topic = fake_send_nutrition_topic
    adapter._nutrition_delivery_receipt = lambda sent: str(sent.message_id)
    deferred = asyncio.run(
        adapter._drain_activation_completion_notices(profile, kst_date=datetime(2026, 9, 2).date())
    )
    check(deferred == () and calls == [] and notice_store.latest()[0].state == "prepared", "activation notice defers before start date")
    drained = asyncio.run(
        adapter._drain_activation_completion_notices(profile, kst_date=datetime(2026, 9, 3).date())
    )
    check(drained == () and len(calls) == 1 and notice_store.latest()[0].state == "sent_audited", "activation notice drain sends and audits once")
    drained_replay = asyncio.run(
        adapter._drain_activation_completion_notices(profile, kst_date=datetime(2026, 9, 3).date())
    )
    check(drained_replay == () and len(calls) == 1 and receipt.reservation_id == notice_store.latest()[0].reservation_id, "activation notice drain replay is at-most-once")
    SCENARIOS["membership_evidence_notice_defer_drain"] = "PASS"


def assert_installed_wheels() -> None:
    import checkin_cli
    import gateway

    site = str(QA_ROOT / "venv" / "lib" / "python3.12" / "site-packages")
    check(str(Path(checkin_cli.__file__).resolve()).startswith(site), "checkin CLI imports from installed candidate wheel")
    check(str(Path(gateway.__file__).resolve()).startswith(site), "gateway imports from installed candidate wheel")


def block_network() -> None:
    def denied_connect(self, address):  # type: ignore[no-untyped-def]
        raise AssertionError(f"network is prohibited in installed-wheel QA: {address!r}")

    socket.socket.connect = denied_connect  # type: ignore[assignment]


def main() -> None:
    if RUN_ROOT.exists():
        for path in sorted(RUN_ROOT.rglob("*"), reverse=True):
            if path.is_symlink() or path.is_file():
                path.parent.chmod(stat.S_IRWXU)
                path.unlink()
            elif path.is_dir():
                path.chmod(stat.S_IRWXU)
                path.rmdir()
    private_dir(RUN_ROOT)
    assert_installed_wheels()
    block_network()
    test_invite_claim_consent_activation()
    test_expiry_uncertain_capacity_symlink_permissions()
    test_22_answers_restart_revise_attest_review()
    test_navigation_pause_resume_save_edit_dedup_safety()
    test_membership_and_notice_defer_drain()
    payload = {
        "status": "PASS",
        "explicit_assertions": CHECKS,
        "scenarios": SCENARIOS,
        "network": "blocked in-process; all delivery used a fake local method",
        "roots": [str(RUN_ROOT)],
    }
    RESULT_PATH.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(json.dumps(payload, sort_keys=True))


if __name__ == "__main__":
    main()