"""Safe customer registry onboarding with explicit activation."""

from __future__ import annotations

import argparse
import json
import os
import hashlib
import re
import uuid
import fcntl
import stat
from contextlib import contextmanager
from contextvars import ContextVar
from types import MappingProxyType
from dataclasses import dataclass
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from typing import Callable, Iterator, Mapping
from threading import local
from pydantic import ValidationError
from zoneinfo import ZoneInfo

from checkin_cli.activation_token_rotation_policy import (
    ActivationChecklistBindings,
    TokenRotationPolicyError,
    validate_token_rotation_policy,
)
from checkin_cli.customer_coaching import (
    CONSENT_VERSION,
    AiProcessingConsent,
    AdaptiveRegistrationInputs,
    CustomerRuntime,
    CustomerTrainingScheduleEntry,
    CustomerRegistryError,
    CustomerSpec,
    RegistryDocument,
    TelegramAddress,
    CustomerRegistry,
    load_customer_registry,
)
from checkin_cli.customer_reporting import PilotKPIJudgement, judge_pilot_kpis
from checkin_cli.customer_schedule import (
    _claims_by_schedule,
    _current_rows,
    _read_fence,
    _read_jsonl,
    _validate_schedule_rows,
)
from checkin_cli.weekly_operations_schedule_host_models_r4 import CustomerScheduleError
from checkin_cli.models import OperatorTask, PaymentKind, PaymentMethod, RecordResult
from checkin_cli.nutrition_readiness import (
    NUTRITION_READINESS_REASON_CODES,
    NutritionReadinessError,
    require_nutrition_start_readiness,
)
from checkin_cli.nutrition_onboarding_projection import (
    apply_nutrition_onboarding_projection as apply_nutrition_onboarding_projection,
    build_legacy_activation_authority as build_legacy_activation_authority,
    build_nutrition_activation_receipt_v2,
    customer_nutrition_projection_digest,
    project_nutrition_document as project_nutrition_document,
    validate_legacy_activation_authority,
)
from checkin_cli.nutrition_onboarding_fs import (
    atomic_write_private_json,
    read_private_json,
    validate_private_file,
    validate_profile_path,
)
from checkin_cli.store import CanonicalEventTransaction, EventStore
from checkin_cli.staff_membership_evidence import (
    StaffMembershipEvidenceError,
    validate_staff_membership_evidence,
)
from checkin_cli.adaptive_nutrition import (
    AdaptiveEventStore,
    append_approved_policy_extension,
    canonical_event_records,
    canonical_json,
    customer_policy_from_onboarding_artifact,
    digest as adaptive_digest,
    feature_config_digest,
    initialize_adaptive_customer,
    _DUAL_COACH_RISK_POLICY_FILE,
    _DUAL_COACH_RISK_POLICY_ID,
    _DUAL_COACH_RISK_POLICY_SCHEMA,
    _dual_coach_risk_policy_document_digest,
    _validate_dual_coach_risk_policy_value,
)

_DUAL_COACH_RISK_POLICY_VALUE: dict[str, object] = {
    "weight_change_percent": {"normal": "<=2", "elevated": ">2-4", "high": ">4"},
    "sleep_hours": {"normal": ">=7", "elevated": "5-<7", "high": "<5"},
    "fatigue": ["low", "moderate", "high"],
    "pain": ["none", "present", "severe"],
    "exercise_feasibility": ["possible", "limited", "impossible"],
    "meal_deviation": ["none", "partial", "material"],
    "score_threshold": 4,
    "hard_overrides": ["pain_override", "exercise_impossible_override"],
    "missing_evidence_reason": "risk_evidence_unavailable",
}


def approve_dual_coach_risk_policy(
    profile_root: Path | str,
    customer_key: str,
    *,
    version: str,
    owner_actor: TelegramAddress,
    approved_at_kst: str | datetime | None = None,
) -> dict[str, object]:
    """Write the one owner-approved risk policy through the authority lock."""
    if type(version) is not str or not version.strip():
        raise CustomerAdminError("dual-coach risk policy version is invalid")
    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    if not isinstance(owner_actor, TelegramAddress):
        raise CustomerAdminError("dual-coach risk policy requires a typed owner actor")
    with profile_authority_lock(root):
        registry_path = _resolve_registry_path(root)
        registry = _read_profile_registry(registry_path, root)
        selected = next(
            (item for item in registry.customers if item.customer_key == key), None
        )
        if selected is None:
            raise CustomerAdminError(f"unknown customer: {key}")
        if owner_actor.model_dump(mode="json") != registry.owner.model_dump(
            mode="json"
        ):
            raise CustomerAdminError(
                "dual-coach risk policy owner actor is not current"
            )
        authority, authority_digest = _registration_activation_authority(
            root, registry_path, registry, selected
        )
        runtime = _registered_customer(registry_path, root, key, None)
        plans_root = runtime.nutrition_plans_root
        if plans_root.is_symlink():
            raise CustomerAdminError("dual-coach risk policy root is unsafe")
        plans_root.mkdir(parents=True, exist_ok=True, mode=0o700)
        root_status = plans_root.stat()
        if (
            not stat.S_ISDIR(root_status.st_mode)
            or root_status.st_uid != os.geteuid()
            or stat.S_IMODE(root_status.st_mode) != 0o700
        ):
            raise CustomerAdminError("dual-coach risk policy root is unsafe")
        policy = _validate_dual_coach_risk_policy_value(_DUAL_COACH_RISK_POLICY_VALUE)
        document: dict[str, object] = {
            "schema_version": _DUAL_COACH_RISK_POLICY_SCHEMA,
            "policy_id": _DUAL_COACH_RISK_POLICY_ID,
            "version": version,
            "policy": policy,
            "policy_digest": adaptive_digest(policy),
            "document_digest": "",
            "approved": True,
            "approved_by": owner_actor.model_dump(mode="json"),
            "approved_at_kst": _registration_kst_timestamp(approved_at_kst),
            "customer_key": key,
            "owner_digest": authority["owner_digest"],
            "registry_digest": authority["registry_digest"],
            "activation_receipt_digest": authority["activation_receipt_digest"],
            "authority_digest": authority_digest,
        }
        document["document_digest"] = _dual_coach_risk_policy_document_digest(document)
        encoded = (canonical_json(document) + "\n").encode("utf-8")
        path = plans_root / _DUAL_COACH_RISK_POLICY_FILE
        if path.is_symlink():
            raise CustomerAdminError("dual-coach risk policy path is unsafe")
        temporary = (
            plans_root / f".{_DUAL_COACH_RISK_POLICY_FILE}.{uuid.uuid4().hex}.tmp"
        )
        descriptor: int | None = None
        try:
            descriptor = os.open(
                temporary,
                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
                0o600,
            )
            os.write(descriptor, encoded)
            os.fsync(descriptor)
            opened = os.fstat(descriptor)
            if (
                not stat.S_ISREG(opened.st_mode)
                or opened.st_uid != os.geteuid()
                or opened.st_nlink != 1
                or stat.S_IMODE(opened.st_mode) != 0o600
            ):
                raise CustomerAdminError("dual-coach risk policy file is unsafe")
            os.close(descriptor)
            descriptor = None
            os.replace(temporary, path)
            directory = os.open(plans_root, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
        except OSError as exc:
            raise CustomerAdminError("dual-coach risk policy write failed") from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            try:
                temporary.unlink()
            except FileNotFoundError:
                pass
        return document


class CustomerAdminError(ValueError):
    pass


@dataclass(frozen=True, slots=True)
class CustomerDraft:
    customer_key: str
    display_name: str
    user_id: str
    chat_id: str
    topic_id: str
    starts_on: date
    daily_time: time
    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


@dataclass(frozen=True, slots=True)
class ActivationResult:
    """Receipt for one guarded customer activation."""

    customer: CustomerSpec
    registry_path: Path
    data_root: Path
    audit_path: Path

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

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

    @property
    def enabled(self) -> bool:
        return self.customer.enabled

    @property
    def spec(self) -> CustomerSpec:
        return self.customer


_CUSTOMER_KEY = re.compile(r"^[a-z0-9][a-z0-9_-]{2,63}$")
_OPAQUE_ID_MIN = 16
_OPAQUE_ID_MAX = 80
_KST = ZoneInfo("Asia/Seoul")
_PROFILE_AUTHORITY_LOCK_FILE = ".adaptive-authority.lock"
_PROFILE_AUTHORITY_LOCK_STATE = local()


@contextmanager
def profile_authority_lock(profile_root: Path | str) -> Iterator[None]:
    """Serialize all profile authority/config writers across processes."""
    root = _resolve_profile_root(Path(profile_root))
    lock_parent = root / "data"
    if lock_parent.is_symlink():
        raise CustomerAdminError(
            "profile authority lock parent symlinks are not allowed"
        )
    lock_parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    lock_parent.chmod(0o700)
    lock_path = lock_parent / _PROFILE_AUTHORITY_LOCK_FILE
    if lock_path.is_symlink():
        raise CustomerAdminError("profile authority lock symlinks are not allowed")
    state = getattr(_PROFILE_AUTHORITY_LOCK_STATE, "locks", None)
    if state is None:
        state = {}
        _PROFILE_AUTHORITY_LOCK_STATE.locks = state
    key = str(lock_path)
    held = state.get(key)
    if held is not None:
        handle, depth = held
        state[key] = (handle, depth + 1)
        try:
            yield
        finally:
            if depth == 0:
                state.pop(key, None)
            else:
                state[key] = (handle, depth)
        return
    descriptor: int | None = None
    try:
        descriptor = os.open(
            lock_path,
            os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC,
            0o600,
        )
        opened = os.fstat(descriptor)
        named = lock_path.lstat()
        identity = (opened.st_dev, opened.st_ino)
        if (
            not stat.S_ISREG(opened.st_mode)
            or opened.st_uid != os.geteuid()
            or opened.st_nlink != 1
            or stat.S_IMODE(opened.st_mode) != 0o600
            or identity != (named.st_dev, named.st_ino)
        ):
            raise CustomerAdminError("profile authority lock is unsafe")
        fcntl.flock(descriptor, fcntl.LOCK_EX)
        if identity != (lock_path.lstat().st_dev, lock_path.lstat().st_ino):
            raise CustomerAdminError("profile authority lock was replaced")
        state[key] = (descriptor, 1)
        try:
            yield
        finally:
            state.pop(key, None)
            fcntl.flock(descriptor, fcntl.LOCK_UN)
    except OSError as exc:
        raise CustomerAdminError("profile authority lock is unavailable") from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)


def _require_customer_key(value: str) -> str:
    if type(value) is not str or _CUSTOMER_KEY.fullmatch(value) is None:
        raise CustomerAdminError("customer key is invalid")
    return value


def _require_date(value: date | str, field_name: str) -> date:
    if type(value) is date:
        return value
    if type(value) is not str:
        raise CustomerAdminError(f"{field_name} must be an ISO date")
    try:
        return date.fromisoformat(value)
    except ValueError as exc:
        raise CustomerAdminError(f"{field_name} must be an ISO date") from exc


def _activation_kst_date(value: date | datetime | None) -> date:
    if value is None:
        return datetime.now(_KST).date()
    if type(value) is date:
        return value
    if type(value) is datetime:
        if value.tzinfo is None:
            return value.replace(tzinfo=_KST).date()
        return value.astimezone(_KST).date()
    raise CustomerAdminError("kst_date must be a date or datetime")


def _require_opaque_id(
    value: str, field_name: str, *, optional: bool = False
) -> str | None:
    if optional and value is None:
        return None
    if type(value) is not str or not _OPAQUE_ID_MIN <= len(value) <= _OPAQUE_ID_MAX:
        raise CustomerAdminError(
            f"{field_name} must be an opaque id between 16 and 80 characters"
        )
    if value.strip() != value or any(
        ord(character) < 0x20 or ord(character) == 0x7F for character in value
    ):
        raise CustomerAdminError(f"{field_name} contains invalid characters")
    return value


def _registered_customer(
    path: Path,
    profile_root: Path,
    customer_key: str,
    data_root: Path | None,
) -> CustomerRuntime:
    key = _require_customer_key(customer_key)
    root = _resolve_profile_root(Path(profile_root))
    canonical_registry = _resolve_registry_path(root)
    requested_registry = Path(path)
    if (
        requested_registry.is_symlink()
        or requested_registry.resolve() != canonical_registry
    ):
        raise CustomerAdminError("registry path does not match the profile registry")
    try:
        registry = load_runtime_customer_registry(root)
    except (CustomerRegistryError, OSError, ValueError) as exc:
        raise CustomerAdminError("customer registry failed closed") from exc
    runtime = next(
        (item for item in registry.customers if item.spec.customer_key == key), None
    )
    if runtime is None:
        raise CustomerAdminError(f"unknown customer: {key}")
    registered_root = runtime.data_root.resolve()
    if (
        not registered_root.exists()
        or not registered_root.is_dir()
        or not registered_root.is_relative_to(root)
        or _path_has_symlink(registered_root, root)
    ):
        raise CustomerAdminError("registered customer data root is invalid")
    if data_root is not None:
        requested_root = Path(data_root)
        if requested_root.is_symlink() or requested_root.resolve() != registered_root:
            raise CustomerAdminError(
                "data root does not match the registered customer root"
            )
    return runtime


def _payment_period(
    spec: CustomerSpec, kind: PaymentKind | str
) -> tuple[PaymentKind, date, date]:
    try:
        payment_kind = PaymentKind(kind)
    except (TypeError, ValueError) as exc:
        raise CustomerAdminError("payment kind must be initial or renewal") from exc
    starts_on = spec.plan.starts_on
    if payment_kind is PaymentKind.INITIAL:
        return payment_kind, starts_on, starts_on + timedelta(days=27)
    return payment_kind, starts_on + timedelta(days=28), starts_on + timedelta(days=55)


def _accepted_record(result: RecordResult, label: str) -> RecordResult:
    if result.outcome not in {"recorded", "duplicate"} or result.event_id is None:
        raise CustomerAdminError(f"{label} record was not accepted")
    return result


def judge_customer_pilot_kpis(
    path: Path,
    customer_key: str,
    *,
    profile_root: Path,
) -> PilotKPIJudgement:
    """Judge the registered pilot customer using only canonical plan and event paths."""
    runtime = _registered_customer(path, profile_root, customer_key, None)
    spec = runtime.spec
    if not spec.enabled:
        raise CustomerAdminError("customer must be enabled for KPI judgement")
    try:
        transaction = CanonicalEventTransaction.for_customer_runtime(runtime)
    except (OSError, ValueError) as exc:
        raise CustomerAdminError(
            "canonical customer event ledger is unavailable"
        ) from exc
    events_path = transaction.events_path
    if (
        events_path.is_symlink()
        or not events_path.exists()
        or not events_path.is_file()
        or transaction.sequence_path.is_symlink()
        or not transaction.sequence_path.exists()
        or not transaction.sequence_path.is_file()
    ):
        raise CustomerAdminError("canonical customer event ledger is unavailable")
    try:
        transaction.read_snapshot_readonly()
        return judge_pilot_kpis(events_path, plan=spec.plan)
    except (OSError, RuntimeError, UnicodeDecodeError, ValueError) as exc:
        raise CustomerAdminError(
            "canonical customer KPI judgement failed closed"
        ) from exc


def _print_kpi_judgement(customer_key: str, judgement: PilotKPIJudgement) -> None:
    payload = {
        "customer_key": customer_key,
        "window": {
            "starts_on": judgement.starts_on.isoformat(),
            "ends_on": judgement.ends_on.isoformat(),
        },
        "passed": judgement.passed,
        "checkin": {
            "passed": judgement.checkin_pass,
            "rate_percent": judgement.checkin_rate_percent,
        },
        "satisfaction": {
            "passed": judgement.satisfaction_pass,
            "score": judgement.satisfaction_score,
        },
        "operator_time": {
            "passed": judgement.operator_time_pass,
            "weekly_minutes": list(judgement.weekly_operator_minutes),
        },
        "renewal": {"passed": judgement.renewal_valid},
        "failure_reasons": list(judgement.failure_reasons),
    }
    print(json.dumps(payload, ensure_ascii=False, sort_keys=True))


def record_payment(
    path: Path,
    customer_key: str,
    *,
    profile_root: Path,
    data_root: Path | None = None,
    amount_krw: int = 150000,
    paid_on: date | str,
    period_start_on: date | str,
    period_end_on: date | str,
    method: PaymentMethod | str = PaymentMethod.BANK_TRANSFER,
    kind: PaymentKind | str,
) -> RecordResult:
    runtime = _registered_customer(path, profile_root, customer_key, data_root)
    spec = runtime.spec
    if type(amount_krw) is not int or amount_krw != 150000:
        raise CustomerAdminError("payment amount must be exactly 150000 KRW")
    try:
        payment_method = PaymentMethod(method)
    except (TypeError, ValueError) as exc:
        raise CustomerAdminError("payment method must be bank_transfer") from exc
    payment_kind, expected_start, expected_end = _payment_period(spec, kind)
    paid = _require_date(paid_on, "paid_on")
    period_start = _require_date(period_start_on, "period_start_on")
    period_end = _require_date(period_end_on, "period_end_on")
    if period_start != expected_start or period_end != expected_end:
        raise CustomerAdminError(
            f"{payment_kind.value} payment period must match the canonical customer plan"
        )
    if paid > expected_end:
        raise CustomerAdminError(
            "payment paid_on must not be after the payment period end"
        )
    try:
        result = EventStore.for_registered(runtime).record_payment(
            customer_key,
            amount_krw=amount_krw,
            paid_on=paid,
            period_start_on=period_start,
            period_end_on=period_end,
            method=payment_method,
            kind=payment_kind,
            plan=spec.plan,
        )
    except (OSError, RuntimeError, ValueError) as exc:
        raise CustomerAdminError("payment record was rejected") from exc
    return _accepted_record(result, "payment")


def record_satisfaction(
    path: Path,
    customer_key: str,
    *,
    profile_root: Path,
    data_root: Path | None = None,
    score_1to10: int | None = None,
    collected_on: date | str | None = None,
    note: str | None = None,
    score: int | None = None,
    collection_date: date | str | None = None,
) -> RecordResult:
    runtime = _registered_customer(path, profile_root, customer_key, data_root)
    selected_score = score_1to10 if score_1to10 is not None else score
    if type(selected_score) is not int or not 1 <= selected_score <= 10:
        raise CustomerAdminError("satisfaction score must be an integer from 1 to 10")
    if score_1to10 is not None and score is not None and score_1to10 != score:
        raise CustomerAdminError("satisfaction score was supplied twice")
    selected_date = collected_on if collected_on is not None else collection_date
    if selected_date is None:
        raise CustomerAdminError("satisfaction collected_on is required")
    collected = _require_date(selected_date, "collected_on")
    if note is not None and (type(note) is not str or len(note) > 2000):
        raise CustomerAdminError(
            "satisfaction note must be text no longer than 2000 characters"
        )
    try:
        result = EventStore.for_registered(runtime).record_satisfaction(
            customer_key,
            score_1to10=selected_score,
            collected_on=collected,
            note=note,
        )
    except (OSError, RuntimeError, ValueError) as exc:
        raise CustomerAdminError("satisfaction record was rejected") from exc
    return _accepted_record(result, "satisfaction")


def record_operator_time(
    path: Path,
    customer_key: str,
    *,
    profile_root: Path,
    data_root: Path | None = None,
    entry_id: str,
    attempt_id: str,
    minutes: int,
    task: OperatorTask | str,
    work_date: date | str | None = None,
    work_on: date | str | None = None,
    supersedes_entry_id: str | None = None,
) -> RecordResult:
    runtime = _registered_customer(path, profile_root, customer_key, data_root)
    entry = _require_opaque_id(entry_id, "entry_id")
    attempt = _require_opaque_id(attempt_id, "attempt_id")
    replacement = _require_opaque_id(
        supersedes_entry_id,
        "supersedes_entry_id",
        optional=True,
    )
    if replacement is not None and replacement == entry:
        raise CustomerAdminError("operator-time correction must use a new entry_id")
    if type(minutes) is not int or not 1 <= minutes <= 600:
        raise CustomerAdminError(
            "operator-time minutes must be an integer from 1 to 600"
        )
    try:
        operator_task = OperatorTask(task)
    except (TypeError, ValueError) as exc:
        raise CustomerAdminError("operator-time task is invalid") from exc
    selected_date = work_date if work_date is not None else work_on
    if selected_date is None:
        raise CustomerAdminError("operator-time work_date is required")
    worked = _require_date(selected_date, "work_date")
    if work_date is not None and work_on is not None and work_date != work_on:
        raise CustomerAdminError("operator-time work date was supplied twice")
    try:
        result = EventStore.for_registered(runtime).record_operator_time(
            customer_key,
            entry_id=entry,
            attempt_id=attempt,
            minutes=minutes,
            task=operator_task,
            work_date=worked,
            supersedes_entry_id=replacement,
        )
    except (OSError, RuntimeError, ValueError) as exc:
        raise CustomerAdminError("operator-time record was rejected") from exc
    return _accepted_record(result, "operator-time")


def _print_record_receipt(result: RecordResult) -> None:
    print(
        json.dumps(
            {"receipt": result.event_id, "status": result.outcome},
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        )
    )


def register_customer(path: Path, draft: CustomerDraft) -> CustomerSpec:
    with profile_authority_lock(_registry_profile_root(path)):
        return _register_customer_locked(path, draft)


def _register_customer_locked(path: Path, draft: CustomerDraft) -> CustomerSpec:
    """Append one disabled, validated customer draft for plan review."""
    document = _read(path)
    if any(item.customer_key == draft.customer_key for item in document.customers):
        raise CustomerAdminError(f"customer already exists: {draft.customer_key}")
    weeks = tuple(
        {
            "week": week,
            "calories_kcal": draft.calories_kcal,
            "protein_g": draft.protein_g,
            "meal_structure": draft.meals,
            "carbohydrate_g": draft.carbohydrate_g,
            "fat_g": draft.fat_g,
            "water_liters": draft.water_liters,
        }
        for week in range(1, 13)
    )
    customer_payload = {
        "customer_key": draft.customer_key,
        "display_name": draft.display_name,
        "enabled": False,
        "telegram": {
            "user_id": draft.user_id,
            "chat_id": draft.chat_id,
            "topic_id": draft.topic_id,
        },
        "schedule": {
            "daily_time": draft.daily_time,
            "weekly_weekday": draft.weekly_weekday,
            "monthly_day": draft.monthly_day,
        },
        "profile": {
            "primary_goal": draft.primary_goal,
            "dietary_restrictions": draft.dietary_restrictions,
            "allergies": draft.allergies,
            "food_preferences": draft.food_preferences,
            "supplements": draft.supplements,
            "digestion_context": draft.digestion_context,
            "sleep_goal_hours": draft.sleep_goal_hours,
            "recovery_goal": draft.recovery_goal,
            "training_context": draft.training_context,
        },
        "plan": {
            "starts_on": draft.starts_on,
            "focus": "nutrition_90_training_10",
            "weeks": weeks,
        },
    }
    customer = CustomerSpec.model_validate(customer_payload)
    try:
        document.validate_pilot_customer_boundary(customer)
    except CustomerRegistryError as exc:
        raise CustomerAdminError("pilot supports only one external customer") from exc
    payload = document.model_dump(mode="json")
    customers = payload["customers"]
    if not isinstance(customers, list):
        raise CustomerAdminError("registry customers must be a list")
    customers.append(customer.model_dump(mode="json"))
    _write(path, RegistryDocument.model_validate(payload))
    return customer


def set_customer_enabled(
    path: Path, customer_key: str, *, enabled: bool
) -> CustomerSpec:
    with profile_authority_lock(_registry_profile_root(path)):
        return _set_customer_enabled_locked(path, customer_key, enabled=enabled)


def _set_customer_enabled_locked(
    path: Path, customer_key: str, *, enabled: bool
) -> CustomerSpec:
    """Disable a customer; enabling is available only through activate_customer."""
    if enabled:
        raise CustomerAdminError(
            "direct enabling is not allowed; use activate_customer"
        )
    document = _read(path)
    selected: CustomerSpec | None = None
    customers: list[CustomerSpec] = []
    for customer in document.customers:
        updated = (
            customer.model_copy(update={"enabled": False})
            if customer.customer_key == customer_key
            else customer
        )
        if customer.customer_key == customer_key:
            selected = updated
        customers.append(updated)
    if selected is None:
        raise CustomerAdminError(f"unknown customer: {customer_key}")
    payload = document.model_dump(mode="json")
    payload["customers"] = [customer.model_dump(mode="json") for customer in customers]
    _write(path, RegistryDocument.model_validate(payload))
    return selected


def disable_customer(path: Path, customer_key: str) -> CustomerSpec:
    """Disable a customer without evaluating activation prerequisites."""
    return set_customer_enabled(path, customer_key, enabled=False)


def set_customer_ai_consent(
    path: Path, customer_key: str, consent: AiProcessingConsent
) -> CustomerSpec:
    with profile_authority_lock(_registry_profile_root(path)):
        return _set_customer_ai_consent_locked(path, customer_key, consent)


def _set_customer_ai_consent_locked(
    path: Path, customer_key: str, consent: AiProcessingConsent
) -> CustomerSpec:
    document = _read(path)
    selected: CustomerSpec | None = None
    customers: list[CustomerSpec] = []
    for customer in document.customers:
        updated = (
            customer.model_copy(update={"ai_processing_consent": consent})
            if customer.customer_key == customer_key
            else customer
        )
        if customer.customer_key == customer_key:
            selected = updated
        customers.append(updated)
    if selected is None:
        raise CustomerAdminError(f"unknown customer: {customer_key}")
    payload = document.model_dump(mode="json")
    payload["customers"] = [customer.model_dump(mode="json") for customer in customers]
    _write(path, RegistryDocument.model_validate(payload))
    return selected


def withdraw_customer(
    profile_root: Path | str,
    customer_key: str,
    *,
    kst_date: date | datetime,
) -> CustomerSpec:
    """Atomically disable a customer and record AI consent withdrawal."""
    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    if kst_date is None:
        raise CustomerAdminError("kst_date is required")
    withdrawal_day = _activation_kst_date(kst_date)
    withdrawn_consent = AiProcessingConsent(
        granted=False,
        recorded_on=withdrawal_day,
        notice_version=CONSENT_VERSION,
    )
    with profile_authority_lock(root):
        registry_path = _resolve_registry_path(root)
        _recover_activation_journal_locked(root, registry_path)
        document = _read_profile_registry(registry_path, root)
        selected = next(
            (item for item in document.customers if item.customer_key == key),
            None,
        )
        if selected is None:
            raise CustomerAdminError(f"unknown customer: {key}")
        withdrawn = selected.model_copy(
            update={
                "enabled": False,
                "ai_processing_consent": withdrawn_consent,
            }
        )
        if withdrawn == selected:
            return selected
        payload = document.model_dump(mode="json")
        payload["customers"] = [
            withdrawn.model_dump(mode="json")
            if item.customer_key == key
            else item.model_dump(mode="json")
            for item in document.customers
        ]
        transitioned = RegistryDocument.model_validate(payload)
        _write(registry_path, transitioned)
        return next(item for item in transitioned.customers if item.customer_key == key)


_REQUIRED_CHECKLIST_ITEMS = (
    "token_rotated",
    "missend_test_passed",
    "provider_terms_checked",
    "withdrawal_deletion_doc",
    "retention_backup_doc",
    "manual_fallback_doc",
)
_PERSONAL_PATH_MARKERS = ("personal", "richard", "owner")
_HistoricalActivationArtifactClassifier = Callable[[Path, bytes], bool]
_activation_historical_artifact_classifier: ContextVar[
    _HistoricalActivationArtifactClassifier | None
] = ContextVar("activation_historical_artifact_classifier", default=None)


@contextmanager
def _activation_historical_artifact_audit(
    classifier: _HistoricalActivationArtifactClassifier,
) -> Iterator[None]:
    """Scope a caller-authenticated historical-artifact exception to one audit."""
    if not callable(classifier):
        raise CustomerAdminError("G1 historical artifact classifier is invalid")
    token = _activation_historical_artifact_classifier.set(classifier)
    try:
        yield
    finally:
        _activation_historical_artifact_classifier.reset(token)


def activate_customer(
    profile_root: Path,
    data_root: Path,
    customer_id: str,
    checklist_evidence_path: Path,
    staff_membership_evidence_path: Path,
    *,
    kst_date: date | datetime | None = None,
) -> ActivationResult:
    root = _resolve_profile_root(Path(profile_root))
    with profile_authority_lock(root):
        return _activate_customer_locked(
            root,
            data_root,
            customer_id,
            checklist_evidence_path,
            staff_membership_evidence_path,
            kst_date=kst_date,
        )


