"""Owner/customer-only Telegram bootstrap authority for DualCoach v1."""

from __future__ import annotations

import fcntl
import hashlib
import hmac
import json
import os
import re
import secrets
import stat
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass, replace
from datetime import datetime, timedelta, timezone
from enum import StrEnum
from pathlib import Path
from typing import Callable, Final, Iterator, Mapping, cast
from urllib.parse import quote


ROOM_BOOTSTRAP_RELATIVE_PATH: Final = Path(
    "data/onboarding/telegram-customer-bootstrap-v1"
)
_INVITE_TTL: Final = timedelta(hours=24)
_TOKEN_RE: Final = re.compile(r"^rc1_(?P<sid>[A-Za-z0-9_-]{22})$")
_SID_RE: Final = re.compile(r"^[A-Za-z0-9_-]{22}$")
_SESSION_RE: Final = re.compile(r"^cb_[A-Za-z0-9_-]{22}$")
_BOT_RE: Final = re.compile(r"^[A-Za-z0-9_]{5,64}$")
_DIGEST_RE: Final = re.compile(r"^[a-f0-9]{64}$")
_ID_RE: Final = re.compile(r"^-?[0-9]+$")
_LEDGER_SCHEMA: Final = "telegram-customer-bootstrap-v1"
_JOURNAL_SCHEMA: Final = "telegram-customer-bootstrap-journal-v1"
_JOURNAL_HEAD_SCHEMA: Final = "telegram-customer-bootstrap-journal-head-v1"
_JOURNAL_MARKER_SCHEMA: Final = "telegram-customer-bootstrap-journal-required-v1"
_TERMINAL_STATES: Final = frozenset({"ACTIVE", "CANCELLED", "EXPIRED", "FAILED"})


def customer_consent_callback(
    customer_key: object,
    notice_version: object,
    action: str,
) -> str:
    """Build the one canonical customer consent callback contract."""
    if action not in {"g", "d"}:
        raise ValueError("unsupported customer consent action")
    key = str(customer_key or "").strip()
    version = str(notice_version or "").strip()
    if not key or not version:
        raise ValueError("customer consent callback inputs are required")
    token = hashlib.sha256(
        f"customer-consent\0{key}\0{version}".encode("utf-8")
    ).hexdigest()[:24]
    callback = f"cc1:{token}:{action}"
    if len(callback.encode("utf-8")) > 64:
        raise ValueError("customer consent callback exceeds Telegram limit")
    return callback


def room_bootstrap_state_dir(profile_root: Path | str) -> Path:
    return Path(profile_root) / ROOM_BOOTSTRAP_RELATIVE_PATH


class BootstrapError(ValueError):
    pass


class BootstrapConflict(BootstrapError):
    pass


class GenerationConflict(BootstrapError):
    pass


class BootstrapState(StrEnum):
    PREPARED = "PREPARED"
    REGISTERING = "REGISTERING"
    AWAITING_CONSENT = "AWAITING_CONSENT"
    AWAITING_ACTIVATION = "AWAITING_ACTIVATION"
    ACTIVE = "ACTIVE"
    CANCELLED = "CANCELLED"
    EXPIRED = "EXPIRED"
    FAILED = "FAILED"


class Role(StrEnum):
    CUSTOMER = "customer"


class RecoverySlot(StrEnum):
    CONSENT_CARD = "consent_card"


@dataclass(frozen=True, slots=True)
class RoleClaim:
    role: Role
    user_id: str
    chat_id: str
    topic_id: str
    message_id: str


@dataclass(frozen=True, slots=True)
class RecoveryAttempt:
    session_id: str
    slot: RecoverySlot
    chat_id: str
    generation: int
    created_at: datetime


@dataclass(frozen=True, slots=True)
class ConsentHandoff:
    update_id: int
    actor_id: int
    chat_id: int
    topic_id: int
    message_id: int
    callback_data: str
    bootstrap_generation: int
    recorded_at: datetime
    provenance_digest: str


@dataclass(frozen=True, slots=True)
class CustomerDraft:
    customer_key: str
    display_name: str
    starts_on: str
    daily_time: str
    weekly_weekday: int
    monthly_day: int
    calories_kcal: int
    protein_g: int
    meals: tuple[str, ...]
    primary_goal: str = "미정"
    dietary_restrictions: tuple[str, ...] = ()
    allergies: tuple[str, ...] = ()
    food_preferences: tuple[str, ...] = ()
    supplements: tuple[str, ...] = ()
    digestion_context: str | None = None
    sleep_goal_hours: float | None = None
    recovery_goal: str | None = None
    training_context: str | None = None
    carbohydrate_g: int | None = None
    fat_g: int | None = None
    water_liters: float | None = None
    customer_user_id: str | None = None

    @property
    def canonical_json(self) -> str:
        return _canonical_json(self.to_dict())

    @property
    def digest(self) -> str:
        return _sha256(self.canonical_json.encode("utf-8"))

    def to_dict(self) -> dict[str, object]:
        return {
            "customer_key": self.customer_key,
            "display_name": self.display_name,
            "starts_on": self.starts_on,
            "daily_time": self.daily_time,
            "weekly_weekday": self.weekly_weekday,
            "monthly_day": self.monthly_day,
            "calories_kcal": self.calories_kcal,
            "protein_g": self.protein_g,
            "meals": list(self.meals),
            "primary_goal": self.primary_goal,
            "dietary_restrictions": list(self.dietary_restrictions),
            "allergies": list(self.allergies),
            "food_preferences": list(self.food_preferences),
            "supplements": list(self.supplements),
            "digestion_context": self.digestion_context,
            "sleep_goal_hours": self.sleep_goal_hours,
            "recovery_goal": self.recovery_goal,
            "training_context": self.training_context,
            "carbohydrate_g": self.carbohydrate_g,
            "fat_g": self.fat_g,
            "water_liters": self.water_liters,
            "customer_user_id": self.customer_user_id,
        }


@dataclass(frozen=True, slots=True)
class RoomBootstrapSession:
    session_id: str
    sid_hash: str
    customer_draft: CustomerDraft
    customer_draft_digest: str
    bot_username: str
    state: BootstrapState
    generation: int
    created_at: datetime
    updated_at: datetime
    expires_at: datetime
    owner_id: str
    first_claim: bool = False
    role_claims: tuple[RoleClaim, ...] = ()
    consent_publication_attempt: int = 0
    consent_card_message_id: str | None = None
    recovery_attempt_generation: int = 0
    recovery_attempts: tuple[RecoveryAttempt, ...] = ()
    failure_code: str | None = None
    committed_consent_card_message_id: str | None = None
    consent_handoff: ConsentHandoff | None = None
    consent_recovery_reconciled: bool = False

    @property
    def customer_key(self) -> str:
        return self.customer_draft.customer_key

    @property
    def terminal(self) -> bool:
        return self.state.value in _TERMINAL_STATES

    @property
    def chat_id(self) -> None:
        """Private-DM bootstrap has no shared room authority."""
        return None

    def role_claim(self, role: Role) -> RoleClaim | None:
        return next((claim for claim in self.role_claims if claim.role is role), None)

    def recovery_attempt(self, slot: RecoverySlot) -> RecoveryAttempt | None:
        return next(
            (attempt for attempt in self.recovery_attempts if attempt.slot is slot),
            None,
        )


