"""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"
_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
    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.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():
                self._read_unlocked()
            else:
                self._write_unlocked(())

    def prepare_rehearsal_customer_invite(
        self,
        draft: CustomerDraft,
        *,
        bot_username: str,
        owner_id: str,
    ) -> 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 draft.customer_user_id is not None:
            raise BootstrapError("customer identity must be claimed through the DM link")
        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,
        )
        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.customer_draft.customer_user_id is not 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")
        session = self.rehearsal_customer_invite_session(token)

        def mutate(current: RoomBootstrapSession, now: datetime) -> RoomBootstrapSession:
            if (
                current.state is not BootstrapState.PREPARED
                or current.role_claims
                or current.customer_draft.customer_user_id is not None
                or user == current.owner_id
            ):
                raise BootstrapError("customer bootstrap invite is unavailable")
            draft = replace(current.customer_draft, customer_user_id=user)
            return replace(
                current,
                customer_draft=draft,
                customer_draft_digest=draft.digest,
                state=BootstrapState.REGISTERING,
                generation=current.generation + 1,
                updated_at=now,
                role_claims=(RoleClaim(Role.CUSTOMER, user, chat, "0", message),),
            )

        return self._mutate(session.session_id, session.generation, mutate)

    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 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:
            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
            ):
                raise GenerationConflict("committed consent 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)
            expected_callback = customer_consent_callback(
                session.customer_key,
                CONSENT_VERSION,
                "g",
            )
            if (
                session.state is not 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)
                != session.committed_consent_card_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 is not None and session.consent_handoff != handoff:
                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:
        if self.state_dir.is_symlink():
            raise BootstrapError("customer bootstrap state directory is a symlink")
        self.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
        self.state_dir.chmod(0o700)
        for path in (self.ledger_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, ...]:
        _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) 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 _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))}
        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,
        "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",
    }
    if set(document) not in {frozenset(expected), frozenset(expected - 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")
        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"),
            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
    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.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
            != (
                session.generation - 1
                if session.state is BootstrapState.ACTIVE
                else session.generation
            )
            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",
]