def _activate_customer_locked(
    profile_root: Path,
    data_root: Path,
    customer_id: str,
    checklist_evidence_path: Path,
    staff_membership_evidence_path: Path,
    *,
    kst_date: date | datetime | None = None,
) -> ActivationResult:
    """Enable one customer only after the complete G1-G5 activation guard."""
    root = _resolve_profile_root(profile_root)
    registry_path = _resolve_registry_path(root)
    pre_activation_document = _read_profile_registry(registry_path, root)
    pre_activation_spec = next(
        (item for item in pre_activation_document.customers if item.customer_key == customer_id),
        None,
    )
    if pre_activation_spec is None:
        raise CustomerAdminError(f"unknown customer: {customer_id}")
    try:
        membership_bindings = validate_staff_membership_evidence(
            Path(staff_membership_evidence_path),
            profile_root=root,
            registry_path=registry_path,
            customer_id=customer_id,
            customer_user_id=pre_activation_spec.telegram.user_id,
        )
    except StaffMembershipEvidenceError as exc:
        raise CustomerAdminError(str(exc)) from exc
    _recover_activation_journal(root, registry_path)
    document = _read_profile_registry(registry_path, root)
    _validate_enabled_activation_receipts(root, registry_path, document)
    try:
        registry = load_customer_registry(registry_path, root)
    except (CustomerRegistryError, OSError, ValueError) as exc:
        raise CustomerAdminError("registry failed activation validation") from exc
    runtime = next(
        (item for item in registry.customers if item.spec.customer_key == customer_id),
        None,
    )
    if runtime is None:
        raise CustomerAdminError(f"unknown customer: {customer_id}")
    spec = runtime.spec
    if spec.enabled:
        raise CustomerAdminError(f"customer is already enabled: {customer_id}")
    activation_day = _activation_kst_date(kst_date)
    window_start = spec.plan.starts_on
    window_end = window_start + timedelta(days=27)
    if not window_start <= activation_day <= window_end:
        raise CustomerAdminError(
            "activation is outside the KST plan window "
            f"[{window_start.isoformat()}, {window_end.isoformat()}]"
        )
    enabled_spec = spec.model_copy(update={"enabled": True})
    try:
        document.validate_pilot_customer_boundary(enabled_spec)
    except (CustomerRegistryError, ValueError) as exc:
        raise CustomerAdminError("pilot supports only one external customer") from exc

    try:
        spec.validate_local_activation_prerequisites()
    except (CustomerRegistryError, ValueError) as exc:
        raise CustomerAdminError(
            "G2/G3 customer activation prerequisites failed"
        ) from exc

    try:
        document.validate_enabled_customer(enabled_spec)
    except (CustomerRegistryError, ValueError) as exc:
        raise CustomerAdminError("G4 customer identity validation failed") from exc

    _read_activation_checklist(
        Path(checklist_evidence_path),
        root,
        runtime.data_root,
        registry_path,
        spec,
    )

    try:
        readiness_audit = require_nutrition_start_readiness(root, customer_id)
    except NutritionReadinessError as exc:
        raise CustomerAdminError(str(exc)) from exc
    classifier = _authenticated_onboarding_artifact_classifier(
        runtime.data_root,
        readiness_audit.digests,
    )
    with _activation_historical_artifact_audit(classifier):
        _validate_activation_data_root(
            root,
            Path(data_root),
            runtime.data_root,
            document,
            spec,
        )
    nutrition_activation_receipt = build_nutrition_activation_receipt_v2(
        customer_key=customer_id,
        readiness_receipt_digest=readiness_audit.digests["receipt"],
        readiness_bundle_digest=_json_digest(dict(readiness_audit.digests)),
        customer_projection_digest=customer_nutrition_projection_digest(
            spec.model_dump(mode="json")
        ),
        input_reconciliation_digest=readiness_audit.digests["input_reconciliation"],
    )

    try:
        initialize_adaptive_customer(runtime.data_root)
        feature_epoch = read_private_json(
            runtime.data_root / "nutrition-plans" / "feature-epoch.json"
        )
        if (
            feature_epoch.get("activation") is not False
            or feature_epoch.get("delivery") is not False
        ):
            raise ValueError("adaptive runtime flags must remain disabled")
        _promote_onboarding_adjustment_policy(
            runtime.data_root,
            owner=document.owner,
            plan_starts_on=spec.plan.starts_on,
            expected_digest=readiness_audit.digests["adjustment_policy"],
        )
    except (OSError, TypeError, ValueError) as exc:
        raise CustomerAdminError("adaptive runtime preparation failed") from exc
    payload = document.model_dump(mode="json")
    payload["customers"] = [
        enabled_spec.model_dump(mode="json")
        if item.customer_key == customer_id
        else item.model_dump(mode="json")
        for item in document.customers
    ]
    try:
        activated_document = RegistryDocument.model_validate(payload)
    except (CustomerRegistryError, ValueError) as exc:
        raise CustomerAdminError("activation registry transition failed") from exc

    registry_before = registry_path.read_bytes()
    audit_path = _activation_audit_path(root)
    audit_before = _read_optional_bytes(audit_path)
    transaction_id = uuid.uuid4().hex
    registry_sha256 = _document_fingerprint(activated_document)
    recorded_at = datetime.now(timezone.utc).isoformat()
    audit_record = _build_activation_audit_record(
        registry_path,
        runtime.data_root,
        customer_id,
        Path(checklist_evidence_path),
        transaction_id=transaction_id,
        registry_sha256=registry_sha256,
        recorded_at=recorded_at,
        nutrition_activation_receipt=nutrition_activation_receipt,
        membership_bindings=membership_bindings,
    )
    pending = {
        "version": 3,
        "state": "prepared",
        "recovery_required": False,
        "transaction_id": transaction_id,
        "customer_id": customer_id,
        "registry_path": str(registry_path.resolve()),
        "data_root": str(runtime.data_root.resolve()),
        "checklist_evidence_path": str(Path(checklist_evidence_path).resolve()),
        "audit_path": str(audit_path.resolve()),
        "registry_sha256": registry_sha256,
        "previous_registry_sha256": _document_fingerprint(document),
        "audit_record_sha256": _json_digest(audit_record),
        "previous_audit_sha256": _bytes_digest(audit_before),
        "previous_registry": document.model_dump(mode="json"),
        "created_at": recorded_at,
        "prepared_at": recorded_at,
        "nutrition_activation_receipt": nutrition_activation_receipt,
        **membership_bindings,
    }
    journal_path = _activation_journal_path(root)
    _write_activation_journal(journal_path, pending)

    try:
        _append_activation_audit(
            root,
            registry_path,
            runtime.data_root,
            customer_id,
            Path(checklist_evidence_path),
            transaction_id=transaction_id,
            registry_sha256=registry_sha256,
            recorded_at=recorded_at,
            nutrition_activation_receipt=nutrition_activation_receipt,
            membership_bindings=membership_bindings,
        )
        _write(registry_path, activated_document)
        try:
            written_registry_sha256 = _document_fingerprint(_read(registry_path))
            written_audit_state = _activation_audit_receipt_state(
                audit_path,
                transaction_id=transaction_id,
                audit_record_sha256=_json_digest(audit_record),
            )
        except Exception as exc:
            raise CustomerAdminError(
                "activation side effects could not be verified"
            ) from exc
        if (
            written_registry_sha256 != registry_sha256
            or written_audit_state != "present"
        ):
            raise CustomerAdminError("activation side effects could not be verified")
        _write_activation_journal(
            journal_path,
            {
                **pending,
                "state": "committed",
                "recovery_required": False,
                "committed_at": datetime.now(timezone.utc).isoformat(),
            },
        )
    except Exception as exc:
        try:
            _rollback_activation(
                registry_path,
                registry_before,
                audit_path,
                audit_before,
                journal_path,
                pending,
            )
        except CustomerAdminError as rollback_error:
            raise rollback_error from exc
        if isinstance(exc, CustomerAdminError):
            raise
        raise CustomerAdminError(
            "activation transaction could not be committed"
        ) from exc

    activated = next(
        item
        for item in activated_document.customers
        if item.customer_key == customer_id
    )
    return ActivationResult(activated, registry_path, runtime.data_root, audit_path)


def _promote_onboarding_adjustment_policy(
    data_root: Path,
    *,
    owner: TelegramAddress,
    plan_starts_on: date,
    expected_digest: str,
) -> None:
    source_path = data_root / "nutrition-onboarding" / "adjustment-policy-v1.json"
    source = read_private_json(source_path)
    source_digest = source.get("digest")
    policy_value = {key: value for key, value in source.items() if key != "digest"}
    if (
        not isinstance(source_digest, str)
        or source_digest != expected_digest
        or _json_digest(policy_value) != source_digest
    ):
        raise ValueError("onboarding adjustment policy digest is stale")
    policy = customer_policy_from_onboarding_artifact(policy_value)
    if policy.starts_on != plan_starts_on:
        raise ValueError("onboarding adjustment policy start is stale")
    promoted = {
        "schema_version": "1.0",
        "version": source["policy_version"],
        "digest": source_digest,
        "approved": True,
        "approved_by": owner.model_dump(mode="json"),
        "approved_at_kst": source["effective_at_kst"],
        "policy": policy_value,
    }
    plans_root = data_root / "nutrition-plans"
    policy_path = plans_root / "policy.json"
    current = read_private_json(policy_path)
    if current == promoted:
        return
    if current != {"schema_version": "1.0", "enabled": False}:
        raise ValueError("adaptive policy promotion conflicts with existing policy")
    for filename in ("policy-revisions.jsonl", "policy-audit.jsonl"):
        journal_path = plans_root / filename
        validate_private_file(journal_path)
        if journal_path.stat().st_size:
            raise ValueError(
                "adaptive policy promotion conflicts with revision history"
            )
    atomic_write_private_json(policy_path, promoted)


def _resolve_profile_root(profile_root: Path) -> Path:
    candidate = Path(profile_root)
    if candidate.is_symlink() or not candidate.exists() or not candidate.is_dir():
        raise CustomerAdminError("profile root must be an existing directory")
    resolved = candidate.resolve()
    if resolved.is_symlink():
        raise CustomerAdminError("profile root symlinks are not allowed")
    return resolved


def _resolve_registry_path(profile_root: Path) -> Path:
    candidates = (
        profile_root / "customers" / "registry.json",
        profile_root / "registry.json",
    )
    for candidate in candidates:
        if candidate.exists():
            if candidate.is_symlink():
                raise CustomerAdminError("registry symlinks are not allowed")
            resolved = candidate.resolve()
            if not resolved.is_relative_to(profile_root):
                raise CustomerAdminError("registry escapes the profile")
            return resolved
    raise CustomerAdminError("profile registry.json was not found")


def _registry_profile_root(path: Path) -> Path:
    requested = Path(path)
    try:
        resolved = requested.resolve()
    except (OSError, RuntimeError) as exc:
        raise CustomerAdminError("registry path is invalid") from exc
    root_candidate = (
        resolved.parent.parent
        if resolved.parent.name == "customers"
        else resolved.parent
    )
    root = _resolve_profile_root(root_candidate)
    if _resolve_registry_path(root) != resolved:
        raise CustomerAdminError("registry path does not match the profile registry")
    return root


def _read_profile_registry(path: Path, profile_root: Path) -> RegistryDocument:
    resolved = path.resolve()
    if not resolved.is_relative_to(profile_root):
        raise CustomerAdminError("registry escapes the profile")
    return _read(path)


_ACTIVATION_JOURNAL_FILE = "customer-activation-journal.json"
_ACTIVATION_AUDIT_FILE = "customer-activation-audit.jsonl"


def _activation_journal_path(profile_root: Path) -> Path:
    return profile_root / "data" / _ACTIVATION_JOURNAL_FILE


def _activation_audit_path(profile_root: Path) -> Path:
    return profile_root / "data" / _ACTIVATION_AUDIT_FILE


def _json_digest(value: object) -> str:
    payload = json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def _bytes_digest(value: bytes | None) -> str | None:
    if value is None:
        return None
    return hashlib.sha256(value).hexdigest()


def _document_fingerprint(document: RegistryDocument) -> str:
    return _json_digest(document.model_dump(mode="json"))


def _ensure_private_parent(path: Path) -> None:
    parent = path.parent
    if parent.is_symlink():
        raise CustomerAdminError(
            "activation persistence parent symlinks are not allowed"
        )
    parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    parent.chmod(0o700)


def _fsync_directory(path: Path) -> None:
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    descriptor = os.open(path, flags)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _atomic_write_bytes(path: Path, content: bytes) -> None:
    if path.is_symlink():
        raise CustomerAdminError("activation persistence symlinks are not allowed")
    _ensure_private_parent(path)
    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
    try:
        with temporary.open("wb") as handle:
            temporary.chmod(0o600)
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        path.chmod(0o600)
        _fsync_directory(path.parent)
    except OSError:
        try:
            temporary.unlink()
        except OSError:
            pass
        raise


def _read_optional_bytes(path: Path) -> bytes | None:
    if path.is_symlink():
        raise CustomerAdminError("activation persistence symlinks are not allowed")
    if not path.exists():
        return None
    if not path.is_file():
        raise CustomerAdminError("activation persistence path must be a regular file")
    try:
        return path.read_bytes()
    except OSError as exc:
        raise CustomerAdminError("activation persistence could not be read") from exc


def _restore_bytes(path: Path, previous: bytes | None) -> None:
    if previous is None:
        if path.is_symlink():
            raise CustomerAdminError("activation persistence symlinks are not allowed")
        if path.exists():
            path.unlink()
            _fsync_directory(path.parent)
        return
    _atomic_write_bytes(path, previous)


def _read_activation_journal(path: Path) -> dict[str, object] | None:
    if path.is_symlink():
        raise CustomerAdminError("activation journal symlinks are not allowed")
    if not path.exists():
        return None
    if not path.is_file():
        raise CustomerAdminError("activation journal must be a regular file")
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CustomerAdminError("activation journal is invalid") from exc
    if not isinstance(payload, dict):
        raise CustomerAdminError("activation journal must be a JSON object")
    return payload


def _write_activation_journal(path: Path, payload: dict[str, object]) -> None:
    state = payload.get("state")
    if state not in {"prepared", "committed", "abandoned"}:
        raise CustomerAdminError("activation journal state is invalid")
    try:
        content = (
            json.dumps(
                payload,
                ensure_ascii=False,
                sort_keys=True,
                indent=2,
            ).encode("utf-8")
            + b"\n"
        )
        _atomic_write_bytes(path, content)
    except (OSError, TypeError, ValueError) as exc:
        if isinstance(exc, CustomerAdminError):
            raise
        raise CustomerAdminError("activation journal could not be persisted") from exc


def _resolve_journal_path(value: object, profile_root: Path, label: str) -> Path:
    if not isinstance(value, str) or not value:
        raise CustomerAdminError(f"activation journal {label} is invalid")
    candidate = Path(value)
    if candidate.is_symlink():
        raise CustomerAdminError(f"activation journal {label} symlinks are not allowed")
    try:
        resolved = candidate.resolve()
    except (OSError, RuntimeError) as exc:
        raise CustomerAdminError(f"activation journal {label} is invalid") from exc
    if not resolved.is_relative_to(profile_root):
        raise CustomerAdminError(f"activation journal {label} escapes the profile")
    return resolved


def _build_activation_audit_record(
    registry_path: Path,
    data_root: Path,
    customer_id: str,
    checklist_path: Path,
    *,
    transaction_id: str | None = None,
    registry_sha256: str | None = None,
    recorded_at: str | None = None,
    nutrition_activation_receipt: Mapping[str, str] | None = None,
    membership_bindings: Mapping[str, object] | None = None,
) -> dict[str, object]:
    record: dict[str, object] = {
        "event": "customer_activation",
        "customer_id": customer_id,
        "enabled": True,
        "registry_path": str(registry_path),
        "data_root": str(data_root),
        "checklist_evidence_path": str(checklist_path.resolve()),
        "recorded_at": recorded_at or datetime.now(timezone.utc).isoformat(),
    }
    if transaction_id is not None:
        record["transaction_id"] = transaction_id
    if registry_sha256 is not None:
        record["registry_sha256"] = registry_sha256
    if nutrition_activation_receipt is not None:
        record["nutrition_activation_receipt"] = dict(nutrition_activation_receipt)
    if membership_bindings is not None:
        record.update(membership_bindings)
    return record


def _activation_failure_detail(
    component: str,
    operation: str,
    error: BaseException,
) -> dict[str, str]:
    return {
        "component": component,
        "operation": operation,
        "error_type": type(error).__name__,
        "error": str(error) or type(error).__name__,
    }


def _record_activation_recovery_required(
    journal_path: Path,
    pending: dict[str, object],
    failures: list[dict[str, str]],
) -> None:
    prior_failures = pending.get("recovery_failures")
    combined_failures = (
        [item for item in prior_failures if isinstance(item, dict)]
        if isinstance(prior_failures, list)
        else []
    ) + failures
    payload = {
        **pending,
        "state": "prepared",
        "recovery_required": True,
        "recovery_failures": combined_failures,
        "recovery_recorded_at": datetime.now(timezone.utc).isoformat(),
    }
    try:
        _write_activation_journal(journal_path, payload)
    except Exception as exc:
        raise CustomerAdminError(
            "activation recovery required and failure evidence could not be persisted"
        ) from exc


def _rollback_activation(
    registry_path: Path,
    registry_before: bytes,
    audit_path: Path,
    audit_before: bytes | None,
    journal_path: Path,
    pending: dict[str, object],
) -> None:
    failures: list[dict[str, str]] = []
    registry_error: BaseException | None = None
    try:
        _restore_bytes(registry_path, registry_before)
    except Exception as exc:
        registry_error = exc
    try:
        registry_after = _read_optional_bytes(registry_path)
    except Exception as exc:
        registry_after = None
        registry_error = registry_error or exc
    if registry_after != registry_before:
        failures.append(
            _activation_failure_detail(
                "registry",
                "restore",
                registry_error
                or CustomerAdminError(
                    "registry bytes do not match the prepared snapshot"
                ),
            )
        )

    audit_error: BaseException | None = None
    try:
        _restore_bytes(audit_path, audit_before)
    except Exception as exc:
        audit_error = exc
    try:
        audit_after = _read_optional_bytes(audit_path)
    except Exception as exc:
        audit_after = object()
        audit_error = audit_error or exc
    if audit_after != audit_before:
        failures.append(
            _activation_failure_detail(
                "audit",
                "restore",
                audit_error
                or CustomerAdminError("audit bytes do not match the prepared snapshot"),
            )
        )

    if failures:
        _record_activation_recovery_required(journal_path, pending, failures)
        raise CustomerAdminError("activation recovery required")

    try:
        _write_activation_journal(
            journal_path,
            {
                **pending,
                "state": "abandoned",
                "recovery_required": False,
                "abandoned_at": datetime.now(timezone.utc).isoformat(),
            },
        )
    except Exception as exc:
        terminalization_failure = _activation_failure_detail(
            "journal",
            "record abandoned state",
            exc,
        )
        try:
            _record_activation_recovery_required(
                journal_path,
                pending,
                [terminalization_failure],
            )
        except CustomerAdminError as evidence_error:
            raise evidence_error from exc
        raise CustomerAdminError("activation recovery required") from exc