def consent_handoff_digest(
    session: RoomBootstrapSession,
    handoff: ConsentHandoff,
) -> str:
    """Bind exact Telegram evidence to one committed bootstrap generation."""
    return _sha256(
        _canonical_bytes(
            {
                "schema": "telegram-nutrition-consent-handoff-v1",
                "session_id": session.session_id,
                "customer_key": session.customer_key,
                "bootstrap_generation": handoff.bootstrap_generation,
                "update_id": handoff.update_id,
                "actor_id": handoff.actor_id,
                "chat_id": handoff.chat_id,
                "topic_id": handoff.topic_id,
                "message_id": handoff.message_id,
                "callback_data": handoff.callback_data,
                "recorded_at": handoff.recorded_at.astimezone(
                    timezone.utc
                ).isoformat(),
            }
        )
    )


@dataclass(frozen=True, slots=True)
class PreparedRehearsalCustomerInvite:
    session: RoomBootstrapSession
    customer_link: str
    expires_at: datetime


_DRAFT_REQUIRED: Final = frozenset(
    {
        "customer_key",
        "display_name",
        "starts_on",
        "daily_time",
        "weekly_weekday",
        "monthly_day",
        "calories_kcal",
        "protein_g",
        "meals",
    }
)
_DRAFT_OPTIONAL: Final = frozenset(CustomerDraft.__dataclass_fields__) - _DRAFT_REQUIRED