def _remove_activation_audit_record(path: Path, transaction_id: str) -> None:
    if path.is_symlink():
        raise CustomerAdminError("activation audit path symlinks are not allowed")
    if not path.exists():
        return
    if not path.is_file():
        raise CustomerAdminError("activation audit path must be a regular file")
    try:
        lines = path.read_bytes().splitlines(keepends=True)
    except OSError as exc:
        raise CustomerAdminError("activation audit could not be read") from exc
    retained: list[bytes] = []
    changed = False
    for line in lines:
        try:
            record = json.loads(line.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            retained.append(line)
            continue
        if isinstance(record, dict) and record.get("transaction_id") == transaction_id:
            changed = True
            continue
        retained.append(line)
    if changed:
        _atomic_write_bytes(path, b"".join(retained))


def _activation_audit_receipt_state(
    audit_path: Path,
    *,
    transaction_id: str,
    audit_record_sha256: str,
) -> str:
    """Return present/absent, rejecting any ambiguous audit ledger state."""
    if audit_path.is_symlink():
        raise CustomerAdminError("activation audit path symlinks are not allowed")
    if not audit_path.exists():
        return "absent"
    if not audit_path.is_file():
        raise CustomerAdminError("activation audit path must be a regular file")
    try:
        lines = audit_path.read_bytes().splitlines()
    except (OSError, UnicodeDecodeError) as exc:
        raise CustomerAdminError("activation audit could not be read") from exc
    matching = 0
    for line in lines:
        if not line.strip():
            continue
        try:
            record = json.loads(line.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CustomerAdminError("activation audit is invalid") from exc
        if not isinstance(record, dict):
            raise CustomerAdminError("activation audit record is invalid")
        if record.get("transaction_id") != transaction_id:
            continue
        matching += 1
        if (
            record.get("enabled") is not True
            or _json_digest(record) != audit_record_sha256
        ):
            raise CustomerAdminError("activation audit receipt is ambiguous")
    if matching > 1:
        raise CustomerAdminError("activation audit receipt is duplicated")
    return "present" if matching == 1 else "absent"


def _recover_activation_journal(profile_root: Path, registry_path: Path) -> None:
    root = _resolve_profile_root(Path(profile_root))
    canonical_registry = _resolve_registry_path(root)
    if Path(registry_path).resolve() != canonical_registry:
        raise CustomerAdminError(
            "activation journal registry does not match the profile"
        )
    with profile_authority_lock(root):
        _recover_activation_journal_locked(root, canonical_registry)


def _recover_activation_journal_locked(profile_root: Path, registry_path: Path) -> None:
    journal_path = _activation_journal_path(profile_root)
    journal = _read_activation_journal(journal_path)
    if journal is None:
        return
    state = journal.get("state")
    if journal.get("version") not in {1, 2, 3}:
        raise CustomerAdminError("activation journal version is invalid")
    if state in {"committed", "abandoned"}:
        return
    if state not in {"prepared", "pending", "recovery_required"}:
        raise CustomerAdminError("activation journal has an unknown state")

    journal_registry = _resolve_journal_path(
        journal.get("registry_path"),
        profile_root,
        "registry_path",
    )
    if journal_registry != registry_path.resolve():
        raise CustomerAdminError(
            "activation journal registry does not match the profile"
        )
    previous_payload = journal.get("previous_registry")
    if not isinstance(previous_payload, dict):
        raise CustomerAdminError("pending activation has no rollback registry")
    try:
        previous = RegistryDocument.model_validate(previous_payload)
    except (CustomerRegistryError, TypeError, ValueError) as exc:
        raise CustomerAdminError(
            "pending activation rollback registry is invalid"
        ) from exc

    transaction_id = journal.get("transaction_id")
    registry_sha256 = journal.get("registry_sha256")
    audit_record_sha256 = journal.get("audit_record_sha256")
    customer_id = journal.get("customer_id")
    if (
        not isinstance(transaction_id, str)
        or not transaction_id
        or not isinstance(registry_sha256, str)
        or not registry_sha256
        or not isinstance(audit_record_sha256, str)
        or not audit_record_sha256
        or not isinstance(customer_id, str)
        or not customer_id
    ):
        raise CustomerAdminError("pending activation journal is invalid")
    previous_customer = next(
        (item for item in previous.customers if item.customer_key == customer_id),
        None,
    )
    if previous_customer is None or previous_customer.enabled:
        raise CustomerAdminError("pending activation rollback customer is invalid")
    previous_sha256 = _document_fingerprint(previous)
    recorded_previous_sha256 = journal.get("previous_registry_sha256")
    if (
        recorded_previous_sha256 is not None
        and recorded_previous_sha256 != previous_sha256
    ):
        raise CustomerAdminError(
            "pending activation rollback registry fingerprint is invalid"
        )
    if registry_sha256 == previous_sha256:
        raise CustomerAdminError(
            "pending activation journal has identical registry states"
        )

    audit_path = _resolve_journal_path(
        journal.get("audit_path"), profile_root, "audit_path"
    )
    if audit_path != _activation_audit_path(profile_root).resolve():
        raise CustomerAdminError("activation journal audit does not match the profile")
    data_root = _resolve_journal_path(
        journal.get("data_root"), profile_root, "data_root"
    )
    expected_data_root = (profile_root / "data" / "customers" / customer_id).resolve()
    if data_root != expected_data_root:
        raise CustomerAdminError(
            "activation journal data root does not match the customer"
        )

    def require_recovery(
        component: str,
        operation: str,
        error: BaseException,
    ) -> None:
        failure = _activation_failure_detail(component, operation, error)
        _record_activation_recovery_required(journal_path, journal, [failure])
        raise CustomerAdminError("pending activation recovery required") from error

    try:
        current = _read(registry_path)
        current_fingerprint = _document_fingerprint(current)
    except Exception as exc:
        require_recovery("registry", "inspect", exc)

    try:
        audit_state = _activation_audit_receipt_state(
            audit_path,
            transaction_id=transaction_id,
            audit_record_sha256=audit_record_sha256,
        )
    except Exception as exc:
        require_recovery("audit", "inspect", exc)

    def terminalize(state_name: str) -> None:
        try:
            _write_activation_journal(
                journal_path,
                {
                    **journal,
                    "state": state_name,
                    "recovery_required": False,
                    f"{state_name}_at": datetime.now(timezone.utc).isoformat(),
                },
            )
        except Exception as exc:
            require_recovery("journal", f"record {state_name} state", exc)

    if current_fingerprint == registry_sha256 and audit_state == "present":
        if (
            not data_root.exists()
            or not data_root.is_dir()
            or _path_has_symlink(data_root, profile_root)
        ):
            require_recovery(
                "data_root",
                "verify committed activation",
                CustomerAdminError("activation data root is missing or unsafe"),
            )
        terminalize("committed")
        return

    if current_fingerprint == previous_sha256 and audit_state == "absent":
        terminalize("abandoned")
        return

    if current_fingerprint == registry_sha256 and audit_state == "absent":
        try:
            _write(registry_path, previous)
        except Exception as exc:
            require_recovery("registry", "restore", exc)
        try:
            restored = _document_fingerprint(_read(registry_path))
        except Exception as exc:
            require_recovery("registry", "verify restore", exc)
        if restored != previous_sha256:
            require_recovery(
                "registry",
                "verify restore",
                CustomerAdminError(
                    "registry restore did not reach the prepared snapshot"
                ),
            )
        try:
            audit_state = _activation_audit_receipt_state(
                audit_path,
                transaction_id=transaction_id,
                audit_record_sha256=audit_record_sha256,
            )
        except Exception as exc:
            require_recovery("audit", "verify compensation", exc)
        if audit_state != "absent":
            require_recovery(
                "audit",
                "verify compensation",
                CustomerAdminError("activation audit receipt remains present"),
            )
        terminalize("abandoned")
        return

    if current_fingerprint == previous_sha256 and audit_state == "present":
        try:
            _remove_activation_audit_record(audit_path, transaction_id)
        except Exception as exc:
            require_recovery("audit", "restore", exc)
        try:
            audit_state = _activation_audit_receipt_state(
                audit_path,
                transaction_id=transaction_id,
                audit_record_sha256=audit_record_sha256,
            )
        except Exception as exc:
            require_recovery("audit", "verify restore", exc)
        if audit_state != "absent":
            require_recovery(
                "audit",
                "verify restore",
                CustomerAdminError("activation audit receipt remains present"),
            )
        terminalize("abandoned")
        return

    require_recovery(
        "activation",
        "reconcile",
        CustomerAdminError("registry and audit states are ambiguous"),
    )


def _audit_contains_receipt(
    audit_path: Path,
    *,
    transaction_id: str,
    audit_record_sha256: str,
) -> bool:
    try:
        return (
            _activation_audit_receipt_state(
                audit_path,
                transaction_id=transaction_id,
                audit_record_sha256=audit_record_sha256,
            )
            == "present"
        )
    except CustomerAdminError:
        return False


def _require_committed_activation_receipt(
    profile_root: Path,
    registry_path: Path,
    document: RegistryDocument,
    spec: CustomerSpec,
) -> str:
    journal = _read_activation_journal(_activation_journal_path(profile_root))
    if journal is None or journal.get("state") != "committed":
        raise CustomerAdminError("committed activation receipt is missing or stale")
    version = journal.get("version")
    if version not in {1, 2, 3} or journal.get("customer_id") != spec.customer_key:
        raise CustomerAdminError("committed activation receipt is missing or stale")
    if journal.get("recovery_required") is True:
        raise CustomerAdminError("committed activation receipt is missing or stale")
    transaction_id = journal.get("transaction_id")
    registry_sha256 = journal.get("registry_sha256")
    audit_record_sha256 = journal.get("audit_record_sha256")
    if (
        not isinstance(transaction_id, str)
        or not transaction_id
        or not isinstance(registry_sha256, str)
        or not isinstance(audit_record_sha256, str)
    ):
        raise CustomerAdminError("committed activation receipt is missing or stale")
    if version in {2, 3}:
        receipt = journal.get("nutrition_activation_receipt")
        if not isinstance(receipt, dict):
            raise CustomerAdminError(
                "nutrition activation v2 receipt is missing or stale"
            )
        try:
            readiness = require_nutrition_start_readiness(
                profile_root, spec.customer_key
            )
            expected_receipt = build_nutrition_activation_receipt_v2(
                customer_key=spec.customer_key,
                readiness_receipt_digest=readiness.digests["receipt"],
                readiness_bundle_digest=_json_digest(dict(readiness.digests)),
                customer_projection_digest=customer_nutrition_projection_digest(
                    spec.model_dump(mode="json")
                ),
                input_reconciliation_digest=readiness.digests["input_reconciliation"],
            )
        except NutritionReadinessError as exc:
            raise CustomerAdminError(str(exc)) from exc
        except (KeyError, ValueError) as exc:
            raise CustomerAdminError(
                "nutrition activation v2 receipt is missing or stale"
            ) from exc
        if receipt != expected_receipt:
            raise CustomerAdminError(
                "nutrition activation v2 receipt is missing or stale"
            )
        if version == 3:
            membership_path = journal.get("staff_membership_evidence_path")
            required_membership_fields = (
                "staff_membership_evidence_sha256",
                "staff_chat_inventory_sha256",
                "membership_subscription_epoch_id",
            )
            if not isinstance(membership_path, str) or any(
                not isinstance(journal.get(field), str) or not journal.get(field)
                for field in required_membership_fields
            ):
                raise CustomerAdminError(
                    "committed staff membership evidence is missing or stale"
                )
            resolved_membership = _resolve_journal_path(
                membership_path, profile_root, "staff_membership_evidence_path"
            )
            try:
                membership_bytes = resolved_membership.read_bytes()
                membership_digest = hashlib.sha256(membership_bytes).hexdigest()
                membership_payload = json.loads(membership_bytes)
            except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
                raise CustomerAdminError(
                    "committed staff membership evidence is missing or stale"
                ) from exc
            if (
                not isinstance(membership_payload, dict)
                or membership_digest != journal.get("staff_membership_evidence_sha256")
                or membership_payload.get("staff_chat_inventory_sha256")
                != journal.get("staff_chat_inventory_sha256")
                or membership_payload.get("subscription_epoch_id")
                != journal.get("membership_subscription_epoch_id")
            ):
                raise CustomerAdminError(
                    "committed staff membership evidence is missing or stale"
                )
    else:
        manifest_path = (
            profile_root
            / "data"
            / "migrations"
            / "nutrition-readiness-v1"
            / "legacy-activation-authority.json"
        )
        try:
            validate_profile_path(manifest_path, profile_root)
            manifest = read_private_json(manifest_path)
        except (OSError, ValueError) as exc:
            raise CustomerAdminError(
                "legacy activation authority manifest is missing or stale"
            ) from exc
        if not isinstance(manifest, dict) or not validate_legacy_activation_authority(
            manifest,
            customer_key=spec.customer_key,
            activation_receipt_digest=_json_digest(journal),
            registry_projection_digest=customer_nutrition_projection_digest(
                spec.model_dump(mode="json")
            ),
            owner_digest=adaptive_digest(document.owner.model_dump(mode="json")),
        ):
            raise CustomerAdminError(
                "legacy activation authority manifest is missing or stale"
            )
    try:
        receipt_registry = _resolve_journal_path(
            journal.get("registry_path"),
            profile_root,
            "registry_path",
        )
        receipt_data = _resolve_journal_path(
            journal.get("data_root"),
            profile_root,
            "data_root",
        )
        receipt_audit = _resolve_journal_path(
            journal.get("audit_path"),
            profile_root,
            "audit_path",
        )
    except CustomerAdminError:
        raise CustomerAdminError("committed activation receipt is missing or stale")
    expected_data = (profile_root / "data" / "customers" / spec.customer_key).resolve()
    if (
        receipt_registry != registry_path.resolve()
        or receipt_data != expected_data
        or receipt_audit != _activation_audit_path(profile_root).resolve()
        or _document_fingerprint(document) != registry_sha256
        or not expected_data.exists()
        or not expected_data.is_dir()
        or _path_has_symlink(expected_data, profile_root)
        or not _audit_contains_receipt(
            receipt_audit,
            transaction_id=transaction_id,
            audit_record_sha256=audit_record_sha256,
        )
    ):
        raise CustomerAdminError("committed activation receipt is missing or stale")
    for key in ("created_at", "committed_at"):
        value = journal.get(key)
        if not isinstance(value, str):
            raise CustomerAdminError("committed activation receipt is missing or stale")
        try:
            datetime.fromisoformat(value)
        except ValueError as exc:
            raise CustomerAdminError(
                "committed activation receipt is missing or stale"
            ) from exc
    return transaction_id


def _validate_enabled_activation_receipts(
    profile_root: Path,
    registry_path: Path,
    document: RegistryDocument,
) -> None:
    for spec in document.customers:
        if spec.enabled:
            _require_committed_activation_receipt(
                profile_root, registry_path, document, spec
            )


def validate_committed_activation(
    profile_root: Path,
    registry_path: Path | None = None,
    customer_id: str | None = None,
) -> bool:
    """Fail closed unless every selected enabled customer has a committed receipt."""
    root = _resolve_profile_root(profile_root)
    resolved_registry = (
        _resolve_registry_path(root)
        if registry_path is None
        else Path(registry_path).resolve()
    )
    _recover_activation_journal(root, resolved_registry)
    document = _read_profile_registry(resolved_registry, root)
    if customer_id is not None:
        selected = next(
            (item for item in document.customers if item.customer_key == customer_id),
            None,
        )
        if selected is None:
            raise CustomerAdminError(f"unknown customer: {customer_id}")
        if selected.enabled:
            _require_committed_activation_receipt(
                root, resolved_registry, document, selected
            )
            journal = _read_activation_journal(_activation_journal_path(root))
            if journal is None:
                raise CustomerAdminError(
                    "committed activation receipt is missing or stale"
                )
            if journal.get("version") == 2:
                try:
                    require_nutrition_start_readiness(root, selected.customer_key)
                except NutritionReadinessError as exc:
                    raise CustomerAdminError(str(exc)) from exc
    else:
        _validate_enabled_activation_receipts(root, resolved_registry, document)
        enabled = tuple(selected for selected in document.customers if selected.enabled)
        if enabled:
            journal = _read_activation_journal(_activation_journal_path(root))
            if journal is None:
                raise CustomerAdminError(
                    "committed activation receipt is missing or stale"
                )
            if journal.get("version") == 2:
                for selected in enabled:
                    try:
                        require_nutrition_start_readiness(root, selected.customer_key)
                    except NutritionReadinessError as exc:
                        raise CustomerAdminError(str(exc)) from exc
    return True


def load_runtime_customer_registry(profile_root: Path) -> CustomerRegistry:
    """Load the registry only when enabled entries have committed activation receipts."""
    root = _resolve_profile_root(profile_root)
    registry_path = _resolve_registry_path(root)
    _recover_activation_journal(root, registry_path)
    document = _read_profile_registry(registry_path, root)
    _validate_enabled_activation_receipts(root, registry_path, document)
    try:
        return load_customer_registry(registry_path, root)
    except (CustomerRegistryError, OSError, ValueError) as exc:
        raise CustomerAdminError("runtime registry failed closed") from exc


_REGISTRATION_SCHEMA_VERSION = "1.0"
_REGISTRATION_REVISION_FILE = "adaptive-registration-inputs.jsonl"
_REGISTRATION_CONFIG_FILE = "adaptive-registration-config.jsonl"
_REGISTRATION_APPROVAL_FILE = "input-approvals.jsonl"
_REGISTRATION_LOCK_FILE = ".adaptive-registration-inputs.lock"
_REGISTRATION_ZERO_DIGEST = "0" * 64
_REGISTRATION_REVISION_FIELDS = frozenset(
    {
        "schema_version",
        "kind",
        "append_sequence",
        "intent_id",
        "state",
        "customer_key",
        "version",
        "revision_digest",
        "input_document",
        "supersedes_digest",
        "activation_receipt_id",
        "activation_receipt_digest",
        "registry_digest",
        "owner_digest",
        "authority_digest",
        "authority",
        "approved_by",
        "approved_at_kst",
        "prepared_digest",
        "row_digest",
    }
)
_REGISTRATION_CONFIG_FIELDS = frozenset(
    {
        "schema_version",
        "kind",
        "append_sequence",
        "intent_id",
        "state",
        "customer_key",
        "version",
        "revision_digest",
        "supersedes_digest",
        "activation_receipt_id",
        "activation_receipt_digest",
        "registry_digest",
        "owner_digest",
        "authority_digest",
        "approved_by",
        "approved_at_kst",
        "prepared_digest",
        "row_digest",
        "artifact_digests",
    }
)
_REGISTRATION_APPROVAL_FIELDS = frozenset(
    {
        "schema_version",
        "kind",
        "append_sequence",
        "intent_id",
        "state",
        "customer_key",
        "registration_digest",
        "artifact_kind",
        "artifact_version",
        "artifact_digest",
        "artifact_document",
        "supersedes_digest",
        "approved_by",
        "approved_at_kst",
        "registry_digest",
        "owner_digest",
        "activation_receipt_digest",
        "authority_digest",
        "prepared_digest",
        "row_digest",
    }
)


def _registration_private_root(data_root: Path) -> Path:
    root = Path(data_root)
    if (
        root.is_symlink()
        or not root.exists()
        or not root.is_dir()
        or root.stat().st_mode & 0o077
    ):
        raise CustomerAdminError("adaptive registration root is unavailable")
    if root.name != "nutrition-plans":
        root = root / "nutrition-plans"
    if (
        root.is_symlink()
        or not root.exists()
        or not root.is_dir()
        or root.stat().st_mode & 0o077
    ):
        raise CustomerAdminError("adaptive registration root is unavailable")
    return root


def _registration_paths(root: Path) -> tuple[Path, Path, Path, Path]:
    return (
        root / _REGISTRATION_REVISION_FILE,
        root / _REGISTRATION_CONFIG_FILE,
        root / _REGISTRATION_APPROVAL_FILE,
        root / _REGISTRATION_LOCK_FILE,
    )


def _registration_owner(value: object) -> dict[str, str]:
    if isinstance(value, TelegramAddress):
        return value.model_dump(mode="json")
    if not isinstance(value, Mapping):
        raise CustomerAdminError(
            "adaptive registration approver must be the full owner identity"
        )
    if set(value) != {"user_id", "chat_id", "topic_id"}:
        raise CustomerAdminError(
            "adaptive registration approver must be the full owner identity"
        )
    try:
        owner = TelegramAddress.model_validate(dict(value))
    except (TypeError, ValueError) as exc:
        raise CustomerAdminError("adaptive registration approver is invalid") from exc
    return owner.model_dump(mode="json")


def _registration_kst_timestamp(value: str | None) -> str:
    timestamp = datetime.now(_KST) if value is None else value
    if isinstance(timestamp, str):
        try:
            parsed = datetime.fromisoformat(timestamp)
        except ValueError as exc:
            raise CustomerAdminError(
                "adaptive registration approval time is invalid"
            ) from exc
    elif isinstance(timestamp, datetime):
        parsed = timestamp
    else:
        raise CustomerAdminError("adaptive registration approval time is invalid")
    if parsed.tzinfo is None or parsed.utcoffset() != timedelta(hours=9):
        raise CustomerAdminError("adaptive registration approval time must be KST")
    return parsed.isoformat()


def _registration_activation_authority(
    profile_root: Path,
    registry_path: Path,
    document: RegistryDocument,
    spec: CustomerSpec,
) -> tuple[dict[str, object], str]:
    if not spec.enabled:
        raise CustomerAdminError(
            "customer must be enabled for adaptive registration inputs"
        )
    transaction_id = _require_committed_activation_receipt(
        profile_root,
        registry_path,
        document,
        spec,
    )
    journal = _read_activation_journal(_activation_journal_path(profile_root))
    if journal is None:
        raise CustomerAdminError("activation receipt is missing or stale")
    for key in ("registry_sha256", "audit_record_sha256"):
        value = journal.get(key)
        if not isinstance(value, str) or not value:
            raise CustomerAdminError("activation receipt is missing or stale")
    owner = document.owner.model_dump(mode="json")
    owner_digest = adaptive_digest(owner)
    registry_digest = _document_fingerprint(document)
    activation_receipt_digest = adaptive_digest(
        {
            "schema_version": _REGISTRATION_SCHEMA_VERSION,
            "customer_key": spec.customer_key,
            "activation_receipt_id": transaction_id,
            "registry_sha256": journal["registry_sha256"],
            "audit_record_sha256": journal["audit_record_sha256"],
        }
    )
    authority = {
        "schema_version": _REGISTRATION_SCHEMA_VERSION,
        "customer_key": spec.customer_key,
        "owner": owner,
        "owner_digest": owner_digest,
        "registry_digest": registry_digest,
        "activation_receipt_id": transaction_id,
        "activation_receipt_digest": activation_receipt_digest,
    }
    return authority, adaptive_digest(authority)


def _registration_jsonl(
    path: Path, *, recover: bool = False
) -> list[dict[str, object]]:
    if path.is_symlink():
        raise CustomerAdminError("adaptive registration journal symlink is not allowed")
    if not path.exists():
        return []
    if not path.is_file():
        raise CustomerAdminError("adaptive registration journal must be a regular file")
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise CustomerAdminError(
            "adaptive registration journal is unavailable"
        ) from exc
    rows: list[dict[str, object]] = []
    complete_end = 0
    for index, line in enumerate(raw.splitlines(keepends=True)):
        final = complete_end + len(line) == len(raw)
        if not line.endswith(b"\n"):
            if recover and final:
                break
            raise CustomerAdminError(
                f"adaptive registration journal has a torn tail at row {index}"
            )
        content = line[:-1]
        if content.endswith(b"\r"):
            content = content[:-1]
        if not content:
            raise CustomerAdminError(
                "adaptive registration journal contains a blank row"
            )
        try:
            parsed = json.loads(content.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CustomerAdminError(
                "adaptive registration journal is invalid"
            ) from exc
        if not isinstance(parsed, dict):
            raise CustomerAdminError("adaptive registration journal row is invalid")
        rows.append(parsed)
        complete_end += len(line)
    if recover and complete_end != len(raw):
        try:
            with path.open("r+b") as handle:
                handle.truncate(complete_end)
            path.chmod(0o600)
        except OSError as exc:
            raise CustomerAdminError(
                "adaptive registration journal recovery failed"
            ) from exc
    elif complete_end != len(raw):
        raise CustomerAdminError("adaptive registration journal has a torn tail")
    return rows


def _registration_row_digest(row: Mapping[str, object]) -> str:
    body = {key: value for key, value in row.items() if key != "row_digest"}
    return adaptive_digest(body)


def _registration_digest(value: object, label: str) -> str:
    if (
        not isinstance(value, str)
        or len(value) != 64
        or any(character not in "0123456789abcdef" for character in value)
    ):
        raise CustomerAdminError(f"adaptive registration {label} digest is invalid")
    return value


def _validate_registration_revision_rows(
    rows: list[dict[str, object]],
) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    states: dict[str, dict[str, object]] = {}
    committed: list[dict[str, object]] = []
    for expected_sequence, row in enumerate(rows, start=1):
        if set(row) != _REGISTRATION_REVISION_FIELDS:
            raise CustomerAdminError(
                "adaptive registration revision journal schema mismatch"
            )
        if row.get("schema_version") != _REGISTRATION_SCHEMA_VERSION:
            raise CustomerAdminError("adaptive registration revision schema mismatch")
        if row.get("kind") != "customer_adaptive_registration":
            raise CustomerAdminError("adaptive registration revision kind is invalid")
        if row.get("append_sequence") != expected_sequence:
            raise CustomerAdminError(
                "adaptive registration revision sequence is not contiguous"
            )
        if row.get("row_digest") != _registration_row_digest(row):
            raise CustomerAdminError(
                "adaptive registration revision row digest mismatch"
            )
        intent_id = row.get("intent_id")
        customer_key = row.get("customer_key")
        state = row.get("state")
        if not isinstance(intent_id, str) or not intent_id:
            raise CustomerAdminError("adaptive registration revision intent is invalid")
        if not isinstance(customer_key, str) or not customer_key:
            raise CustomerAdminError(
                "adaptive registration revision customer is invalid"
            )
        if state not in {"prepared", "committed"}:
            raise CustomerAdminError("adaptive registration revision state is invalid")
        if state == "prepared" and row.get("prepared_digest") is not None:
            raise CustomerAdminError(
                "prepared adaptive registration revision has a terminal digest"
            )
        if state == "committed" and not isinstance(row.get("prepared_digest"), str):
            raise CustomerAdminError(
                "committed adaptive registration revision has no prepared digest"
            )
        revision_digest = _registration_digest(row.get("revision_digest"), "revision")
        supersedes = _registration_digest(row.get("supersedes_digest"), "supersession")
        authority_digest = _registration_digest(
            row.get("authority_digest"), "authority"
        )
        _registration_digest(row.get("registry_digest"), "registry")
        _registration_digest(row.get("owner_digest"), "owner")
        _registration_digest(row.get("activation_receipt_digest"), "activation")
        if not isinstance(row.get("input_document"), Mapping):
            raise CustomerAdminError("adaptive registration input document is invalid")
        if not isinstance(row.get("authority"), Mapping):
            raise CustomerAdminError("adaptive registration authority is invalid")
        if not isinstance(row.get("approved_by"), Mapping):
            raise CustomerAdminError("adaptive registration approver is invalid")
        if not isinstance(row.get("approved_at_kst"), str):
            raise CustomerAdminError("adaptive registration approval time is invalid")
        try:
            parsed_at = datetime.fromisoformat(str(row["approved_at_kst"]))
        except ValueError as exc:
            raise CustomerAdminError(
                "adaptive registration approval time is invalid"
            ) from exc
        if parsed_at.tzinfo is None or parsed_at.utcoffset() != timedelta(hours=9):
            raise CustomerAdminError("adaptive registration approval time must be KST")
        document = row["input_document"]
        if (
            document.get("customer_key") != customer_key
            or document.get("digest") != revision_digest
        ):
            raise CustomerAdminError("adaptive registration revision document mismatch")
        if document.get("supersedes_digest") != supersedes:
            raise CustomerAdminError(
                "adaptive registration revision supersession mismatch"
            )
        if document.get("authority_digest") != authority_digest:
            raise CustomerAdminError(
                "adaptive registration revision authority mismatch"
            )
        if adaptive_digest(dict(row["authority"])) != authority_digest:
            raise CustomerAdminError("adaptive registration authority digest mismatch")
        if state == "committed":
            _registration_digest(row.get("prepared_digest"), "prepared row")
        previous = states.get(intent_id)
        if previous is None:
            if state != "prepared":
                raise CustomerAdminError(
                    "adaptive registration terminal row has no prepared row"
                )
            states[intent_id] = dict(row)
            continue
        if previous.get("state") != "prepared" or state != "committed":
            raise CustomerAdminError(
                "adaptive registration revision transition is invalid"
            )
        if row.get("prepared_digest") != previous.get("row_digest"):
            raise CustomerAdminError("adaptive registration prepared digest mismatch")
        for key in _REGISTRATION_REVISION_FIELDS - {
            "state",
            "append_sequence",
            "prepared_digest",
            "row_digest",
        }:
            if row.get(key) != previous.get(key):
                raise CustomerAdminError(
                    "adaptive registration prepared payload mismatch"
                )
        states[intent_id] = dict(row)
        committed.append(dict(row))
    pending = [row for row in states.values() if row.get("state") == "prepared"]
    by_customer: dict[str, list[dict[str, object]]] = {}
    for row in committed:
        by_customer.setdefault(str(row["customer_key"]), []).append(row)
    for customer_key, customer_rows in by_customer.items():
        expected = _REGISTRATION_ZERO_DIGEST
        for row in customer_rows:
            if row.get("supersedes_digest") != expected:
                raise CustomerAdminError(
                    f"adaptive registration supersession chain is invalid for {customer_key}"
                )
            expected = str(row["revision_digest"])
    return committed, pending


def _validate_registration_config_rows(
    rows: list[dict[str, object]],
) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    states: dict[str, dict[str, object]] = {}
    committed: list[dict[str, object]] = []
    for expected_sequence, row in enumerate(rows, start=1):
        if set(row) != _REGISTRATION_CONFIG_FIELDS:
            raise CustomerAdminError(
                "adaptive registration config journal schema mismatch"
            )
        if row.get("schema_version") != _REGISTRATION_SCHEMA_VERSION:
            raise CustomerAdminError("adaptive registration config schema mismatch")
        if row.get("kind") != "customer_adaptive_registration_config":
            raise CustomerAdminError("adaptive registration config kind is invalid")
        if row.get("append_sequence") != expected_sequence:
            raise CustomerAdminError(
                "adaptive registration config sequence is not contiguous"
            )
        if row.get("row_digest") != _registration_row_digest(row):
            raise CustomerAdminError("adaptive registration config row digest mismatch")
        if row.get("state") not in {"prepared", "committed"}:
            raise CustomerAdminError("adaptive registration config state is invalid")
        intent_id = row.get("intent_id")
        if not isinstance(intent_id, str) or not intent_id:
            raise CustomerAdminError("adaptive registration config intent is invalid")
        for field, label in (
            ("revision_digest", "revision"),
            ("supersedes_digest", "supersession"),
            ("authority_digest", "authority"),
            ("registry_digest", "registry"),
            ("owner_digest", "owner"),
            ("activation_receipt_digest", "activation"),
        ):
            _registration_digest(row.get(field), label)
        artifact_digests = row.get("artifact_digests")
        if not isinstance(artifact_digests, Mapping) or set(artifact_digests) != {
            "base_policy",
            "meal_constraints",
            "catalog",
        }:
            raise CustomerAdminError(
                "adaptive registration config artifacts are invalid"
            )
        for artifact_digest in artifact_digests.values():
            _registration_digest(artifact_digest, "artifact")
        if not isinstance(row.get("approved_by"), Mapping):
            raise CustomerAdminError("adaptive registration config approver is invalid")
        if not isinstance(row.get("approved_at_kst"), str):
            raise CustomerAdminError(
                "adaptive registration config approval time is invalid"
            )
        if row.get("state") == "prepared" and row.get("prepared_digest") is not None:
            raise CustomerAdminError(
                "prepared adaptive registration config has a terminal digest"
            )
        if row.get("state") == "committed" and not isinstance(
            row.get("prepared_digest"), str
        ):
            raise CustomerAdminError(
                "committed adaptive registration config has no prepared digest"
            )
        previous = states.get(intent_id)
        if previous is None:
            if row.get("state") != "prepared":
                raise CustomerAdminError(
                    "adaptive registration config terminal row has no prepared row"
                )
            states[intent_id] = dict(row)
            continue
        if previous.get("state") != "prepared" or row.get("state") != "committed":
            raise CustomerAdminError(
                "adaptive registration config transition is invalid"
            )
        if row.get("prepared_digest") != previous.get("row_digest"):
            raise CustomerAdminError(
                "adaptive registration config prepared digest mismatch"
            )
        for key in _REGISTRATION_CONFIG_FIELDS - {
            "state",
            "append_sequence",
            "prepared_digest",
            "row_digest",
        }:
            if row.get(key) != previous.get(key):
                raise CustomerAdminError(
                    "adaptive registration config prepared payload mismatch"
                )
        states[intent_id] = dict(row)
        committed.append(dict(row))
    pending = [row for row in states.values() if row.get("state") == "prepared"]
    return committed, pending


def _registration_read_artifact_source(
    root: Path,
    artifact_kind: str,
    supplied: object,
) -> Mapping[str, object]:
    """Read approved artifacts; only a missing canonical catalog may use legacy catalog.json."""
    value = supplied
    if value is None:
        # food-catalog.json is canonical. catalog.json is a read-only migration
        # fallback selected only when the canonical path is absent.
        names = {
            "base_policy": ("policy.json",),
            "meal_constraints": ("meal-constraints.json",),
            "catalog": ("food-catalog.json", "catalog.json"),
        }[artifact_kind]
        path = next(
            (
                root / name
                for name in names
                if (root / name).exists() or (root / name).is_symlink()
            ),
            None,
        )
        if path is None:
            raise CustomerAdminError(f"adaptive {artifact_kind} artifact is missing")
        if path.is_symlink() or not path.is_file():
            raise CustomerAdminError(f"adaptive {artifact_kind} artifact is invalid")
        try:
            if path.stat().st_mode & 0o077:
                raise CustomerAdminError(
                    f"adaptive {artifact_kind} artifact permissions are too broad"
                )
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CustomerAdminError(
                f"adaptive {artifact_kind} artifact is invalid"
            ) from exc
    elif isinstance(value, (str, Path)):
        path = Path(value)
        if path.is_symlink() or not path.is_file():
            raise CustomerAdminError(f"adaptive {artifact_kind} artifact is invalid")
        try:
            if path.stat().st_mode & 0o077:
                raise CustomerAdminError(
                    f"adaptive {artifact_kind} artifact permissions are too broad"
                )
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CustomerAdminError(
                f"adaptive {artifact_kind} artifact is invalid"
            ) from exc
    if not isinstance(value, Mapping):
        raise CustomerAdminError(f"adaptive {artifact_kind} artifact must be an object")
    return dict(value)


def _registration_artifact_value(
    document: Mapping[str, object],
    artifact_kind: str,
) -> object:
    aliases = {
        "base_policy": ("policy", "base_policy", "value", "data"),
        "meal_constraints": ("meal_constraints", "constraints", "value", "data"),
        "catalog": ("catalog", "food_catalog", "foods", "value", "data"),
    }[artifact_kind]
    for key in aliases:
        if key in document:
            return document[key]
    metadata = {
        "schema_version",
        "version",
        "digest",
        "approved",
        "approved_by",
        "approved_at_kst",
        "enabled",
        "customer_key",
        "activation_receipt_id",
        "authority_digest",
        "registry_digest",
        "owner_digest",
    }
    value = {
        key: candidate for key, candidate in document.items() if key not in metadata
    }
    if value:
        return value
    raise CustomerAdminError(f"adaptive {artifact_kind} artifact value is missing")


def _registration_validate_external_artifact(
    document: Mapping[str, object],
    artifact_kind: str,
    owner: Mapping[str, str],
) -> tuple[dict[str, object], str, str]:
    if document.get("approved") is not True:
        raise CustomerAdminError(f"adaptive {artifact_kind} artifact is not approved")
    if document.get("approved_by") != dict(owner):
        raise CustomerAdminError(
            f"adaptive {artifact_kind} artifact approver does not match owner"
        )
    version = document.get("version")
    if not isinstance(version, str) or not version.strip():
        raise CustomerAdminError(
            f"adaptive {artifact_kind} artifact version is required"
        )
    approved_at = document.get("approved_at_kst")
    if not isinstance(approved_at, str) or not approved_at.strip():
        raise CustomerAdminError(
            f"adaptive {artifact_kind} artifact approval time is required"
        )
    _registration_kst_timestamp(approved_at)
    declared = document.get("digest")
    if (
        not isinstance(declared, str)
        or len(declared) != 64
        or any(character not in "0123456789abcdef" for character in declared)
    ):
        raise CustomerAdminError(f"adaptive {artifact_kind} artifact digest is invalid")
    value = _registration_artifact_value(document, artifact_kind)
    if adaptive_digest(value) != declared:
        raise CustomerAdminError(
            f"adaptive {artifact_kind} artifact digest does not match content"
        )
    if artifact_kind == "base_policy" and not isinstance(value, Mapping):
        raise CustomerAdminError("adaptive base_policy artifact must contain an object")
    if artifact_kind == "meal_constraints":
        if not isinstance(value, Mapping) or not all(
            key in value for key in ("meal_count", "budget_tier", "cooking_access")
        ):
            raise CustomerAdminError("adaptive meal constraints artifact is incomplete")
    if artifact_kind == "catalog":
        rows = (
            value.get("foods")
            if isinstance(value, Mapping) and "foods" in value
            else value
        )
        if not isinstance(rows, (list, tuple)) or not rows:
            raise CustomerAdminError("adaptive catalog artifact is empty")
    return dict(document), version.strip(), declared


def _registration_derived_meal_constraints(
    payload: Mapping[str, object],
    *,
    owner: Mapping[str, str],
    approved_at_kst: str,
) -> tuple[dict[str, object], str, str]:
    value = {
        "meal_count": payload["meal_count"],
        "budget_tier": payload["budget_band"],
        "cooking_access": payload["cooking_access"],
        "preferences": list(payload["preferences"]),
        "excluded_food_ids": list(payload["exclusions"]),
        "allergies": list(payload["allergies"]),
        "training_time_by_day": [
            {
                "kst_day": entry["date"],
                "training_time": entry["time"],
                "weekday": entry["weekday"],
                "load_category": entry["load_category"],
            }
            for entry in payload["training_schedule"]
        ],
        "strict_inputs": True,
    }
    artifact_digest = adaptive_digest(value)
    document = {
        "schema_version": _REGISTRATION_SCHEMA_VERSION,
        "version": f"{payload['version']}.meal-constraints",
        "digest": artifact_digest,
        "approved": True,
        "approved_by": dict(owner),
        "approved_at_kst": approved_at_kst,
        "meal_constraints": value,
    }
    return document, str(document["version"]), artifact_digest


def _registration_artifact_bundle(
    root: Path,
    payload: Mapping[str, object],
    inputs: object,
    *,
    owner: Mapping[str, str],
    approved_at_kst: str,
) -> tuple[tuple[str, dict[str, object], str, str], ...]:
    if isinstance(inputs, AdaptiveRegistrationInputs):
        raw = inputs.model_dump(mode="json", exclude_none=True)
    elif isinstance(inputs, Mapping):
        raw = dict(inputs)
    else:
        model_dump = getattr(inputs, "model_dump", None)
        dumped = (
            model_dump(mode="json", exclude_none=True) if callable(model_dump) else {}
        )
        raw = dict(dumped) if isinstance(dumped, Mapping) else {}
    policy_document = _registration_read_artifact_source(
        root,
        "base_policy",
        raw.get("base_policy", raw.get("policy")),
    )
    catalog_document = _registration_read_artifact_source(
        root,
        "catalog",
        raw.get("food_catalog", raw.get("catalog")),
    )
    _, policy_version, policy_digest = _registration_validate_external_artifact(
        policy_document,
        "base_policy",
        owner,
    )
    _, catalog_version, catalog_digest = _registration_validate_external_artifact(
        catalog_document,
        "catalog",
        owner,
    )
    constraints_document, constraints_version, constraints_digest = (
        _registration_derived_meal_constraints(
            payload,
            owner=owner,
            approved_at_kst=approved_at_kst,
        )
    )
    constraints_source = raw.get("meal_constraints")
    if constraints_source is not None:
        supplied_constraints = _registration_read_artifact_source(
            root,
            "meal_constraints",
            constraints_source,
        )
        _, supplied_version, supplied_digest = _registration_validate_external_artifact(
            supplied_constraints,
            "meal_constraints",
            owner,
        )
        supplied_value = _registration_artifact_value(
            supplied_constraints,
            "meal_constraints",
        )
        derived_value = _registration_artifact_value(
            constraints_document,
            "meal_constraints",
        )
        if (
            supplied_version != constraints_version
            or supplied_digest != constraints_digest
            or adaptive_digest(supplied_value) != adaptive_digest(derived_value)
        ):
            raise CustomerAdminError(
                "adaptive meal constraints must be derived from the registration inputs"
            )
    return (
        ("base_policy", policy_document, policy_version, policy_digest),
        (
            "meal_constraints",
            constraints_document,
            constraints_version,
            constraints_digest,
        ),
        ("catalog", catalog_document, catalog_version, catalog_digest),
    )


def _registration_approval_row(
    *,
    sequence: int,
    intent_id: str,
    state: str,
    customer_key: str,
    registration_digest: str,
    artifact_kind: str,
    artifact_document: Mapping[str, object],
    artifact_version: str,
    artifact_digest: str,
    supersedes_digest: str,
    authority: Mapping[str, object],
    authority_digest: str,
    owner: Mapping[str, str],
    approved_at_kst: str,
    prepared_digest: str | None,
) -> dict[str, object]:
    body = {
        "schema_version": _REGISTRATION_SCHEMA_VERSION,
        "kind": "adaptive_input_approval",
        "append_sequence": sequence,
        "intent_id": intent_id,
        "state": state,
        "customer_key": customer_key,
        "registration_digest": registration_digest,
        "artifact_kind": artifact_kind,
        "artifact_version": artifact_version,
        "artifact_digest": artifact_digest,
        "artifact_document": dict(artifact_document),
        "supersedes_digest": supersedes_digest,
        "approved_by": dict(owner),
        "approved_at_kst": approved_at_kst,
        "registry_digest": authority["registry_digest"],
        "owner_digest": authority["owner_digest"],
        "activation_receipt_digest": authority["activation_receipt_digest"],
        "authority_digest": authority_digest,
        "prepared_digest": prepared_digest,
    }
    return {**body, "row_digest": _registration_row_digest(body)}


def _validate_registration_approval_rows(
    rows: list[dict[str, object]],
) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    states: dict[str, dict[str, object]] = {}
    committed: list[dict[str, object]] = []
    for expected_sequence, row in enumerate(rows, start=1):
        if set(row) != _REGISTRATION_APPROVAL_FIELDS:
            raise CustomerAdminError("adaptive input approval journal schema mismatch")
        if row.get("schema_version") != _REGISTRATION_SCHEMA_VERSION:
            raise CustomerAdminError("adaptive input approval schema mismatch")
        if row.get("kind") != "adaptive_input_approval":
            raise CustomerAdminError("adaptive input approval kind is invalid")
        if row.get("append_sequence") != expected_sequence:
            raise CustomerAdminError(
                "adaptive input approval sequence is not contiguous"
            )
        if row.get("row_digest") != _registration_row_digest(row):
            raise CustomerAdminError("adaptive input approval row digest mismatch")
        if row.get("state") not in {"prepared", "committed", "abandoned"}:
            raise CustomerAdminError("adaptive input approval state is invalid")
        intent_id = row.get("intent_id")
        if not isinstance(intent_id, str) or not intent_id:
            raise CustomerAdminError("adaptive input approval intent is invalid")
        if row.get("artifact_kind") not in {
            "base_policy",
            "meal_constraints",
            "catalog",
        }:
            raise CustomerAdminError("adaptive input approval artifact kind is invalid")
        for field, label in (
            ("registration_digest", "registration"),
            ("artifact_digest", "artifact"),
            ("supersedes_digest", "supersession"),
            ("registry_digest", "registry"),
            ("owner_digest", "owner"),
            ("activation_receipt_digest", "activation"),
            ("authority_digest", "authority"),
        ):
            _registration_digest(row.get(field), label)
        if not isinstance(row.get("artifact_document"), Mapping):
            raise CustomerAdminError("adaptive input approval artifact is invalid")
        artifact_value = _registration_artifact_value(
            row["artifact_document"],
            str(row["artifact_kind"]),
        )
        if adaptive_digest(artifact_value) != row.get("artifact_digest"):
            raise CustomerAdminError("adaptive input approval artifact digest mismatch")
        if not isinstance(row.get("approved_by"), Mapping):
            raise CustomerAdminError("adaptive input approval owner is invalid")
        _registration_owner(row.get("approved_by"))
        if not isinstance(row.get("approved_at_kst"), str):
            raise CustomerAdminError("adaptive input approval time is invalid")
        _registration_kst_timestamp(row["approved_at_kst"])
        if row.get("state") == "prepared" and row.get("prepared_digest") is not None:
            raise CustomerAdminError(
                "prepared adaptive input approval has a terminal digest"
            )
        if row.get("state") in {"committed", "abandoned"} and not isinstance(
            row.get("prepared_digest"), str
        ):
            raise CustomerAdminError(
                "terminal adaptive input approval has no prepared digest"
            )
        previous = states.get(intent_id)
        if previous is None:
            if row.get("state") != "prepared":
                raise CustomerAdminError(
                    "adaptive input approval terminal row has no prepared row"
                )
            states[intent_id] = dict(row)
            continue
        if previous.get("state") != "prepared" or row.get("state") not in {
            "committed",
            "abandoned",
        }:
            raise CustomerAdminError("adaptive input approval transition is invalid")
        if row.get("prepared_digest") != previous.get("row_digest"):
            raise CustomerAdminError("adaptive input approval prepared digest mismatch")
        for key in _REGISTRATION_APPROVAL_FIELDS - {
            "state",
            "append_sequence",
            "prepared_digest",
            "row_digest",
        }:
            if row.get(key) != previous.get(key):
                raise CustomerAdminError(
                    "adaptive input approval prepared payload mismatch"
                )
        states[intent_id] = dict(row)
        if row.get("state") == "committed":
            committed.append(dict(row))
    pending = [row for row in states.values() if row.get("state") == "prepared"]
    return committed, pending


@contextmanager
def _registration_lock(root: Path) -> Iterator[None]:
    _, _, _, lock_path = _registration_paths(root)
    descriptor: int | None = None
    locked = False
    try:
        try:
            if lock_path.is_symlink():
                raise CustomerAdminError(
                    "adaptive registration lock symlinks are not allowed"
                )
            descriptor = os.open(
                lock_path,
                os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC,
                0o600,
            )
            opened = os.fstat(descriptor)
            named = lock_path.lstat()
            identity = (opened.st_dev, opened.st_ino)
            if (
                not stat.S_ISREG(opened.st_mode)
                or opened.st_uid != os.geteuid()
                or opened.st_nlink != 1
                or stat.S_IMODE(opened.st_mode) != 0o600
                or identity != (named.st_dev, named.st_ino)
            ):
                raise CustomerAdminError("adaptive registration lock is unsafe")
            fcntl.flock(descriptor, fcntl.LOCK_EX)
            locked = True
            if identity != (lock_path.lstat().st_dev, lock_path.lstat().st_ino):
                raise CustomerAdminError("adaptive registration lock was replaced")
        except CustomerAdminError:
            raise
        except OSError as exc:
            raise CustomerAdminError(
                "adaptive registration lock is unavailable"
            ) from exc
        yield
    finally:
        if descriptor is not None:
            try:
                if locked:
                    fcntl.flock(descriptor, fcntl.LOCK_UN)
            finally:
                os.close(descriptor)


def _append_registration_approvals(
    root: Path,
    document: Mapping[str, object],
    bundle: tuple[tuple[str, dict[str, object], str, str], ...],
    authority: Mapping[str, object],
    authority_digest: str,
) -> None:
    _, _, approval_path, _ = _registration_paths(root)
    owner = document.get("approved_by")
    if not isinstance(owner, Mapping):
        raise CustomerAdminError("adaptive input approval owner is invalid")
    try:
        with _registration_lock(root):
            rows = _registration_jsonl(approval_path, recover=True)
            committed, pending = _validate_registration_approval_rows(rows)
            key = str(document["customer_key"])
            registration_digest = str(document["digest"])
            current_by_kind: dict[str, str] = {}
            for row in reversed(committed):
                if row.get("customer_key") != key:
                    continue
                current_by_kind.setdefault(
                    str(row["artifact_kind"]), str(row["artifact_digest"])
                )
            for (
                artifact_kind,
                artifact_document,
                artifact_version,
                artifact_digest,
            ) in bundle:
                matching = next(
                    (
                        row
                        for row in committed
                        if row.get("customer_key") == key
                        and row.get("registration_digest") == registration_digest
                        and row.get("artifact_kind") == artifact_kind
                    ),
                    None,
                )
                if matching is not None:
                    if (
                        matching.get("artifact_digest") != artifact_digest
                        or matching.get("artifact_document") != artifact_document
                        or matching.get("authority_digest") != authority_digest
                    ):
                        raise CustomerAdminError(
                            "conflicting adaptive input approval replay"
                        )
                    continue
                pending_match = next(
                    (
                        row
                        for row in pending
                        if row.get("customer_key") == key
                        and row.get("registration_digest") == registration_digest
                        and row.get("artifact_kind") == artifact_kind
                    ),
                    None,
                )
                supersedes = current_by_kind.get(
                    artifact_kind, _REGISTRATION_ZERO_DIGEST
                )
                prior_kind = next(
                    (
                        row
                        for row in committed
                        if row.get("customer_key") == key
                        and row.get("artifact_kind") == artifact_kind
                        and row.get("artifact_digest") == supersedes
                    ),
                    None,
                )
                if (
                    prior_kind is not None
                    and artifact_digest != supersedes
                    and prior_kind.get("artifact_version") == artifact_version
                ):
                    raise CustomerAdminError(
                        "adaptive input artifact version must change on update"
                    )
                if pending_match is not None:
                    if (
                        pending_match.get("artifact_digest") != artifact_digest
                        or pending_match.get("supersedes_digest") != supersedes
                    ):
                        raise CustomerAdminError(
                            "conflicting incomplete adaptive input approval"
                        )
                    prepared = pending_match
                    commit = _registration_approval_row(
                        sequence=len(rows) + 1,
                        intent_id=str(prepared["intent_id"]),
                        state="committed",
                        customer_key=key,
                        registration_digest=registration_digest,
                        artifact_kind=artifact_kind,
                        artifact_document=artifact_document,
                        artifact_version=artifact_version,
                        artifact_digest=artifact_digest,
                        supersedes_digest=supersedes,
                        authority=authority,
                        authority_digest=authority_digest,
                        owner=owner,
                        approved_at_kst=str(document["approved_at_kst"]),
                        prepared_digest=str(prepared["row_digest"]),
                    )
                    _append_registration_row(approval_path, commit)
                    rows.append(commit)
                    committed, pending = _validate_registration_approval_rows(rows)
                    continue
                intent_id = (
                    f"adaptive-input:{key}:{registration_digest}:{artifact_kind}"
                )
                prepared = _registration_approval_row(
                    sequence=len(rows) + 1,
                    intent_id=intent_id,
                    state="prepared",
                    customer_key=key,
                    registration_digest=registration_digest,
                    artifact_kind=artifact_kind,
                    artifact_document=artifact_document,
                    artifact_version=artifact_version,
                    artifact_digest=artifact_digest,
                    supersedes_digest=supersedes,
                    authority=authority,
                    authority_digest=authority_digest,
                    owner=owner,
                    approved_at_kst=str(document["approved_at_kst"]),
                    prepared_digest=None,
                )
                _append_registration_row(approval_path, prepared)
                committed_row = _registration_approval_row(
                    sequence=len(rows) + 2,
                    intent_id=intent_id,
                    state="committed",
                    customer_key=key,
                    registration_digest=registration_digest,
                    artifact_kind=artifact_kind,
                    artifact_document=artifact_document,
                    artifact_version=artifact_version,
                    artifact_digest=artifact_digest,
                    supersedes_digest=supersedes,
                    authority=authority,
                    authority_digest=authority_digest,
                    owner=owner,
                    approved_at_kst=str(document["approved_at_kst"]),
                    prepared_digest=prepared["row_digest"],
                )
                _append_registration_row(approval_path, committed_row)
                rows.extend((prepared, committed_row))
                committed, pending = _validate_registration_approval_rows(rows)
    except CustomerAdminError:
        raise
    except OSError as exc:
        raise CustomerAdminError("adaptive input approval transition failed") from exc


def _registration_latest_approvals(
    rows: list[dict[str, object]],
    customer_key: str,
    registration_digest: str,
) -> dict[str, dict[str, object]]:
    committed, pending = _validate_registration_approval_rows(rows)
    if pending:
        raise CustomerAdminError("adaptive input approval transition is incomplete")
    selected_rows = [
        row
        for row in committed
        if row.get("customer_key") == customer_key
        and row.get("registration_digest") == registration_digest
    ]
    if (
        len(selected_rows) != 3
        or len({str(row.get("artifact_kind")) for row in selected_rows}) != 3
    ):
        raise CustomerAdminError("adaptive input approval artifacts are incomplete")
    selected = {str(row["artifact_kind"]): row for row in selected_rows}
    for artifact_kind in ("base_policy", "meal_constraints", "catalog"):
        expected = _REGISTRATION_ZERO_DIGEST
        for row in (
            candidate
            for candidate in committed
            if candidate.get("customer_key") == customer_key
            and candidate.get("artifact_kind") == artifact_kind
        ):
            if row.get("supersedes_digest") != expected:
                raise CustomerAdminError(
                    "adaptive input approval supersession chain is invalid"
                )
            expected = str(row["artifact_digest"])
    return selected


def _append_registration_row(path: Path, row: Mapping[str, object]) -> None:
    if path.is_symlink():
        raise CustomerAdminError("adaptive registration journal symlink is not allowed")
    try:
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        path.parent.chmod(0o700)
        path.touch(mode=0o600, exist_ok=True)
        path.chmod(0o600)
        with path.open("a", encoding="utf-8") as handle:
            handle.write(
                json.dumps(
                    dict(row), ensure_ascii=False, sort_keys=True, separators=(",", ":")
                )
                + "\n"
            )
            handle.flush()
            os.fsync(handle.fileno())
    except (OSError, TypeError, ValueError) as exc:
        if isinstance(exc, CustomerAdminError):
            raise
        raise CustomerAdminError(
            "adaptive registration journal could not be appended"
        ) from exc


def _registration_authority_matches(
    document: Mapping[str, object],
    authority: Mapping[str, object],
    authority_digest: str,
    owner: Mapping[str, str],
) -> None:
    if document.get("approved") is not True:
        raise CustomerAdminError("adaptive registration input is not approved")
    if document.get("approved_by") != dict(owner):
        raise CustomerAdminError(
            "adaptive registration input approver does not match owner"
        )
    if document.get("authority_digest") != authority_digest:
        raise CustomerAdminError("adaptive registration authority is stale")
    if document.get("authority") != dict(authority):
        raise CustomerAdminError("adaptive registration authority is stale")
    if document.get("registry_digest") != authority.get("registry_digest"):
        raise CustomerAdminError("adaptive registration registry evidence is stale")
    if document.get("owner_digest") != authority.get("owner_digest"):
        raise CustomerAdminError("adaptive registration owner evidence is stale")
    if document.get("activation_receipt_digest") != authority.get(
        "activation_receipt_digest"
    ):
        raise CustomerAdminError("adaptive registration activation evidence is stale")
    if document.get("activation_receipt_id") != authority.get("activation_receipt_id"):
        raise CustomerAdminError("adaptive registration activation receipt is stale")


def _registration_input_payload(
    inputs: object,
    customer_key: str,
    *,
    default_version: str,
) -> dict[str, object]:
    if type(inputs) is AdaptiveRegistrationInputs:
        raw: dict[str, object] = inputs.model_dump(mode="json", exclude_none=True)
    elif isinstance(inputs, Mapping):
        raw = dict(inputs)
    else:
        raise CustomerAdminError("adaptive registration inputs are invalid")
    supplied_key = raw.get("customer_key")
    if supplied_key is not None and supplied_key != customer_key:
        raise CustomerAdminError("adaptive registration customer key does not match")
    if (
        raw.get("schema_version", _REGISTRATION_SCHEMA_VERSION)
        != _REGISTRATION_SCHEMA_VERSION
    ):
        raise CustomerAdminError("adaptive registration schema version is invalid")
    version = raw.get("version", default_version)
    if not isinstance(version, str) or not version.strip():
        raise CustomerAdminError("adaptive registration version is required")
    version = version.strip()

    def text_values(name: str) -> list[str]:
        if name not in raw:
            raise CustomerAdminError(f"adaptive registration {name} is required")
        value = raw[name]
        if isinstance(value, str):
            value = [value]
        if not isinstance(value, (list, tuple, set, frozenset)):
            raise CustomerAdminError(f"adaptive registration {name} is invalid")
        result: list[str] = []
        for item in value:
            if not isinstance(item, str) or not item.strip() or len(item.strip()) > 200:
                raise CustomerAdminError(f"adaptive registration {name} is invalid")
            normalized = item.strip()
            if normalized not in result:
                result.append(normalized)
        return result

    meal_count = raw.get("meal_count")
    if type(meal_count) is not int or not 1 <= meal_count <= 8:
        raise CustomerAdminError(
            "adaptive registration meal_count is required and invalid"
        )
    budget_band = raw.get("budget_band")
    cooking_access = raw.get("cooking_access")
    if not isinstance(budget_band, str) or not budget_band.strip():
        raise CustomerAdminError("adaptive registration budget_band is required")
    if not isinstance(cooking_access, str) or not cooking_access.strip():
        raise CustomerAdminError("adaptive registration cooking_access is required")

    schedule = raw.get("training_schedule")
    if not isinstance(schedule, (list, tuple)) or not schedule:
        raise CustomerAdminError("adaptive registration training_schedule is required")
    parsed_schedule: list[dict[str, object]] = []
    seen_days: set[date] = set()
    for entry in schedule:
        if type(entry) is CustomerTrainingScheduleEntry:
            candidate = entry.model_dump(mode="json")
        elif isinstance(entry, Mapping):
            candidate = dict(entry)
        else:
            raise CustomerAdminError(
                "adaptive registration training_schedule is invalid"
            )
        if set(candidate) != {"date", "weekday", "time", "load_category"}:
            raise CustomerAdminError(
                "adaptive registration training_schedule is invalid"
            )
        day_value = candidate.get("date")
        weekday_value = candidate.get("weekday")
        time_value = candidate.get("time")
        load_value = candidate.get("load_category")
        if (
            not isinstance(day_value, str)
            or not isinstance(weekday_value, int)
            or isinstance(weekday_value, bool)
        ):
            raise CustomerAdminError(
                "adaptive registration training_schedule date/weekday is invalid"
            )
        if not isinstance(time_value, str) or not time_value.strip():
            raise CustomerAdminError(
                "adaptive registration training_schedule time is invalid"
            )
        if not isinstance(load_value, str) or not load_value.strip():
            raise CustomerAdminError(
                "adaptive registration training_schedule load category is invalid"
            )
        try:
            parsed_day = date.fromisoformat(day_value)
            parsed_time = time.fromisoformat(time_value)
        except ValueError as exc:
            raise CustomerAdminError(
                "adaptive registration training_schedule date/time is invalid"
            ) from exc
        if parsed_day.weekday() != weekday_value:
            raise CustomerAdminError(
                "adaptive registration training_schedule weekday is invalid"
            )
        if parsed_day in seen_days:
            raise CustomerAdminError(
                "adaptive registration training_schedule dates must be unique"
            )
        seen_days.add(parsed_day)
        parsed_schedule.append(
            {
                "date": parsed_day.isoformat(),
                "weekday": weekday_value,
                "time": parsed_time.isoformat(),
                "load_category": load_value.strip(),
            }
        )
    parsed_schedule.sort(key=lambda item: (str(item["date"]), str(item["time"])))
    payload = {
        "schema_version": _REGISTRATION_SCHEMA_VERSION,
        "customer_key": customer_key,
        "version": version,
        "meal_count": meal_count,
        "budget_band": budget_band.strip(),
        "cooking_access": cooking_access.strip(),
        "preferences": text_values("preferences"),
        "exclusions": text_values("exclusions"),
        "allergies": text_values("allergies"),
        "training_schedule": parsed_schedule,
    }
    supplied_digest = raw.get("digest")
    computed = adaptive_digest(payload)
    if supplied_digest is not None and supplied_digest != computed:
        raise CustomerAdminError(
            "adaptive registration input digest does not match content"
        )
    return payload


def _require_onboarding_profile_registration_match(
    payload: Mapping[str, object],
    profile: object,
) -> None:
    if getattr(profile, "starting_context", None) != "nutrition-onboarding-v1":
        return
    for field in ("meal_count", "budget_band", "cooking_access"):
        if payload.get(field) != getattr(profile, field, None):
            raise CustomerAdminError(
                f"adaptive registration {field} differs from onboarding profile"
            )
    expected_lists = {
        "preferences": tuple(getattr(profile, "food_preferences", ())),
        "allergies": tuple(getattr(profile, "allergies", ())),
        "exclusions": tuple(
            (
                *getattr(profile, "dietary_restrictions", ()),
                *getattr(profile, "disliked_foods", ()),
            )
        ),
    }
    for field, expected in expected_lists.items():
        actual = payload.get(field)
        if not isinstance(actual, (list, tuple)) or tuple(actual) != expected:
            raise CustomerAdminError(
                f"adaptive registration {field} differs from onboarding profile"
            )


def _registration_document(
    payload: Mapping[str, object],
    *,
    supersedes_digest: str,
    authority: Mapping[str, object],
    authority_digest: str,
    approved_at_kst: str,
    approved_by: Mapping[str, str],
) -> dict[str, object]:
    result = {
        **dict(payload),
        "digest": adaptive_digest(payload),
        "supersedes_digest": supersedes_digest,
        "approved": True,
        "approved_by": dict(approved_by),
        "approved_at_kst": approved_at_kst,
        "activation_receipt_id": authority["activation_receipt_id"],
        "activation_receipt_digest": authority["activation_receipt_digest"],
        "registry_digest": authority["registry_digest"],
        "owner_digest": authority["owner_digest"],
        "authority_digest": authority_digest,
        "authority": dict(authority),
    }
    try:
        validated = AdaptiveRegistrationInputs.model_validate(result)
    except (CustomerRegistryError, TypeError, ValueError) as exc:
        raise CustomerAdminError("adaptive registration inputs are invalid") from exc
    return validated.model_dump(mode="json", exclude_none=True)


def _registration_revision_row(
    *,
    sequence: int,
    intent_id: str,
    state: str,
    document: Mapping[str, object],
    prepared_digest: str | None,
) -> dict[str, object]:
    body = {
        "schema_version": _REGISTRATION_SCHEMA_VERSION,
        "kind": "customer_adaptive_registration",
        "append_sequence": sequence,
        "intent_id": intent_id,
        "state": state,
        "customer_key": document["customer_key"],
        "version": document["version"],
        "revision_digest": document["digest"],
        "input_document": dict(document),
        "supersedes_digest": document["supersedes_digest"],
        "activation_receipt_id": document["activation_receipt_id"],
        "activation_receipt_digest": document["activation_receipt_digest"],
        "registry_digest": document["registry_digest"],
        "owner_digest": document["owner_digest"],
        "authority_digest": document["authority_digest"],
        "authority": document["authority"],
        "approved_by": document["approved_by"],
        "approved_at_kst": document["approved_at_kst"],
        "prepared_digest": prepared_digest,
    }
    return {**body, "row_digest": _registration_row_digest(body)}


def _registration_config_row(
    *,
    sequence: int,
    intent_id: str,
    state: str,
    document: Mapping[str, object],
    prepared_digest: str | None,
) -> dict[str, object]:
    body = {
        "schema_version": _REGISTRATION_SCHEMA_VERSION,
        "kind": "customer_adaptive_registration_config",
        "append_sequence": sequence,
        "intent_id": intent_id,
        "state": state,
        "customer_key": document["customer_key"],
        "version": document["version"],
        "revision_digest": document["digest"],
        "supersedes_digest": document["supersedes_digest"],
        "activation_receipt_id": document["activation_receipt_id"],
        "activation_receipt_digest": document["activation_receipt_digest"],
        "registry_digest": document["registry_digest"],
        "owner_digest": document["owner_digest"],
        "authority_digest": document["authority_digest"],
        "artifact_digests": document.get("artifact_digests"),
        "approved_by": document["approved_by"],
        "approved_at_kst": document["approved_at_kst"],
        "prepared_digest": prepared_digest,
    }
    return {**body, "row_digest": _registration_row_digest(body)}


def _registration_append_revision_locked(
    root: Path,
    document: Mapping[str, object],
) -> None:
    """Append a registration revision while the registration lock is held."""

    revisions_path, config_path, _, _ = _registration_paths(root)
    revision_rows = _registration_jsonl(revisions_path, recover=True)
    config_rows = _registration_jsonl(config_path, recover=True)
    committed, pending = _validate_registration_revision_rows(revision_rows)
    config_committed, config_pending = _validate_registration_config_rows(config_rows)
    revision_digest = str(document["digest"])
    key = str(document["customer_key"])
    existing = next(
        (
            row
            for row in committed
            if row.get("customer_key") == key
            and row.get("revision_digest") == revision_digest
        ),
        None,
    )
    if existing is not None:
        if existing.get("input_document") != dict(document):
            raise CustomerAdminError("conflicting adaptive registration replay")
        config_existing = next(
            (
                row
                for row in config_committed
                if row.get("customer_key") == key
                and row.get("revision_digest") == revision_digest
            ),
            None,
        )
        if config_existing is None:
            intent_id = f"registration-config:{key}:{revision_digest}"
            config_prepared = _registration_config_row(
                sequence=len(config_rows) + 1,
                intent_id=intent_id,
                state="prepared",
                document=document,
                prepared_digest=None,
            )
            _append_registration_row(config_path, config_prepared)
            config_commit = _registration_config_row(
                sequence=len(config_rows) + 2,
                intent_id=intent_id,
                state="committed",
                document=document,
                prepared_digest=config_prepared["row_digest"],
            )
            _append_registration_row(config_path, config_commit)
        return
    if pending:
        matching = next(
            (
                row
                for row in pending
                if row.get("customer_key") == key
                and row.get("revision_digest") == revision_digest
            ),
            None,
        )
        if matching is None:
            raise CustomerAdminError(
                "another adaptive registration transition requires recovery"
            )
        if matching.get("input_document") != dict(document):
            raise CustomerAdminError(
                "conflicting incomplete adaptive registration replay"
            )
        prepared = matching
        commit = _registration_revision_row(
            sequence=len(revision_rows) + 1,
            intent_id=str(prepared["intent_id"]),
            state="committed",
            document=document,
            prepared_digest=str(prepared["row_digest"]),
        )
        _append_registration_row(revisions_path, commit)
        revision_rows.append(commit)
        committed, pending = _validate_registration_revision_rows(revision_rows)
    else:
        intent_id = f"registration:{key}:{revision_digest}"
        prepared = _registration_revision_row(
            sequence=len(revision_rows) + 1,
            intent_id=intent_id,
            state="prepared",
            document=document,
            prepared_digest=None,
        )
        _append_registration_row(revisions_path, prepared)
        commit = _registration_revision_row(
            sequence=len(revision_rows) + 2,
            intent_id=intent_id,
            state="committed",
            document=document,
            prepared_digest=prepared["row_digest"],
        )
        _append_registration_row(revisions_path, commit)
        revision_rows.extend((prepared, commit))
        committed, pending = _validate_registration_revision_rows(revision_rows)
    config_rows = _registration_jsonl(config_path, recover=True)
    config_committed, config_pending = _validate_registration_config_rows(config_rows)
    config_existing = next(
        (
            row
            for row in config_committed
            if row.get("customer_key") == key
            and row.get("revision_digest") == revision_digest
        ),
        None,
    )
    if config_existing is None:
        if config_pending:
            matching_config = next(
                (
                    row
                    for row in config_pending
                    if row.get("customer_key") == key
                    and row.get("revision_digest") == revision_digest
                ),
                None,
            )
            if matching_config is None:
                raise CustomerAdminError(
                    "another adaptive registration config transition requires recovery"
                )
            config_prepared = matching_config
            config_commit = _registration_config_row(
                sequence=len(config_rows) + 1,
                intent_id=str(config_prepared["intent_id"]),
                state="committed",
                document=document,
                prepared_digest=str(config_prepared["row_digest"]),
            )
            _append_registration_row(config_path, config_commit)
        else:
            config_intent = f"registration-config:{key}:{revision_digest}"
            config_prepared = _registration_config_row(
                sequence=len(config_rows) + 1,
                intent_id=config_intent,
                state="prepared",
                document=document,
                prepared_digest=None,
            )
            _append_registration_row(config_path, config_prepared)
            config_commit = _registration_config_row(
                sequence=len(config_rows) + 2,
                intent_id=config_intent,
                state="committed",
                document=document,
                prepared_digest=config_prepared["row_digest"],
            )
            _append_registration_row(config_path, config_commit)


def _registration_append_revision(
    root: Path,
    document: Mapping[str, object],
) -> None:
    try:
        with _registration_lock(root):
            _registration_append_revision_locked(root, document)
    except CustomerAdminError:
        raise
    except OSError as exc:
        raise CustomerAdminError("adaptive registration transition failed") from exc


def _load_adaptive_registration_inputs(
    profile_root: Path,
    customer_key: str,
    *,
    recover: bool = True,
) -> AdaptiveRegistrationInputs:
    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    registry_path = _resolve_registry_path(root)
    if recover:
        _recover_activation_journal(root, registry_path)
    document = _read_profile_registry(registry_path, root)
    _validate_enabled_activation_receipts(root, registry_path, document)
    selected = next(
        (item for item in document.customers if item.customer_key == key), None
    )
    if selected is None:
        raise CustomerAdminError(f"unknown customer: {key}")
    if not selected.enabled:
        raise CustomerAdminError(
            "customer must be enabled for adaptive registration inputs"
        )
    authority, authority_digest = _registration_activation_authority(
        root,
        registry_path,
        document,
        selected,
    )
    runtime = next(
        (
            item
            for item in load_customer_registry(registry_path, root).customers
            if item.spec.customer_key == key
        ),
        None,
    )
    if runtime is None:
        raise CustomerAdminError("registered customer data root is invalid")
    adaptive_root = _registration_private_root(runtime.data_root)
    revisions_path, config_path, approval_path, _ = _registration_paths(adaptive_root)
    revision_rows = _registration_jsonl(revisions_path)
    config_rows = _registration_jsonl(config_path)
    committed, pending = _validate_registration_revision_rows(revision_rows)
    config_committed, config_pending = _validate_registration_config_rows(config_rows)
    if pending or config_pending:
        raise CustomerAdminError("adaptive registration transition is incomplete")
    selected_rows = [row for row in committed if row.get("customer_key") == key]
    selected_configs = [
        row for row in config_committed if row.get("customer_key") == key
    ]
    if len({str(row.get("revision_digest")) for row in selected_rows}) != len(
        selected_rows
    ) or len({str(row.get("revision_digest")) for row in selected_configs}) != len(
        selected_configs
    ):
        raise CustomerAdminError(
            "adaptive registration journal has duplicate revisions"
        )
    if not selected_rows or not selected_configs:
        raise CustomerAdminError("approved adaptive registration inputs are missing")
    latest = selected_rows[-1]
    latest_config = selected_configs[-1]
    approval_rows = _registration_jsonl(approval_path)
    approval_artifacts = _registration_latest_approvals(
        approval_rows,
        key,
        str(latest["revision_digest"]),
    )
    if latest_config.get("revision_digest") != latest.get("revision_digest"):
        raise CustomerAdminError("adaptive registration config is stale")
    input_document = latest.get("input_document")
    if not isinstance(input_document, Mapping):
        raise CustomerAdminError("adaptive registration input document is invalid")
    _registration_authority_matches(
        input_document,
        authority,
        authority_digest,
        document.owner.model_dump(mode="json"),
    )
    owner_payload = document.owner.model_dump(mode="json")
    revision_digest = _registration_digest(input_document.get("digest"), "revision")
    if revision_digest != latest.get("revision_digest"):
        raise CustomerAdminError("adaptive registration revision is stale")
    try:
        derived_payload = _registration_input_payload(
            input_document,
            key,
            default_version=str(input_document.get("version", "v1")),
        )
        expected_meal_document, expected_meal_version, expected_meal_digest = (
            _registration_derived_meal_constraints(
                derived_payload,
                owner=owner_payload,
                approved_at_kst=str(input_document["approved_at_kst"]),
            )
        )
    except (CustomerAdminError, KeyError, TypeError, ValueError) as exc:
        raise CustomerAdminError("adaptive meal constraints artifact is stale") from exc
    expected_meal_value = _registration_artifact_value(
        expected_meal_document,
        "meal_constraints",
    )
    if not isinstance(expected_meal_value, Mapping):
        raise CustomerAdminError("adaptive meal constraints artifact is invalid")
    supplied_derived = input_document.get("derived_constraints")
    if supplied_derived is not None and (
        not isinstance(supplied_derived, Mapping)
        or dict(supplied_derived) != dict(expected_meal_value)
    ):
        raise CustomerAdminError("adaptive registration derived constraints are stale")
    supplied_derived_digest = input_document.get("derived_constraints_digest")
    if (
        supplied_derived_digest is not None
        and supplied_derived_digest != expected_meal_digest
    ):
        raise CustomerAdminError(
            "adaptive registration derived constraints digest is stale"
        )
    if input_document.get("meal_constraints_digest") != expected_meal_digest:
        raise CustomerAdminError(
            "adaptive registration meal constraints digest is stale"
        )
    if input_document.get("meal_constraints_version") != expected_meal_version:
        raise CustomerAdminError(
            "adaptive registration meal constraints version is stale"
        )
    input_artifacts = input_document.get("artifact_documents")
    if input_artifacts is not None:
        if not isinstance(input_artifacts, Mapping) or set(input_artifacts) != {
            "base_policy",
            "meal_constraints",
            "catalog",
        }:
            raise CustomerAdminError("adaptive registration artifacts are incomplete")
        for artifact_kind, row in approval_artifacts.items():
            artifact_document = row.get("artifact_document")
            supplied_document = input_artifacts.get(artifact_kind)
            if (
                not isinstance(artifact_document, Mapping)
                or not isinstance(supplied_document, Mapping)
                or dict(supplied_document) != dict(artifact_document)
            ):
                raise CustomerAdminError("adaptive registration artifacts are stale")
    artifact_digests = input_document.get("artifact_digests")
    if (
        not isinstance(artifact_digests, Mapping)
        or set(artifact_digests) != {"base_policy", "meal_constraints", "catalog"}
        or any(
            artifact_digests.get(kind) != input_document.get(f"{kind}_digest")
            for kind in ("base_policy", "meal_constraints", "catalog")
        )
        or artifact_digests.get("meal_constraints") != expected_meal_digest
    ):
        raise CustomerAdminError("adaptive registration artifact digests are stale")
    for kind in ("base_policy", "meal_constraints", "catalog"):
        _registration_digest(artifact_digests.get(kind), f"{kind} artifact")
    for artifact_kind, row in approval_artifacts.items():
        digest_field = f"{artifact_kind}_digest"
        if (
            row.get("authority_digest") != authority_digest
            or row.get("registry_digest") != authority.get("registry_digest")
            or row.get("owner_digest") != authority.get("owner_digest")
            or row.get("activation_receipt_digest")
            != authority.get("activation_receipt_digest")
            or row.get("approved_by") != owner_payload
            or row.get("artifact_digest") != input_document.get(digest_field)
            or row.get("artifact_version")
            != input_document.get(f"{artifact_kind}_version")
        ):
            raise CustomerAdminError("adaptive input approval artifacts are stale")
        artifact_document = row.get("artifact_document")
        if not isinstance(artifact_document, Mapping):
            raise CustomerAdminError("adaptive input approval artifact is invalid")
        if artifact_kind == "meal_constraints":
            artifact_value = _registration_artifact_value(
                artifact_document,
                artifact_kind,
            )
            if (
                artifact_document.get("approved") is not True
                or artifact_document.get("approved_by") != owner_payload
                or artifact_document.get("version") != expected_meal_version
                or artifact_document.get("approved_at_kst")
                != input_document.get("approved_at_kst")
                or row.get("artifact_digest") != expected_meal_digest
                or artifact_value != expected_meal_value
                or adaptive_digest(artifact_value) != expected_meal_digest
            ):
                raise CustomerAdminError("adaptive meal constraints artifact is stale")
            approved_at = artifact_document.get("approved_at_kst")
            if not isinstance(approved_at, str):
                raise CustomerAdminError(
                    "adaptive meal constraints artifact metadata is invalid"
                )
            _registration_kst_timestamp(approved_at)
            continue
        _, row_version, row_digest = _registration_validate_external_artifact(
            artifact_document,
            artifact_kind,
            owner_payload,
        )
        if row_version != row.get("artifact_version") or row_digest != row.get(
            "artifact_digest"
        ):
            raise CustomerAdminError("adaptive input approval artifact is stale")
        current_document = _registration_read_artifact_source(
            adaptive_root,
            artifact_kind,
            None,
        )
        _, current_version, current_digest = _registration_validate_external_artifact(
            current_document,
            artifact_kind,
            owner_payload,
        )
        if (
            current_version != row.get("artifact_version")
            or current_digest != row.get("artifact_digest")
            or current_document.get("approved_at_kst")
            != artifact_document.get("approved_at_kst")
            or current_document.get("approved_by") != owner_payload
        ):
            raise CustomerAdminError("adaptive input approval artifact is stale")
    if any(
        latest_config.get(field) != latest.get(field)
        for field in (
            "customer_key",
            "version",
            "revision_digest",
            "supersedes_digest",
            "activation_receipt_id",
            "activation_receipt_digest",
            "registry_digest",
            "owner_digest",
            "authority_digest",
            "approved_by",
            "approved_at_kst",
        )
    ) or latest_config.get("artifact_digests") != input_document.get(
        "artifact_digests"
    ):
        raise CustomerAdminError("adaptive registration config is stale")
    authoritative_document = dict(input_document)
    authoritative_document["derived_constraints"] = dict(expected_meal_value)
    authoritative_document["derived_constraints_digest"] = expected_meal_digest
    authoritative_document["artifact_documents"] = {
        artifact_kind: dict(row["artifact_document"])
        for artifact_kind, row in approval_artifacts.items()
        if isinstance(row.get("artifact_document"), Mapping)
    }
    try:
        result = AdaptiveRegistrationInputs.model_validate(authoritative_document)
    except (CustomerRegistryError, TypeError, ValueError) as exc:
        raise CustomerAdminError(
            "approved adaptive registration inputs are invalid"
        ) from exc
    if (
        result.digest != adaptive_digest(result.value_payload())
        or result.digest != revision_digest
        or result.derived_constraints is None
        or result.derived_constraints_digest != expected_meal_digest
        or result.meal_constraints_digest != expected_meal_digest
    ):
        raise CustomerAdminError(
            "adaptive registration input digest does not match content"
        )
    return result


def approve_adaptive_registration_inputs(
    profile_root: Path,
    customer_key: str,
    *,
    inputs: object,
    approved_by: object,
    approved_at_kst: str | None = None,
    supersedes_digest: str | None = None,
    base_policy: object | None = None,
    policy: object | None = None,
    food_catalog: object | None = None,
    catalog: object | None = None,
    meal_constraints: object | None = None,
) -> AdaptiveRegistrationInputs:
    root = _resolve_profile_root(Path(profile_root))
    with profile_authority_lock(root):
        return _approve_adaptive_registration_inputs_locked(
            root,
            customer_key,
            inputs=inputs,
            approved_by=approved_by,
            approved_at_kst=approved_at_kst,
            supersedes_digest=supersedes_digest,
            base_policy=base_policy,
            policy=policy,
            food_catalog=food_catalog,
            catalog=catalog,
            meal_constraints=meal_constraints,
        )


def _approve_adaptive_registration_inputs_locked(
    profile_root: Path,
    customer_key: str,
    *,
    inputs: object,
    approved_by: object,
    approved_at_kst: str | None = None,
    supersedes_digest: str | None = None,
    base_policy: object | None = None,
    policy: object | None = None,
    food_catalog: object | None = None,
    catalog: object | None = None,
    meal_constraints: object | None = None,
) -> AdaptiveRegistrationInputs:
    """Approve and append one complete customer adaptive-input revision."""

    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    registry_path = _resolve_registry_path(root)
    _recover_activation_journal(root, registry_path)
    document = _read_profile_registry(registry_path, root)
    _validate_enabled_activation_receipts(root, registry_path, document)
    selected = next(
        (item for item in document.customers if item.customer_key == key), None
    )
    if selected is None:
        raise CustomerAdminError(f"unknown customer: {key}")
    if not selected.enabled:
        raise CustomerAdminError(
            "customer must be enabled for adaptive registration inputs"
        )
    authority, authority_digest = _registration_activation_authority(
        root,
        registry_path,
        document,
        selected,
    )
    owner = _registration_owner(approved_by)
    if owner != document.owner.model_dump(mode="json"):
        raise CustomerAdminError("adaptive registration approver does not match owner")
    runtime = next(
        (
            item
            for item in load_customer_registry(registry_path, root).customers
            if item.spec.customer_key == key
        ),
        None,
    )
    if runtime is None:
        raise CustomerAdminError("registered customer data root is invalid")
    adaptive_root = _registration_private_root(runtime.data_root)
    try:
        with _registration_lock(adaptive_root):
            revisions_path, _, _, _ = _registration_paths(adaptive_root)
            revisions = _registration_jsonl(revisions_path, recover=True)
            committed, pending = _validate_registration_revision_rows(revisions)
            selected_committed = [
                row for row in committed if row.get("customer_key") == key
            ]
            latest_digest = (
                str(selected_committed[-1]["revision_digest"])
                if selected_committed
                else _REGISTRATION_ZERO_DIGEST
            )
            latest_version = (
                str(selected_committed[-1]["version"]) if selected_committed else "v0"
            )
            default_version = f"v{len(selected_committed) + 1}"
            payload = _registration_input_payload(
                inputs, key, default_version=default_version
            )
            _require_onboarding_profile_registration_match(
                payload,
                runtime.spec.profile,
            )
            raw_supersedes = supersedes_digest
            if raw_supersedes is None:
                raw_supersedes = (
                    inputs.get("supersedes_digest")
                    if isinstance(inputs, Mapping)
                    else getattr(inputs, "supersedes_digest", None)
                )
            supersedes_digest = (
                latest_digest if raw_supersedes in (None, "") else raw_supersedes
            )
            if supersedes_digest != latest_digest:
                raise CustomerAdminError(
                    "adaptive registration supersession predecessor is stale"
                )
            if payload["version"] == latest_version and selected_committed:
                if adaptive_digest(payload) != latest_digest:
                    payload = {
                        **payload,
                        "version": default_version,
                    }
            approval_time = _registration_kst_timestamp(approved_at_kst)
            registration_document = _registration_document(
                payload,
                supersedes_digest=supersedes_digest,
                authority=authority,
                authority_digest=authority_digest,
                approved_at_kst=approval_time,
                approved_by=owner,
            )
            artifact_inputs: object = inputs
            if any(
                value is not None
                for value in (
                    base_policy,
                    policy,
                    food_catalog,
                    catalog,
                    meal_constraints,
                )
            ):
                if isinstance(inputs, Mapping):
                    artifact_payload = dict(inputs)
                else:
                    model_dump = getattr(inputs, "model_dump", None)
                    dumped = (
                        model_dump(mode="json", exclude_none=True)
                        if callable(model_dump)
                        else {}
                    )
                    artifact_payload = (
                        dict(dumped) if isinstance(dumped, Mapping) else {}
                    )
                if base_policy is not None:
                    artifact_payload["base_policy"] = base_policy
                if meal_constraints is not None:
                    artifact_payload["meal_constraints"] = meal_constraints
                if policy is not None:
                    artifact_payload["policy"] = policy
                if food_catalog is not None:
                    artifact_payload["food_catalog"] = food_catalog
                if catalog is not None:
                    artifact_payload["catalog"] = catalog
                artifact_inputs = artifact_payload
            bundle = _registration_artifact_bundle(
                adaptive_root,
                payload,
                artifact_inputs,
                owner=owner,
                approved_at_kst=approval_time,
            )
            registration_document = {
                **registration_document,
                "base_policy_version": bundle[0][2],
                "base_policy_digest": bundle[0][3],
                "meal_constraints_version": bundle[1][2],
                "meal_constraints_digest": bundle[1][3],
                "derived_constraints": dict(
                    _registration_artifact_value(bundle[1][1], "meal_constraints")
                ),
                "derived_constraints_digest": bundle[1][3],
                "catalog_version": bundle[2][2],
                "catalog_digest": bundle[2][3],
                "artifact_digests": {
                    kind: artifact_digest for kind, _, _, artifact_digest in bundle
                },
                "artifact_documents": {
                    kind: artifact_document for kind, artifact_document, _, _ in bundle
                },
            }
            try:
                registration_document = AdaptiveRegistrationInputs.model_validate(
                    registration_document
                ).model_dump(mode="json", exclude_none=True)
            except (CustomerRegistryError, TypeError, ValueError) as exc:
                raise CustomerAdminError(
                    "adaptive registration artifact metadata is invalid"
                ) from exc
            _registration_append_revision_locked(adaptive_root, registration_document)
    except CustomerAdminError:
        raise
    except OSError as exc:
        raise CustomerAdminError("adaptive registration transition failed") from exc
    _append_registration_approvals(
        adaptive_root,
        registration_document,
        bundle,
        authority,
        authority_digest,
    )
    return AdaptiveRegistrationInputs.model_validate(registration_document)


def update_adaptive_registration_inputs(
    profile_root: Path,
    customer_key: str,
    *,
    inputs: object,
    approved_by: object,
    approved_at_kst: str | None = None,
    supersedes_digest: str | None = None,
    base_policy: object | None = None,
    policy: object | None = None,
    food_catalog: object | None = None,
    catalog: object | None = None,
    meal_constraints: object | None = None,
) -> AdaptiveRegistrationInputs:
    """Create a superseding approved revision; no in-place update is allowed."""

    return approve_adaptive_registration_inputs(
        profile_root,
        customer_key,
        inputs=inputs,
        approved_by=approved_by,
        supersedes_digest=supersedes_digest,
        approved_at_kst=approved_at_kst,
        base_policy=base_policy,
        policy=policy,
        food_catalog=food_catalog,
        catalog=catalog,
        meal_constraints=meal_constraints,
    )


def reapprove_adaptive_registration_inputs(
    profile_root: Path,
    customer_key: str,
    *,
    approved_by: object,
) -> AdaptiveRegistrationInputs:
    """Append an authority-only child approval after activation receipt rotation."""

    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    with profile_authority_lock(root):
        registry_path = _resolve_registry_path(root)
        _recover_activation_journal_locked(root, registry_path)
        document = _read_profile_registry(registry_path, root)
        _validate_enabled_activation_receipts(root, registry_path, document)
        selected = next(
            (item for item in document.customers if item.customer_key == key),
            None,
        )
        if selected is None:
            raise CustomerAdminError(f"unknown customer: {key}")
        if not selected.enabled:
            raise CustomerAdminError(
                "customer must be enabled for adaptive registration inputs"
            )
        authority, authority_digest = _registration_activation_authority(
            root,
            registry_path,
            document,
            selected,
        )
        owner = _registration_owner(approved_by)
        if owner != document.owner.model_dump(mode="json"):
            raise CustomerAdminError(
                "adaptive registration approver does not match owner"
            )
        runtime = next(
            (
                item
                for item in load_customer_registry(registry_path, root).customers
                if item.spec.customer_key == key
            ),
            None,
        )
        if runtime is None:
            raise CustomerAdminError("registered customer data root is invalid")
        adaptive_root = _registration_private_root(runtime.data_root)
        with _registration_lock(adaptive_root):
            revisions_path, config_path, approval_path, _ = _registration_paths(
                adaptive_root
            )
            revision_rows = _registration_jsonl(revisions_path, recover=True)
            config_rows = _registration_jsonl(config_path, recover=True)
            approval_rows = _registration_jsonl(approval_path, recover=True)
            committed, pending = _validate_registration_revision_rows(revision_rows)
            config_committed, config_pending = _validate_registration_config_rows(
                config_rows
            )
            if pending or config_pending:
                raise CustomerAdminError(
                    "adaptive registration transition is incomplete"
                )
            selected_rows = [row for row in committed if row.get("customer_key") == key]
            selected_configs = [
                row for row in config_committed if row.get("customer_key") == key
            ]
            if not selected_rows or not selected_configs:
                raise CustomerAdminError(
                    "approved adaptive registration inputs are missing"
                )
            latest = selected_rows[-1]
            latest_config = selected_configs[-1]
            latest_digest = str(latest["revision_digest"])
            if latest_config.get("revision_digest") != latest_digest:
                raise CustomerAdminError("adaptive registration config is stale")
            input_document = latest.get("input_document")
            if not isinstance(input_document, Mapping):
                raise CustomerAdminError(
                    "adaptive registration input document is invalid"
                )
            payload = _registration_input_payload(
                input_document,
                key,
                default_version=str(input_document.get("version", "v1")),
            )
            if adaptive_digest(payload) != latest_digest:
                raise CustomerAdminError("adaptive registration revision is stale")
            old_authority = input_document.get("authority")
            old_authority_digest = input_document.get("authority_digest")
            expected_authority_fields = {
                "schema_version",
                "customer_key",
                "owner",
                "owner_digest",
                "registry_digest",
                "activation_receipt_id",
                "activation_receipt_digest",
            }
            if (
                not isinstance(old_authority, Mapping)
                or set(old_authority) != expected_authority_fields
                or not isinstance(old_authority_digest, str)
                or adaptive_digest(dict(old_authority)) != old_authority_digest
                or input_document.get("approved_by") != owner
                or old_authority.get("owner") != owner
                or old_authority.get("owner_digest") != authority["owner_digest"]
                or input_document.get("owner_digest")
                != old_authority.get("owner_digest")
                or input_document.get("registry_digest")
                != old_authority.get("registry_digest")
                or input_document.get("activation_receipt_id")
                != old_authority.get("activation_receipt_id")
                or input_document.get("activation_receipt_digest")
                != old_authority.get("activation_receipt_digest")
            ):
                raise CustomerAdminError(
                    "adaptive registration owner authority changed"
                )
            approval_artifacts = _registration_latest_approvals(
                approval_rows,
                key,
                latest_digest,
            )
            artifact_digests = input_document.get("artifact_digests")
            if not isinstance(artifact_digests, Mapping) or set(artifact_digests) != {
                "base_policy",
                "meal_constraints",
                "catalog",
            }:
                raise CustomerAdminError(
                    "adaptive registration artifacts are incomplete"
                )
            approved_at_kst = input_document.get("approved_at_kst")
            if not isinstance(approved_at_kst, str):
                raise CustomerAdminError(
                    "adaptive registration approval time is invalid"
                )
            bundle = _registration_artifact_bundle(
                adaptive_root,
                payload,
                {},
                owner=owner,
                approved_at_kst=approved_at_kst,
            )
            bundle_by_kind = {
                kind: (artifact_document, version, artifact_digest)
                for kind, artifact_document, version, artifact_digest in bundle
            }
            for kind, row in approval_artifacts.items():
                expected_document, expected_version, expected_digest = bundle_by_kind[
                    kind
                ]
                if (
                    row.get("authority_digest") != old_authority_digest
                    or row.get("registry_digest")
                    != old_authority.get("registry_digest")
                    or row.get("owner_digest") != old_authority.get("owner_digest")
                    or row.get("activation_receipt_digest")
                    != old_authority.get("activation_receipt_digest")
                    or row.get("approved_by") != owner
                    or row.get("artifact_digest") != artifact_digests.get(kind)
                    or row.get("artifact_digest") != expected_digest
                    or row.get("artifact_version") != expected_version
                    or row.get("artifact_document") != expected_document
                ):
                    raise CustomerAdminError(
                        "adaptive registration approval artifact changed"
                    )
            if old_authority_digest == authority_digest:
                try:
                    return AdaptiveRegistrationInputs.model_validate(input_document)
                except (CustomerRegistryError, TypeError, ValueError) as exc:
                    raise CustomerAdminError(
                        "approved adaptive registration inputs are invalid"
                    ) from exc
            candidate = len(selected_rows) + 1
            seen_digests = {str(row["revision_digest"]) for row in selected_rows}
            while True:
                rebound_payload = {
                    **payload,
                    "version": f"v{candidate}",
                }
                if adaptive_digest(rebound_payload) not in seen_digests:
                    break
                candidate += 1
            base_policy = bundle_by_kind["base_policy"][0]
            food_catalog = bundle_by_kind["catalog"][0]
        return _approve_adaptive_registration_inputs_locked(
            root,
            key,
            inputs=rebound_payload,
            approved_by=owner,
            supersedes_digest=latest_digest,
            base_policy=base_policy,
            food_catalog=food_catalog,
        )


def validate_adaptive_registration_reapproval(
    profile_root: Path,
    customer_key: str,
    predecessor_digest: str,
) -> bool:
    """Prove the current registration is an authority-only child revision."""

    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    predecessor = str(predecessor_digest or "").strip()
    if len(predecessor) != 64 or any(
        character not in "0123456789abcdef" for character in predecessor
    ):
        raise CustomerAdminError("adaptive registration predecessor digest is invalid")
    current = load_approved_adaptive_registration_inputs(root, key)
    runtime = next(
        (
            item
            for item in load_runtime_customer_registry(root).customers
            if item.spec.customer_key == key
        ),
        None,
    )
    if runtime is None:
        raise CustomerAdminError("registered customer data root is invalid")
    adaptive_root = _registration_private_root(runtime.data_root)
    with profile_authority_lock(root):
        registry_path = _resolve_registry_path(root)
        document = _read_profile_registry(registry_path, root)
        selected = next(
            (item for item in document.customers if item.customer_key == key),
            None,
        )
        if selected is None or not selected.enabled:
            raise CustomerAdminError(
                "customer must be enabled for adaptive registration inputs"
            )
        authority, authority_digest = _registration_activation_authority(
            root,
            registry_path,
            document,
            selected,
        )
        current_document = current.model_dump(mode="json", exclude_none=True)
        _registration_authority_matches(
            current_document,
            authority,
            authority_digest,
            document.owner.model_dump(mode="json"),
        )
        with _registration_lock(adaptive_root):
            revisions_path, config_path, approval_path, _ = _registration_paths(
                adaptive_root
            )
            committed, pending = _validate_registration_revision_rows(
                _registration_jsonl(revisions_path)
            )
            config_committed, config_pending = _validate_registration_config_rows(
                _registration_jsonl(config_path)
            )
            if pending or config_pending:
                raise CustomerAdminError(
                    "adaptive registration transition is incomplete"
                )
            selected_rows = [row for row in committed if row.get("customer_key") == key]
            selected_configs = [
                row for row in config_committed if row.get("customer_key") == key
            ]
            if not selected_rows or not selected_configs:
                raise CustomerAdminError(
                    "approved adaptive registration inputs are missing"
                )
            latest = selected_rows[-1]
            latest_config = selected_configs[-1]
            if (
                latest.get("revision_digest") != current.digest
                or latest_config.get("revision_digest") != current.digest
                or current.supersedes_digest != predecessor
            ):
                raise CustomerAdminError(
                    "adaptive registration reapproval lineage is stale"
                )
            previous = next(
                (
                    row
                    for row in selected_rows
                    if row.get("revision_digest") == predecessor
                ),
                None,
            )
            previous_document = (
                previous.get("input_document")
                if isinstance(previous, Mapping)
                else None
            )
            if not isinstance(previous_document, Mapping):
                raise CustomerAdminError(
                    "adaptive registration reapproval predecessor is unavailable"
                )
            previous_payload = _registration_input_payload(
                previous_document,
                key,
                default_version=str(previous_document.get("version", "v1")),
            )
            if adaptive_digest(previous_payload) != predecessor:
                raise CustomerAdminError(
                    "adaptive registration reapproval predecessor is stale"
                )
            current_payload = current.value_payload()
            previous_payload.pop("version", None)
            current_payload.pop("version", None)
            if (
                previous_payload != current_payload
                or previous_document.get("approved_by")
                != current_document.get("approved_by")
                or previous_document.get("owner_digest") != current.owner_digest
                or previous_document.get("artifact_digests") != current.artifact_digests
            ):
                raise CustomerAdminError(
                    "adaptive registration reapproval changed approved meaning"
                )
            _registration_latest_approvals(
                _registration_jsonl(approval_path),
                key,
                current.digest,
            )
    return True


def load_approved_adaptive_registration_inputs(
    profile_root: Path,
    customer_key: str,
) -> AdaptiveRegistrationInputs:
    """Load the latest approved inputs and their registration-bound artifacts.

    The returned value exposes ``registration_digest``,
    ``derived_constraints``, and ``derived_constraints_digest`` together;
    policy and catalog remain separately owner-approved artifacts.
    """

    return _load_adaptive_registration_inputs(profile_root, customer_key)


_RECON_FEATURE_FLAGS = (
    "analytics_shadow",
    "operator_candidates",
    "activation",
    "delivery",
)
_RECON_EVENT_FLOWS = {
    "morning_checkin": "morning",
    "nutrition_checkin": "nutrition",
    "workout_record": "workout",
    "safety_audit": "safety_audit",
    "check_in_validated": "legacy_combined",
    "history_imported": "history_import",
}


def _reconciliation_registry_document(
    registry: object,
    live: RegistryDocument,
) -> RegistryDocument:
    if isinstance(registry, RegistryDocument):
        candidate = registry
    elif isinstance(registry, CustomerRegistry):
        payload = {
            "version": live.version,
            "owner": registry.owner.model_dump(mode="json"),
            "customers": [
                runtime.spec.model_dump(mode="json") for runtime in registry.customers
            ],
        }
        candidate = RegistryDocument.model_validate(payload)
    else:
        model_dump = getattr(registry, "model_dump", None)
        raw = model_dump(mode="json") if callable(model_dump) else registry
        if not isinstance(raw, Mapping):
            raise CustomerAdminError("registry evidence is invalid")
        candidate = RegistryDocument.model_validate(dict(raw))
    if _document_fingerprint(candidate) != _document_fingerprint(live):
        raise CustomerAdminError("registry evidence is stale")
    return candidate


def _reconciliation_source_records(
    canonical_events: object,
    expected_path: Path,
    *,
    transaction: CanonicalEventTransaction,
) -> tuple[dict[str, object], ...]:
    if not isinstance(transaction, CanonicalEventTransaction):
        raise CustomerAdminError("canonical transaction evidence is invalid")
    if (
        transaction.events_path.is_symlink()
        or transaction.events_path.resolve() != expected_path.resolve()
    ):
        raise CustomerAdminError("canonical EventStore path does not match customer")
    source_path: Path | None = None
    source_values: object = canonical_events
    if isinstance(canonical_events, (str, Path)):
        source_path = Path(canonical_events)
        source_values = None
    elif isinstance(canonical_events, EventStore):
        source_path = Path(getattr(canonical_events, "_events", expected_path))
        source_values = None
    else:
        candidate_path = getattr(canonical_events, "_events", None)
        if candidate_path is not None:
            source_path = Path(candidate_path)
            source_values = None
    if source_path is not None:
        if source_path.is_symlink() or source_path.resolve() != expected_path.resolve():
            raise CustomerAdminError(
                "canonical EventStore path does not match customer"
            )
    try:
        persisted = tuple(
            event.model_dump(mode="json", exclude_none=True)
            for event in transaction.read_snapshot().events
        )
    except (OSError, TypeError, ValueError) as exc:
        raise CustomerAdminError("canonical event evidence is invalid") from exc
    if source_values is None:
        supplied = persisted
    elif isinstance(source_values, Mapping):
        supplied = (dict(source_values),)
    else:
        try:
            supplied = tuple(source_values)  # type: ignore[arg-type]
        except TypeError as exc:
            raise CustomerAdminError("canonical event evidence is invalid") from exc
    try:
        supplied_records = canonical_event_records(supplied)
        persisted_records = canonical_event_records(persisted)
    except (TypeError, ValueError) as exc:
        raise CustomerAdminError("canonical event evidence is invalid") from exc
    if tuple(canonical_json(item) for item in supplied_records) != tuple(
        canonical_json(item) for item in persisted_records
    ):
        raise CustomerAdminError("canonical event evidence is stale")
    return tuple(dict(item) for item in persisted_records)


def _reconciliation_event_plan(
    record: Mapping[str, object],
    customer_key: str,
    epoch: int,
) -> dict[str, object]:
    event_id = record.get("event_id")
    event_type = record.get("event_type")
    occurred = record.get("occurred_at_kst")
    if (
        not isinstance(event_id, str)
        or not event_id
        or not isinstance(event_type, str)
        or not event_type
        or not isinstance(occurred, str)
        or not occurred
    ):
        raise CustomerAdminError("canonical event evidence is invalid")
    provenance = record.get("provenance")
    source_ref: object = None
    if isinstance(provenance, Mapping):
        source_ref = provenance.get("source_ref")
    if source_ref is not None and not isinstance(source_ref, str):
        raise CustomerAdminError("canonical event provenance is invalid")
    source_ref_value = source_ref.strip() if isinstance(source_ref, str) else ""
    source_parts = source_ref_value.split(":")
    if (
        len(source_parts) >= 2
        and source_parts[0] in {"pilot", "customer"}
        and source_parts[1] != customer_key
    ):
        raise CustomerAdminError("canonical event customer scope is invalid")
    if record.get("customer_key") not in (None, customer_key):
        raise CustomerAdminError("canonical event customer scope is invalid")
    for payload_name in (
        "check_in",
        "payment",
        "satisfaction",
        "operator_time",
        "draft",
    ):
        payload = record.get(payload_name)
        if isinstance(payload, Mapping):
            payload_customer = payload.get("customer_key")
            if payload_customer is not None and payload_customer != customer_key:
                raise CustomerAdminError("canonical event customer scope is invalid")
    try:
        timestamp = datetime.fromisoformat(occurred)
        if timestamp.tzinfo is None:
            timestamp = timestamp.replace(tzinfo=_KST)
        observation_day = timestamp.astimezone(_KST).date()
    except (TypeError, ValueError) as exc:
        raise CustomerAdminError("canonical event timestamp is invalid") from exc
    session_id = source_ref_value or str(record.get("dedupe_key") or event_id)
    if not session_id.strip():
        raise CustomerAdminError("canonical event session identity is invalid")
    return {
        "root_event_id": event_id,
        "customer_key": customer_key,
        "mapped_flow": _RECON_EVENT_FLOWS.get(event_type, event_type),
        "observation_kst_day": observation_day,
        "session_id": session_id,
        "writer_epoch": epoch,
        "root_preimage_digest": adaptive_digest(record),
    }


def _reconciliation_feature_config(path: Path) -> tuple[int, str, dict[str, bool]]:
    if path.is_symlink() or not path.exists() or not path.is_file():
        raise CustomerAdminError("feature config is unavailable")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CustomerAdminError("feature config is invalid") from exc
    if not isinstance(value, Mapping):
        raise CustomerAdminError("feature config is invalid")
    flags = {name: value.get(name) for name in _RECON_FEATURE_FLAGS}
    if (
        set(value)
        != {"schema_version", "epoch", "config_digest", *_RECON_FEATURE_FLAGS}
        or value.get("schema_version") != "1.0"
        or type(value.get("epoch")) is not int
        or value["epoch"] < 0
        or any(type(flag) is not bool for flag in flags.values())
        or not isinstance(value.get("config_digest"), str)
        or len(value["config_digest"]) != 64
        or any(
            character not in "0123456789abcdef" for character in value["config_digest"]
        )
    ):
        raise CustomerAdminError("feature config is invalid")
    expected = feature_config_digest(value["epoch"], flags)
    if value["config_digest"] != expected:
        raise CustomerAdminError("feature config digest mismatch")
    return value["epoch"], value["config_digest"], flags


def _reconciliation_journal_digest(rows: object) -> str:
    return adaptive_digest(tuple(rows))


def _reconciliation_owner_authority(
    document: RegistryDocument,
    customer_key: str,
    *,
    activation_receipt_digest: str,
    consent_digest: str,
) -> tuple[dict[str, str], str]:
    for label, value in (
        ("activation receipt", activation_receipt_digest),
        ("consent", consent_digest),
    ):
        if (
            not isinstance(value, str)
            or len(value) != 64
            or any(character not in "0123456789abcdef" for character in value)
        ):
            raise CustomerAdminError(f"{label} evidence is invalid")
    owner = document.owner.model_dump(mode="json")
    if set(owner) != {"user_id", "chat_id", "topic_id"} or any(
        type(value) is not str or not value.strip() for value in owner.values()
    ):
        raise CustomerAdminError("owner authority evidence is incomplete")
    authority = {
        "schema_version": "1.0",
        "customer_key": customer_key,
        "owner": dict(owner),
        "registry_digest": _document_fingerprint(document),
        "activation_receipt_digest": activation_receipt_digest,
        "consent_digest": consent_digest,
    }
    return dict(owner), adaptive_digest(authority)


def reconcile_adaptive_nutrition_journals(
    profile_root: Path,
    customer_key: str,
    *,
    canonical_events: object,
    registry: object,
) -> Mapping[str, object]:
    root = _resolve_profile_root(Path(profile_root))
    with profile_authority_lock(root):
        return _reconcile_adaptive_nutrition_journals_locked(
            root,
            customer_key,
            canonical_events=canonical_events,
            registry=registry,
        )


def _reconcile_adaptive_nutrition_journals_locked(
    profile_root: Path,
    customer_key: str,
    *,
    canonical_events: object,
    registry: object,
) -> Mapping[str, object]:
    """Reconcile adaptive projections from one live canonical customer boundary."""

    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    registry_path = _resolve_registry_path(root)
    _recover_activation_journal(root, registry_path)
    live_document = _read_profile_registry(registry_path, root)
    _validate_enabled_activation_receipts(root, registry_path, live_document)
    evidence = _reconciliation_registry_document(registry, live_document)
    selected = next(
        (item for item in evidence.customers if item.customer_key == key),
        None,
    )
    enabled_keys = tuple(
        sorted(item.customer_key for item in evidence.customers if item.enabled)
    )
    if selected is None or not selected.enabled or key not in enabled_keys:
        raise CustomerAdminError("customer must be enabled for adaptive reconciliation")
    try:
        live_registry = load_customer_registry(registry_path, root)
    except (CustomerRegistryError, OSError, ValueError) as exc:
        raise CustomerAdminError("registry failed adaptive reconciliation") from exc
    runtime = next(
        (item for item in live_registry.customers if item.spec.customer_key == key),
        None,
    )
    if runtime is None or runtime.data_root.is_symlink():
        raise CustomerAdminError("registered customer data root is invalid")
    customer_root = runtime.data_root.resolve()
    if (
        not customer_root.exists()
        or not customer_root.is_dir()
        or not customer_root.is_relative_to(root)
        or _path_has_symlink(customer_root, root)
    ):
        raise CustomerAdminError("registered customer data root is invalid")
    canonical_path = customer_root / "wizard" / "events.jsonl"
    canonical_transaction = CanonicalEventTransaction.for_customer_runtime(runtime)
    activation_receipt_id = _require_committed_activation_receipt(
        root,
        registry_path,
        live_document,
        selected,
    )
    activation_receipt_digest = adaptive_digest(
        {
            "activation_receipt_id": activation_receipt_id,
        }
    )
    consent_digest = adaptive_digest(
        selected.ai_processing_consent.model_dump(mode="json")
    )
    records = _reconciliation_source_records(
        canonical_events,
        canonical_path,
        transaction=canonical_transaction,
    )
    owner, authority_fact_digest = _reconciliation_owner_authority(
        evidence,
        key,
        activation_receipt_digest=activation_receipt_digest,
        consent_digest=consent_digest,
    )
    event_plans = tuple(
        _reconciliation_event_plan(record, key, 0) for record in records
    )
    adaptive_root = customer_root / "nutrition-plans"
    if adaptive_root.is_symlink() or (
        adaptive_root.exists() and not adaptive_root.is_dir()
    ):
        raise CustomerAdminError("adaptive runtime is invalid")
    try:
        _load_adaptive_registration_inputs(root, key)
    except CustomerAdminError as exc:
        raise CustomerAdminError(
            "adaptive reconciliation requires approved complete customer inputs"
        ) from exc
    feature_path = adaptive_root / "feature-epoch.json"
    feature_exists = feature_path.exists()
    if feature_exists:
        _reconciliation_feature_config(feature_path)
    store = AdaptiveEventStore.for_registered(runtime)
    if adaptive_root.exists():
        try:
            store.read()
            store.source_day_rows()
            for kind in ("source_day", "authority", "config_epoch"):
                store.journal_rows(kind)
        except (OSError, UnicodeDecodeError, ValueError) as exc:
            raise CustomerAdminError("adaptive journals failed closed") from exc
    initialize_adaptive_customer(customer_root)
    epoch, config_digest, _ = _reconciliation_feature_config(feature_path)
    event_plans = tuple({**plan, "writer_epoch": epoch} for plan in event_plans)
    expected_config_states = {
        "prepared": {key: "pending" for key in enabled_keys},
        "committed": {key: "committed" for key in enabled_keys},
    }
    for row in store.journal_rows("config_epoch"):
        if row.get("intent_id") != f"epoch:{epoch}":
            continue
        expected_state = row.get("state")
        if expected_state in expected_config_states and (
            row.get("epoch") != epoch
            or row.get("config_digest") != config_digest
            or tuple(row.get("customer_keys", ())) != enabled_keys
            or row.get("customer_state") != expected_config_states[expected_state]
            or row.get("approved_by") != owner
        ):
            raise CustomerAdminError("config epoch does not match enabled registry")
    try:
        canonical_transaction.recover()
        sequence_rows = canonical_transaction.read_snapshot().sequence_rows
        mapping_rows = store.source_day_rows(recover=True)
        source_intent_rows = store.journal_rows("source_day", recover=True)
        authority_rows = store.journal_rows("authority", recover=True)
        config_rows = store.journal_rows("config_epoch", recover=True)
    except (OSError, UnicodeDecodeError, ValueError) as exc:
        raise CustomerAdminError("adaptive journal recovery failed") from exc

    mappings_by_event = {str(row["root_event_id"]): row for row in mapping_rows}
    for plan in event_plans:
        existing = mappings_by_event.get(str(plan["root_event_id"]))
        mapping_epoch = (
            existing.get("writer_epoch")
            if isinstance(existing, Mapping)
            and type(existing.get("writer_epoch")) is int
            and existing["writer_epoch"] >= 0
            else plan["writer_epoch"]
        )
        mapping = store.append_source_day_mapping(
            root_event_id=str(plan["root_event_id"]),
            customer_key=str(plan["customer_key"]),
            mapped_flow=str(plan["mapped_flow"]),
            observation_kst_day=plan["observation_kst_day"],
            session_id=str(plan["session_id"]),
            writer_epoch=mapping_epoch,
            root_preimage_digest=str(plan["root_preimage_digest"]),
        )
        mappings_by_event[str(plan["root_event_id"])] = mapping
        intent_id = f"source-day:{mapping['mapping_id']}"
        intent_payload = {
            key: mapping[key]
            for key in (
                "mapping_id",
                "root_event_id",
                "customer_key",
                "mapped_flow",
                "observation_kst_day",
                "session_id",
                "writer_epoch",
                "root_preimage_digest",
            )
        }
        existing_intents = [
            row for row in source_intent_rows if row.get("intent_id") == intent_id
        ]
        for row in existing_intents:
            if row.get("state") not in {"committed", "abandoned"}:
                continue
            if any(row.get(field) != intent_payload[field] for field in intent_payload):
                raise CustomerAdminError("source-day intent does not match mapping")
        if not any(
            row.get("state") in {"committed", "abandoned"} for row in existing_intents
        ):
            store.prepare_source_day(intent_id, **intent_payload)
            store.commit_source_day(intent_id, **intent_payload)
        source_intent_rows = store.journal_rows("source_day")

    sequence_rows = canonical_transaction.read_snapshot().sequence_rows
    authority_intent_id = f"authority:{key}:{authority_fact_digest}"
    authority_payload = {
        "authority_kind": "owner",
        "canonical_fact_id": f"owner-authority:{key}",
        "canonical_fact_digest": authority_fact_digest,
        "valid_from": datetime.combine(
            selected.plan.starts_on,
            time.min,
            tzinfo=_KST,
        ).isoformat(),
        "adaptive_sequence": (sequence_rows[-1]["sequence"] if sequence_rows else 0),
        "customer_key": key,
        "owner": owner,
        "registry_digest": _document_fingerprint(evidence),
        "activation_receipt_digest": activation_receipt_digest,
        "consent_digest": consent_digest,
    }
    authority_intents = [
        row for row in authority_rows if row.get("intent_id") == authority_intent_id
    ]
    for row in authority_intents:
        if row.get("state") != "committed":
            continue
        if any(
            row.get(field) != authority_payload[field]
            for field in authority_payload
            if field != "adaptive_sequence"
        ):
            raise CustomerAdminError("authority mirror does not match owner evidence")
    if not any(
        row.get("state") in {"committed", "abandoned"} for row in authority_intents
    ):
        store.append_authority_mirror(
            intent_id=authority_intent_id,
            authority_kind=str(authority_payload["authority_kind"]),
            canonical_fact_id=str(authority_payload["canonical_fact_id"]),
            canonical_fact_digest=authority_fact_digest,
            valid_from=str(authority_payload["valid_from"]),
            adaptive_sequence=int(authority_payload["adaptive_sequence"]),
            state="prepared",
            customer_key=key,
            owner=owner,
            registry_digest=str(authority_payload["registry_digest"]),
            activation_receipt_digest=str(
                authority_payload["activation_receipt_digest"]
            ),
            consent_digest=str(authority_payload["consent_digest"]),
        )
        store.append_journal(
            "authority",
            authority_payload,
            intent_id=authority_intent_id,
            state="committed",
        )

    config_intent_id = f"epoch:{epoch}"
    config_rows = store.journal_rows("config_epoch")
    config_for_epoch = [
        row for row in config_rows if row.get("intent_id") == config_intent_id
    ]
    for row in config_for_epoch:
        if row.get("state") != "committed":
            continue
        observed_keys = tuple(row.get("customer_keys", ()))
        observed_states = row.get("customer_state")
        if (
            row.get("epoch") != epoch
            or row.get("config_digest") != config_digest
            or observed_keys != enabled_keys
            or observed_states != {key: "committed" for key in enabled_keys}
            or row.get("approved_by") != owner
        ):
            raise CustomerAdminError("config epoch does not match enabled registry")
    if not any(row.get("state") == "committed" for row in config_for_epoch):
        store.append_config_epoch(
            epoch,
            config_digest,
            enabled_keys,
            state="prepared",
            customer_states={key: "pending" for key in enabled_keys},
            approved_by=owner,
        )
        store.append_config_epoch(
            epoch,
            config_digest,
            enabled_keys,
            state="committed",
            customer_states={key: "committed" for key in enabled_keys},
            approved_by=owner,
        )
    config_rows = store.journal_rows("config_epoch")
    mapping_rows = store.source_day_rows()
    authority_rows = store.journal_rows("authority")
    sequence_rows = canonical_transaction.read_snapshot().sequence_rows
    source_intent_rows = store.journal_rows("source_day")
    return {
        "schema_version": "1.0",
        "customer_key": key,
        "reconciled_at_kst": datetime.now(_KST).isoformat(),
        "epoch": epoch,
        "config_digest": config_digest,
        "canonical_sequence": {
            "count": len(sequence_rows),
            "digest": _reconciliation_journal_digest(sequence_rows),
        },
        "source_day_mappings": {
            "count": len(mapping_rows),
            "digest": _reconciliation_journal_digest(mapping_rows),
        },
        "source_day_intents": {
            "count": len(source_intent_rows),
            "digest": _reconciliation_journal_digest(source_intent_rows),
        },
        "authority_mirror": {
            "count": len(authority_rows),
            "digest": _reconciliation_journal_digest(authority_rows),
        },
        "config_epoch": {
            "count": len(config_rows),
            "digest": _reconciliation_journal_digest(config_rows),
        },
    }


def prepare_adaptive_nutrition_runtime(
    profile_root: Path,
    customer_key: str,
    *,
    extension_through: date | None = None,
) -> Mapping[str, str]:
    root = _resolve_profile_root(Path(profile_root))
    with profile_authority_lock(root):
        return _prepare_adaptive_nutrition_runtime_locked(
            root,
            customer_key,
            extension_through=extension_through,
        )


def _prepare_adaptive_nutrition_runtime_locked(
    profile_root: Path,
    customer_key: str,
    *,
    extension_through: date | None = None,
) -> Mapping[str, str]:
    """Prepare private adaptive files and append an approved customer policy extension."""
    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    registry_path = _resolve_registry_path(root)
    _recover_activation_journal(root, registry_path)
    document = _read_profile_registry(registry_path, root)
    _validate_enabled_activation_receipts(root, registry_path, document)
    registry = load_runtime_customer_registry(root)
    runtime = next(
        (item for item in registry.customers if item.spec.customer_key == key),
        None,
    )
    if runtime is None or runtime.spec.enabled is not True:
        raise CustomerAdminError(
            "customer must be enabled for adaptive runtime preparation"
        )
    if extension_through is not None:
        if type(extension_through) is not date:
            raise CustomerAdminError("adaptive extension date is invalid")
        starts_on = runtime.spec.plan.starts_on
        minimum = starts_on + timedelta(days=28)
        maximum = starts_on + timedelta(days=83)
        if not minimum <= extension_through <= maximum:
            if extension_through == starts_on + timedelta(days=84):
                raise CustomerAdminError(
                    "adaptive extension D+85 is not allowed; valid window is D+29..D+84"
                )
            raise CustomerAdminError("adaptive extension must remain within D+29..D+84")
    activation_receipt_id: str | None = None
    authority: dict[str, object] | None = None
    approved_at_kst: str | None = None
    if extension_through is not None:
        activation_receipt_id = _require_committed_activation_receipt(
            root,
            registry_path,
            document,
            runtime.spec,
        )
        actor = {
            "user_id": document.owner.user_id,
            "chat_id": document.owner.chat_id,
            "topic_id": document.owner.topic_id,
        }
        authority = {
            "schema_version": "1.0",
            "customer_key": key,
            "extension_through": extension_through.isoformat(),
            "activation_receipt_id": activation_receipt_id,
            "actor": actor,
        }
        approved_at_kst = datetime.now(_KST).isoformat()
    try:
        result = initialize_adaptive_customer(runtime.data_root)
        if extension_through is None:
            return result
        assert activation_receipt_id is not None
        assert authority is not None
        assert approved_at_kst is not None
        extension_result = append_approved_policy_extension(
            runtime.data_root,
            customer_key=key,
            extension_through=extension_through,
            activation_receipt_id=activation_receipt_id,
            authority_digest=adaptive_digest(authority),
            authority=authority,
            approved_by=document.owner.user_id,
            approved_at_kst=approved_at_kst,
        )
        return {**result, **extension_result}
    except CustomerAdminError:
        raise
    except (OSError, TypeError, ValueError) as exc:
        raise CustomerAdminError("adaptive runtime preparation failed") from exc


def _validate_activation_data_root(
    profile_root: Path,
    requested_root: Path,
    registered_root: Path,
    document: RegistryDocument,
    spec: CustomerSpec,
) -> None:
    if requested_root.is_symlink():
        raise CustomerAdminError("G1 data root symlinks are not allowed")
    normalized = requested_root.resolve()
    expected = registered_root.resolve()
    if (
        not normalized.exists()
        or not normalized.is_dir()
        or not normalized.is_relative_to(profile_root)
        or normalized != expected
    ):
        raise CustomerAdminError(
            "G1 data root does not match the registered customer root"
        )
    if _path_has_symlink(normalized, profile_root):
        raise CustomerAdminError("G1 data root symlinks are not allowed")
    _audit_data_root(normalized, document, spec)


def _path_has_symlink(path: Path, profile_root: Path) -> bool:
    if profile_root.is_symlink() or not path.is_relative_to(profile_root):
        return True
    current = profile_root
    for part in path.relative_to(profile_root).parts:
        current /= part
        if current.is_symlink():
            return True
    return False


def _audit_data_root(
    data_root: Path,
    document: RegistryDocument,
    spec: CustomerSpec,
) -> None:
    forbidden: set[str] = set()

    def add_identity(
        address: TelegramAddress,
        *,
        allowed_components: frozenset[str] = frozenset(),
    ) -> None:
        for value in address.key:
            if len(value) >= 4 and value not in allowed_components:
                forbidden.add(value)
        forbidden.add("|".join(address.key))

    current_role_components = frozenset(spec.telegram.key)
    add_identity(document.owner, allowed_components=current_role_components)
    for customer in document.customers:
        if customer.customer_key == spec.customer_key:
            continue
        forbidden.add(customer.customer_key)
        if len(customer.display_name) >= 4:
            forbidden.add(customer.display_name)
        add_identity(
            customer.telegram,
            allowed_components=current_role_components,
        )

    historical_classifier = _activation_historical_artifact_classifier.get()
    for path in data_root.rglob("*"):
        if path.is_symlink():
            raise CustomerAdminError("G1 data root contains a symlink")
        if path.is_dir():
            continue
        relative = path.relative_to(data_root).as_posix().lower()
        has_personal_path_marker = any(
            marker in relative for marker in _PERSONAL_PATH_MARKERS
        )
        try:
            content = path.read_bytes()
        except OSError as exc:
            raise CustomerAdminError("G1 data root could not be audited") from exc
        has_forbidden_identifier = any(
            identifier.encode("utf-8") in content for identifier in forbidden
        )
        if not has_personal_path_marker and not has_forbidden_identifier:
            continue
        try:
            authenticated_history = (
                historical_classifier is not None
                and historical_classifier(path, content) is True
            )
        except Exception:
            # An authenticated exception must fail closed; ordinary G1 checks
            # remain authoritative when its verifier is unavailable or corrupt.
            authenticated_history = False
        if authenticated_history:
            continue
        if has_personal_path_marker:
            raise CustomerAdminError("G1 data root contains a personal-flow identifier")
        raise CustomerAdminError(
            "G1 data root contains a foreign or personal identifier"
        )


def _authenticated_onboarding_artifact_classifier(
    data_root: Path,
    readiness_digests: Mapping[str, str],
) -> _HistoricalActivationArtifactClassifier:
    onboarding_root = data_root / "nutrition-onboarding"
    baseline_path = onboarding_root / "baseline-v1.json"
    ready_path = onboarding_root / "ready.json"
    try:
        baseline_bytes = baseline_path.read_bytes()
        baseline = json.loads(baseline_bytes)
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CustomerAdminError(
            "authenticated onboarding artifacts are unavailable",
        ) from exc
    if not isinstance(baseline, dict):
        raise CustomerAdminError(
            "authenticated onboarding artifacts are stale",
        )
    receipt = baseline.get("owner_risk_acceptance_receipt")
    if receipt is None:
        return lambda _path, _content: False
    try:
        ready_bytes = ready_path.read_bytes()
        ready = json.loads(ready_bytes)
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CustomerAdminError(
            "authenticated onboarding artifacts are unavailable",
        ) from exc
    if (
        not isinstance(ready, dict)
        or
        baseline.get("digest") != readiness_digests.get("baseline")
        or ready.get("state") != "ready"
        or ready.get("baseline_digest")
        != readiness_digests.get("baseline")
        or ready.get("readiness_pointer_digest")
        != readiness_digests.get("pointer")
        or (
            receipt is not None
            and ready.get("owner_risk_acceptance_receipts")
            != [receipt]
        )
    ):
        raise CustomerAdminError(
            "authenticated onboarding artifacts are stale",
        )
    allowed = {
        baseline_path.resolve(): baseline_bytes,
        ready_path.resolve(): ready_bytes,
    }

    def classify(path: Path, content: bytes) -> bool:
        try:
            expected = allowed.get(path.resolve())
        except (OSError, RuntimeError):
            return False
        return expected is not None and content == expected

    return classify


def _read_activation_checklist(
    path: Path,
    profile_root: Path,
    data_root: Path,
    registry_path: Path,
    spec: CustomerSpec,
) -> dict[str, object]:
    if path.is_symlink() or not path.exists() or not path.is_file():
        raise CustomerAdminError("G5 checklist evidence must be a regular file")
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CustomerAdminError("G5 checklist evidence is not valid JSON") from exc
    if not isinstance(payload, dict):
        raise CustomerAdminError("G5 checklist evidence must be a JSON object")
    evidence: object = payload.get("checklist", payload.get("evidence", payload))
    if not isinstance(evidence, dict):
        raise CustomerAdminError("G5 checklist evidence object is malformed")

    for key, expected in (
        ("customer_id", spec.customer_key),
        ("customer_key", spec.customer_key),
        ("profile_root", profile_root),
        ("data_root", data_root),
        ("registry_path", registry_path),
    ):
        if key not in payload:
            continue
        actual = payload[key]
        if key in {"profile_root", "data_root", "registry_path"}:
            try:
                if Path(str(actual)).resolve() != Path(expected).resolve():
                    raise CustomerAdminError(
                        f"G5 checklist {key} does not match activation root"
                    )
            except (OSError, RuntimeError) as exc:
                raise CustomerAdminError(f"G5 checklist {key} is invalid") from exc
        elif actual != expected:
            raise CustomerAdminError(f"G5 checklist {key} does not match customer")

    for key in _REQUIRED_CHECKLIST_ITEMS:
        value = evidence.get(key)
        if key == "token_rotated":
            try:
                validate_token_rotation_policy(
                    evidence,
                    payload,
                    ActivationChecklistBindings(
                        customer_key=spec.customer_key,
                        profile_root=profile_root,
                        data_root=data_root,
                        registry_path=registry_path,
                    ),
                )
            except TokenRotationPolicyError as exc:
                raise CustomerAdminError(f"G5 {exc}") from exc
            continue
        if key == "provider_terms_checked":
            if not isinstance(value, dict):
                raise CustomerAdminError(
                    "G3 provider terms check uses the wrong consent version"
                )
            continue
        if value is not True:
            raise CustomerAdminError(f"G5 checklist item is not true: {key}")
    terms = evidence["provider_terms_checked"]
    if not any(
        terms.get(name) is True
        for name in ("checked", "passed", "valid", "value", "recorded")
    ):
        raise CustomerAdminError("G3 provider terms check is not recorded")
    versions = tuple(
        terms[name]
        for name in ("version", "notice_version", "consent_version")
        if name in terms
    )
    if not versions or any(version != CONSENT_VERSION for version in versions):
        raise CustomerAdminError(
            "G3 provider terms check uses the wrong consent version"
        )
    for key in ("consent_version", "notice_version"):
        if key in payload and payload[key] != CONSENT_VERSION:
            raise CustomerAdminError("G3 checklist consent version is not current")
    return payload


def _append_activation_audit(
    profile_root: Path,
    registry_path: Path,
    data_root: Path,
    customer_id: str,
    checklist_path: Path,
    *,
    transaction_id: str | None = None,
    registry_sha256: str | None = None,
    recorded_at: str | None = None,
    nutrition_activation_receipt: Mapping[str, str] | None = None,
    membership_bindings: Mapping[str, object] | None = None,
) -> Path:
    path = _activation_audit_path(profile_root)
    if path.is_symlink():
        raise CustomerAdminError("activation audit path symlinks are not allowed")
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    path.parent.chmod(0o700)
    try:
        previous = path.read_bytes() if path.exists() else b""
        record = _build_activation_audit_record(
            registry_path,
            data_root,
            customer_id,
            checklist_path,
            transaction_id=transaction_id,
            registry_sha256=registry_sha256,
            recorded_at=recorded_at,
            nutrition_activation_receipt=nutrition_activation_receipt,
            membership_bindings=membership_bindings,
        )
        content = previous + (
            json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n"
        ).encode("utf-8")
        _atomic_write_bytes(path, content)
    except (OSError, CustomerAdminError) as exc:
        raise CustomerAdminError("activation audit could not be written") from exc
    return path


def _read(path: Path) -> RegistryDocument:
    if path.is_symlink():
        raise CustomerAdminError("registry symlinks are not allowed")
    return RegistryDocument.model_validate_json(path.read_text(encoding="utf-8"))


def _write(path: Path, document: RegistryDocument) -> None:
    content = (document.model_dump_json(indent=2) + "\n").encode("utf-8")
    _atomic_write_bytes(path, content)


_GATE_PREFLIGHT_SCHEMA = "gate_d_preflight_v1"
_GATE_PREFLIGHT_MAX_COUNT = 1_000_000
_GATE_PREFLIGHT_MAX_EPOCH = 2**63 - 1
_GATE_PREFLIGHT_MAX_REASONS = 64
_GATE_PREFLIGHT_DIGEST = re.compile(r"^[0-9a-f]{64}$")
_GATE_PREFLIGHT_CHECK_KEYS = frozenset(
    {
        "profile_contained",
        "private_modes",
        "enabled_isolation",
        "identity_roles_distinct",
        "review_operator",
        "separate_bot",
        "current_kst_window",
        "activation_receipt",
        "approved_registration",
        "approved_artifacts",
        "canonical_reconciliation",
        "feature_flags_disabled",
        "delivery_disabled",
        "schedule_fence",
        "schedule_ledger",
        "schedule_tombstones",
        "schedule_transitions",
    }
)
_GATE_PREFLIGHT_COUNT_KEYS = frozenset(
    {
        "enabled_customers",
        "canonical_events",
        "canonical_sequences",
        "source_day_mappings",
        "source_day_intents",
        "authority_rows",
        "config_epoch_rows",
        "approved_artifacts",
        "schedule_rows",
        "schedule_tombstones",
    }
)
_GATE_PREFLIGHT_DIGEST_KEYS = frozenset(
    {
        "registry",
        "config",
        "activation_receipt",
        "feature_epoch",
        "feature_config",
        "registration",
        "artifact_base_policy",
        "artifact_meal_constraints",
        "artifact_catalog",
        "canonical_events",
        "canonical_sequences",
        "source_day_mappings",
        "source_day_intents",
        "authority_rows",
        "config_epoch_rows",
        "canonical_prefix",
        "canonical_reconciliation",
        "schedule_ledger",
        "schedule_tombstones",
    }
)
_GATE_PREFLIGHT_REASON_CODES = (
    frozenset(
        {
            "schedule_missing",
            "private_modes_invalid",
            "schedule_fence_invalid",
            "schedule_fence_missing",
            "schedule_fence_not_ready",
            "schedule_ledger_invalid",
            "schedule_transition_evidence_invalid",
            "schedule_transition_evidence_missing",
            "schedule_tombstone_invalid",
            "schedule_tombstone_missing",
            "schedule_tombstone_digest_mismatch",
            "schedule_legacy_claim_stale",
            "schedule_tombstone_reservation_mismatch",
            "schedule_tombstone_unpaired",
            "identity_roles_not_distinct",
            "customer_not_enabled",
            "main_profile_containment",
            "config_missing",
            "config_invalid",
            "review_space_invalid",
            "delivery_enabled",
            "feature_flags_enabled",
            "review_operator_missing",
            "review_operator_invalid",
            "review_space_collision",
            "review_space_collision_customer",
            "review_space_collision_owner_scheduled",
            "review_space_collision_generic",
            "separate_bot_invalid",
            "separate_bot_missing",
            "activation_receipt_missing",
            "plan_window_stale",
            "canonical_reconciliation_missing",
            "canonical_reconciliation_stale",
            "registration_missing",
            "artifacts_unapproved",
            "artifacts_missing",
        }
    )
    | NUTRITION_READINESS_REASON_CODES
)


def _canonical_preflight_timestamp(value: str) -> str:
    if type(value) is not str or len(value) > 32:
        raise ValueError("checked_at_kst must be a bounded timestamp")
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError as exc:
        raise ValueError("checked_at_kst must be an ISO timestamp") from exc
    if parsed.tzinfo is None or parsed.utcoffset() != timedelta(hours=9):
        raise ValueError("checked_at_kst must use the KST offset")
    canonical = parsed.astimezone(_KST).isoformat(timespec="seconds")
    if value != canonical:
        raise ValueError("checked_at_kst must be canonical")
    return canonical


@dataclass(frozen=True, slots=True)
class GateDPreflightReceipt(Mapping[str, object]):
    """Bounded, read-only Gate-D readiness evidence.

    The receipt deliberately contains no Telegram addresses, credentials, raw
    customer values, or filesystem paths. A false receipt is a normal
    preflight result; corrupt profile input and unsafe paths raise before a
    receipt can be trusted.
    """

    schema_version: str
    ready: bool
    checks: Mapping[str, bool]
    counts: Mapping[str, int]
    digests: Mapping[str, str]
    epoch: int | None
    checked_at_kst: str
    reason_codes: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        if self.schema_version != _GATE_PREFLIGHT_SCHEMA:
            raise ValueError("unsupported Gate-D preflight schema")
        if type(self.ready) is not bool:
            raise ValueError("ready must be a boolean")
        if not isinstance(self.checks, Mapping):
            raise ValueError("checks must be a mapping")
        checks = dict(self.checks)
        if set(checks) != _GATE_PREFLIGHT_CHECK_KEYS or any(
            type(value) is not bool for value in checks.values()
        ):
            raise ValueError("checks are not bounded")
        if not isinstance(self.counts, Mapping):
            raise ValueError("counts must be a mapping")
        counts = dict(self.counts)
        if set(counts) != _GATE_PREFLIGHT_COUNT_KEYS or any(
            type(value) is not int or value < 0 or value > _GATE_PREFLIGHT_MAX_COUNT
            for value in counts.values()
        ):
            raise ValueError("counts are not bounded")
        if not isinstance(self.digests, Mapping):
            raise ValueError("digests must be a mapping")
        digests = dict(self.digests)
        if set(digests) - _GATE_PREFLIGHT_DIGEST_KEYS or any(
            type(value) is not str or _GATE_PREFLIGHT_DIGEST.fullmatch(value) is None
            for value in digests.values()
        ):
            raise ValueError("digests are not canonical")
        if self.epoch is not None and (
            type(self.epoch) is not int
            or self.epoch < 0
            or self.epoch > _GATE_PREFLIGHT_MAX_EPOCH
        ):
            raise ValueError("epoch is not bounded")
        reasons = tuple(self.reason_codes)
        if (
            len(reasons) > _GATE_PREFLIGHT_MAX_REASONS
            or len(set(reasons)) != len(reasons)
            or any(
                type(reason) is not str or reason not in _GATE_PREFLIGHT_REASON_CODES
                for reason in reasons
            )
        ):
            raise ValueError("reason codes are not bounded")
        object.__setattr__(self, "checks", MappingProxyType(checks))
        object.__setattr__(self, "counts", MappingProxyType(counts))
        object.__setattr__(self, "digests", MappingProxyType(digests))
        object.__setattr__(self, "reason_codes", tuple(sorted(reasons)))
        object.__setattr__(
            self,
            "checked_at_kst",
            _canonical_preflight_timestamp(self.checked_at_kst),
        )

    def __getitem__(self, key: str) -> object:
        return self.to_dict()[key]

    def __iter__(self) -> Iterator[str]:
        return iter(
            (
                "schema_version",
                "ready",
                "checks",
                "counts",
                "digests",
                "epoch",
                "reason_codes",
                "checked_at_kst",
            )
        )

    def __len__(self) -> int:
        return 8

    def __bool__(self) -> bool:
        return self.ready

    @property
    def passed(self) -> bool:
        return self.ready

    @property
    def failure_reasons(self) -> tuple[str, ...]:
        return self.reason_codes

    def to_dict(self) -> dict[str, object]:
        return {
            "schema_version": self.schema_version,
            "ready": self.ready,
            "checks": dict(self.checks),
            "counts": dict(self.counts),
            "digests": dict(self.digests),
            "epoch": self.epoch,
            "reason_codes": list(self.reason_codes),
            "checked_at_kst": self.checked_at_kst,
        }


def _preflight_digest_bytes(path: Path) -> str:
    try:
        return hashlib.sha256(path.read_bytes()).hexdigest()
    except (OSError, UnicodeDecodeError) as exc:
        raise CustomerAdminError("Gate-D preflight evidence is unavailable") from exc


def _preflight_private(path: Path, *, directory: bool) -> bool:
    if path.is_symlink() or not path.exists():
        return False
    if directory and not path.is_dir():
        return False
    if not directory and not path.is_file():
        return False
    try:
        return path.stat().st_mode & 0o077 == 0
    except OSError:
        return False


def _preflight_tree_private(path: Path) -> bool:
    """Check every descendant for symlinks and group/other permissions."""
    try:
        entries = tuple(path.rglob("*"))
    except OSError:
        return False
    for entry in entries:
        if entry.is_symlink():
            return False
        if entry.is_dir():
            if not _preflight_private(entry, directory=True):
                return False
        elif entry.is_file() and not _preflight_private(entry, directory=False):
            return False
    return True


def _preflight_jsonl(path: Path) -> tuple[list[dict[str, object]], str]:
    if path.is_symlink() or not path.exists() or not path.is_file():
        raise CustomerAdminError(
            "Gate-D canonical reconciliation ledger is unavailable"
        )
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise CustomerAdminError(
            "Gate-D canonical reconciliation ledger is unavailable"
        ) from exc
    rows: list[dict[str, object]] = []
    offset = 0
    for index, line in enumerate(raw.splitlines(keepends=True), start=1):
        if not line.endswith(b"\n"):
            raise CustomerAdminError(
                f"Gate-D canonical reconciliation ledger has a torn tail at row {index}"
            )
        value = line[:-1]
        if value.endswith(b"\r"):
            value = value[:-1]
        if not value:
            raise CustomerAdminError(
                "Gate-D canonical reconciliation ledger contains a blank row"
            )
        try:
            parsed = json.loads(value.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CustomerAdminError(
                "Gate-D canonical reconciliation ledger is corrupt"
            ) from exc
        if not isinstance(parsed, dict):
            raise CustomerAdminError("Gate-D canonical reconciliation row is invalid")
        rows.append(parsed)
        offset += len(line)
    if offset != len(raw):
        raise CustomerAdminError(
            "Gate-D canonical reconciliation ledger has a torn tail"
        )
    return rows, hashlib.sha256(raw).hexdigest()


def _preflight_json_document(path: Path) -> tuple[Mapping[str, object], str]:
    if path.is_symlink() or not path.exists() or not path.is_file():
        raise CustomerAdminError("Gate-D configuration is unavailable")
    try:
        raw = path.read_bytes()
        value = json.loads(raw.decode("utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise CustomerAdminError("Gate-D configuration is invalid") from exc
    if not isinstance(value, Mapping):
        raise CustomerAdminError("Gate-D configuration is invalid")
    return dict(value), hashlib.sha256(raw).hexdigest()


def _preflight_scalar(value: str) -> object:
    import ast

    text = value.strip()
    if not text:
        return {}
    if text in {"null", "Null", "NULL", "~"}:
        return None
    if text.lower() in {"true", "false"}:
        return text.lower() == "true"
    try:
        if text.startswith(("{", "[", "'", '"')):
            try:
                return json.loads(text)
            except json.JSONDecodeError:
                return ast.literal_eval(text)
        return int(text)
    except (ValueError, SyntaxError):
        return text.strip("\"'")


def _preflight_simple_yaml(text: str) -> Mapping[str, object]:
    """Parse the small mapping subset used by profile config without a dependency."""
    lines: list[tuple[int, str]] = []
    raw_lines = text.splitlines()
    index = 0
    while index < len(raw_lines):
        raw = raw_lines[index]
        index += 1
        if not raw.strip() or raw.lstrip().startswith("#"):
            continue
        indentation = len(raw) - len(raw.lstrip(" "))
        content = raw.strip()
        if " #" in content:
            content = content.split(" #", 1)[0].rstrip()
        if ":" in content:
            key, raw_value = content.split(":", 1)
            block_style = raw_value.strip()
            if block_style in {"|", "|-", "|+", ">", ">-", ">+"}:
                block_lines: list[str] = []
                while index < len(raw_lines):
                    candidate = raw_lines[index]
                    candidate_indentation = len(candidate) - len(candidate.lstrip(" "))
                    if candidate.strip() and candidate_indentation <= indentation:
                        break
                    block_lines.append(candidate)
                    index += 1
                content_indents = [
                    len(item) - len(item.lstrip(" "))
                    for item in block_lines
                    if item.strip()
                ]
                block_indent = (
                    min(content_indents) if content_indents else indentation + 1
                )
                value = "\n".join(
                    item[block_indent:] if item.strip() else "" for item in block_lines
                )
                if not block_style.endswith("-"):
                    value += "\n"
                lines.append((indentation, f"{key.strip()}: {value!r}"))
                continue
        lines.append((indentation, content))
    for line_index, (indentation, content) in enumerate(lines[:-1]):
        next_indentation, next_content = lines[line_index + 1]
        if (
            content.endswith(":")
            and next_indentation == indentation
            and next_content.startswith("- ")
        ):
            cursor = line_index + 1
            while cursor < len(lines):
                item_indentation, item_content = lines[cursor]
                if item_indentation < indentation:
                    break
                if item_indentation == indentation and not item_content.startswith(
                    "- "
                ):
                    break
                if item_indentation == indentation:
                    lines[cursor] = (indentation + 1, item_content)
                cursor += 1
    root: dict[str, object] = {}
    stack: list[tuple[int, object]] = [(-1, root)]
    for index, (indentation, content) in enumerate(lines):
        while stack and indentation <= stack[-1][0]:
            stack.pop()
        if not stack:
            raise CustomerAdminError("Gate-D configuration indentation is invalid")
        parent = stack[-1][1]
        if content.startswith("- "):
            if not isinstance(parent, list):
                raise CustomerAdminError("Gate-D configuration list is invalid")
            parent.append(_preflight_scalar(content[2:]))
            continue
        if ":" not in content or not isinstance(parent, dict):
            raise CustomerAdminError("Gate-D configuration mapping is invalid")
        key, raw_value = content.split(":", 1)
        key = key.strip()
        if not key:
            raise CustomerAdminError("Gate-D configuration key is invalid")
        raw_value = raw_value.strip()
        if raw_value:
            parent[key] = _preflight_scalar(raw_value)
            continue
        next_is_list = (
            index + 1 < len(lines)
            and lines[index + 1][0] > indentation
            and lines[index + 1][1].startswith("- ")
        )
        child: object = [] if next_is_list else {}
        parent[key] = child
        stack.append((indentation, child))
    return root


def _preflight_config(root: Path) -> tuple[Mapping[str, object], str, Path]:
    candidates = (root / "config.json", root / "config.yaml", root / "config.yml")
    path = next(
        (
            candidate
            for candidate in candidates
            if candidate.exists() or candidate.is_symlink()
        ),
        None,
    )
    if path is None:
        raise FileNotFoundError("Gate-D configuration is missing")
    if path.is_symlink() or not path.is_file():
        raise CustomerAdminError("Gate-D configuration symlinks are not allowed")
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise CustomerAdminError("Gate-D configuration is unavailable") from exc
    digest_value = hashlib.sha256(raw).hexdigest()
    if path.suffix == ".json":
        try:
            payload = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CustomerAdminError("Gate-D configuration is invalid") from exc
    else:
        try:
            import yaml  # type: ignore
        except ImportError:
            payload = _preflight_simple_yaml(raw.decode("utf-8"))
        else:
            try:
                payload = yaml.safe_load(raw.decode("utf-8"))
            except Exception as exc:
                raise CustomerAdminError("Gate-D configuration is invalid") from exc
    if not isinstance(payload, Mapping):
        raise CustomerAdminError("Gate-D configuration is invalid")
    return dict(payload), digest_value, path


def _preflight_mapping_at(
    payload: Mapping[str, object],
    *paths: tuple[str, ...],
) -> Mapping[str, object] | None:
    for path in paths:
        value: object = payload
        for key in path:
            if not isinstance(value, Mapping):
                break
            value = value.get(key)
        else:
            if isinstance(value, Mapping):
                return value
    return None


def _preflight_address(value: object) -> tuple[str, str, str] | None:
    if isinstance(value, TelegramAddress):
        return value.key
    if not isinstance(value, Mapping):
        return None
    values = tuple(
        str(item) if isinstance(item, int) and not isinstance(item, bool) else item
        for item in (value.get(key) for key in ("user_id", "chat_id", "topic_id"))
    )
    if any(not isinstance(item, str) or not item.strip() for item in values):
        return None
    return values  # type: ignore[return-value]


def _preflight_review_operator(
    value: object,
) -> tuple[tuple[str, str, str], int] | None:
    """Validate the exact AdaptiveReviewOperator runtime shape without aliases."""
    if not isinstance(value, Mapping) or set(value) != {
        "user_id",
        "chat_id",
        "topic_id",
        "version",
    }:
        return None
    user_id = value["user_id"]
    chat_id = value["chat_id"]
    topic_value = value["topic_id"]
    version = value["version"]
    if (
        not isinstance(user_id, str)
        or not user_id.strip()
        or not isinstance(chat_id, str)
        or not chat_id.strip()
        or isinstance(version, bool)
        or type(version) is not int
        or version < 1
    ):
        return None
    if type(topic_value) is int and not isinstance(topic_value, bool):
        topic_id = topic_value
    elif isinstance(topic_value, str) and topic_value.isdigit():
        topic_id = int(topic_value)
    else:
        return None
    if topic_id != 59:
        return None
    return (user_id.strip(), chat_id.strip(), str(topic_id)), version


def _space_pair(value: object) -> tuple[str, str] | None:
    """Normalize one reserved route to its chat/topic pair."""
    if isinstance(value, Mapping):
        raw = value.get("space_key", value.get("key", value))
    else:
        raw = getattr(value, "space_key", None)
        if raw is None:
            raw = getattr(value, "key", value)
    if isinstance(raw, Mapping):
        if "chat_id" not in raw or ("topic_id" not in raw and "thread_id" not in raw):
            return None
        chat_value = raw.get("chat_id")
        topic_value = raw.get("topic_id", raw.get("thread_id"))
    elif isinstance(raw, (tuple, list)):
        if len(raw) == 3:
            _, chat_value, topic_value = raw
        elif len(raw) == 2:
            chat_value, topic_value = raw
        else:
            return None
    else:
        return None
    if isinstance(chat_value, bool) or isinstance(topic_value, bool):
        return None
    chat = str(chat_value).strip() if isinstance(chat_value, (str, int)) else ""
    topic = str(topic_value).strip() if isinstance(topic_value, (str, int)) else ""
    return (chat, topic) if chat and topic else None


def _iter_reserved_space_values(value: object) -> Iterator[object]:
    """Yield route-shaped values while rejecting malformed category entries."""
    if value is None:
        return
    if isinstance(value, Mapping):
        if (
            "chat_id" in value
            and "topic_id" not in value
            and "thread_id" not in value
            and "topics" in value
        ):
            topics = value.get("topics")
            if not isinstance(topics, (tuple, list)):
                raise CustomerAdminError(
                    "reserved route category contains an invalid route"
                )
            for topic in topics:
                if not isinstance(topic, Mapping):
                    raise CustomerAdminError(
                        "reserved route category contains an invalid route"
                    )
                candidate = dict(topic)
                candidate["chat_id"] = value.get("chat_id")
                yield candidate
            return
        if any(
            field in value
            for field in ("chat_id", "topic_id", "thread_id", "space_key", "key")
        ):
            yield value
            return
        for nested in value.values():
            if nested is None:
                yield None
            else:
                yield from _iter_reserved_space_values(nested)
        return
    if isinstance(value, str):
        parts = value.strip().split(":")
        if len(parts) == 3 and parts[0].lower() == "telegram":
            yield {"chat_id": parts[1], "topic_id": parts[2]}
            return
        raise CustomerAdminError("reserved route category contains an invalid route")
    if isinstance(value, (bytes, bytearray)):
        raise CustomerAdminError("reserved route category contains an invalid route")
    if isinstance(value, (tuple, list, set, frozenset)):
        if _space_pair(value) is not None:
            yield value
            return
        for nested in value:
            if nested is None:
                yield None
            else:
                yield from _iter_reserved_space_values(nested)
        return
    yield value


def validate_review_space_disjoint(
    review_operator: object,
    *,
    customer_routes: object = (),
    owner_scheduled_routes: object = (),
    generic_reserved_routes: object = (),
    owner_routes: object | None = None,
    scheduled_routes: object | None = None,
    generic_routes: object | None = None,
) -> bool:
    """Reject a review chat/topic shared by any reserved ingress category."""
    review_raw = getattr(review_operator, "key", review_operator)
    if isinstance(review_raw, Mapping):
        review_raw = tuple(
            review_raw.get(field) for field in ("user_id", "chat_id", "topic_id")
        )
    if not isinstance(review_raw, (tuple, list)) or len(review_raw) != 3:
        raise CustomerAdminError("review operator route is invalid")
    if any(
        isinstance(item, bool)
        or not isinstance(item, (str, int))
        or not str(item).strip()
        for item in review_raw
    ):
        raise CustomerAdminError("review operator route is invalid")
    review_space = (str(review_raw[1]).strip(), str(review_raw[2]).strip())
    categories: tuple[tuple[str, object], ...] = (
        ("customer", customer_routes),
        (
            "owner_scheduled",
            (
                owner_scheduled_routes,
                owner_routes if owner_routes is not None else (),
                scheduled_routes if scheduled_routes is not None else (),
            ),
        ),
        (
            "generic",
            (
                generic_reserved_routes,
                generic_routes if generic_routes is not None else (),
            ),
        ),
    )
    for category, routes in categories:
        try:
            candidates = tuple(_iter_reserved_space_values(routes))
        except CustomerAdminError:
            raise
        for candidate in candidates:
            pair = _space_pair(candidate)
            if pair is None:
                raise CustomerAdminError(f"{category} reserved route is invalid")
            if pair == review_space:
                raise CustomerAdminError(
                    f"review space collides with {category} reserved route"
                )
    return True


def _preflight_route_values(
    value: object,
    *,
    default_topic: str | None = None,
) -> Iterator[object]:
    """Collect config route entries without repairing malformed values."""
    if value is None:
        return
    if isinstance(value, Mapping):
        if (
            "chat_id" in value
            and "topic_id" not in value
            and "thread_id" not in value
            and "topics" in value
        ):
            topics = value.get("topics")
            if not isinstance(topics, (tuple, list)):
                raise CustomerAdminError("Gate-D reserved route is invalid")
            for topic in topics:
                if not isinstance(topic, Mapping):
                    raise CustomerAdminError("Gate-D reserved route is invalid")
                candidate = dict(topic)
                candidate["chat_id"] = value.get("chat_id")
                yield from _preflight_route_values(
                    candidate,
                    default_topic=default_topic,
                )
            return
        if any(
            field in value
            for field in ("chat_id", "topic_id", "thread_id", "space_key", "key")
        ):
            candidate = dict(value)
            if (
                default_topic is not None
                and "chat_id" in candidate
                and "topic_id" not in candidate
                and "thread_id" not in candidate
            ):
                candidate["topic_id"] = default_topic
            yield candidate
            return
        for nested in value.values():
            yield from _preflight_route_values(
                nested,
                default_topic=default_topic,
            )
        return
    if isinstance(value, (tuple, list, set, frozenset)):
        if _space_pair(value) is not None:
            yield value
            return
        for nested in value:
            yield from _preflight_route_values(
                nested,
                default_topic=default_topic,
            )
        return
    if isinstance(value, str):
        parts = value.strip().split(":")
        if len(parts) == 3 and parts[0].lower() == "telegram":
            yield {"chat_id": parts[1], "topic_id": parts[2]}
            return
        raise CustomerAdminError("Gate-D reserved route is invalid")
    yield value


def _preflight_config_routes(
    config: Mapping[str, object],
    telegram: Mapping[str, object],
    telegram_extra: Mapping[str, object],
    *,
    names: tuple[str, ...],
    default_topic: str | None = None,
) -> tuple[object, ...]:
    values: list[object] = []
    for source in (config, telegram, telegram_extra):
        for name in names:
            if name in source:
                values.extend(
                    _preflight_route_values(
                        source.get(name),
                        default_topic=default_topic,
                    )
                )
    return tuple(values)


def _preflight_bot_identity(value: object) -> str | None:
    if not isinstance(value, Mapping):
        return None
    nested = value.get("config")
    if isinstance(nested, Mapping):
        value = nested
    for name in ("bot_id", "bot_username", "username", "name", "config_id"):
        identity = value.get(name)
        if isinstance(identity, str) and identity.strip():
            return identity
    return None


def _preflight_bot_check(
    config: Mapping[str, object],
    adaptive: Mapping[str, object],
) -> tuple[bool, str | None]:
    main_identity = _preflight_bot_identity(config.get("bot"))
    candidates: list[Mapping[str, object]] = []
    for value in (
        adaptive.get("separate_bot"),
        adaptive.get("test_bot"),
        adaptive.get("bot"),
        adaptive.get("separate_bot_config"),
        adaptive.get("bot_config"),
        config.get("separate_bot"),
        config.get("test_bot"),
        config.get("bot"),
        config.get("bot_config"),
    ):
        if isinstance(value, Mapping):
            candidates.append(value)
    for candidate in candidates:
        bot_config = candidate.get("config")
        if not isinstance(bot_config, Mapping):
            bot_config = candidate
        dedicated = (
            bot_config.get("separate") is True or bot_config.get("dedicated") is True
        )
        identity = _preflight_bot_identity(candidate)
        if dedicated and identity is not None:
            if main_identity is not None and identity == main_identity:
                return False, "separate_bot_invalid"
            return True, None
        if identity is not None:
            return False, "separate_bot_invalid"
    return False, "separate_bot_missing"


def _preflight_schedule_state(
    root: Path,
    checks: dict[str, bool],
    counts: dict[str, int],
    digests: dict[str, str],
    reasons: set[str],
) -> None:
    """Validate schedule persistence without locks, recovery writes, or migration."""
    data_root = root / "data"
    ledger_path = data_root / "scheduled-deliveries.jsonl"
    fence_path = data_root / "scheduled-deliveries-fence.json"
    claims_root = data_root / "customer-schedule-claims"
    if data_root.is_symlink() or not data_root.exists() or not data_root.is_dir():
        checks["schedule_fence"] = False
        reasons.add("schedule_missing")
        return
    if not _preflight_private(data_root, directory=True):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")

    fence = None
    try:
        fence = _read_fence(fence_path)
    except CustomerScheduleError:
        reasons.add("schedule_fence_invalid")
    if fence is None:
        reasons.add("schedule_fence_missing")
    elif fence.state != "ready":
        reasons.add("schedule_fence_not_ready")
    else:
        checks["schedule_fence"] = True
    if fence_path.exists() and not _preflight_private(fence_path, directory=False):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")

    try:
        rows = _read_jsonl(ledger_path)
    except CustomerScheduleError:
        rows = []
        reasons.add("schedule_ledger_invalid")
    else:
        checks["schedule_ledger"] = True
    if ledger_path.exists():
        if not _preflight_private(ledger_path, directory=False):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")
        digests["schedule_ledger"] = _preflight_digest_bytes(ledger_path)
    else:
        digests["schedule_ledger"] = hashlib.sha256(b"").hexdigest()
    counts["schedule_rows"] = len(rows)

    rows_valid = True
    try:
        _validate_schedule_rows(rows)
    except CustomerScheduleError:
        rows_valid = False
        checks["schedule_transitions"] = False
        reasons.add("schedule_transition_evidence_invalid")
    if rows_valid:
        checks["schedule_transitions"] = True
        for row in rows:
            state = row.get("state")
            if state in {"delivered", "sent_audited"} and not isinstance(
                row.get("provider_receipt"), str
            ):
                checks["schedule_transitions"] = False
                reasons.add("schedule_transition_evidence_missing")
            if state in {"unknown", "abandoned"} and not isinstance(
                row.get("reason"), str
            ):
                checks["schedule_transitions"] = False
                reasons.add("schedule_transition_evidence_missing")

    try:
        claims = _claims_by_schedule(claims_root)
    except CustomerScheduleError:
        claims = {}
        checks["schedule_tombstones"] = False
        reasons.add("schedule_tombstone_invalid")
    else:
        checks["schedule_tombstones"] = True
    if claims_root.exists():
        if not _preflight_private(claims_root, directory=True):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")
        if not _preflight_tree_private(claims_root):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")
    counts["schedule_tombstones"] = len(claims)
    claim_material = b"".join(
        schedule.encode("utf-8") + b"\0" + hashlib.sha256(pair[2]).digest()
        for schedule, pair in sorted(claims.items())
    )
    digests["schedule_tombstones"] = hashlib.sha256(claim_material).hexdigest()

    if not rows_valid:
        return
    current = _current_rows(rows)
    for row in current.values():
        schedule = str(row["schedule_key"])
        pair = claims.get(schedule)
        if pair is None:
            reasons.add("schedule_tombstone_missing")
            continue
        raw = pair[2]
        if hashlib.sha256(raw).hexdigest() != str(row["legacy_claim_digest"]):
            reasons.add("schedule_tombstone_digest_mismatch")
            continue
        tombstone = pair[3]
        if tombstone is None:
            if raw.startswith(b"scheduled-delivery-tombstone-v1\n"):
                reasons.add("schedule_tombstone_invalid")
            elif (
                row.get("state") != "unknown"
                or row.get("reason") != "legacy_claim_unknown"
            ):
                reasons.add("schedule_legacy_claim_stale")
        elif tombstone != row["reservation_id"]:
            reasons.add("schedule_tombstone_reservation_mismatch")
    for schedule, pair in claims.items():
        if schedule in current:
            continue
        raw = pair[2]
        if pair[3] is not None:
            reasons.add("schedule_tombstone_unpaired")
        elif raw.startswith(b"scheduled-delivery-tombstone-v1\n"):
            reasons.add("schedule_tombstone_invalid")
        else:
            reasons.add("schedule_legacy_claim_stale")


def _preflight_receipt(
    *,
    checks: Mapping[str, bool],
    counts: Mapping[str, int],
    digests: Mapping[str, str],
    epoch: int | None,
    checked_at_kst: str,
    reasons: set[str],
) -> GateDPreflightReceipt:
    normalized_reasons = tuple(sorted(reasons))
    if set(normalized_reasons) - _GATE_PREFLIGHT_REASON_CODES:
        raise CustomerAdminError("Gate-D preflight produced an unbounded reason code")
    return GateDPreflightReceipt(
        schema_version=_GATE_PREFLIGHT_SCHEMA,
        ready=not normalized_reasons and all(checks.values()),
        checks=dict(checks),
        counts=dict(counts),
        digests=dict(digests),
        epoch=epoch,
        reason_codes=normalized_reasons,
        checked_at_kst=checked_at_kst,
    )


def _preflight_checked_at(
    value: date | datetime | None,
) -> tuple[date, str]:
    current_value = datetime.now(_KST) if value is None else value
    if isinstance(current_value, datetime):
        checked = (
            current_value.astimezone(_KST)
            if current_value.tzinfo is not None
            else current_value.replace(tzinfo=_KST)
        )
    elif isinstance(current_value, date):
        checked = datetime.combine(current_value, time.min).replace(tzinfo=_KST)
    else:
        raise CustomerAdminError("kst_now must be a date or datetime")
    return checked.date(), checked.isoformat(timespec="seconds")


def audit_gate_d_preflight(
    profile_root: Path,
    customer_key: str,
    *,
    kst_now: date | datetime | None = None,
) -> GateDPreflightReceipt:
    """Audit a disposable Gate-D profile without creating or enabling anything."""
    root = _resolve_profile_root(Path(profile_root))
    key = _require_customer_key(customer_key)
    current_day, checked_at_kst = _preflight_checked_at(kst_now)
    registry_path = _resolve_registry_path(root)
    reasons: set[str] = set()
    checks: dict[str, bool] = {
        "profile_contained": True,
        "private_modes": True,
        "enabled_isolation": False,
        "identity_roles_distinct": False,
        "review_operator": False,
        "separate_bot": False,
        "current_kst_window": False,
        "activation_receipt": False,
        "approved_registration": False,
        "approved_artifacts": False,
        "canonical_reconciliation": False,
        "feature_flags_disabled": False,
        "delivery_disabled": False,
        "schedule_fence": False,
        "schedule_ledger": False,
        "schedule_tombstones": False,
        "schedule_transitions": False,
    }
    counts: dict[str, int] = {
        "enabled_customers": 0,
        "canonical_events": 0,
        "canonical_sequences": 0,
        "source_day_mappings": 0,
        "source_day_intents": 0,
        "authority_rows": 0,
        "config_epoch_rows": 0,
        "approved_artifacts": 0,
        "schedule_rows": 0,
        "schedule_tombstones": 0,
    }
    digests: dict[str, str] = {}
    epoch: int | None = None

    if not _preflight_private(root, directory=True):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")
    if registry_path.is_symlink() or not registry_path.resolve().is_relative_to(root):
        raise CustomerAdminError("Gate-D registry path escapes the profile")
    if not _preflight_private(registry_path, directory=False):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")
    try:
        document = _read_profile_registry(registry_path, root)
    except (CustomerRegistryError, ValidationError) as exc:
        if any(
            token in str(exc)
            for token in (
                "owner and customer",
            )
        ):
            reasons.add("identity_roles_not_distinct")
            digests["registry"] = _preflight_digest_bytes(registry_path)
            return _preflight_receipt(
                checks=checks,
                counts=counts,
                digests=digests,
                epoch=epoch,
                checked_at_kst=checked_at_kst,
                reasons=reasons,
            )
        raise CustomerAdminError("Gate-D registry is corrupt") from exc
    except (OSError, UnicodeDecodeError, ValueError) as exc:
        raise CustomerAdminError("Gate-D registry is corrupt") from exc
    digests["registry"] = _preflight_digest_bytes(registry_path)
    enabled = tuple(item for item in document.customers if item.enabled)
    counts["enabled_customers"] = len(enabled)
    selected = next(
        (item for item in document.customers if item.customer_key == key), None
    )
    if selected is None:
        raise CustomerAdminError(f"unknown customer: {key}")
    if not selected.enabled:
        reasons.add("customer_not_enabled")
    if len(enabled) != 1 or enabled[0].customer_key != key:
        reasons.add("main_profile_containment")
    else:
        checks["enabled_isolation"] = True

    owner = _preflight_address(document.owner)
    customer = _preflight_address(selected.telegram)
    identities = tuple(
        value for value in (owner, customer) if value is not None
    )
    if (
        owner is None
        or customer is None
        or len(identities) != 2
        or len(set(identities)) != 2
        or len({(value[1], value[2]) for value in identities}) != 2
    ):
        reasons.add("identity_roles_not_distinct")
    else:
        checks["identity_roles_distinct"] = True

    customer_root = root / "data" / "customers" / key
    if not _preflight_tree_private(customer_root):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")
    if (
        customer_root.is_symlink()
        or not customer_root.exists()
        or not customer_root.is_dir()
        or not customer_root.resolve().is_relative_to(root)
        or _path_has_symlink(customer_root.resolve(), root)
    ):
        raise CustomerAdminError("Gate-D customer root is unsafe")
    if not _preflight_private(customer_root, directory=True):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")
    main_root = root / "main-profile"
    if main_root.exists() and customer_root.resolve() == main_root.resolve():
        checks["profile_contained"] = False
        reasons.add("main_profile_containment")

    try:
        config, config_digest, config_path = _preflight_config(root)
    except FileNotFoundError:
        config, config_digest, config_path = {}, "", root / "config.yaml"
        reasons.add("config_missing")
    except CustomerAdminError:
        config, config_digest, config_path = {}, "", root / "config.yaml"
        reasons.add("config_invalid")
    if config_digest:
        digests["config"] = config_digest
        if not _preflight_private(config_path, directory=False):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")
    configured_main = None
    for key_name in ("main_profile_root", "personal_profile_root"):
        value = config.get(key_name)
        if isinstance(value, str) and value.strip():
            configured_main = Path(value).expanduser().resolve()
            break
    if configured_main is not None and (
        configured_main == root or root.is_relative_to(configured_main)
    ):
        checks["profile_contained"] = False
        reasons.add("main_profile_containment")

    adaptive = _preflight_mapping_at(
        config,
        ("platforms", "telegram", "extra", "adaptive_nutrition"),
        ("telegram", "adaptive_nutrition"),
        ("adaptive_nutrition",),
    )
    if adaptive is None:
        adaptive = {}
        reasons.add("config_invalid")
    delivery_value = adaptive.get("delivery_enabled", adaptive.get("delivery"))
    telegram = (
        _preflight_mapping_at(
            config,
            ("platforms", "telegram"),
            ("telegram",),
        )
        or {}
    )
    telegram_extra = (
        telegram.get("extra") if isinstance(telegram.get("extra"), Mapping) else {}
    )
    try:
        owner_scheduled_routes = (owner,) + _preflight_config_routes(
            config,
            telegram,
            telegram_extra,
            names=(
                "owner_scheduled_routes",
                "scheduled_routes",
                "owner_schedule",
                "scheduled_delivery_routes",
                "cron_routes",
                "owner_deliveries",
                "home_channel",
            ),
            default_topic="1",
        )
        generic_reserved_routes = _preflight_config_routes(
            config,
            telegram,
            telegram_extra,
            names=(
                "dm_topics",
                "physique_checkin",
                "generic_reserved_routes",
                "generic_routes",
                "reserved_routes",
                "reserved_spaces",
                "operator_routes",
                "platform_reserved_routes",
            ),
            default_topic="1",
        )
    except CustomerAdminError:
        owner_scheduled_routes = (owner,)
        generic_reserved_routes = ()
        reasons.add("review_space_invalid")
    if delivery_value is True:
        reasons.add("delivery_enabled")
    elif delivery_value is False:
        checks["delivery_disabled"] = True
    else:
        reasons.add("config_invalid")
    if adaptive.get("enabled") is not True:
        reasons.add("config_invalid")
    if any(
        adaptive.get(name) is True
        for name in (
            "analytics_shadow",
            "operator_candidates",
            "activation",
            "delivery",
        )
    ):
        reasons.add("feature_flags_enabled")

    review = adaptive.get("review_operator")
    if not isinstance(review, Mapping):
        reasons.add("review_operator_missing")
    elif set(review) != {"user_id", "chat_id", "topic_id", "version"} and (
        {"user_id", "chat_id", "topic_id", "version"} - set(review)
    ):
        reasons.add("review_operator_missing")
    else:
        review_result = _preflight_review_operator(review)
        if review_result is None:
            reasons.add("review_operator_invalid")
        else:
            review_address, _review_version = review_result
            legacy_pair_valid = True
            if "operator_chat_id" in adaptive:
                legacy_pair_valid = (
                    str(adaptive["operator_chat_id"]).strip() == review_address[1]
                )
            if "operator_topic_id" in adaptive:
                try:
                    legacy_topic = int(adaptive["operator_topic_id"])
                except (TypeError, ValueError):
                    legacy_pair_valid = False
                else:
                    legacy_pair_valid = legacy_pair_valid and legacy_topic == 59
            if not legacy_pair_valid or customer is None:
                reasons.add("review_operator_invalid")
            else:
                try:
                    validate_review_space_disjoint(
                        review_address,
                        customer_routes=(customer,),
                        owner_scheduled_routes=owner_scheduled_routes,
                        generic_reserved_routes=generic_reserved_routes,
                    )
                except CustomerAdminError as exc:
                    detail = str(exc)
                    reasons.add("review_operator_invalid")
                    reasons.add(
                        "review_space_collision"
                        if "collides" in detail
                        else "review_space_invalid"
                    )
                    for category in (
                        "customer",
                        "owner_scheduled",
                        "generic",
                    ):
                        if category in detail:
                            reasons.add(f"review_space_collision_{category}")
                            break
                else:
                    checks["review_operator"] = True
    bot_ok, bot_reason = _preflight_bot_check(config, adaptive)
    if bot_ok:
        checks["separate_bot"] = True
    elif bot_reason is not None:
        reasons.add(bot_reason)

    try:
        activation_id = _require_committed_activation_receipt(
            root,
            registry_path,
            document,
            selected,
        )
    except CustomerAdminError:
        reasons.add("activation_receipt_missing")
    else:
        checks["activation_receipt"] = True
        digests["activation_receipt"] = adaptive_digest(
            {"activation_receipt_id": activation_id}
        )

    starts = selected.plan.starts_on
    ends = starts + timedelta(days=27)
    if starts <= current_day <= ends:
        checks["current_kst_window"] = True
    else:
        reasons.add("plan_window_stale")

    adaptive_root = customer_root / "nutrition-plans"
    if (
        adaptive_root.is_symlink()
        or not adaptive_root.exists()
        or not adaptive_root.is_dir()
        or _path_has_symlink(adaptive_root, root)
    ):
        raise CustomerAdminError("Gate-D adaptive root is unsafe")
    if not _preflight_private(adaptive_root, directory=True):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")
    if not _preflight_tree_private(adaptive_root):
        checks["private_modes"] = False
        reasons.add("private_modes_invalid")

    feature_path = adaptive_root / "feature-epoch.json"
    try:
        _, feature_digest = _preflight_json_document(feature_path)
    except CustomerAdminError:
        reasons.add("canonical_reconciliation_missing")
    else:
        digests["feature_epoch"] = feature_digest
        try:
            feature_epoch, feature_config_digest_value, flags = (
                _reconciliation_feature_config(feature_path)
            )
        except CustomerAdminError:
            reasons.add("canonical_reconciliation_stale")
        else:
            epoch = feature_epoch
            digests["feature_config"] = feature_config_digest_value
            if all(flag is False for flag in flags.values()):
                checks["feature_flags_disabled"] = True
            else:
                reasons.add("feature_flags_enabled")
        if not _preflight_private(feature_path, directory=False):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")

    if checks["activation_receipt"]:
        try:
            approved = _load_adaptive_registration_inputs(root, key, recover=False)
        except CustomerAdminError:
            reasons.add("registration_missing")
        else:
            if approved.digest is None or not approved.approved:
                reasons.add("artifacts_unapproved")
            else:
                checks["approved_registration"] = True
                digests["registration"] = str(approved.digest)
                artifacts = approved.artifact_digests or {}
                if set(artifacts) == {"base_policy", "meal_constraints", "catalog"}:
                    checks["approved_artifacts"] = True
                    counts["approved_artifacts"] = 3
                    digests.update(
                        {
                            f"artifact_{name}": str(value)
                            for name, value in artifacts.items()
                            if isinstance(value, str)
                        }
                    )
                else:
                    reasons.add("artifacts_missing")

    canonical_path = customer_root / "wizard" / "events.jsonl"
    projection_paths = {
        "source_day_mappings": adaptive_root / "source-days.jsonl",
        "source_day_intents": adaptive_root / "source-day-intents.jsonl",
        "authority_rows": adaptive_root / "authority-mirror-intents.jsonl",
        "config_epoch_rows": adaptive_root / "config-epoch-journal.jsonl",
    }
    try:
        _, events_digest = _preflight_jsonl(canonical_path)
        if not _preflight_private(canonical_path, directory=False):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")
        runtime_registry = load_customer_registry(registry_path, root)
        runtime = next(
            (
                item
                for item in runtime_registry.customers
                if item.spec.customer_key == key
            ),
            None,
        )
        if runtime is None:
            raise CustomerAdminError("Gate-D canonical runtime is unavailable")
        transaction = CanonicalEventTransaction.for_customer_runtime(runtime)
        if (
            transaction.sequence_path.is_symlink()
            or not transaction.sequence_path.exists()
            or not transaction.sequence_path.is_file()
        ):
            raise CustomerAdminError("Gate-D canonical sequence ledger is unavailable")
        snapshot = transaction.read_snapshot_readonly()
        events = [
            event.model_dump(mode="json", exclude_none=True)
            for event in snapshot.events
        ]
        counts["canonical_events"] = len(events)
        digests["canonical_events"] = events_digest
        sequence_rows = [dict(row) for row in snapshot.sequence_rows]
        projected: dict[str, list[dict[str, object]]] = {
            "canonical_sequences": sequence_rows,
        }
        counts["canonical_sequences"] = len(sequence_rows)
        digests["canonical_sequences"] = _preflight_digest_bytes(
            transaction.sequence_path
        )
        if not _preflight_private(transaction.sequence_path, directory=False):
            checks["private_modes"] = False
            reasons.add("private_modes_invalid")
        for name, path in projection_paths.items():
            rows, row_digest = _preflight_jsonl(path)
            projected[name] = rows
            counts[name] = len(rows)
            digests[name] = row_digest
            if not _preflight_private(path, directory=False):
                checks["private_modes"] = False
                reasons.add("private_modes_invalid")
        for row in events:
            if row.get("customer_key") not in (None, key):
                reasons.add("canonical_reconciliation_stale")
        for name in (
            "source_day_mappings",
            "source_day_intents",
            "authority_rows",
            "config_epoch_rows",
        ):
            for row in projected[name]:
                if row.get("customer_key") not in (None, key):
                    reasons.add("canonical_reconciliation_stale")
        committed_configs = [
            row
            for row in projected["config_epoch_rows"]
            if row.get("state") == "committed"
        ]
        committed_authority = [
            row
            for row in projected["authority_rows"]
            if row.get("state") == "committed"
        ]
        if not committed_configs or not committed_authority:
            reasons.add("canonical_reconciliation_missing")
        elif epoch is not None and committed_configs[-1].get("epoch") != epoch:
            reasons.add("canonical_reconciliation_stale")
        if not reasons.intersection(
            {
                "canonical_reconciliation_missing",
                "canonical_reconciliation_stale",
            }
        ):
            checks["canonical_reconciliation"] = True
        digests["canonical_reconciliation"] = adaptive_digest(tuple(projected.values()))
    except CustomerAdminError:
        reasons.add("canonical_reconciliation_missing")
    except (OSError, RuntimeError, UnicodeDecodeError, ValueError):
        reasons.add("canonical_reconciliation_stale")

    _preflight_schedule_state(root, checks, counts, digests, reasons)

    return _preflight_receipt(
        checks=checks,
        counts=counts,
        digests=digests,
        epoch=epoch,
        checked_at_kst=checked_at_kst,
        reasons=reasons,
    )


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="python -m checkin_cli.customer_admin")
    parser.add_argument("--registry", type=Path, required=True)
    commands = parser.add_subparsers(dest="command", required=True)
    add = commands.add_parser("add")
    add.add_argument("--key", required=True)
    add.add_argument("--name", required=True)
    add.add_argument("--user-id", required=True)
    add.add_argument("--chat-id", required=True)
    add.add_argument("--topic-id", required=True)
    add.add_argument("--starts-on", type=date.fromisoformat, required=True)
    add.add_argument("--daily-time", type=time.fromisoformat, default=time(8, 0))
    add.add_argument("--weekly-weekday", type=int, default=0)
    add.add_argument("--monthly-day", type=int, default=1)
    add.add_argument("--calories", type=int, required=True)
    add.add_argument("--protein", type=int, required=True)
    add.add_argument("--meals", nargs="+", default=("아침", "점심", "저녁"))
    add.add_argument("--goal", default="미정")
    add.add_argument("--restriction", action="append", default=[])
    add.add_argument("--allergy", action="append", default=[])
    add.add_argument("--preference", action="append", default=[])
    add.add_argument("--supplement", action="append", default=[])
    add.add_argument("--digestion-context")
    add.add_argument("--sleep-goal", type=float)
    add.add_argument("--recovery-goal")
    add.add_argument("--training-context")
    add.add_argument("--carbs", type=int)
    add.add_argument("--fat", type=int)
    add.add_argument("--water", type=float)
    command = commands.add_parser("disable")
    command.add_argument("customer_key")
    activate = commands.add_parser("activate")
    activate.add_argument("customer_id")
    activate.add_argument("--profile-root", type=Path, required=True)
    activate.add_argument("--data-root", type=Path, required=True)
    activate.add_argument("--checklist-evidence", type=Path, required=True)
    activate.add_argument("--staff-membership-evidence", type=Path, required=True)

    def add_record_scope(command: argparse.ArgumentParser) -> None:
        command.add_argument("customer_key")
        command.add_argument("--profile-root", type=Path, required=True)
        command.add_argument("--data-root", type=Path)

    payment = commands.add_parser("record-payment")
    add_record_scope(payment)
    payment.add_argument(
        "--kind", choices=tuple(kind.value for kind in PaymentKind), required=True
    )
    payment.add_argument("--paid-on", type=date.fromisoformat, required=True)
    payment.add_argument("--period-start-on", type=date.fromisoformat, required=True)
    payment.add_argument("--period-end-on", type=date.fromisoformat, required=True)
    payment.add_argument("--amount-krw", type=int, default=150000)
    payment.add_argument(
        "--method",
        choices=(PaymentMethod.BANK_TRANSFER.value,),
        default=PaymentMethod.BANK_TRANSFER.value,
    )

    satisfaction = commands.add_parser("record-satisfaction")
    add_record_scope(satisfaction)
    satisfaction.add_argument(
        "--score", "--score-1to10", dest="score_1to10", type=int, required=True
    )
    satisfaction.add_argument(
        "--collected-on",
        "--collection-date",
        dest="collected_on",
        type=date.fromisoformat,
        required=True,
    )
    satisfaction.add_argument("--note")

    operator_time = commands.add_parser("record-operator-time")
    add_record_scope(operator_time)
    operator_time.add_argument("--entry-id", required=True)
    operator_time.add_argument("--attempt-id", required=True)
    operator_time.add_argument("--minutes", type=int, required=True)
    operator_time.add_argument(
        "--task", choices=tuple(task.value for task in OperatorTask), required=True
    )
    operator_time.add_argument(
        "--work-date",
        "--work-on",
        dest="work_date",
        type=date.fromisoformat,
        required=True,
    )
    operator_time.add_argument("--supersedes-entry-id")
    kpi = commands.add_parser("judge-kpi")
    kpi.add_argument("customer_key")
    kpi.add_argument("--profile-root", type=Path, required=True)
    commands.add_parser("list")
    commands.add_parser("validate")
    consent = commands.add_parser("consent")
    consent.add_argument("customer_key")
    consent.add_argument("--recorded-on", type=date.fromisoformat, required=True)
    consent.add_argument("--notice-version", required=True)
    revoke = commands.add_parser("revoke-consent")
    revoke.add_argument("customer_key")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    path: Path = args.registry
    if args.command == "add":
        customer = register_customer(
            path,
            CustomerDraft(
                customer_key=args.key,
                display_name=args.name,
                user_id=args.user_id,
                chat_id=args.chat_id,
                topic_id=args.topic_id,
                starts_on=args.starts_on,
                daily_time=args.daily_time,
                weekly_weekday=args.weekly_weekday,
                monthly_day=args.monthly_day,
                calories_kcal=args.calories,
                protein_g=args.protein,
                meals=tuple(args.meals),
                primary_goal=args.goal,
                dietary_restrictions=tuple(args.restriction),
                allergies=tuple(args.allergy),
                food_preferences=tuple(args.preference),
                supplements=tuple(args.supplement),
                digestion_context=args.digestion_context,
                sleep_goal_hours=args.sleep_goal,
                recovery_goal=args.recovery_goal,
                training_context=args.training_context,
                carbohydrate_g=args.carbs,
                fat_g=args.fat,
                water_liters=args.water,
            ),
        )
        print(f"created disabled draft: {customer.customer_key}")
        return 0
    if args.command == "disable":
        customer = set_customer_enabled(path, args.customer_key, enabled=False)
        print(f"{customer.customer_key}: disabled")
        return 0
    if args.command == "activate":
        result = activate_customer(
            args.profile_root,
            args.data_root,
            args.customer_id,
            args.checklist_evidence,
            args.staff_membership_evidence,
        )
        print(f"{result.customer_id}: enabled")
        return 0
    if args.command == "record-payment":
        _print_record_receipt(
            record_payment(
                path,
                args.customer_key,
                profile_root=args.profile_root,
                data_root=args.data_root,
                amount_krw=args.amount_krw,
                paid_on=args.paid_on,
                period_start_on=args.period_start_on,
                period_end_on=args.period_end_on,
                method=args.method,
                kind=args.kind,
            )
        )
        return 0
    if args.command == "record-satisfaction":
        _print_record_receipt(
            record_satisfaction(
                path,
                args.customer_key,
                profile_root=args.profile_root,
                data_root=args.data_root,
                score_1to10=args.score_1to10,
                collected_on=args.collected_on,
                note=args.note,
            )
        )
        return 0
    if args.command == "record-operator-time":
        _print_record_receipt(
            record_operator_time(
                path,
                args.customer_key,
                profile_root=args.profile_root,
                data_root=args.data_root,
                entry_id=args.entry_id,
                attempt_id=args.attempt_id,
                minutes=args.minutes,
                task=args.task,
                work_date=args.work_date,
                supersedes_entry_id=args.supersedes_entry_id,
            )
        )
        return 0
    if args.command == "judge-kpi":
        judgement = judge_customer_pilot_kpis(
            path,
            args.customer_key,
            profile_root=args.profile_root,
        )
        _print_kpi_judgement(args.customer_key, judgement)
        return 0
    if args.command == "consent":
        consent = AiProcessingConsent(
            granted=True,
            recorded_on=args.recorded_on,
            notice_version=args.notice_version,
        )
        customer = set_customer_ai_consent(path, args.customer_key, consent)
        print(f"{customer.customer_key}: AI processing consent recorded")
        return 0
    if args.command == "revoke-consent":
        customer = set_customer_ai_consent(
            path, args.customer_key, AiProcessingConsent()
        )
        print(f"{customer.customer_key}: AI processing consent revoked")
        return 0
    if args.command == "validate":
        document = _read(path)
        print(
            f"registry valid: {len(document.customers)} customers; every plan has 12 weeks"
        )
        return 0
    for customer in _read(path).customers:
        print(
            f"{customer.customer_key}\t{customer.display_name}\t{'enabled' if customer.enabled else 'disabled'}"
        )
    return 0


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