def load_customer_draft(path: Path | str) -> CustomerDraft:
    source = Path(path)
    _require_private_regular_file(source, "customer draft")
    try:
        value = json.loads(source.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise BootstrapError("customer draft JSON is invalid") from exc
    if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
        raise BootstrapError("customer draft must be an object")
    keys = set(value)
    if keys - _DRAFT_REQUIRED - _DRAFT_OPTIONAL:
        raise BootstrapError("customer draft has unknown fields")
    if _DRAFT_REQUIRED - keys:
        raise BootstrapError("customer draft is missing fields")
    return _draft_from_mapping(value)


class RoomBootstrapStore:
    """Private, digest-authenticated generation-CAS bootstrap ledger."""

    def __init__(
        self,
        state_dir: Path | str,
        *,
        now: Callable[[], datetime] | None = None,
    ) -> None:
        self.state_dir = Path(state_dir)
        self.ledger_path = self.state_dir / "ledger.json"
        self.journal_path = self.state_dir / "events.jsonl"
        self.journal_head_path = self.state_dir / "events-head.json"
        if (
            self.state_dir.name == Path(ROOM_BOOTSTRAP_RELATIVE_PATH).name
            and len(self.state_dir.parents) >= 3
        ):
            marker_root = (
                self.state_dir.parents[2]
                / ".nutricoach-bootstrap-authority-v1"
            )
        else:
            marker_root = (
                self.state_dir.parent
                / f".{self.state_dir.name}.authority-v1"
            )
        self.journal_marker_path = (
            marker_root / "journal-required.json"
        )
        self.lock_path = self.state_dir / "ledger.lock"
        self._now = now or (lambda: datetime.now(timezone.utc))
        self._prepare_paths()
        with self._locked():
            if (
                self.ledger_path.exists()
                or self.journal_path.exists()
                or self.journal_head_path.exists()
                or self.journal_marker_path.exists()
            ):
                self._read_unlocked()
            else:
                self._write_unlocked(())

    def prepare_rehearsal_customer_invite(
        self,
        draft: CustomerDraft,
        *,
        bot_username: str,
        owner_id: str,
        first_claim: bool = False,
    ) -> PreparedRehearsalCustomerInvite:
        username = bot_username.removeprefix("@").strip()
        if _BOT_RE.fullmatch(username) is None:
            raise BootstrapError("invalid Telegram bot username")
        owner = _telegram_id(owner_id, "owner")
        if type(first_claim) is not bool:
            raise BootstrapError("customer bootstrap claim policy is invalid")
        if first_claim:
            if draft.customer_user_id is not None:
                raise BootstrapError("first-claim invite must be unbound")
        else:
            if draft.customer_user_id is None:
                raise BootstrapError("intended customer identity is required")
            intended_customer = _telegram_id(
                draft.customer_user_id,
                "intended customer",
            )
            if intended_customer == owner:
                raise BootstrapError("customer bootstrap invite is unavailable")
        sid = secrets.token_urlsafe(16)
        if _SID_RE.fullmatch(sid) is None:
            raise RuntimeError("generated customer bootstrap token is invalid")
        now = self._utc_now()
        session = RoomBootstrapSession(
            session_id="cb_" + secrets.token_urlsafe(16),
            sid_hash=_sha256(sid.encode("ascii")),
            customer_draft=draft,
            customer_draft_digest=draft.digest,
            bot_username=username,
            state=BootstrapState.PREPARED,
            generation=1,
            created_at=now,
            updated_at=now,
            expires_at=now + _INVITE_TTL,
            owner_id=owner,
            first_claim=first_claim,
        )
        with self._locked():
            sessions = list(self._expire_values(self._read_unlocked(), now))
            if any(
                not item.terminal and item.customer_key == draft.customer_key
                for item in sessions
            ):
                raise BootstrapConflict(
                    "customer already has a nonterminal bootstrap session"
                )
            sessions.append(session)
            self._write_unlocked(sessions)
        return PreparedRehearsalCustomerInvite(
            session=session,
            customer_link=(
                f"https://t.me/{username}?start=rc1_{quote(sid, safe='-_')}"
            ),
            expires_at=session.expires_at,
        )

    def rehearsal_customer_invite_session(self, token: str) -> RoomBootstrapSession:
        match = _TOKEN_RE.fullmatch(token) if isinstance(token, str) else None
        if match is None:
            raise BootstrapError("customer bootstrap invite is invalid")
        self.expire_unbound()
        session = self.get_by_sid(match.group("sid"))
        if (
            session is None
            or session.state is not BootstrapState.PREPARED
            or session.role_claims
            or (
                session.first_claim
                and session.customer_draft.customer_user_id is not None
            )
            or (
                not session.first_claim
                and session.customer_draft.customer_user_id is None
            )
        ):
            raise BootstrapError("customer bootstrap invite is unavailable")
        return session

    def claim_rehearsal_customer_invite(
        self,
        token: str,
        *,
        user_id: str,
        chat_id: str,
        message_id: str,
    ) -> RoomBootstrapSession:
        user = _telegram_id(user_id, "customer")
        chat = _telegram_id(chat_id, "customer chat")
        message = _telegram_id(message_id, "customer message")
        if user != chat:
            raise BootstrapError("customer bootstrap invite requires a private DM")
        supplied = self.rehearsal_customer_invite_session(token)
        now = self._utc_now()
        with self._locked():
            sessions = list(self._expire_values(self._read_unlocked(), now))
            current = self._find(tuple(sessions), supplied.session_id)
            if (
                current.state is not BootstrapState.PREPARED
                or current.role_claims
                or user == current.owner_id
                or current.generation != supplied.generation
                or any(
                    item.session_id != current.session_id
                    and not item.terminal
                    and any(claim.user_id == user for claim in item.role_claims)
                    for item in sessions
                )
            ):
                raise BootstrapError("customer bootstrap invite is unavailable")
            draft = current.customer_draft
            if current.first_claim:
                if draft.customer_user_id is not None:
                    raise BootstrapError(
                        "customer bootstrap invite is unavailable"
                    )
                draft = replace(draft, customer_user_id=user)
            elif draft.customer_user_id != user:
                raise BootstrapError("customer bootstrap invite is unavailable")
            updated = replace(
                current,
                state=BootstrapState.REGISTERING,
                generation=current.generation + 1,
                updated_at=now,
                customer_draft=draft,
                customer_draft_digest=draft.digest,
                role_claims=(RoleClaim(Role.CUSTOMER, user, chat, "0", message),),
            )
            sessions[sessions.index(current)] = updated
            self._write_unlocked(sessions)
            return updated

    def list_sessions(self) -> tuple[RoomBootstrapSession, ...]:
        with self._locked():
            return self._read_unlocked()

    def get(self, session_id: str) -> RoomBootstrapSession:
        with self._locked():
            return self._find(self._read_unlocked(), session_id)

    def get_by_sid(self, sid: str) -> RoomBootstrapSession | None:
        if _SID_RE.fullmatch(sid) is None:
            return None
        digest = _sha256(sid.encode("ascii"))
        with self._locked():
            matches = [
                item
                for item in self._read_unlocked()
                if hmac.compare_digest(item.sid_hash, digest)
            ]
        if len(matches) > 1:
            raise BootstrapConflict("customer bootstrap token is ambiguous")
        return matches[0] if matches else None

    def get_by_sid_hash(self, sid_hash: str) -> RoomBootstrapSession | None:
        if _DIGEST_RE.fullmatch(sid_hash) is None:
            return None
        with self._locked():
            matches = [
                item
                for item in self._read_unlocked()
                if hmac.compare_digest(item.sid_hash, sid_hash)
            ]
        if len(matches) > 1:
            raise BootstrapConflict("customer bootstrap hash is ambiguous")
        return matches[0] if matches else None

    def get_by_chat(self, chat_id: str) -> RoomBootstrapSession | None:
        chat = _telegram_id(chat_id, "chat")
        with self._locked():
            matches = [
                item
                for item in self._read_unlocked()
                if not item.terminal
                and any(claim.chat_id == chat for claim in item.role_claims)
            ]
        if len(matches) > 1:
            raise BootstrapConflict("chat has multiple bootstrap sessions")
        return matches[0] if matches else None

    def transition(
        self,
        session_id: str,
        *,
        expected_generation: int,
        target: BootstrapState,
    ) -> RoomBootstrapSession:
        allowed = {
            BootstrapState.REGISTERING: frozenset({BootstrapState.AWAITING_CONSENT}),
            BootstrapState.AWAITING_CONSENT: frozenset(
                {BootstrapState.AWAITING_ACTIVATION}
            ),
            BootstrapState.AWAITING_ACTIVATION: frozenset({BootstrapState.ACTIVE}),
        }

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            if target not in allowed.get(session.state, frozenset()):
                raise BootstrapError("customer bootstrap transition is invalid")
            return replace(
                session,
                state=target,
                generation=session.generation + 1,
                updated_at=now,
            )

        return self._mutate(session_id, expected_generation, mutate)

    def activate_bound_customer(
        self,
        session_id: str,
        *,
        expected_generation: int,
        customer_key: str,
        customer_user_id: str,
        owner_id: str,
    ) -> RoomBootstrapSession:
        """Commit the identity-bound activation lifecycle exactly once."""
        if type(expected_generation) is not int or expected_generation < 1:
            raise GenerationConflict("customer bootstrap generation is invalid")
        customer = _telegram_id(customer_user_id, "customer")
        owner = _telegram_id(owner_id, "owner")
        now = self._utc_now()
        with self._locked():
            sessions = list(self._read_unlocked())
            current = self._find(tuple(sessions), session_id)
            claim = current.role_claim(Role.CUSTOMER)
            identity_matches = (
                current.customer_key == customer_key
                and current.customer_draft.customer_user_id == customer
                and current.owner_id == owner
                and claim is not None
                and len(current.role_claims) == 1
                and claim.user_id == customer
                and claim.chat_id == customer
                and claim.topic_id == "0"
            )
            if not identity_matches:
                raise BootstrapConflict(
                    "customer bootstrap activation identity is unavailable"
                )
            active_matches = tuple(
                item
                for item in sessions
                if item.customer_key == customer_key
                and item.state is BootstrapState.ACTIVE
            )
            if current.state is BootstrapState.ACTIVE:
                if (
                    expected_generation not in {current.generation - 1, current.generation}
                    or active_matches != (current,)
                ):
                    raise GenerationConflict(
                        "customer bootstrap activation authority is stale"
                    )
                return current
            if (
                current.state is not BootstrapState.AWAITING_ACTIVATION
                or current.generation != expected_generation
                or active_matches
            ):
                raise GenerationConflict(
                    "customer bootstrap activation authority is stale"
                )
            updated = replace(
                current,
                state=BootstrapState.ACTIVE,
                generation=current.generation + 1,
                updated_at=now,
            )
            sessions[sessions.index(current)] = updated
            self._write_unlocked(sessions)
            return updated

    def bind_private_owner(
        self,
        session_id: str,
        *,
        owner_id: str,
        expected_generation: int,
    ) -> RoomBootstrapSession:
        owner = _telegram_id(owner_id, "owner")

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            if (
                session.state is not BootstrapState.AWAITING_CONSENT
                or session.owner_id != owner
            ):
                raise BootstrapError("canonical owner binding is unavailable")
            return session

        return self._mutate(session_id, expected_generation, mutate)

    def reserve_consent_publication(
        self,
        session_id: str,
        *,
        expected_generation: int,
    ) -> RoomBootstrapSession:
        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            if session.state is not BootstrapState.AWAITING_CONSENT:
                raise BootstrapError("consent publication requires awaiting consent")
            if session.recovery_attempts:
                raise BootstrapConflict("consent publication outcome is unresolved")
            return replace(
                session,
                consent_publication_attempt=session.consent_publication_attempt + 1,
                consent_card_message_id=None,
                generation=session.generation + 1,
                updated_at=now,
            )

        return self._mutate(session_id, expected_generation, mutate)

    def reserve_recovery_attempt(
        self,
        session_id: str,
        *,
        slot: RecoverySlot,
        chat_id: str,
        expected_generation: int,
    ) -> RecoveryAttempt:
        if slot is not RecoverySlot.CONSENT_CARD:
            raise BootstrapError("unsupported customer bootstrap recovery slot")
        chat = _telegram_id(chat_id, "chat")
        captured: list[RecoveryAttempt] = []

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            customer = session.role_claim(Role.CUSTOMER)
            if (
                session.state is not BootstrapState.AWAITING_CONSENT
                or session.recovery_attempts
                or customer is None
                or customer.chat_id != chat
                or session.consent_card_message_id is not None
            ):
                raise BootstrapConflict("consent provider authority is unavailable")
            attempt = RecoveryAttempt(
                session.session_id,
                slot,
                chat,
                session.recovery_attempt_generation + 1,
                now,
            )
            captured.append(attempt)
            return replace(
                session,
                recovery_attempt_generation=attempt.generation,
                recovery_attempts=(attempt,),
                failure_code=None,
                updated_at=now,
            )

        self._mutate(session_id, expected_generation, mutate)
        return captured[0]

    def bind_recovery_receipt(
        self,
        session_id: str,
        *,
        slot: RecoverySlot,
        chat_id: str,
        attempt_generation: int,
        receipt_id: str,
    ) -> RoomBootstrapSession:
        chat = _telegram_id(chat_id, "chat")
        receipt = _telegram_id(receipt_id, "receipt")

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            self._require_attempt(session, slot, chat, attempt_generation)
            if session.consent_card_message_id is not None:
                raise BootstrapConflict("consent receipt is already bound")
            return replace(
                session,
                consent_card_message_id=receipt,
                recovery_attempts=(),
                failure_code=None,
                updated_at=now,
            )

        current = self.get(session_id)
        return self._mutate(session_id, current.generation, mutate)

    def acknowledge_recovery_no_side_effect(
        self,
        session_id: str,
        *,
        slot: RecoverySlot,
        chat_id: str,
        attempt_generation: int,
    ) -> RoomBootstrapSession:
        chat = _telegram_id(chat_id, "chat")

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            self._require_attempt(session, slot, chat, attempt_generation)
            return replace(session, recovery_attempts=(), updated_at=now)

        current = self.get(session_id)
        return self._mutate(session_id, current.generation, mutate)

    def recover_uncertain_consent_publication(
        self,
        session_id: str,
        *,
        expected_generation: int,
    ) -> RoomBootstrapSession:
        """Supersede one unknown card once, then terminate a second unknown."""

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            attempt = session.recovery_attempt(RecoverySlot.CONSENT_CARD)
            customer = session.role_claim(Role.CUSTOMER)
            if (
                session.state is not BootstrapState.AWAITING_CONSENT
                or len(session.recovery_attempts) != 1
                or attempt is None
                or customer is None
                or attempt.chat_id != customer.chat_id
                or attempt.generation != session.recovery_attempt_generation
                or session.consent_card_message_id is not None
                or session.consent_publication_attempt < 1
            ):
                raise BootstrapConflict(
                    "uncertain consent publication authority is unavailable"
                )
            exhausted = session.consent_publication_attempt >= 2
            return replace(
                session,
                state=(
                    BootstrapState.FAILED
                    if exhausted
                    else BootstrapState.AWAITING_CONSENT
                ),
                recovery_attempts=(),
                failure_code=(
                    "consent_publication_uncertain_exhausted"
                    if exhausted
                    else "consent_publication_superseded"
                ),
                generation=session.generation + 1,
                updated_at=now,
            )

        return self._mutate(session_id, expected_generation, mutate)

    def reconcile_committed_consent(
        self,
        session_id: str,
        *,
        expected_generation: int,
        publication_attempt: int,
        consent_card_message_id: str,
    ) -> RoomBootstrapSession:
        receipt = _telegram_id(consent_card_message_id, "consent receipt")

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            handoff = session.consent_handoff
            if (
                session.state is not BootstrapState.AWAITING_CONSENT
                or session.recovery_attempts
                or publication_attempt != session.consent_publication_attempt
                or receipt != session.consent_card_message_id
                or handoff is None
                or handoff.message_id != int(receipt)
                or handoff.bootstrap_generation != session.generation
            ):
                raise GenerationConflict(
                    "committed consent handoff authority is stale"
                )
            return replace(
                session,
                state=BootstrapState.AWAITING_ACTIVATION,
                consent_card_message_id=None,
                committed_consent_card_message_id=receipt,
                generation=session.generation + 1,
                updated_at=now,
            )

        return self._mutate(session_id, expected_generation, mutate)

    def bind_consent_handoff(
        self,
        session_id: str,
        *,
        expected_generation: int,
        handoff: ConsentHandoff,
    ) -> RoomBootstrapSession:
        """Persist one authenticated Telegram consent event for exact replay."""

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            from checkin_cli.customer_coaching import CONSENT_VERSION

            customer = session.role_claim(Role.CUSTOMER)
            message_id = (
                session.consent_card_message_id
                if session.state is BootstrapState.AWAITING_CONSENT
                else session.committed_consent_card_message_id
            )
            expected_callback = customer_consent_callback(
                session.customer_key,
                CONSENT_VERSION,
                "g",
            )
            if (
                session.state
                not in {
                    BootstrapState.AWAITING_CONSENT,
                    BootstrapState.AWAITING_ACTIVATION,
                }
                or customer is None
                or handoff.actor_id != int(customer.user_id)
                or handoff.chat_id != int(customer.chat_id)
                or handoff.topic_id != int(customer.topic_id)
                or str(handoff.message_id) != message_id
                or handoff.bootstrap_generation != session.generation
                or not hmac.compare_digest(
                    handoff.callback_data,
                    expected_callback,
                )
                or handoff.recorded_at.tzinfo is None
                or _DIGEST_RE.fullmatch(handoff.provenance_digest) is None
                or not hmac.compare_digest(
                    handoff.provenance_digest,
                    consent_handoff_digest(session, handoff),
                )
            ):
                raise BootstrapConflict("consent handoff authority is unavailable")
            if session.consent_handoff == handoff:
                return session
            if session.consent_handoff is not None:
                raise BootstrapConflict("consent handoff authority conflicts")
            return replace(session, consent_handoff=handoff, updated_at=now)

        return self._mutate(session_id, expected_generation, mutate)

    def mark_consent_recovery_reconciled(
        self,
        session_id: str,
        *,
        expected_generation: int,
        provenance_digest: str,
    ) -> RoomBootstrapSession:
        """Mark the exact authenticated handoff reconciled after callback success."""

        def mutate(session: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            handoff = session.consent_handoff
            if (
                session.state is not BootstrapState.AWAITING_ACTIVATION
                or handoff is None
                or not hmac.compare_digest(
                    handoff.provenance_digest, provenance_digest
                )
            ):
                raise BootstrapConflict("consent recovery provenance is unavailable")
            return replace(
                session,
                consent_recovery_reconciled=True,
                updated_at=now,
            )

        return self._mutate(session_id, expected_generation, mutate)

    def expire_unbound(self) -> tuple[RoomBootstrapSession, ...]:
        now = self._utc_now()
        with self._locked():
            before = self._read_unlocked()
            after = self._expire_values(before, now)
            if after != before:
                self._write_unlocked(after)
            return after

    def unresolved_attempts(
        self,
        session_id: str | None = None,
    ) -> tuple[RecoveryAttempt, ...]:
        sessions = self.list_sessions()
        if session_id is not None:
            sessions = (self._find(sessions, session_id),)
        return tuple(
            attempt for session in sessions for attempt in session.recovery_attempts
        )

    @staticmethod
    def _require_attempt(
        session: RoomBootstrapSession,
        slot: RecoverySlot,
        chat_id: str,
        generation: int,
    ) -> RecoveryAttempt:
        matches = tuple(
            attempt
            for attempt in session.recovery_attempts
            if attempt.slot is slot
            and attempt.chat_id == chat_id
            and attempt.generation == generation
        )
        if len(matches) != 1:
            raise GenerationConflict("consent provider authority is stale")
        return matches[0]

    def _mutate(
        self,
        session_id: str,
        expected_generation: int,
        mutation: Callable[[RoomBootstrapSession, datetime], RoomBootstrapSession],
    ) -> RoomBootstrapSession:
        if type(expected_generation) is not int or expected_generation < 1:
            raise GenerationConflict("customer bootstrap generation is invalid")
        now = self._utc_now()
        with self._locked():
            sessions = list(self._read_unlocked())
            current = self._find(tuple(sessions), session_id)
            if current.generation != expected_generation:
                raise GenerationConflict("customer bootstrap generation is stale")
            updated = mutation(current, now)
            if updated.session_id != current.session_id:
                raise BootstrapError("customer bootstrap identity changed")
            index = sessions.index(current)
            sessions[index] = updated
            self._write_unlocked(sessions)
            return updated

    @staticmethod
    def _find(
        sessions: tuple[RoomBootstrapSession, ...],
        session_id: str,
    ) -> RoomBootstrapSession:
        matches = tuple(item for item in sessions if item.session_id == session_id)
        if len(matches) != 1:
            raise BootstrapError("customer bootstrap session is unavailable")
        return matches[0]

    @staticmethod
    def _expire_values(
        sessions: tuple[RoomBootstrapSession, ...],
        now: datetime,
    ) -> tuple[RoomBootstrapSession, ...]:
        return tuple(
            replace(
                session,
                state=BootstrapState.EXPIRED,
                generation=session.generation + 1,
                updated_at=now,
            )
            if session.state is BootstrapState.PREPARED and session.expires_at <= now
            else session
            for session in sessions
        )

    def _prepare_paths(self) -> None:
        for directory in (
            self.state_dir,
            self.state_dir.parent,
            self.state_dir.parent.parent,
        ):
            if directory.is_symlink():
                raise BootstrapError(
                    "customer bootstrap directory chain contains a symlink"
                )
            if not directory.exists():
                continue
            info = directory.stat(follow_symlinks=False)
            if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid():
                raise BootstrapError("customer bootstrap directory is unsafe")
            directory.chmod(0o700)
        self.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
        self.state_dir.chmod(0o700)
        marker_root = self.journal_marker_path.parent
        if marker_root.is_symlink():
            raise BootstrapError(
                "customer bootstrap authority marker directory is a symlink"
            )
        if marker_root.exists():
            marker_info = marker_root.stat(follow_symlinks=False)
            if (
                not stat.S_ISDIR(marker_info.st_mode)
                or marker_info.st_uid != os.geteuid()
                or stat.S_IMODE(marker_info.st_mode) not in {0o500, 0o700}
            ):
                raise BootstrapError(
                    "customer bootstrap authority marker directory is unsafe"
                )
        else:
            marker_root.mkdir(mode=0o700)
        for path in (
            self.ledger_path,
            self.journal_path,
            self.journal_head_path,
            self.journal_marker_path,
            self.lock_path,
        ):
            if path.is_symlink():
                raise BootstrapError("customer bootstrap authority path is a symlink")
        descriptor = os.open(
            self.lock_path,
            os.O_CREAT | os.O_RDWR | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0),
            0o600,
        )
        os.fchmod(descriptor, 0o600)
        os.close(descriptor)

    @contextmanager
    def _locked(self) -> Iterator[None]:
        descriptor = os.open(
            self.lock_path,
            os.O_RDWR | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0),
        )
        try:
            fcntl.flock(descriptor, fcntl.LOCK_EX)
            yield
        finally:
            fcntl.flock(descriptor, fcntl.LOCK_UN)
            os.close(descriptor)

    def _read_unlocked(self) -> tuple[RoomBootstrapSession, ...]:
        marker_exists = self.journal_marker_path.exists()
        if marker_exists:
            self._read_journal_marker_unlocked()
            if not self.journal_path.exists() or not self.journal_head_path.exists():
                raise BootstrapError(
                    "customer bootstrap journal required state is missing"
                )
        elif self.journal_path.exists() or self.journal_head_path.exists():
            raise BootstrapError(
                "customer bootstrap journal migration marker is missing"
            )
        journal = self._read_journal_unlocked()
        if not journal and self.journal_head_path.exists():
            raise BootstrapError(
                "customer bootstrap journal is missing beneath terminal head"
            )
        if journal:
            authoritative = journal[-1][1]
            head = self._read_journal_head_unlocked()
            if head != (len(journal), journal[-1][0]):
                raise BootstrapError(
                    "customer bootstrap journal terminal head conflicts"
                )
            try:
                snapshot, _ = self._read_snapshot_unlocked()
            except BootstrapError as exc:
                if self.ledger_path.exists():
                    raise BootstrapError(
                        "customer bootstrap journal terminal witness is invalid"
                    ) from exc
                self._write_projection_unlocked(authoritative)
                return self._sessions_from_document(authoritative)
            if snapshot == authoritative:
                return self._sessions_from_document(authoritative)
            if len(journal) > 1 and snapshot == journal[-2][1]:
                self._write_projection_unlocked(authoritative)
                return self._sessions_from_document(authoritative)
            raise BootstrapError(
                "customer bootstrap journal terminal witness conflicts"
            )
        _, sessions = self._read_snapshot_unlocked()
        return sessions

    def _read_snapshot_unlocked(
        self,
    ) -> tuple[dict[str, object], tuple[RoomBootstrapSession, ...]]:
        _require_private_regular_file(self.ledger_path, "customer bootstrap ledger")
        try:
            value = json.loads(self.ledger_path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise BootstrapError("customer bootstrap ledger is invalid") from exc
        if not isinstance(value, dict):
            raise BootstrapError("customer bootstrap ledger shape is invalid")
        document = cast(dict[str, object], value)
        return document, self._sessions_from_document(document)

    @staticmethod
    def _sessions_from_document(
        value: Mapping[str, object],
    ) -> tuple[RoomBootstrapSession, ...]:
        if not isinstance(value, dict) or set(value) != {"schema", "sessions", "digest"}:
            raise BootstrapError("customer bootstrap ledger shape is invalid")
        payload = {"schema": value["schema"], "sessions": value["sessions"]}
        if (
            value["schema"] != _LEDGER_SCHEMA
            or not isinstance(value["sessions"], list)
            or value["digest"] != _sha256(_canonical_bytes(payload))
        ):
            raise BootstrapError("customer bootstrap ledger digest is invalid")
        sessions = tuple(_session_from_mapping(item) for item in value["sessions"])
        if len({item.session_id for item in sessions}) != len(sessions):
            raise BootstrapError("customer bootstrap session identity is duplicated")
        return sessions

    def _read_journal_unlocked(
        self,
    ) -> tuple[tuple[str, dict[str, object]], ...]:
        if not self.journal_path.exists():
            return ()
        _require_private_regular_file(
            self.journal_path,
            "customer bootstrap journal",
        )
        try:
            lines = self.journal_path.read_text(encoding="utf-8").splitlines()
        except (OSError, UnicodeDecodeError) as exc:
            raise BootstrapError("customer bootstrap journal is invalid") from exc
        if not lines:
            raise BootstrapError("customer bootstrap journal is empty")
        events: list[tuple[str, dict[str, object]]] = []
        previous = "0" * 64
        for sequence, line in enumerate(lines, start=1):
            try:
                raw = json.loads(line)
            except json.JSONDecodeError as exc:
                raise BootstrapError("customer bootstrap journal is invalid") from exc
            if not isinstance(raw, dict) or set(raw) != {
                "schema",
                "sequence",
                "previous_event_digest",
                "ledger",
                "event_digest",
            }:
                raise BootstrapError("customer bootstrap journal shape is invalid")
            event = cast(dict[str, object], raw)
            digest = event["event_digest"]
            ledger = event["ledger"]
            payload = {
                key: event[key]
                for key in (
                    "schema",
                    "sequence",
                    "previous_event_digest",
                    "ledger",
                )
            }
            if (
                event["schema"] != _JOURNAL_SCHEMA
                or event["sequence"] != sequence
                or event["previous_event_digest"] != previous
                or not isinstance(digest, str)
                or digest != _sha256(_canonical_bytes(payload))
                or not isinstance(ledger, dict)
            ):
                raise BootstrapError("customer bootstrap journal chain is invalid")
            document = cast(dict[str, object], ledger)
            self._sessions_from_document(document)
            events.append((digest, document))
            previous = digest
        return tuple(events)

    def _read_journal_head_unlocked(self) -> tuple[int, str]:
        _require_private_regular_file(
            self.journal_head_path,
            "customer bootstrap journal head",
        )
        try:
            raw = json.loads(
                self.journal_head_path.read_text(encoding="utf-8")
            )
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise BootstrapError(
                "customer bootstrap journal head is invalid"
            ) from exc
        if not isinstance(raw, dict) or set(raw) != {
            "schema",
            "sequence",
            "event_digest",
            "digest",
        }:
            raise BootstrapError("customer bootstrap journal head shape is invalid")
        payload = {
            "schema": raw["schema"],
            "sequence": raw["sequence"],
            "event_digest": raw["event_digest"],
        }
        if (
            raw["schema"] != _JOURNAL_HEAD_SCHEMA
            or type(raw["sequence"]) is not int
            or raw["sequence"] < 1
            or not isinstance(raw["event_digest"], str)
            or raw["digest"] != _sha256(_canonical_bytes(payload))
        ):
            raise BootstrapError(
                "customer bootstrap journal head digest is invalid"
            )
        return raw["sequence"], raw["event_digest"]

    def _read_journal_marker_unlocked(self) -> None:
        _require_private_regular_file(
            self.journal_marker_path,
            "customer bootstrap journal marker",
        )
        try:
            raw = json.loads(
                self.journal_marker_path.read_text(encoding="utf-8")
            )
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise BootstrapError(
                "customer bootstrap journal marker is invalid"
            ) from exc
        payload = {
            "schema": _JOURNAL_MARKER_SCHEMA,
            "state_dir": self.state_dir.name,
        }
        if (
            not isinstance(raw, dict)
            or raw != {
                **payload,
                "digest": _sha256(_canonical_bytes(payload)),
            }
        ):
            raise BootstrapError(
                "customer bootstrap journal marker digest is invalid"
            )

    def _write_unlocked(
        self,
        sessions: tuple[RoomBootstrapSession, ...] | list[RoomBootstrapSession],
    ) -> None:
        payload: dict[str, object] = {
            "schema": _LEDGER_SCHEMA,
            "sessions": [_session_to_dict(session) for session in sessions],
        }
        document = {**payload, "digest": _sha256(_canonical_bytes(payload))}
        journal = self._read_journal_unlocked()
        if not journal and self.ledger_path.exists():
            snapshot, _ = self._read_snapshot_unlocked()
            self._append_journal_event_unlocked(snapshot, previous=None, sequence=1)
            journal = self._read_journal_unlocked()
        if journal and journal[-1][1] == document:
            self._write_projection_unlocked(document)
            return
        self._append_journal_event_unlocked(
            document,
            previous=journal[-1][0] if journal else None,
            sequence=len(journal) + 1,
        )
        self._write_projection_unlocked(document)

    def _append_journal_event_unlocked(
        self,
        document: dict[str, object],
        *,
        previous: str | None,
        sequence: int,
    ) -> None:
        payload: dict[str, object] = {
            "schema": _JOURNAL_SCHEMA,
            "sequence": sequence,
            "previous_event_digest": previous or "0" * 64,
            "ledger": document,
        }
        event = {
            **payload,
            "event_digest": _sha256(_canonical_bytes(payload)),
        }
        with self.journal_path.open("ab") as stream:
            stream.write(_canonical_bytes(event) + b"\n")
            stream.flush()
            os.fsync(stream.fileno())
        self.journal_path.chmod(0o600)
        self._write_journal_head_unlocked(
            sequence=sequence,
            event_digest=str(event["event_digest"]),
        )
        self._write_journal_marker_unlocked()

    def _write_journal_marker_unlocked(self) -> None:
        payload: dict[str, object] = {
            "schema": _JOURNAL_MARKER_SCHEMA,
            "state_dir": self.state_dir.name,
        }
        document = {
            **payload,
            "digest": _sha256(_canonical_bytes(payload)),
        }
        if self.journal_marker_path.exists():
            self._read_journal_marker_unlocked()
            return
        descriptor = os.open(
            self.journal_marker_path,
            os.O_WRONLY
            | os.O_CREAT
            | os.O_EXCL
            | os.O_CLOEXEC
            | getattr(os, "O_NOFOLLOW", 0),
            0o600,
        )
        try:
            os.fchmod(descriptor, 0o600)
            with os.fdopen(descriptor, "wb", closefd=True) as stream:
                stream.write(_canonical_bytes(document))
                stream.flush()
                os.fsync(stream.fileno())
            self.journal_marker_path.parent.chmod(0o500)
            parent = os.open(
                self.journal_marker_path.parent,
                os.O_RDONLY | os.O_DIRECTORY,
            )
            try:
                os.fsync(parent)
            finally:
                os.close(parent)
        except Exception:
            try:
                os.close(descriptor)
            except OSError:
                pass
            raise

    def _write_journal_head_unlocked(
        self,
        *,
        sequence: int,
        event_digest: str,
    ) -> None:
        payload: dict[str, object] = {
            "schema": _JOURNAL_HEAD_SCHEMA,
            "sequence": sequence,
            "event_digest": event_digest,
        }
        document = {
            **payload,
            "digest": _sha256(_canonical_bytes(payload)),
        }
        descriptor, temporary_name = tempfile.mkstemp(
            prefix=".events-head.",
            dir=self.state_dir,
        )
        try:
            os.fchmod(descriptor, 0o600)
            with os.fdopen(descriptor, "wb", closefd=True) as stream:
                stream.write(_canonical_bytes(document))
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(temporary_name, self.journal_head_path)
            self.journal_head_path.chmod(0o600)
            directory = os.open(
                self.state_dir,
                os.O_RDONLY | os.O_DIRECTORY,
            )
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
        finally:
            try:
                os.unlink(temporary_name)
            except FileNotFoundError:
                pass

    def _write_projection_unlocked(
        self,
        document: dict[str, object],
    ) -> None:
        descriptor, temporary_name = tempfile.mkstemp(
            prefix=".ledger.", dir=self.state_dir
        )
        try:
            os.fchmod(descriptor, 0o600)
            with os.fdopen(descriptor, "wb", closefd=True) as stream:
                stream.write(_canonical_bytes(document))
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(temporary_name, self.ledger_path)
            self.ledger_path.chmod(0o600)
            directory = os.open(self.state_dir, os.O_RDONLY | os.O_DIRECTORY)
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
        finally:
            try:
                os.unlink(temporary_name)
            except FileNotFoundError:
                pass

    def _utc_now(self) -> datetime:
        value = self._now()
        if not isinstance(value, datetime) or value.tzinfo is None:
            raise BootstrapError("customer bootstrap clock must be timezone-aware")
        return value.astimezone(timezone.utc)


class RoomBootstrapTransport:
    """Container for the owner/customer-only durable bootstrap store."""

    def __init__(
        self,
        store: RoomBootstrapStore,
        *,
        owner_id: str,
        customer_topic_name: str = "고객 체크인",
    ) -> None:
        if not isinstance(store, RoomBootstrapStore):
            raise TypeError("customer bootstrap transport requires its sealed store")
        self.store = store
        self.owner_id = _telegram_id(owner_id, "owner")
        self.customer_topic_name = str(customer_topic_name)


def _session_to_dict(session: RoomBootstrapSession) -> dict[str, object]:
    return {
        "session_id": session.session_id,
        "sid_hash": session.sid_hash,
        "customer_draft": session.customer_draft.to_dict(),
        "customer_draft_digest": session.customer_draft_digest,
        "bot_username": session.bot_username,
        "state": session.state.value,
        "generation": session.generation,
        "created_at": session.created_at.isoformat(),
        "updated_at": session.updated_at.isoformat(),
        "expires_at": session.expires_at.isoformat(),
        "owner_id": session.owner_id,
        "first_claim": session.first_claim,
        "role_claims": [
            {
                "role": claim.role.value,
                "user_id": claim.user_id,
                "chat_id": claim.chat_id,
                "topic_id": claim.topic_id,
                "message_id": claim.message_id,
            }
            for claim in session.role_claims
        ],
        "consent_publication_attempt": session.consent_publication_attempt,
        "consent_card_message_id": session.consent_card_message_id,
        "recovery_attempt_generation": session.recovery_attempt_generation,
        "recovery_attempts": [
            {
                "session_id": attempt.session_id,
                "slot": attempt.slot.value,
                "chat_id": attempt.chat_id,
                "generation": attempt.generation,
                "created_at": attempt.created_at.isoformat(),
            }
            for attempt in session.recovery_attempts
        ],
        "failure_code": session.failure_code,
        "committed_consent_card_message_id": (
            session.committed_consent_card_message_id
        ),
        "consent_handoff": (
            None
            if session.consent_handoff is None
            else {
                "update_id": session.consent_handoff.update_id,
                "actor_id": session.consent_handoff.actor_id,
                "chat_id": session.consent_handoff.chat_id,
                "topic_id": session.consent_handoff.topic_id,
                "message_id": session.consent_handoff.message_id,
                "callback_data": session.consent_handoff.callback_data,
                "bootstrap_generation": (
                    session.consent_handoff.bootstrap_generation
                ),
                "recorded_at": session.consent_handoff.recorded_at.isoformat(),
                "provenance_digest": session.consent_handoff.provenance_digest,
            }
        ),
        "consent_recovery_reconciled": session.consent_recovery_reconciled,
    }


def _strict_string(value: object, label: str) -> str:
    if not isinstance(value, str):
        raise BootstrapError(f"customer bootstrap {label} is invalid")
    return value


def _required_string(value: Mapping[str, object], field: str) -> str:
    return _strict_string(value[field], field)


def _required_int(value: Mapping[str, object], field: str) -> int:
    candidate = value[field]
    if type(candidate) is not int:
        raise BootstrapError(f"customer bootstrap {field} is invalid")
    return candidate


def _optional_int(value: Mapping[str, object], field: str) -> int | None:
    candidate = value[field]
    if candidate is None:
        return None
    if type(candidate) is not int:
        raise BootstrapError(f"customer bootstrap {field} is invalid")
    return candidate


def _optional_float(value: Mapping[str, object], field: str) -> float | None:
    candidate = value[field]
    if candidate is None:
        return None
    if isinstance(candidate, bool) or not isinstance(candidate, (int, float)):
        raise BootstrapError(f"customer bootstrap {field} is invalid")
    return float(candidate)


def _optional_string(value: Mapping[str, object], field: str) -> str | None:
    candidate = value[field]
    if candidate is None:
        return None
    return _strict_string(candidate, field)


def _required_datetime(value: Mapping[str, object], field: str) -> datetime:
    return datetime.fromisoformat(_required_string(value, field))


def _string_tuple(value: Mapping[str, object], field: str) -> tuple[str, ...]:
    candidate = value[field]
    if not isinstance(candidate, list) or any(
        not isinstance(item, str) for item in candidate
    ):
        raise BootstrapError(f"customer bootstrap {field} is invalid")
    return tuple(cast(list[str], candidate))


def _mapping_sequence(
    value: object,
    *,
    label: str,
    expected: set[str],
) -> tuple[Mapping[str, object], ...]:
    if not isinstance(value, list):
        raise BootstrapError(f"{label} are invalid")
    result: list[Mapping[str, object]] = []
    for item in value:
        if (
            not isinstance(item, Mapping)
            or any(not isinstance(key, str) for key in item)
            or set(item) != expected
        ):
            raise BootstrapError(f"{label} are invalid")
        result.append(cast(Mapping[str, object], item))
    return tuple(result)


def _session_from_mapping(value: object) -> RoomBootstrapSession:
    if not isinstance(value, Mapping) or any(
        not isinstance(key, str) for key in value
    ):
        raise BootstrapError("customer bootstrap session is invalid")
    document = cast(Mapping[str, object], value)
    expected = set(_session_to_dict(_placeholder_session()))
    legacy_missing = {
        "committed_consent_card_message_id",
        "consent_handoff",
        "consent_recovery_reconciled",
        "first_claim",
    }
    actual = set(document)
    if (
        actual - expected
        or not expected - actual <= legacy_missing
    ):
        raise BootstrapError("customer bootstrap session fields are invalid")
    try:
        draft_value = document["customer_draft"]
        if (
            not isinstance(draft_value, Mapping)
            or any(not isinstance(key, str) for key in draft_value)
            or set(draft_value) != set(CustomerDraft.__dataclass_fields__)
        ):
            raise BootstrapError("persisted customer draft fields are invalid")
        draft = _draft_from_mapping(draft_value)
        claim_values = _mapping_sequence(
            document["role_claims"],
            label="customer role claims",
            expected={"role", "user_id", "chat_id", "topic_id", "message_id"},
        )
        claims = tuple(
            RoleClaim(
                Role(_required_string(item, "role")),
                _telegram_id(item["user_id"], "claim user"),
                _telegram_id(item["chat_id"], "claim chat"),
                _telegram_id(item["topic_id"], "claim topic"),
                _telegram_id(item["message_id"], "claim message"),
            )
            for item in claim_values
        )
        attempt_values = _mapping_sequence(
            document["recovery_attempts"],
            label="customer recovery attempts",
            expected={"session_id", "slot", "chat_id", "generation", "created_at"},
        )
        attempts = tuple(
            RecoveryAttempt(
                _required_string(item, "session_id"),
                RecoverySlot(_required_string(item, "slot")),
                _telegram_id(item["chat_id"], "attempt chat"),
                _required_int(item, "generation"),
                _required_datetime(item, "created_at"),
            )
            for item in attempt_values
        )
        consent_card_message_id = document["consent_card_message_id"]
        failure_code = document["failure_code"]
        committed_message = document.get("committed_consent_card_message_id")
        handoff_value = document.get("consent_handoff")
        handoff_fields = {
                "update_id",
                "actor_id",
                "chat_id",
                "topic_id",
                "message_id",
                "callback_data",
                "bootstrap_generation",
                "recorded_at",
                "provenance_digest",
        }
        if handoff_value is not None and (
            not isinstance(handoff_value, Mapping)
            or any(not isinstance(key, str) for key in handoff_value)
            or set(handoff_value)
            not in {frozenset(handoff_fields), frozenset(handoff_fields - {"bootstrap_generation"})}
        ):
            raise BootstrapError("consent handoff is invalid")
        persisted_generation = _required_int(document, "generation")
        persisted_state = BootstrapState(_required_string(document, "state"))
        legacy_handoff_generation = (
            persisted_generation - 1
            if persisted_state is BootstrapState.ACTIVE
            else persisted_generation
        )
        handoff = (
            None
            if handoff_value is None
            else ConsentHandoff(
                update_id=_required_int(handoff_value, "update_id"),
                actor_id=_required_int(handoff_value, "actor_id"),
                chat_id=_required_int(handoff_value, "chat_id"),
                topic_id=_required_int(handoff_value, "topic_id"),
                message_id=_required_int(handoff_value, "message_id"),
                callback_data=_required_string(handoff_value, "callback_data"),
                bootstrap_generation=(
                    _required_int(handoff_value, "bootstrap_generation")
                    if "bootstrap_generation" in handoff_value
                    else legacy_handoff_generation
                ),
                recorded_at=_required_datetime(handoff_value, "recorded_at"),
                provenance_digest=_required_string(
                    handoff_value, "provenance_digest"
                ),
            )
        )
        consent_recovery_reconciled = document.get(
            "consent_recovery_reconciled", False
        )
        if not isinstance(consent_recovery_reconciled, bool):
            raise BootstrapError("consent recovery reconciliation flag is invalid")
        first_claim = document.get("first_claim", False)
        if not isinstance(first_claim, bool):
            raise BootstrapError("first-claim policy is invalid")
        session = RoomBootstrapSession(
            session_id=_required_string(document, "session_id"),
            sid_hash=_required_string(document, "sid_hash"),
            customer_draft=draft,
            customer_draft_digest=_required_string(
                document, "customer_draft_digest"
            ),
            bot_username=_required_string(document, "bot_username"),
            state=persisted_state,
            generation=persisted_generation,
            created_at=_required_datetime(document, "created_at"),
            updated_at=_required_datetime(document, "updated_at"),
            expires_at=_required_datetime(document, "expires_at"),
            owner_id=_telegram_id(document["owner_id"], "owner"),
            first_claim=first_claim,
            role_claims=claims,
            consent_publication_attempt=_required_int(
                document, "consent_publication_attempt"
            ),
            consent_card_message_id=(
                None
                if consent_card_message_id is None
                else _telegram_id(consent_card_message_id, "consent receipt")
            ),
            recovery_attempt_generation=_required_int(
                document, "recovery_attempt_generation"
            ),
            recovery_attempts=attempts,
            failure_code=(
                None
                if failure_code is None
                else _strict_string(failure_code, "failure_code")
            ),
            committed_consent_card_message_id=(
                None
                if committed_message is None
                else _telegram_id(committed_message, "committed consent receipt")
            ),
            consent_handoff=handoff,
            consent_recovery_reconciled=consent_recovery_reconciled,
        )
    except (KeyError, TypeError, ValueError) as exc:
        raise BootstrapError("customer bootstrap session value is invalid") from exc
    valid_handoff_generations = {
        BootstrapState.AWAITING_CONSENT: frozenset({session.generation}),
        BootstrapState.AWAITING_ACTIVATION: frozenset(
            {session.generation, session.generation - 1}
        ),
        BootstrapState.ACTIVE: frozenset(
            {session.generation - 1, session.generation - 2}
        ),
    }.get(session.state, frozenset())
    if (
        _SESSION_RE.fullmatch(session.session_id) is None
        or _DIGEST_RE.fullmatch(session.sid_hash) is None
        or session.customer_draft_digest != session.customer_draft.digest
        or (
            session.state is BootstrapState.PREPARED
            and (
                session.first_claim
                == (session.customer_draft.customer_user_id is not None)
            )
        )
        or (
            not session.terminal
            and session.state is not BootstrapState.PREPARED
            and session.customer_draft.customer_user_id is None
        )
        or session.generation < 1
        or session.consent_publication_attempt < 0
        or session.recovery_attempt_generation < 0
        or any(item.session_id != session.session_id for item in session.recovery_attempts)
        or any(item.created_at.tzinfo is None for item in session.recovery_attempts)
        or type(session.consent_recovery_reconciled) is not bool
        or session.consent_recovery_reconciled and session.consent_handoff is None
        or session.consent_handoff is not None
        and (
            session.consent_handoff.recorded_at.tzinfo is None
            or session.consent_handoff.bootstrap_generation < 1
            or session.consent_handoff.bootstrap_generation
            not in valid_handoff_generations
            or _DIGEST_RE.fullmatch(session.consent_handoff.provenance_digest) is None
            or not hmac.compare_digest(
                session.consent_handoff.provenance_digest,
                consent_handoff_digest(session, session.consent_handoff),
            )
        )
        or any(
            timestamp.tzinfo is None
            for timestamp in (session.created_at, session.updated_at, session.expires_at)
        )
    ):
        raise BootstrapError("customer bootstrap session authority is invalid")
    return session


def _placeholder_session() -> RoomBootstrapSession:
    now = datetime(2000, 1, 1, tzinfo=timezone.utc)
    draft = CustomerDraft("x", "x", "2000-01-01", "08:00", 0, 1, 1, 1, ("x",))
    return RoomBootstrapSession(
        "cb_0000000000000000000000",
        "0" * 64,
        draft,
        draft.digest,
        "ownerbot",
        BootstrapState.PREPARED,
        1,
        now,
        now,
        now,
        "1",
    )


def _draft_from_mapping(value: object) -> CustomerDraft:
    if not isinstance(value, Mapping) or any(not isinstance(key, str) for key in value):
        raise BootstrapError("customer draft is invalid")
    document = cast(Mapping[str, object], value)
    required = _DRAFT_REQUIRED
    allowed = required | _DRAFT_OPTIONAL
    if set(document) - allowed or required - set(document):
        raise BootstrapError("customer draft fields are invalid")
    defaults = CustomerDraft(
        "placeholder",
        "placeholder",
        "2000-01-01",
        "08:00",
        0,
        1,
        1,
        1,
        ("placeholder",),
    ).to_dict()
    values = cast(Mapping[str, object], {**defaults, **dict(document)})
    try:
        draft = CustomerDraft(
            customer_key=_required_string(values, "customer_key"),
            display_name=_required_string(values, "display_name"),
            starts_on=_required_string(values, "starts_on"),
            daily_time=_required_string(values, "daily_time"),
            weekly_weekday=_required_int(values, "weekly_weekday"),
            monthly_day=_required_int(values, "monthly_day"),
            calories_kcal=_required_int(values, "calories_kcal"),
            protein_g=_required_int(values, "protein_g"),
            meals=_string_tuple(values, "meals"),
            primary_goal=_required_string(values, "primary_goal"),
            dietary_restrictions=_string_tuple(values, "dietary_restrictions"),
            allergies=_string_tuple(values, "allergies"),
            food_preferences=_string_tuple(values, "food_preferences"),
            supplements=_string_tuple(values, "supplements"),
            digestion_context=_optional_string(values, "digestion_context"),
            sleep_goal_hours=_optional_float(values, "sleep_goal_hours"),
            recovery_goal=_optional_string(values, "recovery_goal"),
            training_context=_optional_string(values, "training_context"),
            carbohydrate_g=_optional_int(values, "carbohydrate_g"),
            fat_g=_optional_int(values, "fat_g"),
            water_liters=_optional_float(values, "water_liters"),
            customer_user_id=_optional_string(values, "customer_user_id"),
        )
    except (TypeError, ValueError) as exc:
        raise BootstrapError("customer draft value is invalid") from exc
    if (
        not draft.customer_key
        or not draft.display_name
        or not draft.meals
        or type(draft.weekly_weekday) is not int
        or not 0 <= draft.weekly_weekday <= 6
        or type(draft.monthly_day) is not int
        or not 1 <= draft.monthly_day <= 28
        or type(draft.calories_kcal) is not int
        or draft.calories_kcal <= 0
        or type(draft.protein_g) is not int
        or draft.protein_g <= 0
    ):
        raise BootstrapError("customer draft value is invalid")
    return draft


def _require_private_regular_file(path: Path, label: str) -> None:
    if path.is_symlink():
        raise BootstrapError(f"{label} must not be a symlink")
    try:
        info = path.stat()
    except OSError as exc:
        raise BootstrapError(f"{label} is unavailable") from exc
    if (
        not stat.S_ISREG(info.st_mode)
        or info.st_nlink != 1
        or stat.S_IMODE(info.st_mode) & 0o077
        or info.st_size > 1_048_576
    ):
        raise BootstrapError(f"{label} must be one private regular file")


def _telegram_id(value: object, label: str) -> str:
    text = str(value)
    if _ID_RE.fullmatch(text) is None or str(int(text)) != text:
        raise BootstrapError(f"{label} Telegram identity is invalid")
    return text


def _canonical_json(value: object) -> str:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    )


def _canonical_bytes(value: object) -> bytes:
    return _canonical_json(value).encode("utf-8")


def _sha256(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


__all__ = [
    "BootstrapConflict",
    "BootstrapError",
    "BootstrapState",
    "ConsentHandoff",
    "customer_consent_callback",
    "CustomerDraft",
    "GenerationConflict",
    "PreparedRehearsalCustomerInvite",
    "RecoveryAttempt",
    "RecoverySlot",
    "Role",
    "RoleClaim",
    "RoomBootstrapSession",
    "RoomBootstrapStore",
    "RoomBootstrapTransport",
    "consent_handoff_digest",
    "load_customer_draft",
    "room_bootstrap_state_dir",
]