"""Deterministic, operator-approved adaptive nutrition planning.

This module never sends messages and never delegates calculations to an LLM.
It produces immutable proposals which an operator may approve by exact digest.
"""
from __future__ import annotations

import fcntl
import os
import stat
from contextlib import contextmanager
from dataclasses import asdict, dataclass, replace
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
from enum import StrEnum
import hashlib
import hmac
import json
from pathlib import Path
import unicodedata
from types import MappingProxyType
from typing import Any, Iterable, Literal, Mapping, Sequence
from zoneinfo import ZoneInfo
from .models import CanonicalAdherenceSignal, ContractCheckin, Event, derive_canonical_adherence_signal, validate_event
from .customer_coaching import CustomerRuntime
from .store import CanonicalEventTransaction, _reject_registered_persistence_root

@contextmanager
def _verified_private_lock(path: Path, *, exclusive: bool, create: bool) -> Any:
    """Lock one private writer inode without following replacement paths."""
    flags = (os.O_RDWR if exclusive else os.O_RDONLY) | os.O_NOFOLLOW | os.O_CLOEXEC
    if create:
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        flags |= os.O_CREAT
    descriptor = os.open(path, flags, 0o600)
    try:
        opened = os.fstat(descriptor)
        named = 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 ValueError("adaptive lock is unsafe")
        fcntl.flock(descriptor, fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
        current = path.lstat()
        if identity != (current.st_dev, current.st_ino):
            raise ValueError("adaptive lock was replaced")
        yield descriptor
    finally:
        fcntl.flock(descriptor, fcntl.LOCK_UN)
        os.close(descriptor)
class _AdaptiveLockToken:
    __slots__ = ()


_ADAPTIVE_LOCK_TOKEN_TYPE = _AdaptiveLockToken
class _OverlayLockToken:
    __slots__ = ("identity",)

    def __init__(self, identity: tuple[int, int]) -> None:
        self.identity = identity


_OVERLAY_LOCK_TOKEN_TYPE = _OverlayLockToken


class Decision(StrEnum):
    OBSERVE = "observe"
    MAINTAIN = "maintain"
    CALORIE_ADJUSTMENT = "calorie_adjustment_candidate"
    MACRO_REDISTRIBUTION = "macro_redistribution_candidate"
    HUMAN_REVIEW = "human_review"
GLOBAL_MIN_CALORIES = 1500
GLOBAL_MAX_CALORIES = 4500
OPERATOR_NOTE_MAX_LENGTH = 2000
GLOBAL_MIN_PROTEIN_G = 120
GLOBAL_MAX_PROTEIN_G = 250
GLOBAL_MIN_FAT_G = 40
GLOBAL_MAX_FAT_G = 150
CARB_CYCLE_DELTA_CALORIES = 200
CARB_CYCLE_VERSION = "1.0"
_DUAL_COACH_RISK_POLICY_FILE = "dual-coach-risk-policy.json"
_DUAL_COACH_RISK_POLICY_SCHEMA = "dual_coach_risk_policy_v1"
_DUAL_COACH_RISK_POLICY_ID = "dual-coach-risk-v1"
_DUAL_COACH_RISK_UNAVAILABLE = "unavailable"


@dataclass(frozen=True, slots=True)
class DualCoachRiskPolicyV1:
    """The sealed, owner-approved policy used by dual-coach risk evaluation."""

    version: str
    policy: Mapping[str, object]
    policy_digest: str
    document_digest: str

    def __post_init__(self) -> None:
        if not isinstance(self.version, str) or not self.version:
            raise ValueError("dual-coach risk policy version is invalid")
        value = _validate_dual_coach_risk_policy_value(self.policy)
        object.__setattr__(self, "policy", MappingProxyType(value))
        _require_digest(self.policy_digest, "dual-coach risk policy")
        _require_digest(self.document_digest, "dual-coach risk policy document")


@dataclass(frozen=True, slots=True)
class DualCoachRiskEvidence:
    """The six normalized inputs accepted by the sealed risk policy."""

    weight_change_percent: Decimal
    sleep_hours: Decimal
    fatigue: str
    pain: str
    exercise_feasibility: str
    meal_deviation: str


@dataclass(frozen=True, slots=True)
class DualCoachRiskEvaluation:
    policy_version: str
    policy_digest: str
    policy_document_digest: str
    score: int | None
    reasons: tuple[str, ...]
    held: bool

    @property
    def evidence_available(self) -> bool:
        return self.score is not None


@dataclass(frozen=True, slots=True)
class DualCoachRiskReviewCandidate:
    """Append-only review payload identity; unavailable policy uses fixed sentinels."""

    customer_key: str
    evaluation_kst_day: date
    reason_code: str
    policy_version: str
    policy_digest: str
    policy_document_digest: str
    terminal_checkin_id: str
    terminal_checkin_digest: str
    source_strategy_digest: str
    epoch: int

    @property
    def dedupe_key(self) -> str:
        return digest((
            self.customer_key,
            self.evaluation_kst_day.isoformat(),
            self.reason_code,
            self.policy_version,
            self.policy_digest,
            self.policy_document_digest,
            self.terminal_checkin_id,
            self.terminal_checkin_digest,
            self.source_strategy_digest,
            self.epoch,
        ))


def held_dual_coach_risk_candidate(
    *,
    customer_key: str,
    evaluation_kst_day: date,
    terminal_checkin_id: str,
    terminal_checkin_digest: str,
    source_strategy_digest: str,
    epoch: int,
) -> DualCoachRiskReviewCandidate:
    """Create the sole dedupe-stable representation for unavailable risk evidence."""
    if (
        type(customer_key) is not str
        or not customer_key
        or type(evaluation_kst_day) is not date
        or type(terminal_checkin_id) is not str
        or not terminal_checkin_id
        or type(source_strategy_digest) is not str
        or type(epoch) is not int
        or epoch < 0
    ):
        raise ValueError("dual-coach risk review candidate is invalid")
    _require_digest(terminal_checkin_digest, "terminal check-in")
    _require_digest(source_strategy_digest, "source strategy")
    return DualCoachRiskReviewCandidate(
        customer_key,
        evaluation_kst_day,
        "risk_evidence_unavailable",
        _DUAL_COACH_RISK_UNAVAILABLE,
        _DUAL_COACH_RISK_UNAVAILABLE,
        _DUAL_COACH_RISK_UNAVAILABLE,
        terminal_checkin_id,
        terminal_checkin_digest,
        source_strategy_digest,
        epoch,
    )


def _dual_coach_risk_policy_document_digest(document: Mapping[str, object]) -> str:
    return digest({key: value for key, value in document.items() if key != "document_digest"})


def _validate_dual_coach_risk_policy_value(policy: object) -> dict[str, object]:
    if not isinstance(policy, Mapping):
        raise ValueError("dual-coach risk policy is invalid")
    required = {
        "weight_change_percent",
        "sleep_hours",
        "fatigue",
        "pain",
        "exercise_feasibility",
        "meal_deviation",
        "score_threshold",
        "hard_overrides",
        "missing_evidence_reason",
    }
    if set(policy) != required or policy.get("score_threshold") != 4:
        raise ValueError("dual-coach risk policy table is invalid")
    if policy.get("hard_overrides") != ["pain_override", "exercise_impossible_override"]:
        raise ValueError("dual-coach risk policy overrides are invalid")
    if policy.get("missing_evidence_reason") != "risk_evidence_unavailable":
        raise ValueError("dual-coach risk policy missing-evidence reason is invalid")
    expected = {
        "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"],
    }
    for key, value in expected.items():
        if policy.get(key) != value:
            raise ValueError("dual-coach risk policy table is invalid")
    return dict(policy)


def load_verified_dual_coach_risk_policy(runtime: CustomerRuntime) -> DualCoachRiskPolicyV1:
    """Load only the current customer-private, owner-approved risk policy."""
    if not isinstance(runtime, CustomerRuntime):
        raise TypeError("dual-coach risk policy requires CustomerRuntime")
    path = runtime.nutrition_plans_root / _DUAL_COACH_RISK_POLICY_FILE
    try:
        descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
        try:
            status = os.fstat(descriptor)
            named = path.lstat()
            if (
                not stat.S_ISREG(status.st_mode)
                or status.st_uid != os.geteuid()
                or status.st_nlink != 1
                or stat.S_IMODE(status.st_mode) != 0o600
                or (status.st_dev, status.st_ino) != (named.st_dev, named.st_ino)
            ):
                raise ValueError("dual-coach risk policy file is unsafe")
            with os.fdopen(descriptor, "rb", closefd=False) as handle:
                document = json.loads(handle.read().decode("utf-8"))
        finally:
            os.close(descriptor)
        if not isinstance(document, Mapping):
            raise ValueError("dual-coach risk policy document is invalid")
        required = {
            "schema_version", "policy_id", "version", "policy", "policy_digest",
            "document_digest", "approved", "approved_by", "approved_at_kst",
            "customer_key", "owner_digest", "registry_digest",
            "activation_receipt_digest", "authority_digest",
        }
        if set(document) != required or document.get("schema_version") != _DUAL_COACH_RISK_POLICY_SCHEMA:
            raise ValueError("dual-coach risk policy schema is invalid")
        if document.get("policy_id") != _DUAL_COACH_RISK_POLICY_ID or document.get("approved") is not True:
            raise ValueError("dual-coach risk policy approval is invalid")
        if not isinstance(document.get("version"), str) or not document["version"]:
            raise ValueError("dual-coach risk policy version is invalid")
        policy = _validate_dual_coach_risk_policy_value(document.get("policy"))
        policy_digest = document.get("policy_digest")
        document_digest = document.get("document_digest")
        _require_digest(policy_digest, "dual-coach risk policy")
        _require_digest(document_digest, "dual-coach risk policy document")
        if not hmac.compare_digest(str(policy_digest), digest(policy)):
            raise ValueError("dual-coach risk policy digest mismatch")
        if not hmac.compare_digest(str(document_digest), _dual_coach_risk_policy_document_digest(document)):
            raise ValueError("dual-coach risk policy document digest mismatch")
        # This lazy import avoids the admin module's existing adaptive import cycle.
        from checkin_cli import customer_admin as admin
        profile_root = runtime.customer_root.parents[2]
        registry_path = admin._resolve_registry_path(profile_root)
        registry = admin._read_profile_registry(registry_path, profile_root)
        selected = next((item for item in registry.customers if item.customer_key == runtime.spec.customer_key), None)
        if selected is None or selected.enabled is not True or document.get("customer_key") != runtime.spec.customer_key:
            raise ValueError("dual-coach risk policy customer is stale")
        authority, authority_digest = admin._registration_activation_authority(
            profile_root, registry_path, registry, selected
        )
        owner = registry.owner.model_dump(mode="json")
        if (
            document.get("approved_by") != owner
            or document.get("owner_digest") != authority.get("owner_digest")
            or document.get("registry_digest") != authority.get("registry_digest")
            or document.get("activation_receipt_digest") != authority.get("activation_receipt_digest")
            or document.get("authority_digest") != authority_digest
        ):
            raise ValueError("dual-coach risk policy authority pins are stale")
        admin._registration_kst_timestamp(str(document.get("approved_at_kst")))
        return DualCoachRiskPolicyV1(str(document["version"]), policy, str(policy_digest), str(document_digest))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError, KeyError, IndexError) as exc:
        raise ValueError("dual-coach risk policy is unavailable") from exc


def _risk_band(value: Decimal, *, normal: Decimal, elevated: Decimal, reverse: bool = False) -> int:
    if reverse:
        return 0 if value >= normal else 1 if value >= elevated else 2
    return 0 if value <= normal else 1 if value <= elevated else 2


def _evaluate_dual_coach_risk(
    policy: DualCoachRiskPolicyV1,
    evidence: DualCoachRiskEvidence,
) -> DualCoachRiskEvaluation:
    """Evaluate the sealed table without accepting fallback policy or loose evidence."""
    if not isinstance(policy, DualCoachRiskPolicyV1) or not isinstance(evidence, DualCoachRiskEvidence):
        raise TypeError("dual-coach risk evaluation requires sealed policy and typed evidence")
    if any(
        isinstance(value, bool) or not isinstance(value, Decimal) or not value.is_finite()
        for value in (evidence.weight_change_percent, evidence.sleep_hours)
    ):
        raise ValueError("dual-coach risk evidence is invalid")
    levels = {
        "fatigue": ("low", "moderate", "high"),
        "pain": ("none", "present", "severe"),
        "exercise_feasibility": ("possible", "limited", "impossible"),
        "meal_deviation": ("none", "partial", "material"),
    }
    text_values = {
        "fatigue": evidence.fatigue,
        "pain": evidence.pain,
        "exercise_feasibility": evidence.exercise_feasibility,
        "meal_deviation": evidence.meal_deviation,
    }
    if any(type(value) is not str or value not in levels[name] for name, value in text_values.items()):
        raise ValueError("dual-coach risk evidence is invalid")
    score = (
        _risk_band(evidence.weight_change_percent.copy_abs(), normal=Decimal("2"), elevated=Decimal("4"))
        + _risk_band(evidence.sleep_hours, normal=Decimal("7"), elevated=Decimal("5"), reverse=True)
        + levels["fatigue"].index(evidence.fatigue)
        + levels["pain"].index(evidence.pain)
        + levels["exercise_feasibility"].index(evidence.exercise_feasibility)
        + levels["meal_deviation"].index(evidence.meal_deviation)
    )
    reasons: list[str] = []
    if evidence.pain != "none":
        reasons.append("pain_override")
    if evidence.exercise_feasibility == "impossible":
        reasons.append("exercise_impossible_override")
    if score >= 4:
        reasons.append("risk_score_threshold")
    return DualCoachRiskEvaluation(
        policy.version, policy.policy_digest, policy.document_digest, score, tuple(reasons), bool(reasons)
    )


def evaluate_customer_dual_coach_risk(
    runtime: CustomerRuntime,
    evidence: DualCoachRiskEvidence,
) -> DualCoachRiskEvaluation:
    """Fail closed before any normalization when customer policy custody is unavailable."""
    try:
        return _evaluate_dual_coach_risk(load_verified_dual_coach_risk_policy(runtime), evidence)
    except (TypeError, ValueError):
        return DualCoachRiskEvaluation(
            _DUAL_COACH_RISK_UNAVAILABLE,
            _DUAL_COACH_RISK_UNAVAILABLE,
            _DUAL_COACH_RISK_UNAVAILABLE,
            None,
            ("risk_evidence_unavailable",),
            True,
        )


@dataclass(frozen=True, slots=True)
class VersionedFoodCatalog:
    """Immutable catalog boundary for production meal-plan compilation."""

    foods: tuple[Food, ...]
    version: str
    digest: str
    approved: bool = True
    approved_by: str = ""
    approved_at_kst: str = ""
    schema_version: str = "1.0"

    def __post_init__(self) -> None:
        foods = tuple(self.foods)
        if not foods:
            raise ValueError("approved catalog is empty")
        if any(not isinstance(food, Food) for food in foods):
            raise ValueError("approved catalog contains an invalid food")
        if len({food.food_id for food in foods}) != len(foods):
            raise ValueError("approved catalog contains duplicate food ids")
        if any(
            not food.food_id.strip()
            or any(isinstance(value, bool) or not isinstance(value, int) for value in (
                food.calories,
                food.carbs_g,
                food.protein_g,
                food.fat_g,
            ))
            or food.calories < 0
            or min(food.carbs_g, food.protein_g, food.fat_g) < 0
            or not _valid_serving_bounds(food)
            for food in foods
        ):
            raise ValueError("approved catalog contains an invalid food")
        if isinstance(self.version, bool) or not isinstance(self.version, str) or not self.version.strip():
            raise ValueError("approved catalog version is required")
        if not isinstance(self.schema_version, str) or self.schema_version not in {"1.0", "2.0"}:
            raise ValueError("approved catalog schema version is invalid")
        if not isinstance(self.approved, bool):
            raise ValueError("approved catalog approval flag is invalid")
        _require_digest(self.digest, "catalog")
        object.__setattr__(self, "foods", foods)
        if self.schema_version == "2.0":
            canonical_foods = tuple(self.foods)
            accepted_digests = {
                digest(canonical_foods),
                digest({"foods": canonical_foods}),
                digest({"foods": canonical_foods, "version": self.version}),
                digest({"schema_version": self.schema_version, "foods": canonical_foods, "version": self.version}),
                digest({"catalog": canonical_foods}),
                digest({"catalog": canonical_foods, "version": self.version}),
                digest({"schema_version": self.schema_version, "catalog": canonical_foods, "version": self.version}),
                digest({"food_catalog": canonical_foods}),
                digest({"food_catalog": canonical_foods, "version": self.version}),
                digest({"schema_version": self.schema_version, "food_catalog": canonical_foods, "version": self.version}),
            }
            if self.digest not in accepted_digests:
                raise ValueError("catalog digest does not match content")

    def __iter__(self):
        return iter(self.foods)

    def __len__(self) -> int:
        return len(self.foods)

    def __getitem__(self, index):
        return self.foods[index]

    @property
    def catalog(self) -> tuple[Food, ...]:
        """Compatibility view for callers that previously consumed a tuple."""
        return self.foods

    @property
    def catalog_digest(self) -> str:
        return self.digest


@dataclass(frozen=True, slots=True)
class DailyObservation:
    kst_day: date
    weight_kg: Decimal | None = None
    calories: int | None = None
    carbs_g: int | None = None
    protein_g: int | None = None
    fat_g: int | None = None
    adherence_ok: bool | None = None
    safety_held: bool = False
    exercise_load: str | None = None  # high | medium | low
    target_calories_kcal: int | None = None
    target_carbohydrate_g: int | None = None
    target_protein_g: int | None = None
    target_fat_g: int | None = None
    actual_calories_kcal: int | None = None
    actual_carbohydrate_g: int | None = None
    actual_protein_g: int | None = None
    actual_fat_g: int | None = None
    adherence_tolerance_percent: Decimal | None = None
    adherence_version: str | None = None
    exercise_session_done: bool | None = None
    exercise_intensity_vs_plan: str | None = None
    target_calories: int | None = None
    target_carbs_g: int | None = None
    target_protein: int | None = None
    target_fat: int | None = None
    actual_calories: int | None = None
    actual_carbs_g: int | None = None
    actual_protein: int | None = None
    actual_fat: int | None = None


@dataclass(frozen=True, slots=True)
class CustomerPolicy:
    starts_on: date
    goal_mode: str
    weekly_rate_min: Decimal | None = None
    weekly_rate_max: Decimal | None = None
    calorie_step: int | None = None
    calorie_floor: int | None = None
    calorie_ceiling: int | None = None
    minimum_adherent_days: int = 5
    extended_through: date | None = None
    adherence_tolerance_percent: Decimal = Decimal("10")
    cooldown_days: int = 0
    carb_schedule_version: str = "1.0"

    @property
    def configured(self) -> bool:
        return all(v is not None for v in (
            self.weekly_rate_min, self.weekly_rate_max, self.calorie_step,
            self.calorie_floor, self.calorie_ceiling,
        ))


@dataclass(frozen=True, slots=True)
class OnboardingPolicyCompatibility:
    requires_human_review: bool
    may_propose: bool


def validate_onboarding_policy_compatibility(
    artifact: Mapping[str, object],
) -> OnboardingPolicyCompatibility:
    requirements = {
        "calorie_step": (100, "100 kcal"),
        "cooldown_days": (7, "7 day"),
        "calorie_floor": (1500, "1500 kcal"),
        "calorie_ceiling": (4500, "4500 kcal"),
        "minimum_current_samples": (4, "current samples"),
        "minimum_total_samples": (10, "total samples"),
        "minimum_adherent_days": (5, "adherent days"),
    }
    for field, (expected, label) in requirements.items():
        value = artifact.get(field)
        if isinstance(value, bool) or type(value) is not int or value != expected:
            raise ValueError(f"onboarding policy requires {label}")
    if artifact.get("maximum_step_kcal") != artifact.get("calorie_step"):
        raise ValueError("onboarding policy maximum step is invalid")
    if type(artifact.get("safety_hold")) is not bool:
        raise ValueError("onboarding policy safety hold is invalid")
    held = bool(artifact["safety_hold"])
    return OnboardingPolicyCompatibility(
        requires_human_review=held,
        may_propose=not held,
    )


def customer_policy_from_onboarding_artifact(
    artifact: Mapping[str, object],
) -> CustomerPolicy:
    if artifact.get("schema_version") != "1.0":
        raise ValueError("onboarding policy schema is invalid")
    if artifact.get("policy_version") != "1.0":
        raise ValueError("onboarding policy version is invalid")
    if artifact.get("approved") is not True:
        raise ValueError("onboarding policy is not approved")
    compatibility = validate_onboarding_policy_compatibility(artifact)
    if compatibility.requires_human_review:
        raise ValueError("onboarding policy safety hold requires human review")
    if artifact.get("minimum_observation_days") != 7:
        raise ValueError("onboarding policy observation window is invalid")
    if artifact.get("weight_trend_method") != "two_non_overlapping_seven_day_means":
        raise ValueError("onboarding policy trend method is invalid")
    if artifact.get("adherence_inputs") != ["calories", "meal_plan"]:
        raise ValueError("onboarding policy adherence inputs are invalid")
    if artifact.get("actual_intake_inputs") != ["calories"]:
        raise ValueError("onboarding policy actual intake inputs are invalid")
    if artifact.get("activity_change_inputs") != ["training"]:
        raise ValueError("onboarding policy activity inputs are invalid")
    if artifact.get("manual_override_receipt_required") is not True:
        raise ValueError("onboarding policy override receipt is invalid")
    effective_at = artifact.get("effective_at_kst")
    if not isinstance(effective_at, str):
        raise ValueError("onboarding policy approval time is invalid")
    try:
        parsed_effective_at = datetime.fromisoformat(effective_at)
    except ValueError as exc:
        raise ValueError("onboarding policy approval time is invalid") from exc
    if (
        parsed_effective_at.tzinfo is None
        or parsed_effective_at.utcoffset() != timedelta(hours=9)
    ):
        raise ValueError("onboarding policy approval time must be KST")
    goal = artifact.get("goal_mode")
    rate_ranges = {
        "loss": (Decimal("-1.00"), Decimal("-0.25"), "fat_loss"),
        "maintain": (Decimal("-0.25"), Decimal("0.25"), "maintenance"),
        "gain": (Decimal("0.10"), Decimal("0.50"), "muscle_gain"),
    }
    if goal not in rate_ranges:
        raise ValueError("onboarding policy goal is invalid")
    expected_thresholds = {
        "minimum_current_samples": 4,
        "minimum_total_samples": 10,
        "minimum_adherent_days": 5,
        "goal_mode": goal,
    }
    if artifact.get("adjustment_thresholds") != expected_thresholds:
        raise ValueError("onboarding policy adjustment threshold is invalid")
    if artifact.get("escalation_rules") != [
        "contradictory_adherence",
        "safety_hold",
    ]:
        raise ValueError("onboarding policy escalation rules are invalid")
    if artifact.get("safety_stop_rules") != [
        "medical_review",
        "eating_disorder_risk",
    ]:
        raise ValueError("onboarding policy safety stop rules are invalid")
    try:
        effective_from = date.fromisoformat(str(artifact["effective_from"]))
        weekly_min = Decimal(str(artifact["desired_weekly_change_min"]))
        weekly_max = Decimal(str(artifact["desired_weekly_change_max"]))
    except (KeyError, ValueError) as exc:
        raise ValueError("onboarding adjustment policy is invalid") from exc
    expected_min, expected_max, goal_mode = rate_ranges[str(goal)]
    if (
        not weekly_min.is_finite()
        or not weekly_max.is_finite()
        or weekly_min > weekly_max
        or weekly_min != expected_min
        or weekly_max != expected_max
    ):
        raise ValueError("onboarding policy weekly rate is invalid")
    extended_through_value = artifact.get("extended_through")
    try:
        extended_through = (
            None
            if extended_through_value is None
            else date.fromisoformat(str(extended_through_value))
        )
    except ValueError as exc:
        raise ValueError("onboarding policy extension is invalid") from exc
    return CustomerPolicy(
        starts_on=effective_from,
        goal_mode=goal_mode,
        weekly_rate_min=weekly_min,
        weekly_rate_max=weekly_max,
        calorie_step=int(artifact["calorie_step"]),
        calorie_floor=int(artifact["calorie_floor"]),
        calorie_ceiling=int(artifact["calorie_ceiling"]),
        minimum_adherent_days=int(artifact["minimum_adherent_days"]),
        cooldown_days=int(artifact["cooldown_days"]),
        extended_through=extended_through,
    )


@dataclass(frozen=True, slots=True)
class TrendSnapshot:
    evaluation_day: date
    d_plus: int
    current_samples: int
    prior_samples: int
    current_mean_kg: Decimal | None
    prior_mean_kg: Decimal | None
    weekly_rate_percent: Decimal | None
    adherent_days: int
    safety_held: bool
    scheduled_loads: tuple[tuple[date, str], ...]
    exercise_sessions: tuple[
        tuple[date, bool | None, str | None],
        ...,
    ] = ()
    adherence_complete_days: int = 0
    adherence_inadequate_days: int = 0
    adherence_missing_days: int = 0
    adherence_contradictory_days: int = 0
    adherence_signal_version: str = "1.0"
    adherence_digest: str | None = None
    schedule_load_version: str = "1.0"
    schedule_ambiguity: bool = False
    canonical_projection: bool = False


@dataclass(frozen=True, slots=True)
class MacroTarget:
    calories: int
    carbs_g: int
    protein_g: int
    fat_g: int
@dataclass(frozen=True, slots=True)
class DailyNutritionTarget:
    """One immutable seven-day category target."""

    kst_day: date
    category: str
    target: MacroTarget

    @property
    def calories(self) -> int:
        return self.target.calories

    @property
    def carbs_g(self) -> int:
        return self.target.carbs_g

    @property
    def protein_g(self) -> int:
        return self.target.protein_g

    @property
    def fat_g(self) -> int:
        return self.target.fat_g


@dataclass(frozen=True, slots=True)
class WeeklyCarbCycle:
    """Deterministic high/medium/low targets reconciled over seven days."""

    targets: tuple[DailyNutritionTarget, ...]
    base_target: MacroTarget
    weekly_calories: int
    weekly_carbs_g: int
    weekly_protein_g: int
    weekly_fat_g: int
    feasible: bool = True
    reason: str | None = None
    version: str = CARB_CYCLE_VERSION

    def __post_init__(self) -> None:
        if type(self.feasible) is not bool or self.version != CARB_CYCLE_VERSION:
            raise ValueError("weekly cycle metadata is invalid")
        normalized = tuple(self.targets)
        if len(normalized) != 7:
            raise ValueError("weekly cycle requires seven daily targets")
        if any(
            not isinstance(item, DailyNutritionTarget)
            or item.category not in {"high", "medium", "low"}
            or not isinstance(item.kst_day, date)
            for item in normalized
        ):
            raise ValueError("weekly cycle target is invalid")
        if len({item.kst_day for item in normalized}) != 7:
            raise ValueError("weekly cycle days must be unique")
        if tuple(sorted(normalized, key=lambda item: item.kst_day)) != normalized:
            raise ValueError("weekly cycle days must be stable and sorted")
        object.__setattr__(self, "targets", normalized)

    @property
    def daily_targets(self) -> tuple[DailyNutritionTarget, ...]:
        return self.targets
    @property
    def daily(self) -> tuple[DailyNutritionTarget, ...]:
        return self.targets

    @property
    def days(self) -> tuple[tuple[date, str], ...]:
        return tuple((item.kst_day, item.category) for item in self.targets)

    @property
    def carb_days(self) -> tuple[tuple[date, str], ...]:
        return self.days

    @property
    def categories(self) -> tuple[tuple[date, str], ...]:
        return self.days

    @property
    def target_by_day(self) -> tuple[tuple[date, MacroTarget], ...]:
        return tuple((item.kst_day, item.target) for item in self.targets)

    def target_for(self, day: date) -> MacroTarget | None:
        return next((item.target for item in self.targets if item.kst_day == day), None)

    def category_target(self, category: str) -> MacroTarget | None:
        values = tuple(item.target for item in self.targets if item.category == category)
        if not values:
            return None
        first = values[0]
        return first if all(value == first for value in values) else None

    @property
    def high(self) -> MacroTarget | None:
        return self.category_target("high")

    @property
    def medium(self) -> MacroTarget | None:
        return self.category_target("medium")

    @property
    def low(self) -> MacroTarget | None:
        return self.category_target("low")

    @property
    def high_target(self) -> MacroTarget | None:
        return self.high

    @property
    def medium_target(self) -> MacroTarget | None:
        return self.medium

    @property
    def low_target(self) -> MacroTarget | None:
        return self.low

    @property
    def total_calories(self) -> int:
        return self.weekly_calories

    @property
    def total_carbs_g(self) -> int:
        return self.weekly_carbs_g

    @property
    def total_protein_g(self) -> int:
        return self.weekly_protein_g

    @property
    def total_fat_g(self) -> int:
        return self.weekly_fat_g

    @property
    def exact_energy_reconciliation(self) -> bool:
        return self.feasible and self.weekly_calories == self.base_target.calories * 7

    @property
    def reconciles_exactly(self) -> bool:
        return (
            self.exact_energy_reconciliation
            and self.weekly_carbs_g == self.base_target.carbs_g * 7
            and self.weekly_protein_g == self.base_target.protein_g * 7
            and self.weekly_fat_g == self.base_target.fat_g * 7
        )

    @property
    def has_redistribution(self) -> bool:
        return len({item.target for item in self.targets}) > 1
    @property
    def high_calories(self) -> int | None:
        return self.high.calories if self.high is not None else None

    @property
    def medium_calories(self) -> int | None:
        return self.medium.calories if self.medium is not None else None

    @property
    def low_calories(self) -> int | None:
        return self.low.calories if self.low is not None else None

    @property
    def weekly_target_calories(self) -> int:
        return self.base_target.calories * 7

    @property
    def weekly_target(self) -> MacroTarget:
        return self.base_target

    @property
    def exact_weekly_energy(self) -> bool:
        return self.exact_energy_reconciliation
    @property
    def category_targets(self) -> tuple[tuple[str, MacroTarget], ...]:
        return tuple(
            (category, target)
            for category, target in (
                ("high", self.high),
                ("medium", self.medium),
                ("low", self.low),
            )
            if target is not None
        )

    @property
    def assignments(self) -> tuple[tuple[date, str], ...]:
        return self.days
    def __getitem__(self, key: object) -> object:
        if isinstance(key, str):
            return self.category_target(key)
        if isinstance(key, date):
            return self.target_for(key)
        return self.targets[key]  # type: ignore[index]


@dataclass(frozen=True, slots=True)
class CooldownResult:
    """Immutable cooldown decision anchored to a committed calorie change."""

    active: bool
    reason: str = ""
    anchor_kst: str | None = None
    cooldown_until_kst: str | None = None
    days_remaining: int = 0
    source_revision_id: str | None = None

    @property
    def blocked(self) -> bool:
        return self.active

    @property
    def eligible(self) -> bool:
        return not self.active

    @property
    def until_kst(self) -> str | None:
        return self.cooldown_until_kst
CarbCycleTarget = DailyNutritionTarget
WeeklyMacroPlan = WeeklyCarbCycle


@dataclass(frozen=True, slots=True)
class NutritionProposal:
    customer_key: str
    snapshot: TrendSnapshot
    decision: Decision
    reasons: tuple[str, ...]
    target: MacroTarget | None = None
    carb_days: tuple[tuple[date, str], ...] = ()
    revision: int = 1
    parent_digest: str | None = None
    operator_note: str = ""
    source_digest: str | None = None
    policy_digest: str | None = None
    meal_constraints_digest: str | None = None
    catalog_digest: str | None = None
    meal_plan: MealPlan | None = None
    operator_body: str | None = None
    customer_body: str | None = None
    operator_body_digest: str | None = None
    customer_body_digest: str | None = None
    adherence_signal_digest: str | None = None
    weekly_carb_cycle: WeeklyCarbCycle | None = None
    weekly_nutrition_plan: WeeklyNutritionPlan | None = None
    cooldown: CooldownResult | None = None
    explanation: str | None = None

    @property
    def operator_digest(self) -> str | None:
        return self.operator_body_digest

    @property
    def customer_digest(self) -> str | None:
        return self.customer_body_digest

    @property
    def operator_render_digest(self) -> str | None:
        return self.operator_body_digest

    @property
    def customer_render_digest(self) -> str | None:
        return self.customer_body_digest
    @property
    def explanation_digest(self) -> str | None:
        return (
            hashlib.sha256(self.explanation.encode("utf-8")).hexdigest()
            if self.explanation is not None
            else None
        )
    @property
    def weekly_cycle(self) -> WeeklyCarbCycle | None:
        return self.weekly_carb_cycle

    @property
    def weekly_nutrition_plan_digest(self) -> str | None:
        return (
            self.weekly_nutrition_plan.digest
            if self.weekly_nutrition_plan is not None
            else None
        )

    @property
    def cooldown_result(self) -> CooldownResult | None:
        return self.cooldown

    @property
    def digest(self) -> str:
        return digest(self)


@dataclass(frozen=True, slots=True)
class CustomerActionContinuity:
    """One operator-approved, append-only customer action commitment."""

    customer_key: str
    approved_proposal_digest: str
    revision: int
    effective_kst_day: date
    action_text: str
    action_atom: str
    criterion_text: str
    criterion_atom: str
    next_check_kst: str
    approved_at_kst: str | None = None
    action_id: str = ""

    def __post_init__(self) -> None:
        if not isinstance(self.customer_key, str) or not self.customer_key:
            raise ValueError("customer action customer key is invalid")
        _require_digest(self.approved_proposal_digest, "customer action proposal")
        if type(self.revision) is not int or self.revision < 1:
            raise ValueError("customer action revision is invalid")
        for value, label, maximum in (
            (self.action_text, "action text", 240),
            (self.criterion_text, "criterion text", 240),
            (self.action_atom, "action atom", 64),
            (self.criterion_atom, "criterion atom", 64),
        ):
            if not isinstance(value, str) or not value.strip() or len(value) > maximum:
                raise ValueError(f"customer {label} is invalid")
        if self.criterion_atom not in {
            "checkin_recorded", "weight_recorded", "adherence_recorded", "exercise_session_recorded",
        }:
            raise ValueError("customer action criterion atom is invalid")
        _kst_timestamp_is_exact(self.next_check_kst)
        if self.approved_at_kst is not None:
            _kst_timestamp_is_exact(self.approved_at_kst)
        expected = digest({
            "customer_key": self.customer_key,
            "approved_proposal_digest": self.approved_proposal_digest,
            "revision": self.revision,
            "effective_kst_day": self.effective_kst_day.isoformat(),
            "action_atom": self.action_atom,
        })
        if self.action_id and self.action_id != expected:
            raise ValueError("customer action id is invalid")
        object.__setattr__(self, "action_id", expected)


@dataclass(frozen=True, slots=True)
class CustomerActionOutcome:
    action_id: str
    action_text: str
    criterion_text: str
    next_check_kst: str
    outcome: Literal["pending", "met", "not_met", "insufficient"]

@dataclass(frozen=True, slots=True)
class Food:
    food_id: str
    label: str
    calories: int
    carbs_g: int
    protein_g: int
    fat_g: int
    allergens: frozenset[str] = frozenset()
    convenience: bool = False
    serving_min_grams: int | None = None
    serving_max_grams: int | None = None
    serving_step_grams: int | None = None
    macro_basis_grams: int | None = None
    tags: frozenset[str] = frozenset()
    equivalent_group: str | None = None
    eating_out: bool = False
    allowed_budget_bands: frozenset[str] | None = None
    allowed_cooking_access: frozenset[str] | None = None


@dataclass(frozen=True, slots=True)
class MealConstraints:
    meal_count: int
    allergies: frozenset[str] = frozenset()
    excluded_food_ids: frozenset[str] = frozenset()
    budget_tier: str | None = None
    cooking_access: str | None = None
    preferences: tuple[str, ...] = ()
    restrictions: frozenset[str] = frozenset()
    digestion_exclusions: frozenset[str] = frozenset()
    training_times: tuple[tuple[date, str], ...] = ()
    training_time_by_day: tuple[tuple[date, str], ...] = ()
    meal_shares_bps: tuple[tuple[str, int], ...] = ()
    meal_slot_calorie_tolerance_percent: Decimal = Decimal("0")
    meal_slot_macro_tolerance_percent: Decimal = Decimal("0")
    calorie_tolerance_percent: Decimal | None = None
    macro_tolerance_percent: Decimal | None = None
    max_foods_per_meal: int = 4
    max_serving_units_per_meal: int = 12
    version: str | None = None
    digest: str | None = None
    approved: bool = True
    approved_by: str = ""
    approved_at_kst: str = ""
    strict_inputs: bool = False

    @property
    def complete(self) -> bool:
        basic = 1 <= self.meal_count <= 8 and self.budget_tier is not None and self.cooking_access is not None
        if not basic:
            return False
        if self.strict_inputs or self.version is not None:
            if not (self.training_times or self.training_time_by_day):
                return False
            if not isinstance(self.approved, bool) or self.approved is not True:
                return False
        return (
            0 <= self.meal_slot_calorie_tolerance_percent
            and 0 <= self.meal_slot_macro_tolerance_percent
            and 1 <= self.max_foods_per_meal <= 4
            and 1 <= self.max_serving_units_per_meal <= 12
        )


@dataclass(frozen=True, slots=True)
class MealSlot:
    name: str
    food_ids: tuple[str, ...]
    calories: int
    carbs_g: int
    protein_g: int
    fat_g: int
    quantities: tuple[int, ...] = ()
    serving_grams: tuple[int, ...] = ()
    target: MacroTarget | None = None

    @property
    def food_quantities(self) -> tuple[tuple[str, int], ...]:
        quantities = self.quantities or tuple(1 for _ in self.food_ids)
        return tuple(zip(self.food_ids, quantities))

    @property
    def serving_count(self) -> int:
        return sum(self.quantities or tuple(1 for _ in self.food_ids))

    @property
    def servings(self) -> tuple[tuple[str, int], ...]:
        return self.food_quantities

    @property
    def food_servings(self) -> tuple[tuple[str, int], ...]:
        return self.food_quantities

    @property
    def total_serving_units(self) -> int:
        return self.serving_count


@dataclass(frozen=True, slots=True)
class MealPlan:
    slots: tuple[MealSlot, ...]
    swaps: tuple[str, ...]
    fallback: tuple[str, ...]
    target: MacroTarget | None = None
    exact: bool = True
    compiler_version: str = "1.0"

    @property
    def calories(self) -> int:
        return sum(slot.calories for slot in self.slots)

    @property
    def carbs_g(self) -> int:
        return sum(slot.carbs_g for slot in self.slots)

    @property
    def protein_g(self) -> int:
        return sum(slot.protein_g for slot in self.slots)

    @property
    def fat_g(self) -> int:
        return sum(slot.fat_g for slot in self.slots)

    @property
    def digest(self) -> str:
        return digest(self)


@dataclass(frozen=True, slots=True)
class DailyNutritionPlan:
    """One immutable day of macros and exact food quantities."""

    kst_day: date
    category: str
    target: MacroTarget
    meal_plan: MealPlan
    nutrition_basis: str = "initial"
    parent_daily_digest: str | None = None

    def __post_init__(self) -> None:
        if (
            not isinstance(self.kst_day, date)
            or self.category not in {"high", "medium", "low"}
            or self.nutrition_basis
            not in {"initial", "recomputed_future", "frozen_parent"}
        ):
            raise ValueError("daily nutrition plan metadata is invalid")
        if not isinstance(self.target, MacroTarget) or not isinstance(
            self.meal_plan,
            MealPlan,
        ):
            raise ValueError("daily nutrition plan value is invalid")
        if (
            not self.meal_plan.exact
            or self.meal_plan.target != self.target
            or self.meal_plan.calories != self.target.calories
            or self.meal_plan.carbs_g != self.target.carbs_g
            or self.meal_plan.protein_g != self.target.protein_g
            or self.meal_plan.fat_g != self.target.fat_g
        ):
            raise ValueError("daily nutrition meal plan must match target exactly")
        if self.nutrition_basis == "frozen_parent":
            if not self.parent_daily_digest:
                raise ValueError("frozen daily nutrition plan requires parent digest")
        elif self.parent_daily_digest is not None:
            raise ValueError("non-frozen daily nutrition plan cannot pin parent")

    @property
    def digest(self) -> str:
        return digest(self)


@dataclass(frozen=True, slots=True)
class WeeklyNutritionPlan:
    """Seven immutable daily plans with exact weekly reconciliation."""

    horizon_start: date
    as_of_kst_day: date
    frozen_through: date | None
    base_target: MacroTarget
    days: tuple[DailyNutritionPlan, ...]
    parent_plan_digest: str | None = None
    planned_schedule_digest: str | None = None
    exercise_evidence_digest: str | None = None
    effective_schedule_digest: str | None = None
    version: str = "weekly-nutrition-plan-v1"
    compiler_version: str = "1.0"

    def __post_init__(self) -> None:
        if (
            self.version != "weekly-nutrition-plan-v1"
            or not isinstance(self.horizon_start, date)
            or not isinstance(self.as_of_kst_day, date)
            or not isinstance(self.base_target, MacroTarget)
        ):
            raise ValueError("weekly nutrition plan metadata is invalid")
        normalized = tuple(self.days)
        expected_days = tuple(
            self.horizon_start + timedelta(days=offset)
            for offset in range(7)
        )
        if (
            len(normalized) != 7
            or any(not isinstance(item, DailyNutritionPlan) for item in normalized)
            or tuple(item.kst_day for item in normalized) != expected_days
        ):
            raise ValueError(
                "weekly nutrition plan requires seven consecutive sorted days"
            )
        if self.frozen_through is not None and not (
            self.horizon_start <= self.frozen_through <= expected_days[-1]
        ):
            raise ValueError("weekly nutrition frozen boundary is invalid")
        if self.parent_plan_digest is None and any(
            item.nutrition_basis != "initial"
            for item in normalized
        ):
            raise ValueError("initial weekly nutrition plan requires initial days")
        if self.parent_plan_digest is not None and self.frozen_through is None:
            raise ValueError("reconciled weekly nutrition plan requires cutoff")
        object.__setattr__(self, "days", normalized)
        if not self.reconciles_exactly:
            raise ValueError("weekly nutrition plan does not reconcile exactly")

    @property
    def horizon_end(self) -> date:
        return self.horizon_start + timedelta(days=6)

    @property
    def weekly_calories(self) -> int:
        return sum(item.target.calories for item in self.days)

    @property
    def weekly_carbs_g(self) -> int:
        return sum(item.target.carbs_g for item in self.days)

    @property
    def weekly_protein_g(self) -> int:
        return sum(item.target.protein_g for item in self.days)

    @property
    def weekly_fat_g(self) -> int:
        return sum(item.target.fat_g for item in self.days)

    @property
    def reconciles_exactly(self) -> bool:
        return (
            self.weekly_calories == self.base_target.calories * 7
            and self.weekly_carbs_g == self.base_target.carbs_g * 7
            and self.weekly_protein_g == self.base_target.protein_g * 7
            and self.weekly_fat_g == self.base_target.fat_g * 7
        )

    @property
    def daily_plan_digest(self) -> str:
        return digest(tuple(item.digest for item in self.days))

    @property
    def digest(self) -> str:
        return digest(self)


@dataclass(frozen=True, slots=True)
class ApprovedAdaptiveArtifacts:
    """Approved, digested customer-owned adaptive inputs."""

    policy: CustomerPolicy
    meal_constraints: MealConstraints
    catalog: VersionedFoodCatalog
    policy_version: str
    meal_constraints_version: str
    catalog_version: str
    policy_digest: str
    meal_constraints_digest: str
    catalog_digest: str
    approved_by: str
    approved_at_kst: str


def _plain(value: object) -> object:
    if isinstance(value, Decimal):
        return format(value, "f")
    if isinstance(value, (date, StrEnum)):
        return str(value)
    if isinstance(value, frozenset):
        return sorted(value)
    if isinstance(value, tuple):
        return [_plain(v) for v in value]
    if isinstance(value, list):
        return [_plain(v) for v in value]
    if isinstance(value, dict):
        return {str(k): _plain(v) for k, v in value.items()}
    model_dump = getattr(value, "model_dump", None)
    if callable(model_dump):
        return _plain(model_dump(mode="json", exclude_none=False))
    if hasattr(value, "__dataclass_fields__"):
        return _plain(asdict(value))
    return value


def canonical_json(value: object) -> str:
    return unicodedata.normalize("NFC", json.dumps(_plain(value), ensure_ascii=False, sort_keys=True, separators=(",", ":")))


def digest(value: object) -> str:
    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
def _require_config_digest(value: object) -> str:
    if not isinstance(value, str) or len(value) != 64 or any(
        character not in "0123456789abcdefABCDEF" for character in value
    ):
        raise ValueError("config epoch digest is invalid")
    return value
_FEATURE_FLAG_NAMES = (
    "analytics_shadow",
    "operator_candidates",
    "activation",
    "delivery",
)
_FEATURE_CONFIG_FIELDS = frozenset({
    "schema_version",
    "epoch",
    "config_digest",
    *_FEATURE_FLAG_NAMES,
})


def feature_config_digest_preimage(
    epoch: object,
    flags: Mapping[str, object],
) -> Mapping[str, object]:
    """Return the one canonical preimage used for feature-config digests."""

    if type(epoch) is not int or epoch < 0:
        raise ValueError("feature config epoch is invalid")
    if not isinstance(flags, Mapping) or set(flags) != set(_FEATURE_FLAG_NAMES):
        raise ValueError("feature config flags are invalid")
    normalized: dict[str, bool] = {}
    for name in _FEATURE_FLAG_NAMES:
        value = flags[name]
        if type(value) is not bool:
            raise ValueError("feature config flags are invalid")
        normalized[name] = value
    return {"epoch": epoch, **normalized}


def feature_config_digest(epoch: object, flags: Mapping[str, object]) -> str:
    """Digest the canonical feature-config preimage."""

    return digest(feature_config_digest_preimage(epoch, flags))


def _validate_feature_config_document(document: Mapping[str, object]) -> dict[str, object]:
    if not isinstance(document, Mapping) or set(document) != _FEATURE_CONFIG_FIELDS:
        raise ValueError("feature config document is invalid")
    if document.get("schema_version") != "1.0":
        raise ValueError("feature config schema is invalid")
    flags = {name: document.get(name) for name in _FEATURE_FLAG_NAMES}
    epoch = document.get("epoch")
    declared = _require_digest(document.get("config_digest"), "feature config")
    expected = feature_config_digest(epoch, flags)
    if declared != expected:
        raise ValueError("feature config digest mismatch")
    return {
        "schema_version": "1.0",
        "epoch": epoch,
        "config_digest": declared,
        **flags,
    }
_ARTIFACT_METADATA = frozenset({
    "schema_version",
    "version",
    "digest",
    "approved",
    "approved_by",
    "approved_at_kst",
    "enabled",
    "extension_through",
    "customer_key",
    "supersedes_digest",
    "activation_receipt_id",
    "authority_digest",
})
_POLICY_SCHEMA_VERSION = "1.0"
_POLICY_REVISION_FILE = "policy-revisions.jsonl"
_POLICY_AUDIT_FILE = "policy-audit.jsonl"
_KST = ZoneInfo("Asia/Seoul")


def _require_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 ValueError(f"{label} digest is invalid")
    return value


def _approved_value(document: Mapping[str, object], label: str) -> tuple[object, str, str, str]:
    version = document.get("version")
    if isinstance(version, bool) or version in (None, ""):
        raise ValueError(f"{label} version is required")
    if document.get("approved") is not True:
        raise ValueError(f"{label} is not approved")
    approved_by = document.get("approved_by")
    approved_at = document.get("approved_at_kst")
    if isinstance(approved_by, str):
        approver = approved_by.strip()
    elif isinstance(approved_by, Mapping):
        if set(approved_by) != {"user_id", "chat_id", "topic_id"} or any(
            not isinstance(approved_by[key], str) or not approved_by[key].strip()
            for key in ("user_id", "chat_id", "topic_id")
        ):
            raise ValueError(f"{label} approver is required")
        approver = canonical_json(dict(approved_by))
    else:
        approver = ""
    if not approver:
        raise ValueError(f"{label} approver is required")
    if not isinstance(approved_at, str) or not approved_at.strip():
        raise ValueError(f"{label} approval time is required")
    try:
        datetime.fromisoformat(approved_at)
    except ValueError as exc:
        raise ValueError(f"{label} approval time is invalid") from exc
    declared = _require_digest(document.get("digest"), label)
    aliases = {
        "policy": ("policy", "value", "data"),
        "meal constraints": ("meal_constraints", "constraints", "value", "data"),
        "catalog": ("catalog", "food_catalog", "foods", "value", "data"),
    }[label]
    value: object
    for key in aliases:
        if key in document:
            value = document[key]
            break
    else:
        value = {
            key: candidate
            for key, candidate in document.items()
            if key not in _ARTIFACT_METADATA
        }
    if value is None:
        raise ValueError(f"{label} value is missing")
    candidates = [value]
    if isinstance(value, Mapping) and "data" in value:
        candidates.append(value["data"])
    stripped = {
        key: candidate
        for key, candidate in document.items()
        if key not in _ARTIFACT_METADATA
    }
    candidates.append(stripped)
    candidates.append({
        key: candidate
        for key, candidate in document.items()
        if key != "digest"
    })
    if not any(digest(candidate) == declared for candidate in candidates):
        raise ValueError(f"{label} digest does not match content")
    if isinstance(document.get("enabled"), bool) and not document["enabled"]:
        raise ValueError(f"{label} is disabled")
    return value, str(version), approver, approved_at.strip()


def _private_adaptive_root(data_root: Path) -> Path:
    root = Path(data_root)
    if root.is_symlink() or not root.exists() or not root.is_dir():
        raise ValueError("adaptive customer data root is invalid")
    adaptive_root = root if root.name == "nutrition-plans" else root / "nutrition-plans"
    if (
        adaptive_root.is_symlink()
        or not adaptive_root.exists()
        or not adaptive_root.is_dir()
        or adaptive_root.stat().st_mode & 0o077
    ):
        raise ValueError("adaptive nutrition runtime is unavailable")
    return adaptive_root


def _read_approved_document(path: Path, label: str) -> tuple[object, str, str, str]:
    if path.is_symlink() or not path.exists() or not path.is_file():
        raise ValueError(f"{label} file is unavailable")
    try:
        if path.stat().st_mode & 0o077:
            raise ValueError(f"{label} file permissions are too broad")
    except OSError as exc:
        raise ValueError(f"{label} file is unavailable") from exc
    try:
        document = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, ValueError) as exc:
        raise ValueError(f"{label} file is invalid") from exc
    if not isinstance(document, Mapping):
        raise ValueError(f"{label} file must contain an object")
    return _approved_value(document, label)
_POLICY_REVISION_FIELDS = frozenset({
    "schema_version",
    "kind",
    "append_sequence",
    "intent_id",
    "state",
    "customer_key",
    "policy_revision_digest",
    "policy_document",
    "supersedes_digest",
    "activation_receipt_id",
    "authority_digest",
    "authority",
    "approved_by",
    "approved_at_kst",
    "extension_through",
    "actor",
    "dedupe_key",
    "prepared_digest",
    "row_digest",
})
_POLICY_AUDIT_FIELDS = frozenset({
    "schema_version",
    "kind",
    "append_sequence",
    "customer_key",
    "policy_revision_digest",
    "prior_digest",
    "activation_receipt_id",
    "authority_digest",
    "extension_through",
    "actor",
    "approved_at_kst",
    "dedupe_key",
    "row_digest",
})


@contextmanager
def _policy_locked(root: Path) -> Any:
    with _verified_private_lock(
        root / ".adaptive.lock",
        exclusive=True,
        create=True,
    ):
        yield


def _read_json_document(path: Path, label: str) -> dict[str, object]:
    if path.is_symlink() or not path.exists() or not path.is_file():
        raise ValueError(f"{label} file is unavailable")
    try:
        if path.stat().st_mode & 0o077:
            raise ValueError(f"{label} file permissions are too broad")
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} file is invalid") from exc
    if not isinstance(value, Mapping):
        raise ValueError(f"{label} file must contain an object")
    return dict(value)


def _policy_value(document: Mapping[str, object]) -> tuple[dict[str, object], str | None]:
    for key in ("policy", "value", "data"):
        candidate = document.get(key)
        if isinstance(candidate, Mapping):
            return dict(candidate), key
    metadata = _ARTIFACT_METADATA | {
        "customer_key",
        "extension_through",
        "supersedes_digest",
        "activation_receipt_id",
        "authority_digest",
    }
    value = {
        key: candidate
        for key, candidate in document.items()
        if key not in metadata
    }
    if not value:
        raise ValueError("policy value is missing")
    return value, None


def _policy_document_with_value(
    document: Mapping[str, object],
    value: Mapping[str, object],
) -> dict[str, object]:
    result = dict(document)
    for key in ("policy", "value", "data"):
        if key in result:
            result[key] = dict(value)
            return result
    metadata = _ARTIFACT_METADATA | {
        "customer_key",
        "extension_through",
        "supersedes_digest",
        "activation_receipt_id",
        "authority_digest",
    }
    for key in tuple(result):
        if key not in metadata:
            del result[key]
    result.update(value)
    return result


def _kst_timestamp_is_exact(value: object) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError("policy approval time is invalid")
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError as exc:
        raise ValueError("policy approval time is invalid") from exc
    if parsed.tzinfo is None or parsed.utcoffset() != timedelta(hours=9):
        raise ValueError("policy approval time must be KST")
    return parsed.isoformat()


def _policy_date(value: object, label: str) -> date:
    if type(value) is not str:
        raise ValueError(f"{label} is invalid")
    try:
        parsed = date.fromisoformat(value)
    except ValueError as exc:
        raise ValueError(f"{label} is invalid") from exc
    if parsed.isoformat() != value:
        raise ValueError(f"{label} is invalid")
    return parsed


def _policy_actor(value: object) -> dict[str, str]:
    if not isinstance(value, Mapping):
        raise ValueError("policy authority actor is invalid")
    actor = {
        key: value.get(key)
        for key in ("user_id", "chat_id", "topic_id")
    }
    if any(not isinstance(candidate, str) or not candidate for candidate in actor.values()):
        raise ValueError("policy authority actor is invalid")
    return actor


def _policy_rows_path(root: Path) -> Path:
    return root / _POLICY_REVISION_FILE


def _policy_audit_path(root: Path) -> Path:
    return root / _POLICY_AUDIT_FILE


def _recover_policy_jsonl(path: Path) -> None:
    if path.is_symlink():
        raise ValueError("adaptive policy journal symlink is not allowed")
    if not path.exists():
        return
    if not path.is_file():
        raise ValueError("adaptive policy journal must be a regular file")
    try:
        if path.stat().st_mode & 0o077:
            raise ValueError("adaptive policy journal permissions are too broad")
    except OSError as exc:
        raise ValueError("adaptive policy journal is unavailable") from exc
    rows, complete_end = _read_jsonl_bytes(path, allow_truncated_tail=True)
    current_size = path.stat().st_size
    if complete_end < current_size:
        with path.open("r+b") as handle:
            handle.truncate(complete_end)
        path.chmod(0o600)
    if path == _policy_rows_path(path.parent):
        _validate_policy_revision_rows(rows, allow_pending=True)
    else:
        for expected_sequence, row in enumerate(rows, start=1):
            _validate_policy_audit_row(row)
            if row.get("append_sequence") != expected_sequence:
                raise ValueError("policy audit sequence is not contiguous")


def _read_policy_jsonl(path: Path) -> list[dict[str, object]]:
    if path.is_symlink():
        raise ValueError("adaptive policy journal symlink is not allowed")
    if not path.exists():
        return []
    if not path.is_file():
        raise ValueError("adaptive policy journal must be a regular file")
    try:
        if path.stat().st_mode & 0o077:
            raise ValueError("adaptive policy journal permissions are too broad")
    except OSError as exc:
        raise ValueError("adaptive policy journal is unavailable") from exc
    return _read_jsonl_rows(path)

def _append_policy_jsonl(path: Path, row: Mapping[str, object]) -> dict[str, object]:
    if path.is_symlink():
        raise ValueError("adaptive policy journal symlink is not allowed")
    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(canonical_json(dict(row)) + "\n")
        handle.flush()
        os.fsync(handle.fileno())
    return dict(row)


def _validate_policy_document(
    document: Mapping[str, object],
    *,
    customer_key: str,
    revision_digest: str,
    extension_through: str,
    supersedes_digest: str,
    activation_receipt_id: str,
    authority_digest: str,
    approved_by: str,
    approved_at_kst: str,
    authority: Mapping[str, object],
    actor: Mapping[str, str],
) -> None:
    if document.get("schema_version", _POLICY_SCHEMA_VERSION) != _POLICY_SCHEMA_VERSION:
        raise ValueError("policy revision schema mismatch")
    if document.get("customer_key") != customer_key:
        raise ValueError("policy revision customer binding mismatch")
    if document.get("approved") is not True:
        raise ValueError("policy revision is not approved")
    if document.get("approved_by") != approved_by or document.get("approved_at_kst") != approved_at_kst:
        raise ValueError("policy revision approval metadata mismatch")
    if document.get("extension_through") != extension_through:
        raise ValueError("policy revision extension date mismatch")
    if document.get("supersedes_digest") != supersedes_digest:
        raise ValueError("policy revision predecessor mismatch")
    if document.get("activation_receipt_id") != activation_receipt_id:
        raise ValueError("policy revision activation receipt mismatch")
    if document.get("authority_digest") != authority_digest:
        raise ValueError("policy revision authority mismatch")
    _require_digest(revision_digest, "policy revision")
    if document.get("digest") != revision_digest:
        raise ValueError("policy revision digest mismatch")
    _policy_date(extension_through, "policy extension date")
    _kst_timestamp_is_exact(approved_at_kst)
    if not isinstance(activation_receipt_id, str) or not activation_receipt_id:
        raise ValueError("policy revision activation receipt is invalid")
    _require_digest(supersedes_digest, "policy predecessor")
    _require_digest(authority_digest, "policy authority")
    if not isinstance(approved_by, str) or not approved_by:
        raise ValueError("policy revision approver is invalid")
    parsed_actor = _policy_actor(actor)
    if dict(parsed_actor) != dict(actor):
        raise ValueError("policy revision actor mismatch")
    authority_value = dict(authority)
    if set(authority_value) != {
        "schema_version",
        "customer_key",
        "extension_through",
        "activation_receipt_id",
        "actor",
    }:
        raise ValueError("policy revision authority schema mismatch")
    if authority_value.get("schema_version") != _POLICY_SCHEMA_VERSION:
        raise ValueError("policy revision authority schema mismatch")
    if authority_value.get("customer_key") != customer_key:
        raise ValueError("policy revision authority customer mismatch")
    if authority_value.get("extension_through") != extension_through:
        raise ValueError("policy revision authority date mismatch")
    if authority_value.get("activation_receipt_id") != activation_receipt_id:
        raise ValueError("policy revision authority receipt mismatch")
    if dict(_policy_actor(authority_value.get("actor"))) != dict(actor):
        raise ValueError("policy revision authority actor mismatch")
    if digest(authority_value) != authority_digest:
        raise ValueError("policy revision authority digest mismatch")
    value, _ = _policy_value(document)
    declared_value_digest = _require_digest(document.get("digest"), "policy")
    if digest(value) != declared_value_digest:
        raise ValueError("policy revision policy digest mismatch")
    if value.get("extended_through") != extension_through:
        raise ValueError("policy revision policy window mismatch")
    _parse_policy(value)


def _validate_policy_revision_row(row: Mapping[str, object]) -> None:
    if set(row) != _POLICY_REVISION_FIELDS:
        raise ValueError("policy revision journal schema mismatch")
    body = {key: value for key, value in row.items() if key != "row_digest"}
    if row.get("schema_version") != _POLICY_SCHEMA_VERSION or row.get("kind") != "policy_revision":
        raise ValueError("policy revision journal schema mismatch")
    sequence = row.get("append_sequence")
    if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence <= 0:
        raise ValueError("policy revision journal sequence is invalid")
    if row.get("row_digest") != digest(body):
        raise ValueError("policy revision journal row digest mismatch")
    customer_key = row.get("customer_key")
    intent_id = row.get("intent_id")
    state = row.get("state")
    if not isinstance(customer_key, str) or not customer_key:
        raise ValueError("policy revision customer is invalid")
    if not isinstance(intent_id, str) or not intent_id:
        raise ValueError("policy revision intent is invalid")
    if state not in {"prepared", "committed"}:
        raise ValueError("policy revision state is invalid")
    if state == "prepared" and row.get("prepared_digest") is not None:
        raise ValueError("prepared policy revision has a terminal digest")
    if state == "committed":
        _require_digest(row.get("prepared_digest"), "policy prepared row")
    revision_digest = _require_digest(row.get("policy_revision_digest"), "policy revision")
    extension_through = row.get("extension_through")
    _policy_date(extension_through, "policy extension date")
    supersedes_digest = _require_digest(row.get("supersedes_digest"), "policy predecessor")
    authority_digest = _require_digest(row.get("authority_digest"), "policy authority")
    approved_at_kst = _kst_timestamp_is_exact(row.get("approved_at_kst"))
    actor = _policy_actor(row.get("actor"))
    authority = row.get("authority")
    if not isinstance(authority, Mapping):
        raise ValueError("policy revision authority is invalid")
    approved_by = row.get("approved_by")
    activation_receipt_id = row.get("activation_receipt_id")
    if not isinstance(approved_by, str) or not approved_by:
        raise ValueError("policy revision approver is invalid")
    if not isinstance(activation_receipt_id, str) or not activation_receipt_id:
        raise ValueError("policy revision activation receipt is invalid")
    dedupe_key = row.get("dedupe_key")
    if not isinstance(dedupe_key, str) or not dedupe_key:
        raise ValueError("policy revision dedupe key is invalid")
    document = row.get("policy_document")
    if not isinstance(document, Mapping):
        raise ValueError("policy revision document is invalid")
    _validate_policy_document(
        document,
        customer_key=customer_key,
        revision_digest=revision_digest,
        extension_through=extension_through,
        supersedes_digest=supersedes_digest,
        activation_receipt_id=activation_receipt_id,
        authority_digest=authority_digest,
        approved_by=approved_by,
        approved_at_kst=approved_at_kst,
        authority=authority,
        actor=actor,
    )


def _validate_policy_revision_rows(
    rows: Sequence[Mapping[str, object]],
    *,
    base_digest: str | None = None,
    allow_pending: bool = False,
) -> 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):
        _validate_policy_revision_row(row)
        if row.get("append_sequence") != expected_sequence:
            raise ValueError("policy revision journal sequence is not contiguous")
        intent_id = str(row["intent_id"])
        previous = states.get(intent_id)
        if previous is None:
            if row["state"] != "prepared":
                raise ValueError("policy revision terminal row has no prepared row")
            states[intent_id] = dict(row)
            continue
        if previous["state"] != "prepared" or row["state"] != "committed":
            raise ValueError("policy revision transition is invalid")
        if row.get("prepared_digest") != previous.get("row_digest"):
            raise ValueError("policy revision prepared digest mismatch")
        for key in _POLICY_REVISION_FIELDS - {
            "state",
            "append_sequence",
            "prepared_digest",
            "row_digest",
        }:
            if row.get(key) != previous.get(key):
                raise ValueError("policy revision prepared payload mismatch")
        states[intent_id] = dict(row)
        committed.append(dict(row))
    pending = [row for row in states.values() if row.get("state") == "prepared"]
    if pending and not allow_pending:
        raise ValueError("policy revision is incomplete")
    if base_digest is not None:
        _require_digest(base_digest, "policy base")
        expected_parent = base_digest
        for row in committed:
            if row.get("supersedes_digest") != expected_parent:
                raise ValueError("policy revision chain is invalid")
            expected_parent = str(row["policy_revision_digest"])
    return committed, pending


def _validate_policy_audit_row(row: Mapping[str, object]) -> None:
    if set(row) != _POLICY_AUDIT_FIELDS:
        raise ValueError("policy audit schema mismatch")
    body = {key: value for key, value in row.items() if key != "row_digest"}
    if row.get("schema_version") != _POLICY_SCHEMA_VERSION or row.get("kind") != "policy_extension":
        raise ValueError("policy audit schema mismatch")
    sequence = row.get("append_sequence")
    if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence <= 0:
        raise ValueError("policy audit sequence is invalid")
    if row.get("row_digest") != digest(body):
        raise ValueError("policy audit row digest mismatch")
    if not isinstance(row.get("customer_key"), str) or not row["customer_key"]:
        raise ValueError("policy audit customer is invalid")
    _require_digest(row.get("policy_revision_digest"), "policy audit revision")
    _require_digest(row.get("prior_digest"), "policy audit predecessor")
    _require_digest(row.get("authority_digest"), "policy audit authority")
    _policy_date(row.get("extension_through"), "policy audit extension date")
    _kst_timestamp_is_exact(row.get("approved_at_kst"))
    _policy_actor(row.get("actor"))
    if not isinstance(row.get("activation_receipt_id"), str) or not row["activation_receipt_id"]:
        raise ValueError("policy audit activation receipt is invalid")
    if not isinstance(row.get("dedupe_key"), str) or not row["dedupe_key"]:
        raise ValueError("policy audit dedupe key is invalid")


def _validate_policy_audit_rows(
    rows: Sequence[Mapping[str, object]],
    committed: Sequence[Mapping[str, object]],
    *,
    require_all: bool,
) -> dict[str, dict[str, object]]:
    receipts: dict[str, dict[str, object]] = {}
    for expected_sequence, row in enumerate(rows, start=1):
        _validate_policy_audit_row(row)
        if row.get("append_sequence") != expected_sequence:
            raise ValueError("policy audit sequence is not contiguous")
        revision_digest = str(row["policy_revision_digest"])
        if revision_digest in receipts:
            raise ValueError("policy audit receipt is duplicated")
        receipts[revision_digest] = dict(row)
    committed_by_digest = {
        str(row["policy_revision_digest"]): row
        for row in committed
    }
    if any(revision not in committed_by_digest for revision in receipts):
        raise ValueError("policy audit references an unknown revision")
    for revision_digest, row in committed_by_digest.items():
        receipt = receipts.get(revision_digest)
        if receipt is None:
            if require_all:
                raise ValueError("policy audit receipt is missing")
            continue
        if any(
            receipt.get(audit_key) != row.get(row_key)
            for audit_key, row_key in (
                ("customer_key", "customer_key"),
                ("prior_digest", "supersedes_digest"),
                ("activation_receipt_id", "activation_receipt_id"),
                ("authority_digest", "authority_digest"),
                ("extension_through", "extension_through"),
                ("actor", "actor"),
                ("approved_at_kst", "approved_at_kst"),
                ("dedupe_key", "dedupe_key"),
            )
        ):
            raise ValueError("policy audit receipt does not match revision")
    return receipts


def _resolve_policy_document(root: Path) -> dict[str, object]:
    policy_path = root / "policy.json"
    base_document = _read_json_document(policy_path, "policy")
    _approved_value(base_document, "policy")
    base_digest = _require_digest(base_document.get("digest"), "policy")
    revisions_path = _policy_rows_path(root)
    audit_path = _policy_audit_path(root)
    rows = _read_policy_jsonl(revisions_path)
    committed, pending = _validate_policy_revision_rows(
        rows,
        base_digest=base_digest,
        allow_pending=False,
    )
    if pending:
        raise ValueError("policy revision is incomplete")
    audit_rows = _read_policy_jsonl(audit_path)
    _validate_policy_audit_rows(audit_rows, committed, require_all=True)
    if not committed:
        return base_document
    return dict(committed[-1]["policy_document"])


def _policy_row(
    *,
    append_sequence: int,
    intent_id: str,
    state: str,
    customer_key: str,
    policy_revision_digest: str,
    policy_document: Mapping[str, object],
    supersedes_digest: str,
    activation_receipt_id: str,
    authority_digest: str,
    authority: Mapping[str, object],
    approved_by: str,
    approved_at_kst: str,
    extension_through: str,
    actor: Mapping[str, str],
    dedupe_key: str,
    prepared_digest: str | None,
) -> dict[str, object]:
    body = {
        "schema_version": _POLICY_SCHEMA_VERSION,
        "kind": "policy_revision",
        "append_sequence": append_sequence,
        "intent_id": intent_id,
        "state": state,
        "customer_key": customer_key,
        "policy_revision_digest": policy_revision_digest,
        "policy_document": dict(policy_document),
        "supersedes_digest": supersedes_digest,
        "activation_receipt_id": activation_receipt_id,
        "authority_digest": authority_digest,
        "authority": dict(authority),
        "approved_by": approved_by,
        "approved_at_kst": approved_at_kst,
        "extension_through": extension_through,
        "actor": dict(actor),
        "dedupe_key": dedupe_key,
        "prepared_digest": prepared_digest,
    }
    return {**body, "row_digest": digest(body)}


def _policy_audit_row(
    *,
    append_sequence: int,
    customer_key: str,
    policy_revision_digest: str,
    prior_digest: str,
    activation_receipt_id: str,
    authority_digest: str,
    extension_through: str,
    actor: Mapping[str, str],
    approved_at_kst: str,
    dedupe_key: str,
) -> dict[str, object]:
    body = {
        "schema_version": _POLICY_SCHEMA_VERSION,
        "kind": "policy_extension",
        "append_sequence": append_sequence,
        "customer_key": customer_key,
        "policy_revision_digest": policy_revision_digest,
        "prior_digest": prior_digest,
        "activation_receipt_id": activation_receipt_id,
        "authority_digest": authority_digest,
        "extension_through": extension_through,
        "actor": dict(actor),
        "approved_at_kst": approved_at_kst,
        "dedupe_key": dedupe_key,
    }
    return {**body, "row_digest": digest(body)}


def append_approved_policy_extension(
    data_root: Path,
    *,
    customer_key: str,
    extension_through: date,
    activation_receipt_id: str,
    authority_digest: str,
    authority: Mapping[str, object],
    approved_by: str,
    approved_at_kst: str,
) -> Mapping[str, str]:
    """Append one customer-bound approved policy revision without rewriting history."""
    if type(extension_through) is not date:
        raise ValueError("adaptive extension date is invalid")
    extension_day = extension_through.isoformat()
    if not isinstance(customer_key, str) or not customer_key:
        raise ValueError("policy customer key is invalid")
    if not isinstance(activation_receipt_id, str) or not activation_receipt_id:
        raise ValueError("policy activation receipt is invalid")
    if not isinstance(approved_by, str) or not approved_by:
        raise ValueError("policy approver is invalid")
    approved_at_kst = _kst_timestamp_is_exact(approved_at_kst)
    actor = _policy_actor(authority.get("actor") if isinstance(authority, Mapping) else None)
    authority = dict(authority)
    if digest(authority) != authority_digest:
        raise ValueError("policy authority digest mismatch")
    _require_digest(authority_digest, "policy authority")
    root = _private_adaptive_root(data_root)
    policy_path = root / "policy.json"
    rows_path = _policy_rows_path(root)
    audit_path = _policy_audit_path(root)
    with _policy_locked(root):
        _recover_policy_jsonl(rows_path)
        _recover_policy_jsonl(audit_path)
        base_document = _read_json_document(policy_path, "policy")
        _approved_value(base_document, "policy")
        base_digest = _require_digest(base_document.get("digest"), "policy")
        rows = _read_policy_jsonl(rows_path)
        committed, pending = _validate_policy_revision_rows(
            rows,
            base_digest=base_digest,
            allow_pending=True,
        )
        audit_rows = _read_policy_jsonl(audit_path)
        receipts = _validate_policy_audit_rows(
            audit_rows,
            committed,
            require_all=False,
        )
        dedupe_key = (
            f"adaptive-policy-extension:{customer_key}:"
            f"{extension_day}:{authority_digest}"
        )
        matching = next(
            (row for row in committed if row.get("dedupe_key") == dedupe_key),
            None,
        )
        if matching is not None:
            if (
                matching.get("customer_key") != customer_key
                or matching.get("extension_through") != extension_day
                or matching.get("activation_receipt_id") != activation_receipt_id
                or matching.get("authority_digest") != authority_digest
                or matching.get("actor") != actor
            ):
                raise ValueError("conflicting adaptive policy extension replay")
            if str(matching["policy_revision_digest"]) not in receipts:
                audit = _policy_audit_row(
                    append_sequence=len(audit_rows) + 1,
                    customer_key=str(matching["customer_key"]),
                    policy_revision_digest=str(matching["policy_revision_digest"]),
                    prior_digest=str(matching["supersedes_digest"]),
                    activation_receipt_id=str(matching["activation_receipt_id"]),
                    authority_digest=str(matching["authority_digest"]),
                    extension_through=str(matching["extension_through"]),
                    actor=_policy_actor(matching["actor"]),
                    approved_at_kst=str(matching["approved_at_kst"]),
                    dedupe_key=str(matching["dedupe_key"]),
                )
                _append_policy_jsonl(audit_path, audit)
            return {
                _POLICY_REVISION_FILE: hashlib.sha256(rows_path.read_bytes()).hexdigest(),
                _POLICY_AUDIT_FILE: hashlib.sha256(audit_path.read_bytes()).hexdigest(),
                "policy.json": hashlib.sha256(policy_path.read_bytes()).hexdigest(),
            }
        pending_match = next(
            (row for row in pending if row.get("dedupe_key") == dedupe_key),
            None,
        )
        if pending_match is not None:
            if (
                pending_match.get("customer_key") != customer_key
                or pending_match.get("extension_through") != extension_day
                or pending_match.get("activation_receipt_id") != activation_receipt_id
                or pending_match.get("authority_digest") != authority_digest
                or pending_match.get("actor") != actor
            ):
                raise ValueError("conflicting incomplete adaptive policy extension")
            commit = _policy_row(
                append_sequence=len(rows) + 1,
                intent_id=str(pending_match["intent_id"]),
                state="committed",
                customer_key=str(pending_match["customer_key"]),
                policy_revision_digest=str(pending_match["policy_revision_digest"]),
                policy_document=pending_match["policy_document"],
                supersedes_digest=str(pending_match["supersedes_digest"]),
                activation_receipt_id=str(pending_match["activation_receipt_id"]),
                authority_digest=str(pending_match["authority_digest"]),
                authority=pending_match["authority"],
                approved_by=str(pending_match["approved_by"]),
                approved_at_kst=str(pending_match["approved_at_kst"]),
                extension_through=str(pending_match["extension_through"]),
                actor=_policy_actor(pending_match["actor"]),
                dedupe_key=str(pending_match["dedupe_key"]),
                prepared_digest=str(pending_match["row_digest"]),
            )
            _append_policy_jsonl(rows_path, commit)
            rows.append(commit)
            committed, pending = _validate_policy_revision_rows(
                rows,
                base_digest=base_digest,
                allow_pending=True,
            )
            if str(commit["policy_revision_digest"]) not in receipts:
                audit = _policy_audit_row(
                    append_sequence=len(audit_rows) + 1,
                    customer_key=str(commit["customer_key"]),
                    policy_revision_digest=str(commit["policy_revision_digest"]),
                    prior_digest=str(commit["supersedes_digest"]),
                    activation_receipt_id=str(commit["activation_receipt_id"]),
                    authority_digest=str(commit["authority_digest"]),
                    extension_through=str(commit["extension_through"]),
                    actor=_policy_actor(commit["actor"]),
                    approved_at_kst=str(commit["approved_at_kst"]),
                    dedupe_key=str(commit["dedupe_key"]),
                )
                _append_policy_jsonl(audit_path, audit)
            return {
                _POLICY_REVISION_FILE: hashlib.sha256(rows_path.read_bytes()).hexdigest(),
                _POLICY_AUDIT_FILE: hashlib.sha256(audit_path.read_bytes()).hexdigest(),
                "policy.json": hashlib.sha256(policy_path.read_bytes()).hexdigest(),
            }
        if pending and pending_match is None:
            raise ValueError("another adaptive policy extension requires recovery")
        current_document = (
            dict(committed[-1]["policy_document"])
            if committed
            else base_document
        )
        current_digest = _require_digest(current_document.get("digest"), "policy")
        current_value, _ = _policy_value(current_document)
        prior_extension = current_value.get("extended_through")
        if prior_extension is not None and extension_day <= str(prior_extension):
            raise ValueError("adaptive extension must be append-only")
        base_value, _ = _policy_value(current_document)
        new_value = dict(base_value)
        new_value.pop("extension_through", None)
        new_value["extended_through"] = extension_day
        _, current_version, _, _ = _approved_value(current_document, "policy")
        new_document = _policy_document_with_value(current_document, new_value)
        new_document.update({
            "schema_version": _POLICY_SCHEMA_VERSION,
            "customer_key": customer_key,
            "version": f"{current_version}.extension-{len(committed) + 1}",
            "digest": digest(new_value),
            "approved": True,
            "approved_by": approved_by,
            "approved_at_kst": approved_at_kst,
            "extension_through": extension_day,
            "supersedes_digest": current_digest,
            "activation_receipt_id": activation_receipt_id,
            "authority_digest": authority_digest,
        })
        revision_digest = str(new_document["digest"])
        intent_id = f"policy:{customer_key}:{revision_digest}"
        prepared = _policy_row(
            append_sequence=len(rows) + 1,
            intent_id=intent_id,
            state="prepared",
            customer_key=customer_key,
            policy_revision_digest=revision_digest,
            policy_document=new_document,
            supersedes_digest=current_digest,
            activation_receipt_id=activation_receipt_id,
            authority_digest=authority_digest,
            authority=authority,
            approved_by=approved_by,
            approved_at_kst=approved_at_kst,
            extension_through=extension_day,
            actor=actor,
            dedupe_key=dedupe_key,
            prepared_digest=None,
        )
        _append_policy_jsonl(rows_path, prepared)
        committed_row = _policy_row(
            append_sequence=len(rows) + 2,
            intent_id=intent_id,
            state="committed",
            customer_key=customer_key,
            policy_revision_digest=revision_digest,
            policy_document=new_document,
            supersedes_digest=current_digest,
            activation_receipt_id=activation_receipt_id,
            authority_digest=authority_digest,
            authority=authority,
            approved_by=approved_by,
            approved_at_kst=approved_at_kst,
            extension_through=extension_day,
            actor=actor,
            dedupe_key=dedupe_key,
            prepared_digest=str(prepared["row_digest"]),
        )
        _append_policy_jsonl(rows_path, committed_row)
        audit = _policy_audit_row(
            append_sequence=len(audit_rows) + 1,
            customer_key=customer_key,
            policy_revision_digest=revision_digest,
            prior_digest=current_digest,
            activation_receipt_id=activation_receipt_id,
            authority_digest=authority_digest,
            extension_through=extension_day,
            actor=actor,
            approved_at_kst=approved_at_kst,
            dedupe_key=dedupe_key,
        )
        _append_policy_jsonl(audit_path, audit)
    return {
        _POLICY_REVISION_FILE: hashlib.sha256(rows_path.read_bytes()).hexdigest(),
        _POLICY_AUDIT_FILE: hashlib.sha256(audit_path.read_bytes()).hexdigest(),
        "policy.json": hashlib.sha256(policy_path.read_bytes()).hexdigest(),
    }




def _parse_policy(value: object) -> CustomerPolicy:
    if not isinstance(value, Mapping):
        raise ValueError("policy value must be an object")
    raw = dict(value)
    raw.pop("enabled", None)
    try:
        if isinstance(raw.get("starts_on"), str):
            raw["starts_on"] = date.fromisoformat(raw["starts_on"])
        if "extended_through" not in raw and isinstance(raw.get("extension_through"), str):
            raw["extended_through"] = raw.pop("extension_through")
        if isinstance(raw.get("extended_through"), str):
            raw["extended_through"] = date.fromisoformat(raw["extended_through"])
        for key in ("weekly_rate_min", "weekly_rate_max", "adherence_tolerance_percent"):
            if raw.get(key) is not None and not isinstance(raw[key], Decimal):
                raw[key] = Decimal(str(raw[key]))
        return CustomerPolicy(**raw)
    except (ArithmeticError, TypeError, ValueError) as exc:
        raise ValueError("approved policy is invalid") from exc


def _parse_constraints(value: object) -> MealConstraints:
    if not isinstance(value, Mapping):
        raise ValueError("meal constraints value must be an object")
    raw = dict(value)
    for key in ("allergies", "excluded_food_ids", "restrictions", "digestion_exclusions"):
        if raw.get(key) is not None:
            raw[key] = frozenset(str(item) for item in raw[key])
    if raw.get("preferences") is not None:
        raw["preferences"] = tuple(str(item) for item in raw["preferences"])
    for time_key in ("training_times", "training_time_by_day"):
        if raw.get(time_key) is not None:
            source = raw[time_key]
            if isinstance(source, Mapping):
                source = source.items()
            if not isinstance(source, (list, tuple)):
                raise ValueError("approved training times are invalid")
            parsed_times = []
            for item in source:
                if isinstance(item, Mapping):
                    day = item.get("kst_day", item.get("day"))
                    training_time = item.get("training_time", item.get("time"))
                else:
                    try:
                        day, training_time = item
                    except (TypeError, ValueError) as exc:
                        raise ValueError("approved training times are invalid") from exc
                parsed_times.append((date.fromisoformat(str(day)[:10]), str(training_time)))
            raw[time_key] = tuple(sorted(parsed_times))
    if raw.get("meal_shares_bps") is not None:
        shares = raw["meal_shares_bps"]
        if isinstance(shares, Mapping):
            shares = shares.items()
        raw["meal_shares_bps"] = tuple((str(name), int(value)) for name, value in shares)
    for key in (
        "meal_slot_calorie_tolerance_percent",
        "meal_slot_macro_tolerance_percent",
        "calorie_tolerance_percent",
        "macro_tolerance_percent",
    ):
        if raw.get(key) is not None:
            raw[key] = Decimal(str(raw[key]))
    raw["strict_inputs"] = bool(raw.get("strict_inputs", False) or raw.get("version") is not None)
    try:
        constraints = MealConstraints(**raw)
    except (TypeError, ValueError, ArithmeticError) as exc:
        raise ValueError("approved meal constraints are invalid") from exc
    if not constraints.complete:
        raise ValueError("approved meal constraints are incomplete")
    return constraints


def _parse_catalog(value: object) -> tuple[Food, ...]:
    rows = value.get("foods") if isinstance(value, Mapping) and "foods" in value else value
    if not isinstance(rows, (list, tuple)) or not rows:
        raise ValueError("approved catalog is empty or invalid")
    foods: list[Food] = []
    for row in rows:
        if not isinstance(row, Mapping):
            raise ValueError("approved catalog contains an invalid food")
        raw = dict(row)
        for key in (
            "allergens",
            "tags",
            "allowed_budget_bands",
            "allowed_cooking_access",
        ):
            if raw.get(key) is not None:
                raw[key] = frozenset(str(item) for item in raw[key])
        try:
            food = Food(**raw)
        except (TypeError, ValueError) as exc:
            raise ValueError("approved catalog contains an invalid food") from exc
        if not food.food_id.strip() or food.calories < 0 or min(
            food.carbs_g, food.protein_g, food.fat_g
        ) < 0 or not _valid_serving_bounds(food):
            raise ValueError("approved catalog contains an invalid food")
        foods.append(food)
    if len({food.food_id for food in foods}) != len(foods):
        raise ValueError("approved catalog contains duplicate food ids")
    return tuple(foods)


def load_approved_adaptive_artifacts(data_root: Path) -> ApprovedAdaptiveArtifacts:
    """Load only approved, digest-verified private adaptive artifacts."""
    root = _private_adaptive_root(data_root)
    policy_document = _resolve_policy_document(root)
    policy_value, policy_version, approved_by, approved_at = _approved_value(
        policy_document,
        "policy",
    )
    if (
        isinstance(policy_value, Mapping)
        and policy_document.get("extension_through") is not None
        and "extended_through" not in policy_value
    ):
        policy_value = {
            **policy_value,
            "extended_through": policy_document["extension_through"],
        }
    constraints_value, constraints_version, constraints_by, constraints_at = _read_approved_document(
        root / "meal-constraints.json", "meal constraints"
    )
    catalog_path = root / "food-catalog.json"
    if not (catalog_path.exists() or catalog_path.is_symlink()):
        catalog_path = root / "catalog.json"
    catalog_value, catalog_version, catalog_by, catalog_at = _read_approved_document(
        catalog_path, "catalog"
    )
    constraints_document = json.loads(
        (root / "meal-constraints.json").read_text(encoding="utf-8")
    )
    catalog_document = json.loads(catalog_path.read_text(encoding="utf-8"))
    policy_digest = _require_digest(policy_document.get("digest"), "policy")
    constraints_digest = _require_digest(
        constraints_document.get("digest"), "meal constraints"
    )
    catalog_digest = _require_digest(catalog_document.get("digest"), "catalog")
    catalog = VersionedFoodCatalog(
        schema_version=str(catalog_document.get("schema_version", "1.0")),
        foods=_parse_catalog(catalog_value),
        version=catalog_version,
        digest=catalog_digest,
        approved=True,
        approved_by=catalog_by,
        approved_at_kst=catalog_at,
    )
    policy = (
        customer_policy_from_onboarding_artifact(policy_value)
        if isinstance(policy_value, Mapping)
        and any(
            marker in policy_value
            for marker in ("policy_version", "effective_from", "maximum_step_kcal")
        )
        else _parse_policy(policy_value)
    )
    return ApprovedAdaptiveArtifacts(
        policy,
        _parse_constraints(constraints_value),
        catalog,
        policy_version,
        constraints_version,
        catalog_version,
        policy_digest,
        constraints_digest,
        catalog_digest,
        approved_by,
        approved_at,
    )


def canonical_event_records(events: Iterable[object]) -> tuple[Mapping[str, object], ...]:
    """Return deterministic JSON records for canonical Event models."""
    records: list[Mapping[str, object]] = []
    by_id: dict[str, Mapping[str, object]] = {}
    for event in events:
        model_dump = getattr(event, "model_dump", None)
        if callable(model_dump):
            value = model_dump(mode="json", exclude_none=True)
        else:
            value = event
        if not isinstance(value, Mapping):
            raise ValueError("canonical event is not a mapping")
        event_id = value.get("event_id")
        if not isinstance(event_id, str) or not event_id:
            raise ValueError("canonical event id is invalid")
        record = dict(value)
        if event_id in by_id:
            if canonical_json(by_id[event_id]) != canonical_json(record):
                raise ValueError("conflicting duplicate canonical event id")
            continue
        by_id[event_id] = record
        records.append(record)
    return tuple(sorted(records, key=lambda value: str(value["event_id"])))


def canonical_event_digest(events: Iterable[object]) -> str:
    return digest(canonical_event_records(events))
def root_intent_preimage(
    *,
    customer_key: str,
    session_id: str,
    mapped_flow: str,
    observation_kst_day: date | str,
    event_projection: Mapping[str, object],
) -> Mapping[str, object]:
    projection = {
        key: value
        for key, value in dict(event_projection).items()
        if key not in {"event_id", "occurred_at_kst", "recorded_at_kst"}
    }
    return {
        "schema_version": "1.0",
        "customer_key": str(customer_key),
        "session_id": str(session_id),
        "mapped_flow": str(mapped_flow),
        "observation_kst_day": _iso_day(observation_kst_day),
        "event_projection": projection,
    }


def root_intent_preimage_digest(**kwargs: object) -> str:
    return digest(root_intent_preimage(**kwargs))


canonical_root_intent_digest = root_intent_preimage_digest
root_intent_digest = root_intent_preimage_digest
canonical_root_intent_preimage = root_intent_preimage


def validate_typed_safety(events: Iterable[object]) -> None:
    """Require typed safety reasons for every canonical coaching hold."""
    for event in events:
        status = getattr(event, "status", None)
        safety = getattr(event, "safety", None)
        if isinstance(event, Mapping):
            status = event.get("status")
            safety = event.get("safety")
        status_value = getattr(status, "value", status)
        held = bool(getattr(safety, "coaching_held", False))
        if isinstance(safety, Mapping):
            held = bool(safety.get("coaching_held"))
        if status_value != "unsafe" and not held:
            continue
        reasons = getattr(safety, "reasons", None)
        if isinstance(safety, Mapping):
            reasons = safety.get("reasons")
        if not isinstance(reasons, (tuple, list)) or not reasons:
            raise ValueError("canonical safety hold is missing typed reasons")
        for reason in reasons:
            if isinstance(reason, Mapping):
                rule_id = reason.get("rule_id")
                matched_field = reason.get("matched_field")
                excerpt = reason.get("excerpt")
            else:
                rule_id = getattr(reason, "rule_id", None)
                matched_field = getattr(reason, "matched_field", None)
                excerpt = getattr(reason, "excerpt", None)
            if not all((rule_id, matched_field, isinstance(excerpt, str) and excerpt.strip())):
                raise ValueError("canonical safety reason is not typed")



ADHERENCE_SIGNAL_VERSION = "1.0"
CARB_LOAD_VERSION = "1.0"


AdherenceSignal = CanonicalAdherenceSignal


def _coerce_int(value: object) -> int | None:
    if value is None or isinstance(value, bool):
        return None
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return None
    return parsed if parsed >= 0 else None


def _observation_metric(row: DailyObservation, name: str, *, target: bool) -> int | None:
    if target:
        names = {
            "calories": ("target_calories_kcal", "target_calories"),
            "carbs": ("target_carbohydrate_g", "target_carbs_g"),
            "protein": ("target_protein_g", "target_protein"),
            "fat": ("target_fat_g", "target_fat"),
        }[name]
    else:
        names = {
            "calories": ("actual_calories_kcal", "actual_calories", "calories"),
            "carbs": ("actual_carbohydrate_g", "actual_carbs_g", "carbs_g"),
            "protein": ("actual_protein_g", "actual_protein", "protein_g"),
            "fat": ("actual_fat_g", "actual_fat", "fat_g"),
        }[name]
    values = [getattr(row, candidate, None) for candidate in names]
    present = [candidate for candidate in values if candidate is not None]
    if not present:
        return None
    parsed = _coerce_int(present[0])
    if parsed is None or any(_coerce_int(candidate) != parsed for candidate in present[1:]):
        raise ValueError(f"contradictory {name} adherence values")
    return parsed


def _contradictory_adherence_signal(
    *,
    tolerance: Decimal,
    reason: str,
    source_event_ids: Sequence[str] = (),
) -> AdherenceSignal:
    return AdherenceSignal(
        schema_version=ADHERENCE_SIGNAL_VERSION,
        source_event_ids=tuple(sorted(set(str(value) for value in source_event_ids))),
        tolerance_percent=tolerance,
        complete=False,
        adherent=None,
        status="contradictory",
        reason=reason,
    )


def _legacy_adherence_signal(
    *,
    result: bool,
    tolerance: Decimal,
    reason: str,
    source_event_ids: Sequence[str] = (),
) -> AdherenceSignal:
    return AdherenceSignal(
        schema_version=ADHERENCE_SIGNAL_VERSION,
        source_event_ids=tuple(sorted(set(str(value) for value in source_event_ids))),
        tolerance_percent=tolerance,
        complete=True,
        adherent=result,
        status="adequate" if result else "inadequate",
        reason=reason,
    )


def _mapping_adherence_inputs(raw: Mapping[str, object]) -> tuple[dict[str, object], dict[str, object]]:
    """Split canonical target and actual maps without dropping partial evidence."""
    target_source = next(
        (
            value for key in ("target", "check_in_target", "nutrition_target")
            if isinstance((value := raw.get(key)), Mapping)
        ),
        None,
    )
    actual_source = next(
        (
            value for key in ("actual", "check_in", "nutrition_actual")
            if isinstance((value := raw.get(key)), Mapping)
        ),
        None,
    )
    if target_source is None:
        target_source = {
            key: value for key, value in raw.items()
            if str(key).startswith("target_")
        }
    if actual_source is None:
        actual_source = {
            key: value for key, value in raw.items()
            if str(key).startswith("actual_")
        }
    # A flat legacy mapping with unprefixed values is an actual-only check-in.
    if actual_source is None or not actual_source:
        actual_source = {
            key: value for key, value in raw.items()
            if key in {"calories", "calories_kcal", "carbs", "carbs_g", "carbohydrate_g",
                       "protein", "protein_g", "fat", "fat_g"}
        }
    return dict(target_source or {}), dict(actual_source or {})


def derive_adherence_signal(
    row: DailyObservation | Mapping[str, object] | CanonicalAdherenceSignal,
    *,
    tolerance_percent: Decimal | int | str = Decimal("10"),
    source_event_ids: Sequence[str] = (),
) -> AdherenceSignal:
    """Adapt observations to the canonical, lossless adherence model."""
    if isinstance(row, CanonicalAdherenceSignal):
        return row
    try:
        tolerance = Decimal(str(tolerance_percent))
    except ArithmeticError as exc:
        raise ValueError("adherence tolerance is invalid") from exc
    if tolerance < 0 or tolerance > 100:
        raise ValueError("adherence tolerance is invalid")
    mapping_source_event_ids = tuple(str(value) for value in source_event_ids)

    if isinstance(row, DailyObservation):
        try:
            target = {
                "calories_kcal": _observation_metric(row, "calories", target=True),
                "carbohydrate_g": _observation_metric(row, "carbs", target=True),
                "protein_g": _observation_metric(row, "protein", target=True),
                "fat_g": _observation_metric(row, "fat", target=True),
            }
            actual = {
                "calories_kcal": _observation_metric(row, "calories", target=False),
                "carbohydrate_g": _observation_metric(row, "carbs", target=False),
                "protein_g": _observation_metric(row, "protein", target=False),
                "fat_g": _observation_metric(row, "fat", target=False),
            }
        except ValueError:
            return _contradictory_adherence_signal(
                tolerance=tolerance,
                reason="contradictory_target_actual",
                source_event_ids=source_event_ids,
            )
        has_pairs = any(value is not None for value in (*target.values(), *actual.values()))
        if not has_pairs and row.adherence_ok is not None:
            return _legacy_adherence_signal(
                result=bool(row.adherence_ok),
                tolerance=tolerance,
                reason=(
                    "versioned_canonical_signal"
                    if row.adherence_version is not None
                    else "legacy_pinned_signal"
                ),
                source_event_ids=source_event_ids,
            )
    else:
        raw = dict(row)
        raw_source_event_ids = raw.get("source_event_ids")
        if isinstance(raw_source_event_ids, (list, tuple)):
            mapping_source_event_ids = tuple(str(value) for value in raw_source_event_ids)
        embedded = raw.get("adherence_signal")
        if isinstance(embedded, Mapping):
            try:
                return AdherenceSignal.model_validate(embedded)
            except Exception as exc:
                raise ValueError("canonical adherence signal is invalid") from exc
        if (
            "schema_version" in raw
            and "status" in raw
            and "complete" in raw
            and "adherent" in raw
        ):
            try:
                return AdherenceSignal.model_validate(raw)
            except Exception as exc:
                raise ValueError("canonical adherence signal is invalid") from exc
        target, actual = _mapping_adherence_inputs(raw)
        if (
            not target
            and not actual
            and isinstance(raw.get("adherence_ok"), bool)
        ):
            return _legacy_adherence_signal(
                result=bool(raw["adherence_ok"]),
                tolerance=tolerance,
                reason=(
                    "versioned_canonical_signal"
                    if raw.get("adherence_version") is not None
                    else "legacy_pinned_signal"
                ),
                source_event_ids=mapping_source_event_ids,
            )

    try:
        return derive_canonical_adherence_signal(
            target,
            actual,
            tolerance_percent=tolerance,
            source_event_ids=mapping_source_event_ids,
        )
    except ValueError:
        raise
    except Exception as exc:
        raise ValueError("canonical adherence signal is invalid") from exc


def derive_adherence_from_target_actual(
    target: Mapping[str, object] | None,
    actual: Mapping[str, object] | None,
    *,
    tolerance_percent: Decimal | int | str = Decimal("10"),
    source_event_ids: Sequence[str] = (),
) -> AdherenceSignal:
    return derive_canonical_adherence_signal(
        target,
        actual,
        tolerance_percent=tolerance_percent,
        source_event_ids=tuple(str(value) for value in source_event_ids),
    )


canonical_adherence_signal = derive_adherence_from_target_actual


def adherence_signal_from_observation(
    row: DailyObservation | Mapping[str, object] | CanonicalAdherenceSignal,
    **kwargs: Any,
) -> AdherenceSignal:
    return derive_adherence_signal(row, **kwargs)


def _exercise_load_from_observation(row: DailyObservation) -> str | None:
    explicit = row.exercise_load
    derived: str | None = None
    if row.exercise_session_done is not None or row.exercise_intensity_vs_plan is not None:
        if row.exercise_session_done is not True:
            return None
        derived = {
            "below": "low",
            "as_planned": "medium",
            "above": "high",
        }.get(str(getattr(row.exercise_intensity_vs_plan, "value", row.exercise_intensity_vs_plan)))
        if derived is None:
            return None
    if explicit is not None and explicit not in {"low", "medium", "high"}:
        return None
    if explicit is not None and derived is not None and explicit != derived:
        return None
    return derived or explicit


_ADHERENCE_ROW_FIELDS = (
    "calories",
    "carbs_g",
    "protein_g",
    "fat_g",
    "target_calories_kcal",
    "target_carbohydrate_g",
    "target_protein_g",
    "target_fat_g",
    "actual_calories_kcal",
    "actual_carbohydrate_g",
    "actual_protein_g",
    "actual_fat_g",
    "target_calories",
    "target_carbs_g",
    "target_protein",
    "target_fat",
    "actual_calories",
    "actual_carbs_g",
    "actual_protein",
    "actual_fat",
)


def _is_adherence_observation(row: DailyObservation) -> bool:
    """Exclude exercise-only rows from adherence evidence."""
    return (
        row.adherence_ok is not None
        or row.adherence_version is not None
        or any(getattr(row, name, None) is not None for name in _ADHERENCE_ROW_FIELDS)
    )

def build_snapshot(
    observations: Iterable[DailyObservation],
    evaluation_day: date,
    starts_on: date,
    *,
    adherence_tolerance_percent: Decimal | int | str = Decimal("10"),
    canonical_projection: bool = False,
) -> TrendSnapshot:
    all_rows = tuple(observations)
    window_rows = tuple(
        row for row in all_rows
        if evaluation_day - timedelta(days=13) <= row.kst_day <= evaluation_day
    )
    weights: dict[date, Decimal] = {}
    signal_by_day: dict[date, AdherenceSignal] = {}
    signal_conflicts: set[date] = set()
    for row in window_rows:
        if row.weight_kg is not None:
            weights[row.kst_day] = row.weight_kg
        if not _is_adherence_observation(row):
            continue
        try:
            signal = derive_adherence_signal(
                row,
                tolerance_percent=(
                    row.adherence_tolerance_percent
                    if row.adherence_tolerance_percent is not None
                    else adherence_tolerance_percent
                ),
            )
        except ValueError:
            signal = _contradictory_adherence_signal(
                tolerance=Decimal(str(
                    row.adherence_tolerance_percent
                    if row.adherence_tolerance_percent is not None
                    else adherence_tolerance_percent
                )),
                reason="contradictory_target_actual",
            )
        previous = signal_by_day.get(row.kst_day)
        if previous is not None and (
            previous.status != signal.status
            or previous.digest != signal.digest
        ):
            signal_conflicts.add(row.kst_day)
        else:
            signal_by_day[row.kst_day] = signal
    current = [
        weights[d] for d in sorted(weights)
        if evaluation_day - timedelta(days=6) <= d <= evaluation_day
    ]
    prior = [
        weights[d] for d in sorted(weights)
        if evaluation_day - timedelta(days=13) <= d <= evaluation_day - timedelta(days=7)
    ]
    current_mean = sum(current, Decimal(0)) / len(current) if current else None
    prior_mean = sum(prior, Decimal(0)) / len(prior) if prior else None
    rate = None
    if current_mean is not None and prior_mean not in (None, Decimal(0)):
        rate = ((current_mean - prior_mean) / prior_mean * 100).quantize(
            Decimal("0.01"), rounding=ROUND_HALF_UP,
        )
    schedule_by_day: dict[date, str] = {}
    exercise_sessions: list[tuple[date, bool | None, str | None]] = []
    schedule_ambiguity = False
    for row in all_rows:
        if (
            evaluation_day - timedelta(days=6)
            <= row.kst_day
            <= evaluation_day + timedelta(days=7)
            and (
                row.exercise_session_done is not None
                or row.exercise_intensity_vs_plan is not None
            )
        ):
            exercise_sessions.append(
                (
                    row.kst_day,
                    row.exercise_session_done,
                    row.exercise_intensity_vs_plan,
                )
            )
        if evaluation_day <= row.kst_day <= evaluation_day + timedelta(days=7):
            has_exercise_fields = (
                row.exercise_load is not None
                or row.exercise_session_done is not None
                or row.exercise_intensity_vs_plan is not None
            )
            if not has_exercise_fields:
                continue
            load = _exercise_load_from_observation(row)
            if load is None:
                schedule_ambiguity = True
                continue
            existing = schedule_by_day.get(row.kst_day)
            if existing is not None and existing != load:
                schedule_by_day[row.kst_day] = "ambiguous"
                schedule_ambiguity = True
            else:
                schedule_by_day[row.kst_day] = load
    trailing_day = evaluation_day + timedelta(days=7)
    if len(schedule_by_day) == 8 and trailing_day in schedule_by_day:
        del schedule_by_day[trailing_day]
    current_signals = {
        day: signal
        for day, signal in signal_by_day.items()
        if evaluation_day - timedelta(days=6) <= day <= evaluation_day
    }
    for day in signal_conflicts:
        if evaluation_day - timedelta(days=6) <= day <= evaluation_day:
            current_signals[day] = _contradictory_adherence_signal(
                tolerance=Decimal(str(adherence_tolerance_percent)),
                reason="same_day_adherence_conflict",
            )
    complete_days = sum(1 for signal in current_signals.values() if signal.complete)
    inadequate_days = sum(1 for signal in current_signals.values() if signal.status == "inadequate")
    missing_days = sum(1 for signal in current_signals.values() if signal.status == "missing")
    contradictory_days = sum(1 for signal in current_signals.values() if signal.status == "contradictory")
    signals_digest = digest(tuple(
        (day, signal) for day, signal in sorted(current_signals.items())
    )) if current_signals else None
    return TrendSnapshot(
        evaluation_day=evaluation_day,
        d_plus=(evaluation_day - starts_on).days + 1,
        current_samples=len(current),
        prior_samples=len(prior),
        current_mean_kg=current_mean,
        prior_mean_kg=prior_mean,
        weekly_rate_percent=rate,
        adherent_days=sum(
            1 for signal in current_signals.values()
            if signal.status == "adequate"
        ),
        safety_held=any(row.safety_held for row in window_rows),
        scheduled_loads=tuple(sorted(schedule_by_day.items())),
        exercise_sessions=tuple(sorted(exercise_sessions)),
        adherence_complete_days=complete_days,
        adherence_inadequate_days=inadequate_days,
        adherence_missing_days=missing_days,
        adherence_contradictory_days=contradictory_days,
        adherence_signal_version=ADHERENCE_SIGNAL_VERSION,
        adherence_digest=signals_digest,
        schedule_load_version=CARB_LOAD_VERSION,
        schedule_ambiguity=schedule_ambiguity,
        canonical_projection=canonical_projection,
    )


def build_adherence_signal(row: DailyObservation, **kwargs: Any) -> AdherenceSignal:
    return derive_adherence_signal(row, **kwargs)



_EVENT_FLOW_BY_TYPE = {
    "morning_checkin": "morning",
    "nutrition_checkin": "nutrition",
    "workout_record": "workout",
    "exercise_session_record": "exercise_session",
    "safety_audit": "safety_audit",
    "check_in_validated": "legacy_combined",
    "history_imported": "history_import",
}
_HISTORY_SCHEMA_VERSIONS = {"1.0", "2.0"}


def _history_root_day(root: Mapping[str, object]) -> date:
    try:
        root = validate_event(root).model_dump(mode="json", exclude_none=True)
    except ValueError as exc:
        raise ValueError("history import root is not a committed canonical event") from exc
    if str(root.get("status", "")) != "archived":
        raise ValueError("history import root must be archived")
    provenance = root.get("provenance")
    if (
        not isinstance(provenance, Mapping)
        or provenance.get("source_type") != "historical_markdown"
    ):
        raise ValueError("history import root provenance is invalid")
    occurred_at = root.get("occurred_at_kst")
    if not isinstance(occurred_at, str):
        raise ValueError("history import occurrence is invalid")
    try:
        occurred = datetime.fromisoformat(occurred_at)
    except ValueError as exc:
        raise ValueError("history import occurrence is invalid") from exc
    if (
        occurred.tzinfo is None
        or occurred.utcoffset() != timedelta(hours=9)
        or occurred.isoformat() != occurred_at
    ):
        raise ValueError("history import occurrence is invalid")
    check_in = root.get("check_in")
    if not isinstance(check_in, Mapping):
        raise ValueError("history import check-in is invalid")
    try:
        ContractCheckin.model_validate(check_in)
    except ValueError as exc:
        raise ValueError("history import check-in is invalid") from exc
    schema_version = root.get("schema_version", "1.0")
    if schema_version not in _HISTORY_SCHEMA_VERSIONS:
        raise ValueError("history import schema version is invalid")
    if schema_version == "1.0":
        if root.get("import_manifest") is not None:
            raise ValueError("legacy history import manifest is invalid")
    else:
        manifest = root.get("import_manifest")
        if not isinstance(manifest, Mapping) or set(manifest) != {"observation_kst_day"}:
            raise ValueError("history import manifest is invalid")
        manifest_day = manifest.get("observation_kst_day")
        if not isinstance(manifest_day, str):
            raise ValueError("history import manifest is invalid")
        try:
            parsed_manifest_day = date.fromisoformat(manifest_day)
        except ValueError as exc:
            raise ValueError("history import manifest is invalid") from exc
        if parsed_manifest_day.isoformat() != manifest_day or parsed_manifest_day != occurred.date():
            raise ValueError("history import source day disagrees with occurrence")
    return occurred.date()


def _history_terminal_check_in(row: Mapping[str, object]) -> Mapping[str, object]:
    check_in = row.get("check_in")
    if not isinstance(check_in, Mapping):
        raise ValueError("history import terminal check-in is invalid")
    try:
        ContractCheckin.model_validate(check_in)
    except ValueError as exc:
        raise ValueError("history import terminal check-in is invalid") from exc
    return check_in




def _event_field(value: object, *names: str) -> object:
    if isinstance(value, Mapping):
        for name in names:
            if name in value:
                return value[name]
    for name in names:
        candidate = getattr(value, name, None)
        if candidate is not None:
            return candidate
    return None


def _exercise_payload_load(payload: object) -> tuple[str | None, bool | None, str | None]:
    done = _event_field(payload, "session_done", "done")
    intensity = _event_field(payload, "intensity_vs_plan", "intensity")
    done_value = done if isinstance(done, bool) else None
    intensity_value = getattr(intensity, "value", intensity)
    load = {
        "below": "low",
        "as_planned": "medium",
        "above": "high",
    }.get(str(intensity_value))
    if done_value is not True or load is None:
        return None, done_value, str(intensity_value) if intensity is not None else None
    return load, done_value, str(intensity_value)
@dataclass(frozen=True, slots=True)
class CarbLoadSignal:
    version: str
    load: str | None
    session_done: bool | None
    intensity_vs_plan: str | None
    source_event_id: str | None = None

    @property
    def value(self) -> str | None:
        return self.load

    @property
    def digest(self) -> str:
        return digest(self)


VersionedCarbLoad = CarbLoadSignal


def versioned_exercise_carb_load(
    payload: object,
    *,
    source_event_id: str | None = None,
) -> CarbLoadSignal:
    load, done, intensity = _exercise_payload_load(payload)
    return CarbLoadSignal(
        CARB_LOAD_VERSION,
        load,
        done,
        intensity,
        source_event_id,
    )


def exercise_session_carb_load(payload: object) -> str | None:
    return _exercise_payload_load(payload)[0]


exercise_intensity_carb_load = exercise_session_carb_load
map_exercise_session_carb_load = exercise_session_carb_load


def _canonical_adherence_values(row: Mapping[str, object], check_in: Mapping[str, object]) -> dict[str, object]:
    signal = row.get("adherence_signal")
    signal = signal if isinstance(signal, Mapping) else {}

    def mappings(*values: object) -> tuple[Mapping[str, object], ...]:
        return tuple(value for value in values if isinstance(value, Mapping))

    payload = row.get("payload")
    payload = payload if isinstance(payload, Mapping) else {}
    target_sources = list(mappings(
        row.get("target"),
        row.get("check_in_target"),
        row.get("nutrition_target"),
        payload.get("target"),
        payload.get("check_in_target"),
        payload.get("nutrition_target"),
        signal.get("target"),
        signal.get("target_values"),
    ))
    actual_sources = list(mappings(
        check_in,
        row.get("actual"),
        row.get("actual_values"),
        payload.get("actual"),
        payload.get("actual_values"),
        signal.get("actual"),
        signal.get("actual_values"),
    ))
    target_names = {
        "calories": ("target_calories_kcal", "target_calories", "calories_kcal", "calories"),
        "carbs": ("target_carbohydrate_g", "target_carbs_g", "carbohydrate_g", "carbs_g", "carbs"),
        "protein": ("target_protein_g", "target_protein", "protein_g", "protein"),
        "fat": ("target_fat_g", "target_fat", "fat_g", "fat"),
    }
    actual_names = {
        "calories": ("actual_calories_kcal", "actual_calories", "calories_kcal", "calories"),
        "carbs": ("actual_carbohydrate_g", "actual_carbs_g", "carbohydrate_g", "carbs_g", "carbs"),
        "protein": ("actual_protein_g", "actual_protein", "protein_g", "protein"),
        "fat": ("actual_fat_g", "actual_fat", "fat_g", "fat"),
    }
    result: dict[str, object] = {}
    for metric, names in target_names.items():
        for source in target_sources:
            for name in names:
                if name in source:
                    result[f"target_{metric}"] = source[name]
                    break
            if f"target_{metric}" in result:
                break
        if f"target_{metric}" not in result:
            for name in names:
                if name in signal:
                    result[f"target_{metric}"] = signal[name]
                    break
    for metric, names in actual_names.items():
        for source in actual_sources:
            for name in names:
                if name in source:
                    result[f"actual_{metric}"] = source[name]
                    break
            if f"actual_{metric}" in result:
                break
        if f"actual_{metric}" not in result:
            for name in names:
                if name in signal:
                    result[f"actual_{metric}"] = signal[name]
                    break
    signal_version = signal.get("schema_version", signal.get("version"))
    if signal_version is not None:
        result["adherence_version"] = signal_version
    status = str(signal.get("status", "")).lower()
    if status in {"adequate", "adherent"}:
        result["adherence_ok"] = True
    elif status in {"inadequate", "not_adherent"}:
        result["adherence_ok"] = False
    elif isinstance(signal.get("adherent"), bool) and status not in {"missing", "contradictory"}:
        result["adherence_ok"] = signal["adherent"]
    if "tolerance_percent" in signal:
        result["adherence_tolerance_percent"] = signal["tolerance_percent"]
    return result


def project_canonical_events(
    events: Iterable[object],
    evaluation_day: date,
    starts_on: date,
) -> TrendSnapshot:
    """Project real canonical Event dumps without rewriting their history."""
    rows = canonical_event_records(events)
    by_id = {
        str(row.get("event_id")): row
        for row in rows
        if isinstance(row.get("event_id"), str)
    }
    terminal_ids = set(by_id)
    for event_id in sorted(by_id):
        path: set[str] = set()
        current = event_id
        while True:
            if current in path:
                raise ValueError("canonical correction cycle detected")
            path.add(current)
            parent = by_id[current].get("supersedes")
            if parent is None:
                break
            if not isinstance(parent, str) or parent not in by_id:
                raise ValueError("orphan canonical correction")
            current = parent
    child_count: dict[str, int] = {}
    for row in rows:
        parent = row.get("supersedes")
        if isinstance(parent, str):
            child_count[parent] = child_count.get(parent, 0) + 1
            if child_count[parent] > 1:
                raise ValueError("forked canonical correction")
            terminal_ids.discard(parent)

    held_anywhere = any(
        str(row.get("status", "")) == "unsafe"
        or str(row.get("event_type", "")) in {"safety_flag", "safety_audit"}
        or (
            isinstance(row.get("safety"), Mapping)
            and bool(
                row["safety"].get("coaching_held")
                or row["safety"].get("reasons")
                or row["safety"].get("signal")
            )
        )
        for row in rows
    )
    observations: list[DailyObservation] = []
    for event_id in sorted(terminal_ids):
        row = by_id[event_id]
        root_id = event_id
        seen: set[str] = set()
        while by_id[root_id].get("supersedes") is not None:
            if root_id in seen:
                raise ValueError("canonical correction cycle detected")
            seen.add(root_id)
            root_id = str(by_id[root_id]["supersedes"])
        root = by_id[root_id]
        root_type = str(root.get("event_type", ""))
        event_type = str(row.get("event_type", ""))
        mapped_flow = _EVENT_FLOW_BY_TYPE.get(root_type)
        if root_type == "history_imported":
            root_day = _history_root_day(root)
            if event_type == "history_imported":
                if event_id != root_id or str(row.get("status", "")) != "archived":
                    raise ValueError("history import root-terminal conflict")
                check_in = _history_terminal_check_in(row)
                status = "archived"
            elif event_type == "correction":
                status = str(row.get("status", ""))
                if status not in {"accepted", "unsafe"}:
                    raise ValueError("history import correction is not terminal")
                check_in = _history_terminal_check_in(row)
            else:
                raise ValueError("history import root-terminal conflict")
            event_day = root_day
        else:
            status = str(row.get("status", "accepted"))
            if status not in {"accepted", "unsafe"}:
                continue
            if event_type == "correction":
                mapped_flow = _EVENT_FLOW_BY_TYPE.get(root_type)
            if mapped_flow is None:
                continue
            day_row = root if event_type == "correction" else row
            raw_day = (
                day_row.get("observation_kst_day")
                or day_row.get("source_day")
                or day_row.get("occurred_at_kst")
                or day_row.get("occurred_at")
            )
            try:
                event_day = date.fromisoformat(str(raw_day)[:10])
            except (TypeError, ValueError) as exc:
                raise ValueError("invalid canonical observation day") from exc
            payload = row.get("check_in")
            check_in = payload if isinstance(payload, Mapping) else {}
        safety = row.get("safety")
        held = status == "unsafe" or (
            isinstance(safety, Mapping)
            and bool(
                safety.get("coaching_held")
                or safety.get("reasons")
                or safety.get("signal")
            )
        )
        if event_type in {"safety_flag", "safety_audit"} or root_type in {"safety_flag", "safety_audit"}:
            held = True
        values = _canonical_adherence_values(row, check_in)
        exercise_payload = row.get("exercise_session")
        exercise_load: str | None = None
        exercise_done: bool | None = None
        exercise_intensity: str | None = None
        if mapped_flow == "exercise_session" or exercise_payload is not None:
            exercise_load, exercise_done, exercise_intensity = _exercise_payload_load(exercise_payload or {})
        observations.append(DailyObservation(
            event_day,
            weight_kg=Decimal(str(check_in["body_weight_kg"]))
            if check_in.get("body_weight_kg") is not None else None,
            calories=_coerce_int(check_in.get("calories_kcal")),
            carbs_g=_coerce_int(check_in.get("carbohydrate_g")),
            protein_g=_coerce_int(check_in.get("protein_g")),
            fat_g=_coerce_int(check_in.get("fat_g")),
            adherence_ok=values.pop("adherence_ok", None),
            safety_held=held,
            exercise_load=exercise_load,
            target_calories_kcal=_coerce_int(values.pop("target_calories", None)),
            target_carbohydrate_g=_coerce_int(values.pop("target_carbs", None)),
            target_protein_g=_coerce_int(values.pop("target_protein", None)),
            target_fat_g=_coerce_int(values.pop("target_fat", None)),
            actual_calories_kcal=_coerce_int(values.pop("actual_calories", None)),
            actual_carbohydrate_g=_coerce_int(values.pop("actual_carbs", None)),
            actual_protein_g=_coerce_int(values.pop("actual_protein", None)),
            actual_fat_g=_coerce_int(values.pop("actual_fat", None)),
            adherence_tolerance_percent=(
                Decimal(str(values.pop("adherence_tolerance_percent")))
                if values.get("adherence_tolerance_percent") is not None
                else None
            ),
            adherence_version=str(values.pop("adherence_version"))
            if values.get("adherence_version") is not None else None,
            exercise_session_done=exercise_done,
            exercise_intensity_vs_plan=exercise_intensity,
        ))
    snapshot = build_snapshot(
        observations,
        evaluation_day,
        starts_on,
        canonical_projection=True,
    )
    return replace(snapshot, safety_held=snapshot.safety_held or held_anywhere)


def _target_within_global_bounds(target: MacroTarget | None) -> bool:
    if not isinstance(target, MacroTarget):
        return False
    return (
        type(target.calories) is int
        and type(target.carbs_g) is int
        and type(target.protein_g) is int
        and type(target.fat_g) is int
        and GLOBAL_MIN_CALORIES <= target.calories <= GLOBAL_MAX_CALORIES
        and 0 <= target.carbs_g
        and GLOBAL_MIN_PROTEIN_G <= target.protein_g <= GLOBAL_MAX_PROTEIN_G
        and GLOBAL_MIN_FAT_G <= target.fat_g <= GLOBAL_MAX_FAT_G
    )


def solve_macros(calories: int, protein_g: int, fat_g: int) -> MacroTarget | None:
    if any(type(value) is not int for value in (calories, protein_g, fat_g)):
        return None
    remaining = calories - 4 * protein_g - 9 * fat_g
    if (
        calories < GLOBAL_MIN_CALORIES
        or calories > GLOBAL_MAX_CALORIES
        or protein_g < GLOBAL_MIN_PROTEIN_G
        or protein_g > GLOBAL_MAX_PROTEIN_G
        or fat_g < GLOBAL_MIN_FAT_G
        or fat_g > GLOBAL_MAX_FAT_G
        or remaining < 0
        or remaining % 4
    ):
        return None
    target = MacroTarget(calories, remaining // 4, protein_g, fat_g)
    return target if _target_within_global_bounds(target) else None
def _normalise_cycle_day(value: object) -> date | None:
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    if isinstance(value, str):
        try:
            return date.fromisoformat(value[:10])
        except ValueError:
            return None
    return None


def _normalise_cycle_load(
    value: object,
    *,
    preserve_rest: bool = False,
) -> str | None:
    if isinstance(value, DailyObservation):
        return _exercise_load_from_observation(value)
    if isinstance(value, CarbLoadSignal):
        return value.load
    if isinstance(value, Mapping):
        if any(name in value for name in ("session_done", "done", "intensity_vs_plan", "intensity")):
            return _exercise_payload_load(value)[0]
        for name in ("load", "exercise_load", "category", "carb_load"):
            candidate = value.get(name)
            if candidate is not None:
                return _normalise_cycle_load(candidate, preserve_rest=preserve_rest)
        for name in ("exercise_session", "session", "payload"):
            candidate = value.get(name)
            if candidate is not None:
                return _normalise_cycle_load(candidate, preserve_rest=preserve_rest)
        return None
    candidate = getattr(value, "value", value)
    normalized = str(candidate).strip().lower()
    if normalized in {"high", "medium", "low"}:
        return normalized
    if normalized == "rest":
        return "rest" if preserve_rest else "low"
    return {
        "above": "high",
        "as_planned": "medium",
        "below": "low",
    }.get(normalized)


def _cycle_schedule_rows(
    source: object,
    *,
    evaluation_day: date | None,
) -> tuple[tuple[date, object], ...]:
    if source is None:
        return ()
    if isinstance(source, DailyObservation):
        return ((source.kst_day, source),)
    if isinstance(source, Mapping):
        if any(name in source for name in ("kst_day", "day", "date")):
            rows = (source,)
        else:
            rows = tuple(source.items())
    else:
        try:
            rows = tuple(source)  # type: ignore[arg-type]
        except TypeError:
            return ()
    result: list[tuple[date, object]] = []
    sequence_values: list[object] = []
    for item in rows:
        if isinstance(item, DailyObservation):
            day = item.kst_day
            value = item
        elif isinstance(item, Mapping):
            day = _normalise_cycle_day(
                item.get("kst_day", item.get("day", item.get("date")))
            )
            value = item
        else:
            try:
                candidate_day, candidate_value = item
            except (TypeError, ValueError):
                day = None
                candidate_value = item
            else:
                day = _normalise_cycle_day(candidate_day)
            value = candidate_value
        if day is None:
            sequence_values.append(value)
            continue
        result.append((day, value))
    if sequence_values:
        if evaluation_day is None:
            return ()
        for offset, value in enumerate(sequence_values):
            result.append((evaluation_day + timedelta(days=offset), value))
    return tuple(result)


def _normalise_cycle_schedule(
    source: object,
    *,
    evaluation_day: date | None,
    preserve_rest: bool = False,
) -> dict[date, str | None]:
    return {
        day: _normalise_cycle_load(value, preserve_rest=preserve_rest)
        for day, value in _cycle_schedule_rows(source, evaluation_day=evaluation_day)
    }


def _resolve_actual_cycle_load(
    value: object,
    *,
    planned_load: str | None,
) -> str | None:
    if not isinstance(value, Mapping) or not any(
        name in value
        for name in ("session_done", "done", "intensity_vs_plan", "intensity")
    ):
        return _normalise_cycle_load(value)
    done = _event_field(value, "session_done", "done")
    if done is False:
        return "low"
    if done is not True or planned_load not in {"high", "medium", "low"}:
        return None
    intensity = _event_field(value, "intensity_vs_plan", "intensity")
    intensity_value = str(getattr(intensity, "value", intensity)).strip().lower()
    return {
        ("low", "below"): "low",
        ("low", "as_planned"): "low",
        ("low", "above"): "medium",
        ("medium", "below"): "low",
        ("medium", "as_planned"): "medium",
        ("medium", "above"): "high",
        ("high", "below"): "medium",
        ("high", "as_planned"): "high",
        ("high", "above"): "high",
    }.get((planned_load, intensity_value))


def _cycle_schedule_inputs(
    *,
    evaluation_day: date | None,
    scheduled_loads: object,
    planned_sessions: object,
    actual_sessions: object,
    planned_loads: object,
    actual_loads: object,
) -> tuple[tuple[date, str], ...] | None:
    planned_sources = (planned_sessions, planned_loads, scheduled_loads)
    actual_sources = (actual_sessions, actual_loads)
    planned: dict[date, str | None] = {}
    actual: dict[date, str | None] = {}
    for source in planned_sources:
        for day, value in _normalise_cycle_schedule(
            source,
            evaluation_day=evaluation_day,
            preserve_rest=True,
        ).items():
            if day in planned and planned[day] != value:
                planned[day] = None
            else:
                planned[day] = value
    for source in actual_sources:
        for day, raw_value in _cycle_schedule_rows(source, evaluation_day=evaluation_day):
            value = _resolve_actual_cycle_load(raw_value, planned_load=planned.get(day))
            if day in actual and actual[day] != value:
                actual[day] = None
            else:
                actual[day] = value
    days = sorted(set(planned) | set(actual))
    if evaluation_day is not None:
        expected_days = tuple(evaluation_day + timedelta(days=index) for index in range(7))
        if tuple(days) != expected_days:
            return None
    elif len(days) != 7:
        return None
    if len(days) != 7:
        return None
    selected = []
    for day in days:
        # A present actual row, including an unfinished/ambiguous row, takes
        # precedence over the planned value and cannot silently fall back.
        value = actual[day] if day in actual else planned.get(day)
        if value == "rest":
            value = "low"
        if value not in {"high", "medium", "low"}:
            return None
        selected.append((day, value))
    return tuple(selected)


def _cycle_macro_target(
    calories: int,
    base: MacroTarget,
) -> MacroTarget | None:
    if calories % 4 != base.calories % 4:
        return None
    target = MacroTarget(calories, 0, base.protein_g, base.fat_g)
    remaining = calories - 4 * base.protein_g - 9 * base.fat_g
    if remaining < 0 or remaining % 4:
        return None
    target = MacroTarget(calories, remaining // 4, base.protein_g, base.fat_g)
    return target if _target_within_global_bounds(target) else None


def _cycle_category_calories(
    base: MacroTarget,
    categories: Sequence[str],
    *,
    category_delta_calories: int,
    min_category_delta_calories: int,
    max_category_delta_calories: int,
    require_distinct_categories: bool,
) -> dict[str, int] | None:
    counts = {category: categories.count(category) for category in ("high", "medium", "low")}
    if any(counts[category] and not _target_within_global_bounds(base) for category in counts):
        return None
    if type(category_delta_calories) is not int or category_delta_calories < 0:
        return None
    if type(min_category_delta_calories) is not int or min_category_delta_calories < 0:
        return None
    if type(max_category_delta_calories) is not int or max_category_delta_calories < 0:
        return None
    if min_category_delta_calories > max_category_delta_calories:
        return None
    # Category deltas remain congruent with the base energy so protein/fat can
    # stay immutable and carbohydrate totals reconcile without rounding.
    delta = max(0, category_delta_calories)
    delta -= delta % 4
    min_delta = min_category_delta_calories + (-min_category_delta_calories % 4)
    max_delta = max_category_delta_calories - (max_category_delta_calories % 4)
    if min_delta > max_delta:
        return None
    center_high = base.calories + delta
    center_low = base.calories - delta
    total = base.calories * 7
    present = tuple(category for category, count in counts.items() if count)
    if not present:
        return None
    if len(present) == 1:
        return {present[0]: base.calories}
    span = max_delta + 8
    step = 4
    def values(center: int) -> tuple[int, ...]:
        low = max(GLOBAL_MIN_CALORIES, center - span)
        high = min(GLOBAL_MAX_CALORIES, center + span)
        first = low + ((base.calories - low) % step)
        return tuple(range(first, high + 1, step))
    high_values = values(center_high) if counts["high"] else (None,)
    low_values = values(center_low) if counts["low"] else (None,)
    medium_values = values(base.calories) if counts["medium"] else (None,)
    candidates: list[tuple[tuple[int, ...], dict[str, int]]] = []
    for high in high_values:
        for low in low_values:
            if counts["medium"]:
                for medium in medium_values:
                    category_values = {"high": high, "medium": medium, "low": low}
                    weighted = sum(
                        counts[name] * int(category_values[name])
                        for name in counts
                        if category_values[name] is not None
                    )
                    if weighted != total:
                        continue
                    if high is not None and medium is not None and high < medium:
                        continue
                    if medium is not None and low is not None and medium < low:
                        continue
                    candidate = {name: int(value) for name, value in category_values.items() if value is not None}
                    spread = (
                        candidate.get("high", candidate.get("medium", base.calories))
                        - candidate.get("low", candidate.get("medium", base.calories))
                    )
                    if spread < min_delta or spread > max_delta:
                        continue
                    if require_distinct_categories and len(candidate) == 3 and len(set(candidate.values())) != 3:
                        continue
                    if any(abs(value - base.calories) > max_delta for value in candidate.values()):
                        continue
                    score = tuple(
                        abs(candidate.get(name, base.calories) - (
                            center_high if name == "high" else center_low if name == "low" else base.calories
                        ))
                        for name in ("high", "medium", "low")
                    ) + tuple(candidate.get(name, base.calories) for name in ("high", "medium", "low"))
                    candidates.append((score, candidate))
            else:
                category_values = {"high": high, "low": low}
                weighted = sum(
                    counts[name] * int(value)
                    for name, value in category_values.items()
                    if value is not None
                )
                if weighted != total:
                    continue
                if high is not None and low is not None and high < low:
                    continue
                candidate = {name: int(value) for name, value in category_values.items() if value is not None}
                spread = candidate.get("high", candidate.get("low", base.calories)) - candidate.get("low", candidate.get("high", base.calories))
                if spread < min_delta or spread > max_delta:
                    continue
                if require_distinct_categories and len(candidate) == 2 and len(set(candidate.values())) != 2:
                    continue
                if any(abs(value - base.calories) > max_delta for value in candidate.values()):
                    continue
                score = tuple(
                    abs(candidate.get(name, base.calories) - (
                        center_high if name == "high" else center_low if name == "low" else base.calories
                    ))
                    for name in ("high", "medium", "low")
                ) + tuple(candidate.get(name, base.calories) for name in ("high", "medium", "low"))
                candidates.append((score, candidate))
    if not candidates:
        return None
    candidates.sort(key=lambda item: item[0])
    return candidates[0][1]


def compile_weekly_carb_cycle(
    base_target: MacroTarget | None = None,
    scheduled_loads: object = None,
    *,
    planned_sessions: object = None,
    actual_sessions: object = None,
    planned_loads: object = None,
    actual_loads: object = None,
    evaluation_day: date | str | None = None,
    category_delta_calories: int = CARB_CYCLE_DELTA_CALORIES,
    high_delta_calories: int | None = None,
    low_delta_calories: int | None = None,
    min_category_delta_calories: int = 0,
    max_category_delta_calories: int = 400,
    require_distinct_categories: bool = False,
    **kwargs: object,
) -> WeeklyCarbCycle | None:
    """Compile deterministic category targets with exact seven-day reconciliation."""
    if planned_sessions is None:
        planned_sessions = kwargs.pop("planned_schedule", kwargs.pop("planned_session_loads", None))
    if actual_sessions is None:
        actual_sessions = kwargs.pop("actual_schedule", kwargs.pop("actual_session_loads", None))
    if scheduled_loads is None:
        scheduled_loads = kwargs.pop("schedule", kwargs.pop("session_loads", None))
    if evaluation_day is None:
        evaluation_day = kwargs.pop("start_day", kwargs.pop("kst_day", None))  # type: ignore[assignment]

    if base_target is None:
        base_target = kwargs.pop("target", kwargs.pop("current_target", None))  # type: ignore[assignment]
    if base_target is None:
        base_calories = kwargs.pop("base_calories", kwargs.pop("calories", None))
        base_carbs = kwargs.pop("base_carbs_g", kwargs.pop("carbs_g", None))
        base_protein = kwargs.pop("base_protein_g", kwargs.pop("protein_g", None))
        base_fat = kwargs.pop("base_fat_g", kwargs.pop("fat_g", None))
        if all(type(value) is int for value in (base_calories, base_carbs, base_protein, base_fat)):
            base_target = MacroTarget(base_calories, base_carbs, base_protein, base_fat)
    if not isinstance(base_target, MacroTarget) or not _target_within_global_bounds(base_target):
        return None
    if high_delta_calories is not None:
        category_delta_calories = high_delta_calories
    if low_delta_calories is not None and high_delta_calories is None:
        category_delta_calories = max(abs(category_delta_calories), abs(low_delta_calories))
    parsed_evaluation_day = _normalise_cycle_day(evaluation_day)
    schedule = _cycle_schedule_inputs(
        evaluation_day=parsed_evaluation_day,
        scheduled_loads=scheduled_loads,
        planned_sessions=planned_sessions,
        actual_sessions=actual_sessions,
        planned_loads=planned_loads,
        actual_loads=actual_loads,
    )
    if schedule is None:
        return None
    categories = tuple(value for _, value in schedule)
    category_values = _cycle_category_calories(
        base_target,
        categories,
        category_delta_calories=category_delta_calories,
        min_category_delta_calories=min_category_delta_calories,
        max_category_delta_calories=max_category_delta_calories,
        require_distinct_categories=require_distinct_categories,
    )
    if category_values is None:
        return None
    category_targets: dict[str, MacroTarget] = {}
    for category, calories in category_values.items():
        target = _cycle_macro_target(calories, base_target)
        if target is None:
            return None
        category_targets[category] = target
    daily = tuple(
        DailyNutritionTarget(day, category, category_targets[category])
        for day, category in schedule
    )
    weekly_values = (
        sum(item.target.calories for item in daily),
        sum(item.target.carbs_g for item in daily),
        sum(item.target.protein_g for item in daily),
        sum(item.target.fat_g for item in daily),
    )
    if weekly_values != (
        base_target.calories * 7,
        base_target.carbs_g * 7,
        base_target.protein_g * 7,
        base_target.fat_g * 7,
    ):
        return None
    return WeeklyCarbCycle(daily, base_target, *weekly_values)
build_weekly_carb_cycle = compile_weekly_carb_cycle
compute_weekly_carb_cycle = compile_weekly_carb_cycle
def _cooldown_record_mapping(value: object) -> Mapping[str, object] | None:
    if isinstance(value, Mapping):
        return value
    if hasattr(value, "__dataclass_fields__"):
        try:
            return asdict(value)
        except TypeError:
            return None
    return None


def _nested_calories(value: object) -> int | None:
    if isinstance(value, MacroTarget):
        return value.calories
    if type(value) is int:
        return value
    if isinstance(value, Mapping):
        for name in (
            "calories",
            "calorie",
            "calories_kcal",
            "target_calories",
            "target_calories_kcal",
            "calories_after",
            "current_calories",
        ):
            candidate = value.get(name)
            if type(candidate) is int:
                return candidate
        for name in ("target", "proposal", "payload", "overlay", "value"):
            candidate = value.get(name)
            nested = _nested_calories(candidate)
            if nested is not None:
                return nested
    return None


def _cooldown_is_calorie_changing(record: Mapping[str, object]) -> bool:
    explicit: bool | None = None
    for name in ("calorie_changing", "calories_changed", "is_calorie_change"):
        if name in record and type(record[name]) is bool:
            explicit = bool(record[name])
            break
    payload = record.get("payload")
    nested_payload = payload if isinstance(payload, Mapping) else {}
    for name in ("calorie_changing", "calories_changed", "is_calorie_change"):
        if name in nested_payload and type(nested_payload[name]) is bool:
            explicit = bool(nested_payload[name])
            break
    for name in ("delta_calories", "calorie_delta", "calories_delta"):
        candidate = record.get(name, nested_payload.get(name))
        if isinstance(candidate, (int, float, Decimal)) and candidate != 0:
            return True
    candidates = (
        record.get("prior_target"),
        record.get("previous_target"),
        record.get("old_target"),
        record.get("prior_calories"),
        record.get("previous_calories"),
        record.get("old_calories"),
        record.get("calories_before"),
        nested_payload.get("prior_target"),
        nested_payload.get("previous_target"),
        nested_payload.get("old_target"),
        nested_payload.get("prior_calories"),
        nested_payload.get("previous_calories"),
        nested_payload.get("old_calories"),
        nested_payload.get("calories_before"),
    )
    current = (
        _nested_calories(record.get("target"))
        or _nested_calories(record.get("proposal"))
        or _nested_calories(nested_payload.get("target"))
        or _nested_calories(nested_payload.get("proposal"))
        or _nested_calories(record)
    )
    previous = next(
        (
            _nested_calories(value)
            for value in candidates
            if _nested_calories(value) is not None
        ),
        None,
    )
    if previous is not None and current is not None:
        return previous != current
    if explicit is not None:
        return explicit
    decision = str(record.get("decision", nested_payload.get("decision", ""))).lower()
    if "macro_redistribution" in decision or "maintain" in decision:
        return False
    return False


def _cooldown_committed(record: Mapping[str, object]) -> bool:
    if record.get("committed") is True or record.get("is_committed") is True:
        return True
    state = str(record.get("state", "")).lower()
    event_type = str(record.get("event_type", record.get("kind", ""))).lower()
    payload = record.get("payload")
    if isinstance(payload, Mapping) and (
        payload.get("committed") is True or payload.get("is_committed") is True
    ):
        return True
    if isinstance(payload, Mapping):
        state = state or str(payload.get("state", "")).lower()
        event_type = event_type or str(payload.get("event_type", payload.get("kind", ""))).lower()
    return (
        state in {"committed", "effective", "approved"}
        or event_type in {
            "activation_committed",
            "adaptive_plan_effective",
            "adaptive_overlay_effective",
            "plan_approved",
            "overlay_committed",
        }
    )


def _cooldown_anchor(record: Mapping[str, object]) -> datetime | None:
    payload = record.get("payload")
    payload = payload if isinstance(payload, Mapping) else {}
    for name in (
        "committed_at_kst",
        "committed_at",
        "effective_from_kst",
        "effective_from",
        "occurred_at_kst",
        "recorded_at_kst",
        "as_of_kst",
    ):
        value = record.get(name, payload.get(name))
        if value in (None, ""):
            continue
        try:
            return _kst_compare(value)  # type: ignore[arg-type]
        except (TypeError, ValueError):
            return None
    return None
def _strict_evaluation_day(value: date | datetime | str | None) -> date:
    """Normalize an explicit evaluation day without falling back to wall-clock time."""

    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    if value is None:
        raise ValueError("evaluation day is required")
    if not isinstance(value, str) or not value:
        raise ValueError("evaluation day is invalid")
    try:
        parsed_day = date.fromisoformat(value)
    except ValueError:
        try:
            parsed_day = datetime.fromisoformat(value).date()
        except ValueError as exc:
            raise ValueError("evaluation day is invalid") from exc
    if parsed_day.isoformat() != value and "T" not in value and " " not in value:
        raise ValueError("evaluation day is invalid")
    return parsed_day


def evaluate_cooldown(
    evaluation_day: date | datetime | str | None = None,
    overlays: object = (),
    cooldown_days: int = 0,
    *,
    history: object = None,
    overlay_history: object = None,
    latest_committed_overlay: object = None,
    as_of_kst: date | datetime | str | None = None,
    committed_overlays: object = None,
) -> CooldownResult:
    """Evaluate cooldown using the latest committed calorie-changing overlay only."""
    if evaluation_day is None and as_of_kst is not None:
        evaluation_day = as_of_kst
    query_day = _strict_evaluation_day(evaluation_day)
    if committed_overlays is not None:
        overlays = committed_overlays
    if history is not None:
        overlays = history
    if overlay_history is not None:
        overlays = overlay_history
    if latest_committed_overlay is not None:
        if isinstance(overlays, Mapping) or _cooldown_record_mapping(overlays) is not None:
            existing_overlays = (overlays,)
        else:
            try:
                existing_overlays = tuple(overlays)  # type: ignore[arg-type]
            except TypeError:
                existing_overlays = ()
        overlays = existing_overlays + (latest_committed_overlay,)
    if type(cooldown_days) is not int or cooldown_days < 0:
        return CooldownResult(True, "cooldown_configuration_invalid")
    query = _kst_compare(query_day)
    if cooldown_days == 0:
        return CooldownResult(False)
    if isinstance(overlays, Mapping) or _cooldown_record_mapping(overlays) is not None:
        values: Iterable[object] = (overlays,)
    else:
        try:
            values = tuple(overlays)  # type: ignore[arg-type]
        except TypeError:
            values = ()
    candidates: list[tuple[datetime, int, Mapping[str, object]]] = []
    for index, value in enumerate(values):
        record = _cooldown_record_mapping(value)
        if record is None:
            continue
        if not _cooldown_committed(record) or not _cooldown_is_calorie_changing(record):
            continue
        anchor = _cooldown_anchor(record)
        if anchor is None:
            continue
        sequence = record.get("append_sequence", record.get("sequence", index))
        sequence_value = sequence if type(sequence) is int else index
        candidates.append((anchor, sequence_value, record))
    if not candidates:
        return CooldownResult(False)
    anchor, _, record = max(candidates, key=lambda item: (item[0], item[1]))
    until = anchor + timedelta(days=cooldown_days)
    source_revision = record.get("revision_id")
    if query >= until:
        return CooldownResult(
            False,
            anchor_kst=anchor.isoformat(),
            cooldown_until_kst=until.isoformat(),
            source_revision_id=str(source_revision) if source_revision is not None else None,
        )
    remaining = max(1, (until.date() - query.date()).days)
    return CooldownResult(
        True,
        "cooldown_active",
        anchor_kst=anchor.isoformat(),
        cooldown_until_kst=until.isoformat(),
        days_remaining=remaining,
        source_revision_id=str(source_revision) if source_revision is not None else None,
    )
check_cooldown = evaluate_cooldown


def _make_proposal(
    customer_key: str,
    snapshot: TrendSnapshot,
    decision: Decision,
    reasons: tuple[str, ...],
    *,
    target: MacroTarget | None = None,
    carb_days: tuple[tuple[date, str], ...] = (),
    revision: int = 1,
    parent_digest: str | None = None,
    operator_note: str = "",
    source_digest: str | None = None,
    policy_digest: str | None = None,
    meal_constraints_digest: str | None = None,
    catalog_digest: str | None = None,
    meal_plan: MealPlan | None = None,
    weekly_carb_cycle: WeeklyCarbCycle | None = None,
    weekly_nutrition_plan: WeeklyNutritionPlan | None = None,
    cooldown: CooldownResult | None = None,
    explanation: str | None = None,
) -> NutritionProposal:
    operator_note = _bounded_operator_note(operator_note)
    proposal = NutritionProposal(
        customer_key,
        snapshot,
        decision,
        reasons,
        target,
        carb_days,
        revision,
        parent_digest,
        operator_note,
        source_digest,
        policy_digest,
        meal_constraints_digest,
        catalog_digest,
        meal_plan,
    )
    proposal = replace(
        proposal,
        weekly_carb_cycle=weekly_carb_cycle,
        weekly_nutrition_plan=weekly_nutrition_plan,
        cooldown=cooldown,
    )
    validated_explanation = validate_explanation(explanation, proposal)
    proposal = replace(proposal, explanation=validated_explanation)
    operator_body = render_operator_card(proposal)
    customer_body = render_customer_body(proposal)
    return replace(
        proposal,
        operator_body=operator_body,
        customer_body=customer_body,
        operator_body_digest=hashlib.sha256(operator_body.encode("utf-8")).hexdigest(),
        customer_body_digest=hashlib.sha256(customer_body.encode("utf-8")).hexdigest(),
        adherence_signal_digest=snapshot.adherence_digest,
    )


def propose(
    customer_key: str,
    snapshot: TrendSnapshot,
    policy: CustomerPolicy,
    *,
    current_target: MacroTarget,
    protein_g: int,
    fat_g: int,
    meal_constraints: MealConstraints | None = None,
    catalog: VersionedFoodCatalog | Sequence[Food] | None = None,
    source_digest: str | None = None,
    policy_digest: str | None = None,
    planned_sessions: object = None,
    actual_sessions: object = None,
    planned_loads: object = None,
    actual_loads: object = None,
    overlay_history: object = None,
    cooldown_history: object = None,
    explanation: object = None,
    llm_explanation: object = None,
    explanation_provider: object = None,
    **kwargs: object,
) -> NutritionProposal:
    supplied_explanation = explanation
    if planned_sessions is None:
        planned_sessions = kwargs.get("planned_schedule", kwargs.get("planned_session_records"))
    if actual_sessions is None:
        actual_sessions = kwargs.get(
            "actual_schedule",
            kwargs.get("actual_session_records", kwargs.get("actual_session")),
        )
    if supplied_explanation is None:
        supplied_explanation = llm_explanation
    if supplied_explanation is None and callable(explanation_provider):
        try:
            supplied_explanation = explanation_provider(current_target)
        except Exception:
            supplied_explanation = None

    def review(reason: str, *aliases: str) -> NutritionProposal:
        reasons = (reason, *aliases)
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.HUMAN_REVIEW,
            reasons,
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )

    if snapshot.safety_held:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.HUMAN_REVIEW,
            ("safety_hold",),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )
    if (
        (
            snapshot.schedule_ambiguity
            or any(load == "ambiguous" for _, load in snapshot.scheduled_loads)
        )
        and actual_sessions is None
        and actual_loads is None
    ):
        return review("training_schedule_ambiguous")
    if snapshot.d_plus > 84:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.HUMAN_REVIEW,
            ("post_plan_review_required",),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )
    if snapshot.d_plus <= 0 or (
        snapshot.d_plus > 28
        and (policy.extended_through is None or snapshot.evaluation_day > policy.extended_through)
    ):
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.OBSERVE,
            ("outside_approved_window",),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )
    if not policy.configured:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.OBSERVE,
            ("customer_goal_input_required",),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )
    if snapshot.current_samples < 4 or snapshot.current_samples + snapshot.prior_samples < 10 or snapshot.weekly_rate_percent is None:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.OBSERVE,
            ("insufficient_weight_samples",),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )
    if snapshot.adherence_contradictory_days:
        return review("adherence_evidence_contradictory")
    if snapshot.adherence_missing_days:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.OBSERVE,
            ("adherence_evidence_required",),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )

    if (
        type(protein_g) is not int
        or protein_g < GLOBAL_MIN_PROTEIN_G
        or protein_g > GLOBAL_MAX_PROTEIN_G
        or type(fat_g) is not int
        or fat_g < GLOBAL_MIN_FAT_G
        or fat_g > GLOBAL_MAX_FAT_G
    ):
        reason = (
            "global_floor_violation"
            if (
                type(protein_g) is int
                and type(fat_g) is int
                and (
                    protein_g < GLOBAL_MIN_PROTEIN_G
                    or fat_g < GLOBAL_MIN_FAT_G
                )
            )
            else "global_ceiling_violation"
        )
        return review(reason)
    if (
        policy.calorie_floor is None
        or policy.calorie_ceiling is None
        or policy.calorie_floor > GLOBAL_MAX_CALORIES
    ):
        return review("global_floor_violation")
    if (
        policy.calorie_ceiling < GLOBAL_MIN_CALORIES
        or policy.calorie_ceiling > GLOBAL_MAX_CALORIES
    ):
        return review("global_ceiling_violation")
    low, high = policy.weekly_rate_min, policy.weekly_rate_max
    if low is None or high is None or policy.calorie_step is None or policy.calorie_step <= 0:
        return review("customer_goal_input_required")
    rate = snapshot.weekly_rate_percent
    assert rate is not None

    schedule_supplied = bool(
        snapshot.scheduled_loads
        or planned_sessions is not None
        or actual_sessions is not None
        or planned_loads is not None
        or actual_loads is not None
    )

    def cycle_for(target_value: MacroTarget) -> WeeklyCarbCycle | None:
        if not schedule_supplied:
            return None
        return compile_weekly_carb_cycle(
            target_value,
            snapshot.scheduled_loads,
            planned_sessions=planned_sessions,
            actual_sessions=actual_sessions,
            planned_loads=planned_loads,
            actual_loads=actual_loads,
            evaluation_day=(
                snapshot.evaluation_day
                if (
                    planned_sessions is not None
                    or actual_sessions is not None
                    or planned_loads is not None
                    or actual_loads is not None
                )
                else None
            ),
            **{
                key: value
                for key, value in kwargs.items()
                if key in {
                    "category_delta_calories",
                    "high_delta_calories",
                    "low_delta_calories",
                    "min_category_delta_calories",
                    "max_category_delta_calories",
                    "require_distinct_categories",
                }
            },
        )

    def plan_for(target_value: MacroTarget) -> tuple[MealPlan | None, str | None, str | None]:
        if meal_constraints is None and catalog is None:
            return None, None, None
        if meal_constraints is None or catalog is None:
            return None, None, "meal_plan_inputs_required"
        meal_plan_value = compile_meal_plan(target_value, meal_constraints, catalog)
        if meal_plan_value is None:
            return None, None, "meal_plan_no_solution"
        return (
            meal_plan_value,
            meal_constraints.digest or digest(meal_constraints),
            catalog.digest if isinstance(catalog, VersionedFoodCatalog) else digest(tuple(catalog)),
        )

    cycle = cycle_for(current_target)
    if schedule_supplied and cycle is None:
        return review("schedule_evidence_incomplete", "training_schedule_required")
    weekly_plan = None
    if (
        cycle is not None
        and meal_constraints is not None
        and isinstance(catalog, VersionedFoodCatalog)
    ):
        weekly_plan = compile_weekly_nutrition_plan(
            cycle,
            meal_constraints,
            catalog,
            as_of_kst_day=cycle.targets[0].kst_day,
            exercise_evidence_digest=digest(cycle.days),
        )
        if weekly_plan is None:
            return review("weekly_meal_plan_no_solution")

    if low <= rate <= high:
        if not _target_within_global_bounds(current_target):
            if (
                isinstance(current_target, MacroTarget)
                and (
                    current_target.calories < GLOBAL_MIN_CALORIES
                    or current_target.protein_g < GLOBAL_MIN_PROTEIN_G
                    or current_target.fat_g < GLOBAL_MIN_FAT_G
                )
            ):
                return review("global_floor_violation")
            return review("global_ceiling_violation")
        if cycle is not None and cycle.has_redistribution:
            meal_plan, constraints_digest, catalog_digest = plan_for(current_target)
            if meal_constraints is not None or catalog is not None:
                if constraints_digest is None:
                    return review("meal_plan_inputs_required")
                if meal_plan is None:
                    return review("meal_plan_no_solution")
            return _make_proposal(
                customer_key,
                snapshot,
                Decision.MACRO_REDISTRIBUTION,
                ("macro_redistribution_candidate",),
                target=current_target,
                carb_days=cycle.days,
                meal_plan=meal_plan,
                weekly_carb_cycle=cycle,
                weekly_nutrition_plan=weekly_plan,
                source_digest=source_digest,
                policy_digest=policy_digest,
                meal_constraints_digest=constraints_digest,
                catalog_digest=catalog_digest,
                explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
            )
        meal_plan, constraints_digest, catalog_digest = plan_for(current_target)
        if meal_constraints is not None or catalog is not None:
            if constraints_digest is None:
                return review("meal_plan_inputs_required")
            if meal_plan is None:
                return review("meal_plan_no_solution")
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.MAINTAIN,
            ("within_goal_band",),
            target=current_target,
            meal_plan=meal_plan,
            source_digest=source_digest,
            policy_digest=policy_digest,
            meal_constraints_digest=constraints_digest,
            catalog_digest=catalog_digest,
            weekly_carb_cycle=cycle,
            weekly_nutrition_plan=weekly_plan,
            carb_days=cycle.days if cycle is not None else (),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )

    if snapshot.adherent_days < policy.minimum_adherent_days:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.MAINTAIN,
            ("low_adherence_investigate_barriers",),
            target=current_target,
            source_digest=source_digest,
            policy_digest=policy_digest,
            weekly_carb_cycle=cycle,
            carb_days=cycle.days if cycle is not None else (),
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )

    history = cooldown_history if cooldown_history is not None else overlay_history
    cooldown = evaluate_cooldown(
        snapshot.evaluation_day,
        history or (),
        policy.cooldown_days,
        latest_committed_overlay=kwargs.get("latest_committed_overlay"),
    )
    if cooldown.active:
        return _make_proposal(
            customer_key,
            snapshot,
            Decision.MAINTAIN,
            ("cooldown_active",),
            target=current_target,
            carb_days=cycle.days if cycle is not None else (),
            weekly_carb_cycle=cycle,
            cooldown=cooldown,
            source_digest=source_digest,
            policy_digest=policy_digest,
            explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
        )

    delta = policy.calorie_step if rate < low else -policy.calorie_step
    absolute_floor = max(GLOBAL_MIN_CALORIES, policy.calorie_floor)
    policy_ceiling = min(GLOBAL_MAX_CALORIES, policy.calorie_ceiling)
    if policy_ceiling < absolute_floor:
        return review("global_floor_violation")
    calories = max(absolute_floor, min(policy_ceiling, current_target.calories + delta))
    target = solve_macros(calories, protein_g, fat_g)
    if target is None:
        return review("macro_energy_infeasible")
    cycle = cycle_for(target)
    if schedule_supplied and cycle is None:
        return review("schedule_evidence_incomplete", "training_schedule_required")
    meal_plan, constraints_digest, catalog_digest = plan_for(target)
    if meal_constraints is not None or catalog is not None:
        if constraints_digest is None:
            return review("meal_plan_inputs_required")
        if meal_plan is None:
            return review("meal_plan_no_solution")
    return _make_proposal(
        customer_key,
        snapshot,
        Decision.CALORIE_ADJUSTMENT,
        ("calorie_candidate",),
        target=target,
        carb_days=cycle.days if cycle is not None else (),
        weekly_carb_cycle=cycle,
        cooldown=cooldown,
        source_digest=source_digest,
        policy_digest=policy_digest,
        meal_constraints_digest=constraints_digest,
        catalog_digest=catalog_digest,
        meal_plan=meal_plan,
        explanation=supplied_explanation if isinstance(supplied_explanation, str) else None,
    )


def _valid_serving_bounds(food: Food) -> bool:
    bounds = (food.serving_min_grams, food.serving_max_grams, food.serving_step_grams)
    if all(value is None for value in bounds):
        return True
    if any(value is None or isinstance(value, bool) for value in bounds):
        return False
    try:
        minimum, maximum, step = (int(value) for value in bounds)
    except (TypeError, ValueError):
        return False
    return minimum > 0 and maximum >= minimum and step > 0


def _food_variants(food: Food) -> tuple[tuple[int, int, int, int, int, int], ...]:
    """Return ``grams, calories, carbs, protein, fat, units`` variants."""

    if not _valid_serving_bounds(food):
        return ()
    if food.serving_min_grams is None:
        return ((0, food.calories, food.carbs_g, food.protein_g, food.fat_g, 1),)
    minimum = int(food.serving_min_grams)
    maximum = int(food.serving_max_grams)
    step = int(food.serving_step_grams)
    assert maximum is not None and step is not None
    result: list[tuple[int, int, int, int, int, int]] = []
    for grams in range(minimum, maximum + 1, step):
        basis = int(food.macro_basis_grams or minimum)
        if basis <= 0:
            continue
        values = (
            Decimal(food.calories) * Decimal(grams) / Decimal(basis),
            Decimal(food.carbs_g) * Decimal(grams) / Decimal(basis),
            Decimal(food.protein_g) * Decimal(grams) / Decimal(basis),
            Decimal(food.fat_g) * Decimal(grams) / Decimal(basis),
        )
        if any(value != value.to_integral_value() for value in values):
            continue
        calories, carbs, protein, fat = (int(value) for value in values)
        # A minimum-gram row counts as one unit; larger rows consume one
        # catalog serving unit as well, keeping the aggregate cap explicit.
        result.append((grams, calories, carbs, protein, fat, 1))
    return tuple(result)


def _allocate_slot_targets(target: MacroTarget, constraints: MealConstraints) -> tuple[MacroTarget, ...] | None:
    if constraints.meal_shares_bps:
        shares = tuple(int(value) for _, value in constraints.meal_shares_bps)
        if len(shares) != constraints.meal_count or any(value < 0 for value in shares) or sum(shares) != 10_000:
            return None
    else:
        base, remainder = divmod(10_000, constraints.meal_count)
        shares = tuple(base + (1 if index < remainder else 0) for index in range(constraints.meal_count))

    def allocate(value: int) -> tuple[int, ...]:
        floors = [value * share // 10_000 for share in shares]
        residual = value - sum(floors)
        order = sorted(
            range(len(shares)),
            key=lambda index: (-(value * shares[index] % 10_000), index),
        )
        for index in order[:residual]:
            floors[index] += 1
        return tuple(floors)

    calories = allocate(target.calories)
    carbs = allocate(target.carbs_g)
    protein = allocate(target.protein_g)
    fat = allocate(target.fat_g)
    return tuple(
        MacroTarget(calories[index], carbs[index], protein[index], fat[index])
        for index in range(constraints.meal_count)
    )


def _constraint_calorie_tolerance(constraints: MealConstraints) -> Decimal:
    return Decimal(str(
        constraints.calorie_tolerance_percent
        if constraints.calorie_tolerance_percent is not None
        else constraints.meal_slot_calorie_tolerance_percent
    ))


def _constraint_macro_tolerance(constraints: MealConstraints) -> Decimal:
    return Decimal(str(
        constraints.macro_tolerance_percent
        if constraints.macro_tolerance_percent is not None
        else constraints.meal_slot_macro_tolerance_percent
    ))

def _within_tolerance(actual: int, expected: int, tolerance: Decimal) -> bool:
    try:
        boundary = Decimal(str(tolerance))
    except ArithmeticError:
        return False
    if expected == 0:
        return actual == 0
    deviation = abs(Decimal(actual) - Decimal(expected)) / Decimal(expected) * Decimal("100")
    return deviation <= boundary


def _compile_slot_candidates(
    slot_target: MacroTarget,
    allowed: Sequence[Food],
    constraints: MealConstraints,
    *,
    strict_energy: bool,
) -> tuple[MealSlot, ...]:
    variants: list[tuple[str, int, int, int, int, int, int]] = []
    for food in sorted(allowed, key=lambda item: item.food_id):
        for grams, calories, carbs, protein, fat, units in _food_variants(food):
            variants.append((food.food_id, grams, calories, carbs, protein, fat, units))
    variant_grams_by_food = {
        food.food_id: frozenset(item[0] for item in _food_variants(food))
        for food in allowed
    }
    if not variants:
        return ()
    max_foods = constraints.max_foods_per_meal
    max_units = constraints.max_serving_units_per_meal
    candidates: dict[tuple[tuple[str, ...], tuple[int, ...], tuple[int, ...]], MealSlot] = {}

    def visit(
        start: int,
        chosen: tuple[tuple[str, int, int, int, int, int, int], ...],
        calories: int,
        carbs: int,
        protein: int,
        fat: int,
        units: int,
    ) -> None:
        if chosen:
            grouped: dict[str, list[int]] = {}
            for item in chosen:
                values = grouped.setdefault(item[0], [0, 0])
                values[0] += item[1]
                values[1] += item[6]
            grouped_ids = tuple(sorted(grouped))
            grouped_grams = tuple(grouped[food_id][0] for food_id in grouped_ids)
            grouped_quantities = tuple(grouped[food_id][1] for food_id in grouped_ids)
            combined_valid = all(
                not variant_grams_by_food[food_id]
                or grams in variant_grams_by_food[food_id]
                for food_id, grams in zip(grouped_ids, grouped_grams)
            )
            if (
                combined_valid
                and len(grouped_ids) <= max_foods
                and units <= max_units
                and _within_tolerance(carbs, slot_target.carbs_g, _constraint_macro_tolerance(constraints))
                and _within_tolerance(protein, slot_target.protein_g, _constraint_macro_tolerance(constraints))
                and _within_tolerance(fat, slot_target.fat_g, _constraint_macro_tolerance(constraints))
            ):
                actual_calories = 4 * carbs + 4 * protein + 9 * fat if strict_energy else calories
                if _within_tolerance(
                    actual_calories,
                    slot_target.calories,
                    _constraint_calorie_tolerance(constraints),
                ):
                    key = (grouped_ids, grouped_grams, grouped_quantities)
                    candidates[key] = MealSlot(
                        name="",
                        food_ids=grouped_ids,
                        calories=actual_calories,
                        carbs_g=carbs,
                        protein_g=protein,
                        fat_g=fat,
                        quantities=grouped_quantities,
                        serving_grams=grouped_grams,
                        target=slot_target,
                    )
        if units >= max_units:
            return
        for index in range(start, len(variants)):
            item = variants[index]
            next_units = units + item[6]
            if next_units > max_units:
                break
            ids = {candidate[0] for candidate in chosen}
            if item[0] not in ids and len(ids) >= max_foods:
                continue
            next_carbs = carbs + item[3]
            next_protein = protein + item[4]
            next_fat = fat + item[5]
            # No non-negative serving combination can recover an overshoot.
            if next_carbs > slot_target.carbs_g and _constraint_macro_tolerance(constraints) == 0:
                continue
            if next_protein > slot_target.protein_g and _constraint_macro_tolerance(constraints) == 0:
                continue
            if next_fat > slot_target.fat_g and _constraint_macro_tolerance(constraints) == 0:
                continue
            visit(index, chosen + (item,), calories + item[2], next_carbs, next_protein, next_fat, next_units)

    visit(0, (), 0, 0, 0, 0, 0)
    values = list(candidates.values())
    values.sort(
        key=lambda slot: (
            sum(
                abs(Decimal(actual) - Decimal(expected)) / Decimal(max(expected, 1))
                for actual, expected in (
                    (slot.carbs_g, slot_target.carbs_g),
                    (slot.protein_g, slot_target.protein_g),
                    (slot.fat_g, slot_target.fat_g),
                )
            ),
            len(slot.food_ids),
            slot.serving_count,
            slot.food_ids,
            slot.serving_grams,
        )
    )
    return tuple(values[:512])


def compile_meal_plan(
    target: MacroTarget,
    constraints: MealConstraints,
    catalog: VersionedFoodCatalog | Sequence[Food],
    *,
    shadow_test_only: bool = False,
) -> MealPlan | None:
    """Compile bounded integer servings with exact daily macro reconciliation."""

    if not constraints.complete:
        return None
    if isinstance(catalog, VersionedFoodCatalog):
        if catalog.approved is not True:
            raise ValueError("approved VersionedFoodCatalog required")
        foods = catalog.foods
        strict_energy = catalog.schema_version == "2.0" or any(
            food.serving_min_grams is not None for food in foods
        )
    elif shadow_test_only:
        try:
            foods = tuple(catalog)
        except TypeError as exc:
            raise ValueError("shadow catalog must be a food sequence") from exc
        if not all(isinstance(food, Food) for food in foods):
            raise ValueError("shadow catalog contains an invalid food")
        strict_energy = any(food.serving_min_grams is not None for food in foods)
    else:
        raise ValueError("approved VersionedFoodCatalog required")
    allowed = [
        food for food in foods
        if not (food.allergens & constraints.allergies)
        and food.food_id not in constraints.excluded_food_ids
        and not (food.tags & constraints.restrictions)
        and not (food.tags & constraints.digestion_exclusions)
    ]
    if not allowed:
        return None
    slot_targets = _allocate_slot_targets(target, constraints)
    if slot_targets is None:
        return None
    if strict_energy:
        slot_targets = tuple(
            MacroTarget(4 * slot.carbs_g + 4 * slot.protein_g + 9 * slot.fat_g, slot.carbs_g, slot.protein_g, slot.fat_g)
            for slot in slot_targets
        )
    slot_options = [
        _compile_slot_candidates(slot_target, allowed, constraints, strict_energy=strict_energy)
        for slot_target in slot_targets
    ]
    if any(not options for options in slot_options):
        return None

    selected: list[MealSlot] | None = None

    def choose(index: int, chosen: list[MealSlot], calories: int, carbs: int, protein: int, fat: int) -> None:
        nonlocal selected
        if selected is not None:
            return
        if index == len(slot_options):
            if (calories, carbs, protein, fat) == (
                target.calories,
                target.carbs_g,
                target.protein_g,
                target.fat_g,
            ):
                selected = list(chosen)
            return
        for option in slot_options[index]:
            next_values = (
                calories + option.calories,
                carbs + option.carbs_g,
                protein + option.protein_g,
                fat + option.fat_g,
            )
            if next_values[0] > target.calories or next_values[1] > target.carbs_g or next_values[2] > target.protein_g or next_values[3] > target.fat_g:
                continue
            choose(index + 1, chosen + [option], *next_values)

    choose(0, [], 0, 0, 0, 0)
    if selected is None:
        return None
    slots = tuple(
        replace(slot, name=f"Meal {index}")
        for index, slot in enumerate(selected, 1)
    )
    selected_ids = {food_id for slot in slots for food_id in slot.food_ids}
    selected_food_groups = {
        food.equivalent_group
        for food in allowed
        if food.food_id in selected_ids and food.equivalent_group is not None
    }
    equivalent: list[str] = []
    fallback: list[str] = []
    for food in sorted(allowed, key=lambda item: item.food_id):
        if food.food_id in selected_ids:
            continue
        variants = _food_variants(food)

        def matches(slot: MealSlot, variant: tuple[int, int, int, int, int, int]) -> bool:
            return (
                _within_tolerance(variant[3], slot.carbs_g, _constraint_macro_tolerance(constraints))
                and _within_tolerance(variant[4], slot.protein_g, _constraint_macro_tolerance(constraints))
                and _within_tolerance(variant[5], slot.fat_g, _constraint_macro_tolerance(constraints))
                and _within_tolerance(
                    4 * variant[3] + 4 * variant[4] + 9 * variant[5]
                    if strict_energy else variant[2],
                    slot.calories,
                    _constraint_calorie_tolerance(constraints),
                )
            )

        equivalent_match = (
            not selected_food_groups
            or (
                food.equivalent_group is not None
                and food.equivalent_group in selected_food_groups
            )
        )
        if equivalent_match and any(matches(slot, variant) for slot in slots for variant in variants):
            equivalent.append(food.food_id)
        if (food.convenience or food.eating_out) and equivalent_match and any(
            matches(slot, variant) for slot in slots for variant in variants
        ):
            fallback.append(food.food_id)
    return MealPlan(slots, tuple(equivalent), tuple(fallback), target=target, exact=True)
def meal_plan_reconciles_exactly(plan: MealPlan, target: MacroTarget) -> bool:
    return (
        plan.exact
        and plan.calories == target.calories
        and plan.carbs_g == target.carbs_g
        and plan.protein_g == target.protein_g
        and plan.fat_g == target.fat_g
    )


exact_meal_plan_reconciliation = meal_plan_reconciles_exactly


def _weekly_compatible_catalog(
    constraints: MealConstraints,
    catalog: VersionedFoodCatalog,
) -> VersionedFoodCatalog | None:
    budget = constraints.budget_tier
    cooking = constraints.cooking_access
    if budget is None or cooking is None:
        return None
    if catalog.schema_version == "1.0":
        legacy_foods = tuple(sorted(catalog.foods, key=lambda food: food.food_id))
        return VersionedFoodCatalog(
            foods=legacy_foods,
            version=catalog.version,
            digest=catalog.digest,
            approved=catalog.approved,
            approved_by=catalog.approved_by,
            approved_at_kst=catalog.approved_at_kst,
            schema_version="1.0",
        )
    if catalog.schema_version != "2.0":
        return None
    compatible_foods = tuple(
        sorted(
            (
                food
                for food in catalog.foods
                if food.allowed_budget_bands is not None
                and budget in food.allowed_budget_bands
                and food.allowed_cooking_access is not None
                and cooking in food.allowed_cooking_access
            ),
            key=lambda food: food.food_id,
        )
    )
    if not compatible_foods:
        return None
    return VersionedFoodCatalog(
        foods=compatible_foods,
        version=catalog.version,
        digest=digest(compatible_foods),
        approved=catalog.approved,
        approved_by=catalog.approved_by,
        approved_at_kst=catalog.approved_at_kst,
        schema_version="2.0",
    )


def _direct_single_food_plan(
    target: MacroTarget,
    constraints: MealConstraints,
    catalog: VersionedFoodCatalog,
) -> MealPlan | None:
    if constraints.meal_count != 1:
        return None
    matches = tuple(
        food
        for food in catalog.foods
        if food.serving_min_grams is None
        and food.serving_max_grams is None
        and food.serving_step_grams is None
        and food.macro_basis_grams is None
        and food.food_id not in constraints.excluded_food_ids
        and not (food.allergens & constraints.allergies)
        and not (food.tags & constraints.restrictions)
        and not (food.tags & constraints.digestion_exclusions)
        and (
            food.calories,
            food.carbs_g,
            food.protein_g,
            food.fat_g,
        )
        == (
            target.calories,
            target.carbs_g,
            target.protein_g,
            target.fat_g,
        )
    )
    if not matches:
        return None
    selected = min(matches, key=lambda food: food.food_id)
    slot = MealSlot(
        "Meal 1",
        (selected.food_id,),
        selected.calories,
        selected.carbs_g,
        selected.protein_g,
        selected.fat_g,
        quantities=(1,),
        serving_grams=(0,),
        target=target,
    )
    return MealPlan((slot,), (), (), target=target, exact=True)


def compile_weekly_nutrition_plan(
    cycle: WeeklyCarbCycle,
    constraints: MealConstraints,
    catalog: VersionedFoodCatalog,
    *,
    as_of_kst_day: date,
    exercise_evidence_digest: str | None = None,
) -> WeeklyNutritionPlan | None:
    """Compile all seven exact daily meals or fail without a partial plan."""
    if (
        not isinstance(cycle, WeeklyCarbCycle)
        or not cycle.reconciles_exactly
        or not isinstance(constraints, MealConstraints)
        or not constraints.complete
        or not isinstance(catalog, VersionedFoodCatalog)
        or catalog.approved is not True
        or not isinstance(as_of_kst_day, date)
    ):
        return None
    compatible_catalog = _weekly_compatible_catalog(constraints, catalog)
    if compatible_catalog is None:
        return None
    daily_plans: list[DailyNutritionPlan] = []
    meal_plans: dict[MacroTarget, MealPlan] = {}
    for daily_target in cycle.targets:
        meal_plan = meal_plans.get(daily_target.target)
        if meal_plan is None:
            meal_plan = _direct_single_food_plan(
                daily_target.target,
                constraints,
                compatible_catalog,
            )
        if meal_plan is None:
            meal_plan = compile_meal_plan(
                daily_target.target,
                constraints,
                compatible_catalog,
            )
        if meal_plan is None or not meal_plan_reconciles_exactly(
            meal_plan,
            daily_target.target,
        ):
            return None
        meal_plans[daily_target.target] = meal_plan
        daily_plans.append(
            DailyNutritionPlan(
                kst_day=daily_target.kst_day,
                category=daily_target.category,
                target=daily_target.target,
                meal_plan=meal_plan,
            )
        )
    try:
        return WeeklyNutritionPlan(
            horizon_start=cycle.targets[0].kst_day,
            as_of_kst_day=as_of_kst_day,
            frozen_through=None,
            base_target=cycle.base_target,
            days=tuple(daily_plans),
            planned_schedule_digest=digest(cycle.days),
            exercise_evidence_digest=exercise_evidence_digest,
            effective_schedule_digest=digest(cycle.days),
        )
    except ValueError:
        return None


def reconcile_weekly_nutrition_plan(
    parent: WeeklyNutritionPlan,
    effective_cycle: WeeklyCarbCycle,
    constraints: MealConstraints,
    catalog: VersionedFoodCatalog,
    *,
    actual_day: date,
    as_of_kst_day: date,
    exercise_evidence_digest: str,
) -> WeeklyNutritionPlan | None:
    """Freeze elapsed nutrition and rebalance only the remaining horizon."""
    if (
        not isinstance(parent, WeeklyNutritionPlan)
        or not isinstance(effective_cycle, WeeklyCarbCycle)
        or effective_cycle.base_target != parent.base_target
        or not isinstance(actual_day, date)
        or not isinstance(as_of_kst_day, date)
        or not exercise_evidence_digest
    ):
        return None
    horizon_days = tuple(item.kst_day for item in parent.days)
    if (
        tuple(item.kst_day for item in effective_cycle.targets) != horizon_days
        or actual_day not in horizon_days
        or as_of_kst_day not in horizon_days
        or as_of_kst_day < actual_day
    ):
        return None
    boundaries = tuple(
        day
        for day in (parent.frozen_through, actual_day, as_of_kst_day)
        if day is not None
    )
    frozen_through = max(boundaries)
    compatible_catalog = _weekly_compatible_catalog(constraints, catalog)
    if compatible_catalog is None:
        return None

    effective_by_day = {
        item.kst_day: item
        for item in effective_cycle.targets
    }
    target_by_day: dict[date, MacroTarget] = {
        item.kst_day: effective_by_day[item.kst_day].target
        for item in parent.days
        if item.kst_day > frozen_through
    }
    frozen = tuple(item for item in parent.days if item.kst_day <= frozen_through)
    mutable_days = tuple(day for day in horizon_days if day > frozen_through)
    calorie_residual = (
        parent.base_target.calories * 7
        - sum(item.target.calories for item in frozen)
        - sum(target_by_day[day].calories for day in mutable_days)
    )
    carb_residual = (
        parent.base_target.carbs_g * 7
        - sum(item.target.carbs_g for item in frozen)
        - sum(target_by_day[day].carbs_g for day in mutable_days)
    )
    protein_residual = (
        parent.base_target.protein_g * 7
        - sum(item.target.protein_g for item in frozen)
        - sum(target_by_day[day].protein_g for day in mutable_days)
    )
    fat_residual = (
        parent.base_target.fat_g * 7
        - sum(item.target.fat_g for item in frozen)
        - sum(target_by_day[day].fat_g for day in mutable_days)
    )
    if (
        protein_residual != 0
        or fat_residual != 0
        or calorie_residual != 4 * carb_residual
    ):
        return None
    step = 1 if carb_residual > 0 else -1
    category_order = (
        ("high", "medium", "low")
        if step > 0
        else ("low", "medium", "high")
    )
    ordered_mutable_days = tuple(
        sorted(
            mutable_days,
            key=lambda day: (
                category_order.index(effective_by_day[day].category),
                day,
            ),
        )
    )
    remaining = carb_residual
    while remaining:
        progressed = False
        for day in ordered_mutable_days:
            if remaining == 0:
                break
            current = target_by_day[day]
            candidate = MacroTarget(
                current.calories + 4 * step,
                current.carbs_g + step,
                current.protein_g,
                current.fat_g,
            )
            if _target_within_global_bounds(candidate):
                target_by_day[day] = candidate
                remaining -= step
                progressed = True
        if not progressed:
            return None

    child_days: list[DailyNutritionPlan] = []
    for parent_day in parent.days:
        if parent_day.kst_day <= frozen_through:
            child_days.append(
                DailyNutritionPlan(
                    kst_day=parent_day.kst_day,
                    category=parent_day.category,
                    target=parent_day.target,
                    meal_plan=parent_day.meal_plan,
                    nutrition_basis="frozen_parent",
                    parent_daily_digest=parent_day.digest,
                )
            )
            continue
        target = target_by_day[parent_day.kst_day]
        meal_plan = compile_meal_plan(target, constraints, compatible_catalog)
        if meal_plan is None or not meal_plan_reconciles_exactly(meal_plan, target):
            return None
        child_days.append(
            DailyNutritionPlan(
                kst_day=parent_day.kst_day,
                category=effective_by_day[parent_day.kst_day].category,
                target=target,
                meal_plan=meal_plan,
                nutrition_basis="recomputed_future",
            )
        )
    try:
        return WeeklyNutritionPlan(
            horizon_start=parent.horizon_start,
            as_of_kst_day=as_of_kst_day,
            frozen_through=frozen_through,
            base_target=parent.base_target,
            days=tuple(child_days),
            parent_plan_digest=parent.digest,
            planned_schedule_digest=parent.planned_schedule_digest,
            exercise_evidence_digest=exercise_evidence_digest,
            effective_schedule_digest=digest(effective_cycle.days),
        )
    except ValueError:
        return None


class AdaptiveEventStore:
    """Append-only exact-revision audit store with recovery-aware journals."""

    def __init__(
        self,
        path: Path,
        *,
        canonical_events_path: Path | None = None,
        canonical_transaction: CanonicalEventTransaction | None = None,
        root: Path | None = None,
    ) -> None:
        self.path = Path(path)
        self.root = Path(root) if root is not None else self.path.parent
        if canonical_transaction is None and canonical_events_path is not None:
            raise TypeError("canonical adaptive stores require a canonical transaction")
        if canonical_transaction is None:
            _reject_registered_persistence_root(
                self.root,
                label="ordinary AdaptiveEventStore",
            )
            _reject_registered_persistence_root(
                self.path,
                label="ordinary AdaptiveEventStore",
            )
        if canonical_transaction is not None:
            if not isinstance(canonical_transaction, CanonicalEventTransaction):
                raise TypeError("canonical transaction type is invalid")
            if (
                canonical_events_path is not None
                and Path(canonical_events_path).resolve() != canonical_transaction.events_path.resolve()
            ):
                raise ValueError("canonical event path does not match its transaction")
            if canonical_transaction.sequence_path.parent.resolve() != self.root.resolve():
                raise ValueError("canonical transaction root does not match adaptive root")
            self.canonical_events_path = canonical_transaction.events_path
            self.canonical_sequence_path = canonical_transaction.sequence_path
        else:
            self.canonical_events_path = self.root / "canonical-events.jsonl"
            self.canonical_sequence_path = self.root / "canonical-sequence.jsonl"
        self._canonical_transaction = canonical_transaction
        self.source_day_path = self.root / "source-days.jsonl"
        self.source_intent_path = self.root / "source-day-intents.jsonl"
        self.authority_path = self.root / "authority-mirror-intents.jsonl"
        self.config_epoch_path = self.root / "config-epoch-journal.jsonl"
        self.overlay_path = self.root / "adaptive-overlays.jsonl"
        self.source_day_journal_path = self.source_day_path
        self.source_day_intents_path = self.source_intent_path
        self.authority_mirror_path = self.authority_path
        self.config_epoch_journal_path = self.config_epoch_path
        self._active_adaptive_lock_token: _AdaptiveLockToken | None = None

    @classmethod
    def for_registered(cls, runtime: CustomerRuntime) -> AdaptiveEventStore:
        if not isinstance(runtime, CustomerRuntime):
            raise TypeError("registered adaptive stores require CustomerRuntime")
        transaction = CanonicalEventTransaction.for_customer_runtime(runtime)
        return cls(
            runtime.nutrition_plans_root / "events.jsonl",
            canonical_transaction=transaction,
            root=runtime.nutrition_plans_root,
        )

    @contextmanager
    def locked(self) -> Any:
        """Hold the adaptive ledger lock across a guarded lifecycle transition."""
        if self._active_adaptive_lock_token is not None:
            raise RuntimeError("adaptive lock is already held")
        with _verified_private_lock(
            self.root / ".adaptive.lock",
            exclusive=True,
            create=True,
        ):
            token = _AdaptiveLockToken()
            self._active_adaptive_lock_token = token
            try:
                yield token
            finally:
                self._active_adaptive_lock_token = None

    def _require_adaptive_token(self, token: object) -> None:
        if (
            not isinstance(token, _ADAPTIVE_LOCK_TOKEN_TYPE)
            or token is not self._active_adaptive_lock_token
        ):
            raise TypeError("adaptive lock token is invalid")

    @contextmanager
    def read_locked(self) -> Any:
        """Hold the existing adaptive writer inode shared without mutation."""
        with _verified_private_lock(
            self.root / ".adaptive.lock",
            exclusive=False,
            create=False,
        ):
            yield

    read_adaptive_locked = read_locked

    def _append_row(self, path: Path, row: Mapping[str, object]) -> Mapping[str, object]:
        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(canonical_json(dict(row)) + "\n")
            handle.flush()
            os.fsync(handle.fileno())
        return dict(row)

    def append(
        self,
        event_type: str,
        payload: Mapping[str, object],
        *,
        dedupe_key: str,
    ) -> Mapping[str, object]:
        with self.locked() as token:
            self._recover_locked(token, path=self.path)
            existing = self.read()
            try:
                normalized_payload = _plain(dict(payload))
            except (TypeError, ValueError) as exc:
                raise ValueError("adaptive event payload is invalid") from exc
            body = {
                "event_type": str(event_type),
                "payload": normalized_payload,
                "dedupe_key": str(dedupe_key),
            }
            try:
                event_id = digest(body)
            except (TypeError, ValueError) as exc:
                raise ValueError("adaptive event payload is invalid") from exc
            row = {**body, "event_id": event_id}
            _validate_adaptive_event_row(row)
            for existing_row in existing:
                if existing_row.get("dedupe_key") == dedupe_key:
                    if (
                        existing_row.get("payload") != normalized_payload
                        or existing_row.get("event_type") != event_type
                    ):
                        raise ValueError("conflicting adaptive event replay")
                    return existing_row
            self._append_row(self.path, row)
            return row
    def append_customer_action_continuity(
        self,
        continuity: CustomerActionContinuity,
    ) -> Mapping[str, object]:
        """Append one approved action under the existing recovery-aware ledger."""
        if not isinstance(continuity, CustomerActionContinuity):
            raise TypeError("customer action continuity must be typed")
        payload = {
            "customer_key": continuity.customer_key,
            "approved_proposal_digest": continuity.approved_proposal_digest,
            "revision": continuity.revision,
            "effective_kst_day": continuity.effective_kst_day.isoformat(),
            "action_id": continuity.action_id,
            "action_text": continuity.action_text,
            "action_atom": continuity.action_atom,
            "criterion_text": continuity.criterion_text,
            "criterion_atom": continuity.criterion_atom,
            "next_check_kst": continuity.next_check_kst,
            "approved_at_kst": continuity.approved_at_kst,
        }
        return self.append(
            "customer_action_continuity",
            payload,
            dedupe_key=f"customer-action:{continuity.action_id}",
        )

    def project_customer_action_outcomes(
        self,
        *,
        customer_key: str,
        canonical_events: Iterable[object],
        as_of_kst_day: date,
    ) -> tuple[CustomerActionOutcome, ...]:
        """Project later canonical evidence without creating a second memory store."""
        if not isinstance(customer_key, str) or not customer_key or type(as_of_kst_day) is not date:
            raise ValueError("customer action projection input is invalid")
        rows = canonical_event_records(canonical_events)
        ledger_rows = self.read()
        approved_identities = {
            (
                payload.get("customer_key"),
                payload.get("proposal_digest", payload.get("digest")),
                payload.get("revision"),
            )
            for row in ledger_rows
            if isinstance(row, Mapping)
            and row.get("event_type") == "plan_approved"
            and isinstance((payload := row.get("payload")), Mapping)
        }
        outcomes: list[CustomerActionOutcome] = []
        for row in ledger_rows:
            if row.get("event_type") != "customer_action_continuity":
                continue
            payload = row.get("payload")
            if not isinstance(payload, Mapping) or payload.get("customer_key") != customer_key:
                continue
            if (
                customer_key,
                payload.get("approved_proposal_digest"),
                payload.get("revision"),
            ) not in approved_identities:
                continue
            continuity = CustomerActionContinuity(
                customer_key=customer_key,
                approved_proposal_digest=str(payload["approved_proposal_digest"]),
                revision=payload["revision"],
                effective_kst_day=date.fromisoformat(str(payload["effective_kst_day"])),
                action_id=str(payload["action_id"]),
                action_text=str(payload["action_text"]),
                action_atom=str(payload["action_atom"]),
                criterion_text=str(payload["criterion_text"]),
                criterion_atom=str(payload["criterion_atom"]),
                next_check_kst=str(payload["next_check_kst"]),
                approved_at_kst=(
                    str(payload["approved_at_kst"])
                    if payload.get("approved_at_kst") is not None
                    else None
                ),
            )
            evidence = _customer_action_evidence(
                rows,
                continuity.criterion_atom,
                continuity.effective_kst_day,
                as_of_kst_day,
                customer_key=customer_key,
                approved_at_kst=continuity.approved_at_kst,
            )
            due_day = date.fromisoformat(continuity.next_check_kst[:10])
            outcome: Literal["pending", "met", "not_met", "insufficient"]
            if evidence is True:
                outcome = "met"
            elif evidence is False:
                outcome = "not_met"
            elif as_of_kst_day < due_day:
                outcome = "pending"
            else:
                outcome = "insufficient"
            outcomes.append(CustomerActionOutcome(
                continuity.action_id, continuity.action_text, continuity.criterion_text,
                continuity.next_check_kst, outcome,
            ))
        return tuple(outcomes)


    def read(self, *, recover: bool = False) -> list[dict[str, object]]:
        if recover:
            self.recover(path=self.path)
        rows = _read_jsonl_rows(self.path, allow_truncated_tail=False)
        _validate_adaptive_event_rows(rows)
        return rows

    def recover(self, *, path: Path | None = None) -> Mapping[str, object]:
        """Recover torn final records under one adaptive writer lock."""
        with self.locked() as token:
            return self._recover_locked(token, path=path)

    def _recover_locked(
        self,
        token: object,
        *,
        path: Path | None = None,
    ) -> Mapping[str, object]:
        """Recover only adaptive-owned journals while the lock token is held."""
        self._require_adaptive_token(token)
        owned_targets = {
            self.path,
            self.source_day_path,
            self.source_intent_path,
            self.authority_path,
            self.config_epoch_path,
        }
        targets = [Path(path)] if path is not None else list(owned_targets)
        overlay_path = self.root / "adaptive-overlays.jsonl"

        def is_overlay_target(target: Path) -> bool:
            if (
                target.name == overlay_path.name
                or os.path.abspath(target) == os.path.abspath(overlay_path)
            ):
                return True
            try:
                target_stat = target.lstat()
                overlay_stat = overlay_path.lstat()
            except OSError:
                return False
            return (
                target_stat.st_dev,
                target_stat.st_ino,
            ) == (
                overlay_stat.st_dev,
                overlay_stat.st_ino,
            )

        if any(is_overlay_target(target) for target in targets):
            raise ValueError("adaptive recovery cannot mutate overlay journal")
        if any(target not in owned_targets for target in targets):
            raise ValueError("adaptive recovery cannot mutate a foreign journal domain")
        recovered: dict[str, object] = {}
        for target in targets:
            if target.is_symlink():
                raise ValueError("adaptive recovery journal symlink is not allowed")
            if not target.exists():
                continue
            rows, complete_end = _read_jsonl_bytes(target, allow_truncated_tail=True)
            current_size = target.stat().st_size
            if complete_end < current_size:
                with target.open("r+b") as handle:
                    handle.truncate(complete_end)
                target.chmod(0o600)
                recovered[str(target)] = "truncated_torn_tail"
            else:
                recovered[str(target)] = len(rows)
            if target == self.path:
                _validate_adaptive_event_rows(rows)
            elif target == self.source_day_path:
                _validate_flat_journal_rows(rows, "source-day")
            elif target == self.source_intent_path:
                _validate_journal_rows(rows, "source_day")
            elif target == self.authority_path:
                _validate_journal_rows(rows, "authority")
            elif target == self.config_epoch_path:
                _validate_journal_rows(rows, "config_epoch")
        return recovered

    def _append_journal_row(
        self,
        kind: str,
        path: Path,
        payload: Mapping[str, object],
        *,
        intent_id: str,
        state: str,
    ) -> Mapping[str, object]:
        if state not in {"prepared", "committed", "abandoned"}:
            raise ValueError("invalid journal transition")
        if path not in {
            self.source_intent_path,
            self.authority_path,
            self.config_epoch_path,
        }:
            raise ValueError("unknown adaptive journal path")
        normalized = _plain(dict(payload))
        body: dict[str, object] = {
            "schema_version": "1.0",
            "kind": kind,
            "intent_id": str(intent_id),
            "state": state,
        }
        if isinstance(normalized, Mapping):
            for key, value in normalized.items():
                if key not in {"schema_version", "kind", "intent_id", "state", "row_digest"}:
                    body[str(key)] = value
        with self.locked() as token:
            self._recover_locked(token, path=path)
            all_rows = _read_jsonl_rows(path)
            _validate_journal_rows(all_rows, kind)
            existing_rows = [
                existing for existing in all_rows
                if existing.get("intent_id") == intent_id
            ]
            if state in {"committed", "abandoned"}:
                prepared = next(
                    (existing for existing in existing_rows if existing.get("state") == "prepared"),
                    None,
                )
                if prepared is None:
                    raise ValueError("adaptive journal terminal row has no prepared row")
                prepared_digest = prepared.get("row_digest")
                supplied_digest = body.get("prepared_digest")
                if supplied_digest is not None and supplied_digest != prepared_digest:
                    raise ValueError("adaptive journal prepared digest mismatch")
                body["prepared_digest"] = prepared_digest
            row = {**body, "row_digest": digest(body)}
            for existing in existing_rows:
                existing_state = existing.get("state")
                if existing_state == state:
                    if existing.get("row_digest") == row["row_digest"]:
                        return existing
                    raise ValueError("conflicting adaptive journal replay")
                if existing_state in {"committed", "abandoned"}:
                    raise ValueError("adaptive journal transition follows terminal state")
            return self._append_row(path, row)

    def _append_journal(
        self,
        kind: str,
        payload: Mapping[str, object],
        *,
        intent_id: str,
        state: str,
    ) -> Mapping[str, object]:
        paths = {
            "source_day": self.source_intent_path,
            "source-day": self.source_intent_path,
            "authority": self.authority_path,
            "authority_mirror": self.authority_path,
        }
        path = paths.get(kind)
        if path is None:
            raise ValueError("unknown adaptive journal")
        return self._append_journal_row(
            kind,
            path,
            payload,
            intent_id=intent_id,
            state=state,
        )

    def append_journal(
        self,
        kind: str,
        payload: Mapping[str, object],
        *,
        intent_id: str,
        state: str,
    ) -> Mapping[str, object]:
        if kind in {"config_epoch", "config-epoch"}:
            raise ValueError("config epoch requires append_config_epoch")
        return self._append_journal(
            kind,
            payload,
            intent_id=intent_id,
            state=state,
        )

    def validate_journals(self) -> Mapping[str, object]:
        result: dict[str, object] = {
            "adaptive_events": tuple(self.read()),
        }
        for kind in ("source_day", "authority", "config_epoch"):
            result[kind] = tuple(self.journal_rows(kind))
        result["source_day_mappings"] = tuple(self.source_day_rows())
        result["canonical_sequence"] = self.validate_canonical_prefix()
        return result

    def journal_rows(self, kind: str, *, recover: bool = False) -> list[dict[str, object]]:
        path = {
            "source_day": self.source_intent_path,
            "source-day": self.source_intent_path,
            "authority": self.authority_path,
            "authority_mirror": self.authority_path,
            "config_epoch": self.config_epoch_path,
            "config-epoch": self.config_epoch_path,
        }.get(kind)
        if path is None:
            raise ValueError("unknown adaptive journal")
        if recover:
            self.recover(path=path)
        rows = _read_jsonl_rows(path)
        _validate_journal_rows(rows, kind)
        return rows

    def source_day_rows(self, *, recover: bool = False) -> list[dict[str, object]]:
        if recover:
            self.recover(path=self.source_day_path)
        rows = _read_jsonl_rows(self.source_day_path)
        _validate_flat_journal_rows(rows, "source-day")
        return rows

    read_source_day_mappings = source_day_rows

    def _require_canonical_transaction(self) -> CanonicalEventTransaction:
        if self._canonical_transaction is None:
            raise RuntimeError("canonical event operations require a registered transaction")
        return self._canonical_transaction

    def append_canonical_event(
        self,
        event: object,
        *,
        intent_id: str | None = None,
    ) -> Mapping[str, object]:
        transaction = self._require_canonical_transaction()
        try:
            candidate = Event.model_validate(_canonical_event_record(event))
        except (TypeError, ValueError) as exc:
            raise ValueError("canonical event is invalid") from exc
        return transaction.append(candidate, intent_id=intent_id)

    def validate_canonical_prefix(self) -> tuple[Mapping[str, object], ...]:
        transaction = self._require_canonical_transaction()
        return transaction.read_snapshot().sequence_rows

    def freeze_legacy_prefix(self) -> Mapping[str, object]:
        raise ValueError("legacy canonical prefix is unsupported")

    def append_source_day_mapping(
        self,
        *,
        root_event_id: str,
        customer_key: str,
        mapped_flow: str,
        observation_kst_day: date | str,
        session_id: str,
        writer_epoch: int = 0,
        root_preimage_digest: str = "",
    ) -> Mapping[str, object]:
        day = _iso_day(observation_kst_day)
        body = {
            "schema_version": "1.0",
            "mapping_id": digest({
                "root_event_id": root_event_id,
                "customer_key": customer_key,
                "mapped_flow": mapped_flow,
                "observation_kst_day": day,
                "session_id": session_id,
                "writer_epoch": writer_epoch,
                "root_preimage_digest": root_preimage_digest,
            }),
            "root_event_id": str(root_event_id),
            "customer_key": str(customer_key),
            "mapped_flow": str(mapped_flow),
            "observation_kst_day": day,
            "session_id": str(session_id),
            "writer_epoch": int(writer_epoch),
            "root_preimage_digest": str(root_preimage_digest),
        }
        row = {**body, "row_digest": digest(body)}
        with self.locked() as token:
            self._recover_locked(token, path=self.source_day_path)
            existing_rows = _read_jsonl_rows(self.source_day_path)
            _validate_flat_journal_rows(existing_rows, "source-day")
            for existing in existing_rows:
                if existing.get("root_event_id") == root_event_id:
                    if existing.get("row_digest") == row["row_digest"]:
                        return existing
                    raise ValueError("conflicting source-day mapping")
            return self._append_row(self.source_day_path, row)

    source_day_mapping = append_source_day_mapping
    def prepare_source_day(self, intent_id: str, **payload: object) -> Mapping[str, object]:
        return self.append_journal("source_day", payload, intent_id=intent_id, state="prepared")

    def commit_source_day(self, intent_id: str, **payload: object) -> Mapping[str, object]:
        return self.append_journal("source_day", payload, intent_id=intent_id, state="committed")

    def abandon_source_day(self, intent_id: str, **payload: object) -> Mapping[str, object]:
        return self.append_journal("source_day", payload, intent_id=intent_id, state="abandoned")
    prepare_source_intent = prepare_source_day
    commit_source_intent = commit_source_day
    abandon_source_intent = abandon_source_day

    def append_authority_mirror(
        self,
        *,
        intent_id: str,
        authority_kind: str,
        canonical_fact_id: str,
        canonical_fact_digest: str,
        valid_from: str | date | datetime,
        adaptive_sequence: int,
        state: str = "prepared",
        **extra: object,
    ) -> Mapping[str, object]:
        if type(authority_kind) is not str or not authority_kind.strip():
            raise ValueError("authority mirror kind is invalid")
        if type(canonical_fact_id) is not str or not canonical_fact_id.strip():
            raise ValueError("authority mirror fact id is invalid")
        _require_digest(canonical_fact_digest, "authority mirror fact")
        if type(adaptive_sequence) is not int or adaptive_sequence < 0:
            raise ValueError("authority mirror sequence is invalid")
        if type(state) is not str or state not in {"prepared", "committed", "abandoned"}:
            raise ValueError("authority mirror state is invalid")
        payload = {
            "authority_kind": authority_kind,
            "canonical_fact_id": canonical_fact_id,
            "canonical_fact_digest": canonical_fact_digest,
            "valid_from": _kst_timestamp(valid_from),
            "adaptive_sequence": adaptive_sequence,
            **extra,
        }
        return self.append_journal(
            "authority",
            payload,
            intent_id=intent_id,
            state=state,
        )
    def mirror_authority(self, intent_id: str, **payload: object) -> Mapping[str, object]:
        return self.append_journal("authority", payload, intent_id=intent_id, state="committed")

    def append_config_epoch(
        self,
        epoch: int,
        config_digest: str,
        customer_keys: Sequence[str],
        *,
        state: str = "prepared",
        customer_states: Mapping[str, str] | None = None,
        prepared_at_kst: str | None = None,
        committed_at_kst: str | None = None,
        approved_by: Mapping[str, str] | None = None,
    ) -> Mapping[str, object]:
        approved_authority = (
            _validate_full_authority(approved_by, "config epoch approved_by")
            if approved_by is not None
            else None
        )
        epoch_value, digest_value, keys, states = _validate_config_epoch_inputs(
            epoch,
            config_digest,
            customer_keys,
            customer_states,
            state,
        )
        if state == "committed":
            prior_rows = self.journal_rows("config_epoch")
            prepared = next(
                (
                    row for row in prior_rows
                    if row.get("intent_id") == f"epoch:{epoch_value}"
                    and row.get("state") == "prepared"
                ),
                None,
            )
            if prepared is not None:
                prepared_keys = tuple(prepared.get("customer_keys", ()))
                expected_values = {
                    "epoch": epoch_value,
                    "config_digest": digest_value,
                    "customer_keys": keys,
                }
                observed_values = {
                    "epoch": prepared.get("epoch"),
                    "config_digest": prepared.get("config_digest"),
                    "customer_keys": prepared_keys,
                }
                if approved_authority is not None:
                    expected_values["approved_by"] = approved_authority
                if "approved_by" in prepared:
                    observed_values["approved_by"] = prepared["approved_by"]
                if observed_values != expected_values:
                    raise ValueError("config epoch prepared payload mismatch")
        payload: dict[str, object] = {
            "epoch": epoch_value,
            "config_digest": digest_value,
            "customer_keys": keys,
            "customer_state": states,
        }
        if approved_authority is not None:
            payload["approved_by"] = approved_authority
        if prepared_at_kst is not None:
            payload["prepared_at_kst"] = _kst_timestamp(prepared_at_kst)
        if committed_at_kst is not None:
            payload["committed_at_kst"] = _kst_timestamp(committed_at_kst)
        return self._append_journal_row(
            "config_epoch",
            self.config_epoch_path,
            payload,
            intent_id=f"epoch:{epoch_value}",
            state=state,
        )

    def append_overlay(self, overlay: AdaptiveOverlay | Mapping[str, object]) -> Mapping[str, object]:
        return OverlayJournal(self.overlay_path, root=self.root).append(overlay)

    def replace_overlay(
        self,
        old_revision_id: str | None,
        overlay: AdaptiveOverlay | Mapping[str, object],
    ) -> Mapping[str, object]:
        return OverlayJournal(self.overlay_path, root=self.root).replace(old_revision_id, overlay)

    def rollback_overlay(
        self,
        revision_id: str,
        *,
        as_of_kst: str | date | datetime,
        reason: str = "operator_rollback",
    ) -> Mapping[str, object]:
        return OverlayJournal(self.overlay_path, root=self.root).rollback(
            revision_id,
            as_of_kst=as_of_kst,
            reason=reason,
        )

    def resolve_overlay(
        self,
        *,
        effective_kst: str | date | datetime,
        as_of_sequence: int | None = None,
    ) -> AdaptiveOverlay | None:
        return OverlayJournal(self.overlay_path, root=self.root).resolve(
            effective_kst=effective_kst,
            as_of_sequence=as_of_sequence,
        )

    def project_schedule_baseline(
        self,
        *,
        customer_key: str,
        source_reference_id: str,
        source_reference_digest: str,
        policy_version: str,
        policy_digest: str,
        policy_document_digest: str,
        epoch: int,
        parent_digest: str,
    ) -> Mapping[str, object]:
        """Persist the deliberately restrictive pre-confirmation strategy once."""
        for label, value in (
            ("source reference", source_reference_digest), ("policy", policy_digest),
            ("policy document", policy_document_digest), ("parent", parent_digest),
        ):
            _require_digest(value, label)
        if type(epoch) is not int or epoch < 0 or not customer_key:
            raise ValueError("baseline projection pins are invalid")
        payload = {
            "customer_key": customer_key, "strategy_state": "schedule_unconfirmed",
            "source_reference_id": source_reference_id, "source_reference_digest": source_reference_digest,
            "policy_version": policy_version, "policy_digest": policy_digest,
            "policy_document_digest": policy_document_digest, "epoch": epoch,
            "parent_digest": parent_digest,
            "meal_principles": ("regular_meals", "protein_each_meal", "hydration", "regularity"),
            "review_required": True,
        }
        baseline_preimage = {
            "customer_key": customer_key,
            "source_reference_id": source_reference_id,
            "source_reference_digest": source_reference_digest,
            "policy_digest": policy_digest,
            "policy_document_digest": policy_document_digest,
            "epoch": epoch,
            "parent_digest": parent_digest,
        }
        return self.append(
            "schedule_strategy_baseline",
            payload,
            dedupe_key=f"schedule-baseline:{digest(baseline_preimage)}",
        )

    def project_confirmed_schedule_strategy(
        self,
        *,
        customer_key: str,
        source_reference_id: str,
        source_reference_digest: str,
        confirmation_id: str,
        source_day_mapping_digest: str,
        policy_version: str,
        policy_digest: str,
        policy_document_digest: str,
        epoch: int,
        parent_digest: str,
        categories: Sequence[str],
        last_change_note: str,
    ) -> Mapping[str, object]:
        """Persist one exact-confirmation, seven-day strategy projection."""
        if len(categories) != 7 or any(value not in {"training", "rest"} for value in categories):
            raise ValueError("confirmed strategy requires exactly seven training/rest categories")
        if not isinstance(last_change_note, str) or not 1 <= len(last_change_note) <= 500:
            raise ValueError("last change note is invalid")
        for label, value in (
            ("source reference", source_reference_digest), ("source-day mapping", source_day_mapping_digest),
            ("policy", policy_digest), ("policy document", policy_document_digest), ("parent", parent_digest),
        ):
            _require_digest(value, label)
        if type(epoch) is not int or epoch < 0 or not customer_key or not confirmation_id:
            raise ValueError("confirmed strategy pins are invalid")
        payload = {
            "customer_key": customer_key, "strategy_state": "schedule_confirmed",
            "source_reference_id": source_reference_id, "source_reference_digest": source_reference_digest,
            "confirmation_id": confirmation_id, "source_day_mapping_digest": source_day_mapping_digest,
            "policy_version": policy_version, "policy_digest": policy_digest,
            "policy_document_digest": policy_document_digest, "epoch": epoch,
            "parent_digest": parent_digest, "categories": tuple(categories),
            "last_change_note": last_change_note, "exception_behavior": "operator_review_required",
            "meal_principles": ("regular_meals", "protein_each_meal", "hydration", "regularity"),
            "review_required": False,
        }
        key_preimage = {
            "customer_key": customer_key,
            "source_reference_id": source_reference_id,
            "source_reference_digest": source_reference_digest,
            "confirmation_id": confirmation_id,
            "source_day_mapping_digest": source_day_mapping_digest,
            "policy_version": policy_version,
            "policy_digest": policy_digest,
            "policy_document_digest": policy_document_digest,
            "epoch": epoch,
            "parent_digest": parent_digest,
        }
        return self.append(
            "schedule_strategy_confirmed",
            payload,
            dedupe_key=f"schedule-strategy:{digest(key_preimage)}",
        )
    def append_revision(
        self,
        proposal: NutritionProposal,
        *,
        parent_digest: str,
        operator_id: str = "",
    ) -> Mapping[str, object]:
        if proposal.parent_digest != parent_digest:
            raise ValueError("stale proposal parent digest")
        return self.append(
            "plan_revision_created",
            {
                "proposal_digest": proposal.digest,
                "parent_digest": parent_digest,
                "revision": proposal.revision,
                "operator_id": operator_id,
            },
            dedupe_key=f"revision:{proposal.digest}",
        )
    def approve(
        self,
        proposal: NutritionProposal,
        *,
        operator_id: str,
        expected_digest: str,
    ) -> Mapping[str, object]:
        if proposal.digest != expected_digest:
            raise ValueError("stale proposal digest")
        return self.append(
            "plan_approved",
            {
                "proposal_digest": proposal.digest,
                "revision": proposal.revision,
                "operator_id": operator_id,
                "topic_id": 59,
                "execution_mode": "shadow_test_only",
                "customer_body_digest": proposal.customer_body_digest,
                "meal_plan_digest": digest(proposal.meal_plan) if proposal.meal_plan is not None else None,
            },
            dedupe_key=f"approve:{proposal.digest}:{operator_id}",
        )


def render_operator_card(proposal: NutritionProposal) -> str:
    s = proposal.snapshot
    lines = [
        f"{proposal.customer_key} · {s.evaluation_day.isoformat()} · D+{s.d_plus}",
        f"결정: {proposal.decision.value}",
        f"근거: {', '.join(proposal.reasons)}",
    ]
    if s.weekly_rate_percent is not None:
        lines.append(f"주간 체중 변화: {s.weekly_rate_percent}% ({s.current_samples}+{s.prior_samples}일)")
    if s.adherence_complete_days or s.adherence_missing_days or s.adherence_contradictory_days:
        lines.append(
            "순응 근거: "
            f"완료 {s.adherence_complete_days}일 · 적합 {s.adherent_days}일 · "
            f"부족 {s.adherence_missing_days}일 · 충돌 {s.adherence_contradictory_days}일"
        )
    if proposal.target:
        t = proposal.target
        lines.append(f"제안: {t.calories}kcal · 탄 {t.carbs_g}g · 단 {t.protein_g}g · 지 {t.fat_g}g")
    if proposal.carb_days:
        lines.append(
            "탄수화물 부하: "
            + ", ".join(f"{day.isoformat()}={load}" for day, load in proposal.carb_days)
        )
    if proposal.weekly_carb_cycle is not None and proposal.weekly_carb_cycle.has_redistribution:
        cycle = proposal.weekly_carb_cycle
        lines.append(
            "주간 목표: "
            + ", ".join(
                f"{category}={target.calories}kcal/{target.carbs_g}g/{target.protein_g}g/{target.fat_g}g"
                for category, target in cycle.category_targets
            )
        )
    if proposal.cooldown is not None and proposal.cooldown.active:
        lines.append("쿨다운: cooldown_active")
    if proposal.meal_plan:
        for slot in proposal.meal_plan.slots:
            quantities = ", ".join(
                f"{food_id}×{quantity}" for food_id, quantity in slot.food_quantities
            )
            lines.append(f"{slot.name}: {quantities}")
    note = _bounded_operator_note(proposal.operator_note)
    if note:
        lines.append(f"운영자 메모: {note}")
    lines.append("상태: 운영자 승인 전 · 고객에게 자동 전달하지 않음")
    return "\n".join(lines)


def _bounded_operator_note(value: object) -> str:
    if value is None:
        return ""
    if not isinstance(value, str):
        raise ValueError("operator note is invalid")
    if len(value) > OPERATOR_NOTE_MAX_LENGTH:
        raise ValueError("operator note is too long")
    return value


def render_customer_body(proposal: NutritionProposal) -> str:
    """Render structured facts plus a validated deterministic explanation; no operator-only evidence leaks."""

    lines = ["식단 제안"]
    lines.append(validate_explanation(proposal.explanation, proposal))
    if proposal.target is None:
        lines.append("현재는 입력을 확인한 뒤 다음 단계를 안내드릴게요.")
    else:
        target = proposal.target
        lines.append(f"하루 목표: {target.calories}kcal · 탄수화물 {target.carbs_g}g · 단백질 {target.protein_g}g · 지방 {target.fat_g}g")
    if proposal.meal_plan is not None:
        for slot in proposal.meal_plan.slots:
            entries = []
            for food_id, quantity in slot.food_quantities:
                grams = ""
                if slot.serving_grams:
                    index = slot.food_ids.index(food_id)
                    serving_grams = slot.serving_grams[index]
                    if serving_grams > 0:
                        grams = f" ({serving_grams}g)"
                entries.append(f"{food_id} {quantity}회{grams}")
            lines.append(f"{slot.name}: " + ", ".join(entries))
    if proposal.weekly_nutrition_plan is not None:
        lines.append("요일별 탄수·식사 계획")
        for daily in proposal.weekly_nutrition_plan.days:
            entries = []
            for slot in daily.meal_plan.slots:
                for index, (food_id, quantity) in enumerate(slot.food_quantities):
                    grams = (
                        slot.serving_grams[index]
                        if index < len(slot.serving_grams)
                        else 0
                    )
                    amount = f"{grams}g" if grams > 0 else f"{quantity}회"
                    entries.append(f"{food_id} {amount}")
            lines.append(
                f"{daily.kst_day.isoformat()} [{daily.category}] "
                f"{daily.target.calories}kcal · 탄수 {daily.target.carbs_g}g · "
                f"단백질 {daily.target.protein_g}g · 지방 {daily.target.fat_g}g"
            )
            lines.append("식사: " + ", ".join(entries))
    if proposal.decision in {Decision.OBSERVE, Decision.HUMAN_REVIEW}:
        lines.append("운영자 확인 후 안내드리겠습니다.")
    return "\n".join(lines)
_EXPLANATION_MAX_LENGTH = 1000


def _explanation_clauses(proposal: NutritionProposal) -> tuple[str, ...]:
    """Build the only customer-facing clauses permitted for this proposal."""

    target = proposal.target
    if target is None:
        return ("현재 제안은 운영자 확인이 필요합니다.",)
    clauses = (
        f"하루 목표는 {target.calories}kcal, 탄수화물 {target.carbs_g}g, "
        f"단백질 {target.protein_g}g, 지방 {target.fat_g}g입니다.",
    )
    cycle = proposal.weekly_carb_cycle
    if cycle is None or not cycle.has_redistribution:
        return clauses
    category_lines = []
    for category in ("high", "medium", "low"):
        category_target = cycle.category_target(category)
        if category_target is not None:
            category_lines.append(
                f"{category} {category_target.calories}kcal/{category_target.carbs_g}g/"
                f"{category_target.protein_g}g/{category_target.fat_g}g"
            )
    if not category_lines:
        return clauses
    return clauses + ("주간 목표: " + ", ".join(category_lines),)


def render_explanation_fallback(proposal: NutritionProposal) -> str:
    """Render only deterministic proposal facts for customer-facing explanation."""

    return "\n".join(_explanation_clauses(proposal))


def _explanation_allowlist(proposal: NutritionProposal) -> frozenset[str]:
    """Return exact renderings derived exclusively from immutable proposal fields."""

    clauses = _explanation_clauses(proposal)
    return frozenset({"\n".join(clauses)})


def validate_explanation(
    text: object,
    proposal: NutritionProposal,
    *,
    max_length: int = _EXPLANATION_MAX_LENGTH,
) -> str:
    """Accept only an exact deterministic rendering; otherwise fail closed."""

    fallback = render_explanation_fallback(proposal)
    if not isinstance(max_length, int) or max_length <= 0 or max_length > _EXPLANATION_MAX_LENGTH:
        return fallback
    if not isinstance(text, str):
        return fallback
    candidate = unicodedata.normalize("NFC", text)
    if not candidate or len(candidate) > max_length:
        return fallback
    if candidate not in _explanation_allowlist(proposal):
        return fallback
    return candidate
validate_llm_explanation = validate_explanation
fallback_explanation = render_explanation_fallback


def render_operator_body(proposal: NutritionProposal) -> str:
    return render_operator_card(proposal)


def render_customer_card(proposal: NutritionProposal) -> str:
    return render_customer_body(proposal)


def render_operator_digest(proposal: NutritionProposal) -> str:
    return hashlib.sha256(render_operator_card(proposal).encode("utf-8")).hexdigest()


def render_customer_digest(proposal: NutritionProposal) -> str:
    return hashlib.sha256(render_customer_body(proposal).encode("utf-8")).hexdigest()


def create_proposal_revision(
    proposal: NutritionProposal,
    *,
    target: MacroTarget | None = None,
    meal_plan: MealPlan | None = None,
    operator_note: str | None = None,
    reasons: Sequence[str] | None = None,
    explanation: str | None = None,
) -> NutritionProposal:
    """Create an immutable child revision pinned to its exact parent digest."""

    child = replace(
        proposal,
        target=proposal.target if target is None else target,
        meal_plan=proposal.meal_plan if meal_plan is None else meal_plan,
        weekly_carb_cycle=proposal.weekly_carb_cycle if target is None else None,
        weekly_nutrition_plan=(
            proposal.weekly_nutrition_plan
            if target is None
            else None
        ),
        operator_note=_bounded_operator_note(
            proposal.operator_note if operator_note is None else operator_note
        ),
        reasons=proposal.reasons if reasons is None else tuple(str(reason) for reason in reasons),
        revision=proposal.revision + 1,
        parent_digest=proposal.digest,
        operator_body=None,
        customer_body=None,
        operator_body_digest=None,
        customer_body_digest=None,
        adherence_signal_digest=None,
        explanation=None,
    )
    validated_explanation = validate_explanation(explanation, child)
    child = replace(child, explanation=validated_explanation)
    operator_body = render_operator_card(child)
    customer_body = render_customer_body(child)
    return replace(
        child,
        operator_body=operator_body,
        customer_body=customer_body,
        operator_body_digest=hashlib.sha256(operator_body.encode("utf-8")).hexdigest(),
        customer_body_digest=hashlib.sha256(customer_body.encode("utf-8")).hexdigest(),
        adherence_signal_digest=child.snapshot.adherence_digest,
    )


revise_proposal = create_proposal_revision
new_proposal_revision = create_proposal_revision

def initialize_adaptive_customer(
    data_root: Path,
    *,
    extension_through: date | None = None,
) -> Mapping[str, str]:
    """Idempotently create the disabled adaptive-nutrition customer boundary."""
    root_data = Path(data_root)
    if root_data.is_symlink():
        raise ValueError("adaptive customer data root is invalid")
    if root_data.exists():
        if not root_data.is_dir():
            raise ValueError("adaptive customer data root is invalid")
    else:
        try:
            root_data.mkdir(parents=True, mode=0o700)
        except OSError as exc:
            if root_data.is_symlink() or not root_data.is_dir():
                raise ValueError("adaptive customer data root is invalid") from exc
    if root_data.is_symlink() or not root_data.is_dir():
        raise ValueError("adaptive customer data root is invalid")
    root_data.chmod(0o700)
    root = root_data / "nutrition-plans"
    if root.is_symlink():
        raise ValueError("adaptive nutrition runtime symlink is not allowed")
    if root.exists() and not root.is_dir():
        raise ValueError("adaptive nutrition runtime is not a directory")
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    if root.is_symlink() or not root.is_dir():
        raise ValueError("adaptive nutrition runtime is not a directory")
    root.chmod(0o700)
    policy_default: dict[str, object] = {"schema_version": "1.0", "enabled": False}
    if extension_through is not None:
        if not isinstance(extension_through, date):
            raise ValueError("adaptive extension date is invalid")
        policy_default["extension_through"] = extension_through.isoformat()
    defaults = {
        "events.jsonl": "",
        "canonical-sequence.jsonl": "",
        "source-days.jsonl": "",
        "source-day-intents.jsonl": "",
        "authority-mirror-intents.jsonl": "",
        "config-epoch-journal.jsonl": "",
        "adaptive-overlays.jsonl": "",
        ".adaptive.lock": "",
        ".authority-transition.lock": "",
        _POLICY_REVISION_FILE: "",
        _POLICY_AUDIT_FILE: "",
        "policy.json": canonical_json(policy_default) + "\n",
        "meal-constraints.json": canonical_json({"schema_version": "1.0", "approved": False}) + "\n",
        "food-catalog.json": canonical_json({"schema_version": "1.0", "approved": False, "catalog": []}) + "\n",
        "feature-epoch.json": canonical_json({
            "schema_version": "1.0",
            "epoch": 0,
            "config_digest": feature_config_digest(0, {
                "analytics_shadow": False,
                "operator_candidates": False,
                "activation": False,
                "delivery": False,
            }),
            "analytics_shadow": False,
            "operator_candidates": False,
            "activation": False,
            "delivery": False,
        }) + "\n",
    }
    canonical_catalog_path = root / "food-catalog.json"
    legacy_catalog_path = root / "catalog.json"
    if (
        not (canonical_catalog_path.exists() or canonical_catalog_path.is_symlink())
        and (legacy_catalog_path.exists() or legacy_catalog_path.is_symlink())
    ):
        defaults.pop("food-catalog.json")
    result: dict[str, str] = {}
    for name, content in defaults.items():
        path = root / name
        if path.is_symlink():
            raise ValueError("adaptive runtime file symlink is not allowed")
        if path.exists() and not path.is_file():
            raise ValueError("adaptive runtime entry is not a file")
        if not path.exists():
            path.write_text(content, encoding="utf-8")
        path.chmod(0o600)
        result[name] = hashlib.sha256(path.read_bytes()).hexdigest()
    return result
def _parse_jsonl_bytes(
    data: bytes,
    path: Path,
    *,
    allow_truncated_tail: bool,
) -> tuple[list[dict[str, object]], int]:
    rows: list[dict[str, object]] = []
    offset = 0
    for index, raw_line in enumerate(data.splitlines(keepends=True)):
        is_final = offset + len(raw_line) == len(data)
        if not raw_line.endswith(b"\n"):
            if is_final and allow_truncated_tail:
                break
            raise ValueError(f"truncated JSONL record in {path}")
        line = raw_line[:-1]
        if line.endswith(b"\r"):
            line = line[:-1]
        if not line:
            raise ValueError(f"blank JSONL record in {path}")
        try:
            value = json.loads(line.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise ValueError(f"corrupt JSONL record in {path} at index {index}") from exc
        if not isinstance(value, dict):
            raise ValueError(f"JSONL record is not an object in {path}")
        rows.append(value)
        offset += len(raw_line)
    return rows, offset


def _read_jsonl_bytes(
    path: Path,
    *,
    allow_truncated_tail: bool,
) -> tuple[list[dict[str, object]], int]:
    if not path.exists():
        return [], 0
    return _parse_jsonl_bytes(
        path.read_bytes(),
        path,
        allow_truncated_tail=allow_truncated_tail,
    )


def _read_jsonl_rows(path: Path, *, allow_truncated_tail: bool = False) -> list[dict[str, object]]:
    rows, complete_end = _read_jsonl_bytes(path, allow_truncated_tail=allow_truncated_tail)
    if path.exists() and complete_end != path.stat().st_size:
        raise ValueError(f"truncated JSONL record in {path}")
    return rows


def _customer_action_evidence(
    rows: Iterable[Mapping[str, object]],
    criterion_atom: str,
    effective_day: date,
    as_of_day: date,
    *,
    approved_at_kst: str | None,
    customer_key: str,
) -> bool | None:
    """Return only evidence observable in later canonical customer records."""
    saw_relevant = False
    for row in rows:
        row_customer_key = row.get("customer_key")
        if row_customer_key is not None and row_customer_key != customer_key:
            raise ValueError("customer action evidence crosses customer boundary")
        raw_day = row.get("observation_kst_day") or row.get("source_day") or row.get("occurred_at_kst")
        try:
            observed_day = date.fromisoformat(str(raw_day)[:10])
        except (TypeError, ValueError):
            continue
        if not effective_day <= observed_day <= as_of_day:
            continue
        if approved_at_kst is not None:
            raw_timestamp = row.get("occurred_at_kst") or row.get("recorded_at_kst")
            try:
                if (
                    raw_timestamp is None
                    or datetime.fromisoformat(str(raw_timestamp))
                    <= datetime.fromisoformat(approved_at_kst)
                ):
                    continue
            except (TypeError, ValueError):
                continue
        check_in = row.get("check_in")
        check_in = check_in if isinstance(check_in, Mapping) else {}
        if criterion_atom == "checkin_recorded" and check_in:
            return True
        if criterion_atom == "weight_recorded" and check_in.get("body_weight_kg") is not None:
            return True
        if criterion_atom == "adherence_recorded":
            values = _canonical_adherence_values(row, check_in)
            if values.get("adherence_ok") is True:
                return True
            if values.get("adherence_ok") is False:
                saw_relevant = True
        if criterion_atom == "exercise_session_recorded" and isinstance(row.get("exercise_session"), Mapping):
            return True
    return False if saw_relevant else None

_ADAPTIVE_EVENT_TYPES = frozenset({
    "plan_proposed",
    "revision_created",
    "plan_revision_created",
    "schedule_strategy_baseline",
    "schedule_strategy_confirmed",
    "schedule_confirmation_projection",
    "dual_coach_risk_review",
    "missing_checkin_reminder_review",
    "plan_edited",
    "plan_held",
    "plan_released",
    "plan_approved",
    "callback_issued",
    "activation_prepared",
    "adaptive_plan_effective",
    "activation_committed",
    "activation_abandoned",
    "rollback_prepared",
    "overlay_rolled_back",
    "rollback_committed",
    "rollback_abandoned",
    "delivery_attempt_started",
    "delivery_receipt_recorded",
    "delivery_unknown",
    "delivery_preflight_rejected",
    "delivered",
    "audit_pending",
    "sent_audited",
    "transition_prepared",
    "transition_committed",
    "transition_aborted",
    "delivery_attempt_consumed",
    "adaptive_plan_activated",
    "adaptive_plan_rolled_back",
    "customer_action_continuity",
})
_ADAPTIVE_EVENT_SCHEMA_VERSION = "1.0"
_ADAPTIVE_EVENT_ROW_FIELDS = frozenset({
    "event_type",
    "payload",
    "dedupe_key",
    "event_id",
    "schema_version",
})
_DELIVERY_REFERENCE_FIELDS = frozenset({
    "receipt",
    "receipt_id",
    "provider_receipt",
    "message_id",
    "attempt_id",
    "attempt_event_id",
    "delivery_attempt_id",
    "started_event_id",
    "previous_event_id",
    "delivered_event_id",
    "reason",
    "transport_receipt",
    "previous",
    "prior",
    "source",
    "attempt",
    "started",
})


def _payload_revision(payload: Mapping[str, object]) -> int:
    revision = payload.get("revision")
    if isinstance(revision, bool) or not isinstance(revision, int) or revision < 1:
        raise ValueError("adaptive event revision is invalid")
    return revision


def _payload_digest(payload: Mapping[str, object], *names: str) -> str:
    for name in names:
        if name in payload:
            value = payload[name]
            _require_digest(value, f"adaptive {name}")
            return value
    raise ValueError("adaptive event digest is missing")


def _validate_adaptive_event_payload(event_type: str, payload: Mapping[str, object]) -> None:
    if not payload:
        raise ValueError("adaptive event payload is invalid")
    if event_type == "plan_revision_created":
        required = {"proposal_digest", "parent_digest", "revision", "operator_id"}
        if not required.issubset(payload):
            raise ValueError("adaptive event payload schema mismatch")
        _payload_digest(payload, "proposal_digest")
        _payload_digest(payload, "parent_digest")
        _payload_revision(payload)
        if not isinstance(payload["operator_id"], str) or len(payload["operator_id"]) > 256:
            raise ValueError("adaptive event operator is invalid")
    elif event_type in {"plan_proposed", "revision_created"}:
        if "proposal_digest" in payload or "digest" in payload:
            _payload_digest(payload, "proposal_digest", "digest")
        elif isinstance(payload.get("proposal"), Mapping) and "digest" in payload["proposal"]:
            _require_digest(payload["proposal"]["digest"], "adaptive proposal")
        elif isinstance(payload.get("proposal"), str) and len(payload["proposal"]) == 64:
            _require_digest(payload["proposal"], "adaptive proposal")
        else:
            raise ValueError("adaptive event digest is missing")
        _payload_revision(payload)
        if "proposal" in payload and not isinstance(payload["proposal"], (str, Mapping)):
            raise ValueError("adaptive event proposal is invalid")
        if "proposal_digest" in payload or "digest" in payload:
            declared_digest = payload.get("proposal_digest", payload.get("digest"))
            embedded = payload.get("proposal")
            embedded_digest = embedded.get("digest") if isinstance(embedded, Mapping) else (
                embedded if isinstance(embedded, str) and len(embedded) == 64 else None
            )
            if embedded_digest is not None and embedded_digest != declared_digest:
                raise ValueError("adaptive proposal digest mismatch")
        if "parent_digest" in payload:
            _payload_digest(payload, "parent_digest")
    elif event_type in {"plan_edited", "plan_held", "plan_released"}:
        _payload_digest(payload, "proposal_digest", "digest")
        _payload_digest(payload, "parent_digest")
        _payload_revision(payload)
    elif event_type == "plan_approved":
        _payload_digest(payload, "proposal_digest", "digest")
        _payload_revision(payload)
        operator_id = payload.get("operator_id")
        if not isinstance(operator_id, str) or not operator_id or len(operator_id) > 256:
            raise ValueError("adaptive approval operator is required")
        if payload.get("topic_id") != 59:
            raise ValueError("adaptive approval payload is invalid")
        if "execution_mode" in payload and payload["execution_mode"] not in {
            "shadow_test_only",
            "production",
        }:
            raise ValueError("adaptive approval payload is invalid")
        for key in ("customer_body_digest", "meal_plan_digest"):
            if key in payload and payload[key] is not None:
                _require_digest(payload[key], f"adaptive {key}")
    elif event_type == "customer_action_continuity":
        required = {
            "customer_key", "approved_proposal_digest", "revision", "effective_kst_day",
            "action_id", "action_text", "action_atom", "criterion_text", "criterion_atom",
            "next_check_kst",
        }
        if (
            not required.issubset(payload)
            or set(payload) - required not in (set(), {"approved_at_kst"})
        ):
            raise ValueError("customer action payload schema mismatch")
        CustomerActionContinuity(
            customer_key=payload["customer_key"],
            approved_proposal_digest=payload["approved_proposal_digest"],
            revision=payload["revision"],
            effective_kst_day=date.fromisoformat(str(payload["effective_kst_day"])),
            action_id=payload["action_id"],
            action_text=payload["action_text"],
            action_atom=payload["action_atom"],
            criterion_text=payload["criterion_text"],
            criterion_atom=payload["criterion_atom"],
            next_check_kst=payload["next_check_kst"],
            approved_at_kst=(
                str(payload["approved_at_kst"])
                if payload.get("approved_at_kst") is not None
                else None
            ),
        )
    elif event_type == "callback_issued":
        _payload_digest(payload, "proposal_digest", "digest")
        _payload_revision(payload)
        for key in ("token", "action"):
            if not isinstance(payload.get(key), str) or not payload[key]:
                raise ValueError("adaptive callback payload is invalid")
    elif event_type == "delivery_attempt_started":
        required = {"delivery_id", "proposal_digest", "customer_body_digest"}
        if not required.issubset(payload):
            raise ValueError("adaptive delivery payload is invalid")
        if not isinstance(payload["delivery_id"], str) or not payload["delivery_id"]:
            raise ValueError("adaptive delivery payload is invalid")
        destination = payload.get("destination")
        if isinstance(destination, Mapping):
            if any(
                not isinstance(destination.get(key), str) or not destination[key]
                for key in ("user_id", "chat_id", "topic_id")
            ):
                raise ValueError("adaptive delivery payload is invalid")
        elif not (
            isinstance(payload.get("chat_id"), str)
            and payload["chat_id"]
            and payload.get("topic_id") == 59
        ):
            raise ValueError("adaptive delivery payload is invalid")
        _payload_digest(payload, "proposal_digest")
        _payload_digest(payload, "customer_body_digest")
        if payload.get("meal_plan_digest") is not None:
            _require_digest(payload["meal_plan_digest"], "adaptive meal plan")
    elif event_type == "delivery_receipt_recorded":
        delivery_id = payload.get("delivery_id")
        if not isinstance(delivery_id, str) or not delivery_id:
            raise ValueError("adaptive delivery payload is invalid")
        receipt_values: list[str] = []
        for key in ("provider_receipt", "message_id"):
            if key not in payload:
                continue
            value = payload[key]
            if isinstance(value, bool) or not isinstance(value, (str, int)):
                raise ValueError("adaptive delivery receipt is invalid")
            normalized = str(value).strip()
            if not normalized or len(normalized) > 128:
                raise ValueError("adaptive delivery receipt is invalid")
            receipt_values.append(normalized)
        if not receipt_values or len(set(receipt_values)) > 1:
            raise ValueError("adaptive delivery receipt is invalid")
        receipt = payload.get("receipt")
        if receipt is not None:
            if not isinstance(receipt, Mapping) or receipt.get("ok") is not True:
                raise ValueError("adaptive delivery receipt is invalid")
            nested = receipt.get("message_id")
            if isinstance(nested, bool) or not isinstance(nested, (str, int)):
                raise ValueError("adaptive delivery receipt is invalid")
            if str(nested).strip() != receipt_values[0]:
                raise ValueError("adaptive delivery receipt is invalid")
        if "receipt_id" in payload:
            _require_digest(payload["receipt_id"], "adaptive receipt")
    elif event_type in {
        "delivery_unknown",
        "delivery_preflight_rejected",
        "delivered",
        "audit_pending",
        "sent_audited",
    }:
        if not isinstance(payload.get("delivery_id"), str) or not payload["delivery_id"]:
            raise ValueError("adaptive delivery payload is invalid")
        if not any(
            key in payload and payload[key] not in (None, "")
            for key in _DELIVERY_REFERENCE_FIELDS
        ):
            raise ValueError("adaptive delivery predecessor is missing")
    elif event_type == "delivery_attempt_consumed":
        if not isinstance(payload.get("delivery_id"), str) or not payload["delivery_id"]:
            raise ValueError("adaptive delivery payload is invalid")
    elif event_type in {
        "activation_prepared",
        "adaptive_plan_effective",
        "activation_committed",
        "activation_abandoned",
        "rollback_prepared",
        "overlay_rolled_back",
        "rollback_committed",
        "rollback_abandoned",
        "transition_prepared",
        "transition_committed",
        "transition_aborted",
        "adaptive_plan_activated",
        "adaptive_plan_rolled_back",
    }:
        identity_fields = {
            "intent_id",
            "transaction_id",
            "activation_id",
            "rollback_id",
            "revision_id",
            "proposal_digest",
            "overlay_digest",
            "customer_key",
        }
        if not any(key in payload and payload[key] not in (None, "") for key in identity_fields):
            raise ValueError("adaptive transition identity is missing")
        if "proposal_digest" in payload and payload["proposal_digest"] is not None:
            _require_digest(payload["proposal_digest"], "adaptive proposal")
        if "revision" in payload:
            _payload_revision(payload)
    if "customer_body_digest" in payload and payload["customer_body_digest"] is not None:
        _require_digest(payload["customer_body_digest"], "adaptive customer body")
    if "meal_plan_digest" in payload and payload["meal_plan_digest"] is not None:
        _require_digest(payload["meal_plan_digest"], "adaptive meal plan")


def _validate_adaptive_event_row(row: Mapping[str, object]) -> None:
    if not isinstance(row, Mapping):
        raise ValueError("adaptive event row is not an object")
    if set(row) - _ADAPTIVE_EVENT_ROW_FIELDS:
        raise ValueError("adaptive event row schema mismatch")
    if row.get("schema_version", _ADAPTIVE_EVENT_SCHEMA_VERSION) != _ADAPTIVE_EVENT_SCHEMA_VERSION:
        raise ValueError("adaptive event schema mismatch")
    event_type = row.get("event_type")
    if not isinstance(event_type, str) or event_type not in _ADAPTIVE_EVENT_TYPES:
        raise ValueError("unknown adaptive event type")
    payload = row.get("payload")
    if not isinstance(payload, Mapping):
        raise ValueError("adaptive event payload is invalid")
    _validate_adaptive_event_payload(event_type, payload)
    dedupe_key = row.get("dedupe_key")
    if not isinstance(dedupe_key, str) or not dedupe_key.strip() or len(dedupe_key) > 256:
        raise ValueError("adaptive event dedupe key is invalid")
    event_id = row.get("event_id")
    _require_digest(event_id, "adaptive event")
    body = {
        "event_type": event_type,
        "payload": dict(payload),
        "dedupe_key": dedupe_key,
    }
    if event_id != digest(body):
        raise ValueError("adaptive event id digest mismatch")


def _validate_adaptive_event_rows(rows: Sequence[Mapping[str, object]]) -> None:
    event_ids: set[str] = set()
    dedupe_keys: set[str] = set()
    for row in rows:
        _validate_adaptive_event_row(row)
        event_id = str(row["event_id"])
        dedupe_key = str(row["dedupe_key"])
        if event_id in event_ids or dedupe_key in dedupe_keys:
            raise ValueError("duplicate adaptive event row")
        event_ids.add(event_id)
        dedupe_keys.add(dedupe_key)


_CONFIG_EPOCH_STATE_MAP = {
    "prepared": "pending",
    "committed": "committed",
    "abandoned": "abandoned",
}


def _validate_full_authority(value: object, label: str) -> dict[str, str]:
    if not isinstance(value, Mapping) or set(value) != {"user_id", "chat_id", "topic_id"}:
        raise ValueError(f"{label} is invalid")
    result = {
        key: candidate
        for key, candidate in value.items()
    }
    if any(type(candidate) is not str or not candidate.strip() for candidate in result.values()):
        raise ValueError(f"{label} is invalid")
    return result

def _validate_config_epoch_inputs(
    epoch: object,
    config_digest: object,
    customer_keys: object,
    customer_states: object,
    state: object,
) -> tuple[int, str, tuple[str, ...], dict[str, str]]:
    if type(epoch) is not int or epoch < 0:
        raise ValueError("config epoch is invalid")
    if type(state) is not str or state not in _CONFIG_EPOCH_STATE_MAP:
        raise ValueError("config epoch state is invalid")
    normalized_digest = _require_config_digest(config_digest)
    if isinstance(customer_keys, (str, bytes)) or not isinstance(customer_keys, Sequence):
        raise ValueError("config epoch customer keys are invalid")
    keys: list[str] = []
    seen: set[str] = set()
    for key in customer_keys:
        if type(key) is not str or not key.strip() or key != key.strip() or key in seen:
            raise ValueError("config epoch customer keys are invalid")
        seen.add(key)
        keys.append(key)
    if not isinstance(customer_states, Mapping):
        raise ValueError("config epoch customer states are invalid")
    if set(customer_states) != set(keys):
        raise ValueError("config epoch customer states are invalid")
    states: dict[str, str] = {}
    for key, value in customer_states.items():
        if type(key) is not str or type(value) is not str:
            raise ValueError("config epoch customer states are invalid")
        if value != _CONFIG_EPOCH_STATE_MAP.get(state):
            raise ValueError("config epoch customer states are invalid")
        states[key] = value
    return epoch, normalized_digest, tuple(sorted(keys)), {
        key: states[key] for key in sorted(states)
    }


def _validate_config_epoch_rows(rows: Sequence[Mapping[str, object]]) -> None:
    prepared: dict[str, Mapping[str, object]] = {}
    for row in rows:
        epoch = row.get("epoch")
        config_digest = row.get("config_digest")
        keys = row.get("customer_keys")
        states = row.get("customer_state")
        state = row.get("state")
        if "approved_by" in row:
            _validate_full_authority(row["approved_by"], "config epoch approved_by")
        normalized_epoch, normalized_digest, normalized_keys, normalized_states = (
            _validate_config_epoch_inputs(epoch, config_digest, keys, states, state)
        )
        intent_id = row.get("intent_id")
        if intent_id != f"epoch:{normalized_epoch}":
            raise ValueError("config epoch intent is invalid")
        if row.get("epoch") != normalized_epoch or row.get("config_digest") != normalized_digest:
            raise ValueError("config epoch payload is invalid")
        if tuple(keys) != normalized_keys or dict(states) != normalized_states:
            raise ValueError("config epoch payload is invalid")
        if state == "prepared":
            prepared[str(intent_id)] = row
            continue
        prior = prepared.get(str(intent_id))
        if prior is None:
            raise ValueError("config epoch terminal row has no prepared row")
        for field in ("epoch", "config_digest", "customer_keys", "approved_by"):
            if field in prior or field in row:
                if row.get(field) != prior.get(field):
                    raise ValueError("config epoch prepared payload mismatch")


def _validate_source_day_mapping_row(row: Mapping[str, object]) -> None:
    expected_fields = {
        "schema_version",
        "mapping_id",
        "root_event_id",
        "customer_key",
        "mapped_flow",
        "observation_kst_day",
        "session_id",
        "writer_epoch",
        "root_preimage_digest",
        "row_digest",
    }
    if set(row) != expected_fields:
        raise ValueError("source-day journal schema mismatch")
    if row.get("schema_version") != "1.0":
        raise ValueError("source-day journal schema mismatch")
    _require_digest(row.get("mapping_id"), "source-day mapping")
    if row.get("root_preimage_digest") != "":
        _require_digest(row.get("root_preimage_digest"), "source-day root preimage")
    _require_digest(row.get("row_digest"), "source-day row")
    for field in ("root_event_id", "customer_key", "mapped_flow", "session_id", "observation_kst_day"):
        if type(row.get(field)) is not str or not row[field].strip():
            raise ValueError("source-day mapping is invalid")
    if type(row.get("writer_epoch")) is not int or row["writer_epoch"] < 0:
        raise ValueError("source-day mapping epoch is invalid")
    expected_mapping_id = digest({
        "root_event_id": row["root_event_id"],
        "customer_key": row["customer_key"],
        "mapped_flow": row["mapped_flow"],
        "observation_kst_day": row["observation_kst_day"],
        "session_id": row["session_id"],
        "writer_epoch": row["writer_epoch"],
        "root_preimage_digest": row["root_preimage_digest"],
    })
    if row["mapping_id"] != expected_mapping_id:
        raise ValueError("source-day mapping digest mismatch")


def _validate_authority_rows(rows: Sequence[Mapping[str, object]]) -> None:
    required = {
        "owner",
        "customer_key",
        "registry_digest",
        "activation_receipt_digest",
        "consent_digest",
    }
    for row in rows:
        if not (
            {"owner", "activation_receipt_digest", "consent_digest"} & set(row)
        ):
            continue
        if not required.issubset(row):
            raise ValueError("authority mirror evidence is incomplete")
        _validate_full_authority(row["owner"], "authority mirror owner")
        for field in ("registry_digest", "activation_receipt_digest", "consent_digest"):
            _require_digest(row[field], f"authority mirror {field}")
        if type(row["customer_key"]) is not str or not row["customer_key"].strip():
            raise ValueError("authority mirror customer is invalid")


def _validate_journal_rows(rows: Sequence[Mapping[str, object]], kind: str) -> None:
    accepted_kinds = {
        kind,
        {
            "source-day": "source_day",
            "authority_mirror": "authority",
            "config-epoch": "config_epoch",
        }.get(kind, kind),
    }
    states: dict[str, str] = {}
    prepared_digests: dict[str, str] = {}
    for row in rows:
        if row.get("schema_version") != "1.0" or row.get("kind") not in accepted_kinds:
            raise ValueError("adaptive journal schema mismatch")
        intent_id = row.get("intent_id")
        state = row.get("state")
        row_digest = row.get("row_digest")
        if not isinstance(intent_id, str) or state not in {"prepared", "committed", "abandoned"}:
            raise ValueError("adaptive journal transition is invalid")
        if not isinstance(row_digest, str) or row_digest != digest({key: value for key, value in row.items() if key != "row_digest"}):
            raise ValueError("adaptive journal row digest mismatch")
        prior = states.get(intent_id)
        if prior is None:
            if state != "prepared":
                raise ValueError("adaptive journal terminal row has no prepared row")
            states[intent_id] = state
            prepared_digests[intent_id] = row_digest
        elif prior == "prepared":
            if state not in {"committed", "abandoned"}:
                raise ValueError("adaptive journal terminal transition is invalid")
            if row.get("prepared_digest") != prepared_digests[intent_id]:
                raise ValueError("adaptive journal prepared digest mismatch")
            states[intent_id] = state
        else:
            raise ValueError("adaptive journal has multiple terminal rows")
    normalized_kind = {
        "source-day": "source_day",
        "authority_mirror": "authority",
        "config-epoch": "config_epoch",
    }.get(kind, kind)
    if normalized_kind == "config_epoch":
        _validate_config_epoch_rows(rows)
    elif normalized_kind == "authority":
        _validate_authority_rows(rows)


def _validate_flat_journal_rows(rows: Sequence[Mapping[str, object]], label: str) -> None:
    for row in rows:
        if row.get("schema_version") != "1.0":
            raise ValueError(f"{label} journal schema mismatch")
        row_digest = row.get("row_digest")
        if not isinstance(row_digest, str) or row_digest != digest(
            {key: value for key, value in row.items() if key != "row_digest"}
        ):
            raise ValueError(f"{label} journal row digest mismatch")
        if label == "source-day":
            _validate_source_day_mapping_row(row)

def _canonical_event_record(event: object) -> dict[str, object]:
    model_dump = getattr(event, "model_dump", None)
    value = model_dump(mode="json", exclude_none=True) if callable(model_dump) else event
    if not isinstance(value, Mapping):
        raise ValueError("canonical event is not a mapping")
    row = dict(value)
    event_id = row.get("event_id")
    if not isinstance(event_id, str) or not event_id:
        raise ValueError("canonical event id is invalid")
    return row




class CanonicalSequenceJournal:
    """Read-only compatibility facade over the registered canonical transaction."""

    def __init__(
        self,
        events_path: Path,
        sequence_path: Path | None = None,
        *,
        lock_path: Path | None = None,
    ) -> None:
        if sequence_path is None:
            sequence_path = Path(events_path)
            events_path = sequence_path.with_name("events.jsonl")
        self._transaction = CanonicalEventTransaction(
            Path(events_path),
            Path(sequence_path),
            lock_path=lock_path,
        )
        self.events_path = self._transaction.events_path
        self.sequence_path = self._transaction.sequence_path
        self.lock_path = self._transaction.lock_path
    def read_snapshot(self) -> Any:
        """Return an immutable canonical snapshot without mutating storage."""
        return self._transaction.read_snapshot_readonly()

    @contextmanager
    def read_locked(self) -> Any:
        """Yield the canonical pair while retaining its shared readonly lock."""
        with self._transaction.read_locked() as snapshot:
            yield snapshot

    def validate(self) -> tuple[Mapping[str, object], ...]:
        return self._transaction.read_snapshot_readonly().sequence_rows





@dataclass(frozen=True, slots=True)
class AdaptiveOverlay:
    revision_id: str
    proposal_digest: str
    effective_from: str | date | datetime
    registration_digest: str | None = None
    effective_through: str | date | datetime | None = None
    supersedes_revision_id: str | None = None
    authority_snapshot_id: str = ""
    state: str = "effective"
    append_sequence: int = 0
    calorie_changing: bool = False
    calories_before: int | None = None
    calories_after: int | None = None
    committed_at_kst: str | date | datetime | None = None

    @property
    def digest(self) -> str:
        return digest(self)
    @property
    def revision_digest(self) -> str:
        return self.proposal_digest

    @property
    def overlay_digest(self) -> str:
        return self.digest

    @property
    def effective_from_kst(self) -> str:
        return _kst_timestamp(self.effective_from)

    @property
    def effective_through_kst(self) -> str | None:
        return _kst_timestamp(self.effective_through) if self.effective_through is not None else None


EffectiveOverlay = AdaptiveOverlay


def _kst_timestamp(value: str | date | datetime) -> str:
    if isinstance(value, date) and not isinstance(value, datetime):
        return f"{value.isoformat()}T00:00:00+09:00"
    if isinstance(value, datetime):
        parsed = value
    else:
        parsed = datetime.fromisoformat(value)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone(timedelta(hours=9)))
    else:
        parsed = parsed.astimezone(timezone(timedelta(hours=9)))
    return parsed.isoformat()


def _kst_compare(value: str | date | datetime) -> datetime:
    return datetime.fromisoformat(_kst_timestamp(value))


class OverlayJournal:
    """Append-only effective/superseded overlay resolver."""

    def __init__(self, path: Path, *, root: Path | None = None) -> None:
        self.path = Path(path)
        self.root = Path(root) if root is not None else self.path.parent
        self.lock_path = self.root / ".authority-transition.lock"
        self._active_overlay_lock_token: _OverlayLockToken | None = None

    _EVENT_TYPES = frozenset({
        "adaptive_overlay_effective",
        "adaptive_overlay_replaced",
        "adaptive_overlay_deactivated",
    })

    _PAYLOAD_FIELDS = {
        "adaptive_overlay_effective": (
            frozenset({"overlay"}),
            frozenset({"intent_id", "operator_id", "authority_snapshot_id"}),
        ),
        "adaptive_overlay_replaced": (
            frozenset({
                "old_revision_id",
                "new_revision_id",
                "effective_from",
                "new_digest",
                "overlay",
            }),
            frozenset({"intent_id", "operator_id", "authority_snapshot_id"}),
        ),
        "adaptive_overlay_deactivated": (
            frozenset({"revision_id", "valid_from", "reason", "restore_revision_id"}),
            frozenset({"intent_id", "operator_id"}),
        ),
    }

    @classmethod
    def _validate_row(cls, row: Mapping[str, object]) -> None:
        expected_fields = {
            "schema_version",
            "event_type",
            "append_sequence",
            "payload",
            "row_digest",
        }
        if set(row) != expected_fields:
            raise ValueError("overlay journal row schema mismatch")
        if row.get("schema_version") != "1.0":
            raise ValueError("overlay journal schema mismatch")
        event_type = row.get("event_type")
        if not isinstance(event_type, str) or event_type not in cls._EVENT_TYPES:
            raise ValueError("unknown overlay event type")
        sequence = row.get("append_sequence")
        if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence <= 0:
            raise ValueError("overlay journal sequence is invalid")
        if not isinstance(row.get("payload"), Mapping):
            raise ValueError("overlay payload is invalid")
        row_digest = row.get("row_digest")
        if not isinstance(row_digest, str) or row_digest != digest(
            {key: value for key, value in row.items() if key != "row_digest"}
        ):
            raise ValueError("overlay journal row digest mismatch")

    def _validate_rows(self, rows: Sequence[Mapping[str, object]]) -> None:
        for expected_sequence, row in enumerate(rows, start=1):
            self._validate_row(row)
            if row.get("append_sequence") != expected_sequence:
                raise ValueError("overlay journal sequence is not contiguous")
        self._replay_rows(rows)
    @contextmanager
    def _locked(self) -> Any:
        self._require_overlay_domain()
        if self._active_overlay_lock_token is not None:
            raise RuntimeError("overlay lock is already held")
        with _verified_private_lock(
            self.lock_path,
            exclusive=True,
            create=True,
        ) as descriptor:
            opened = os.fstat(descriptor)
            token = _OverlayLockToken((opened.st_dev, opened.st_ino))
            self._active_overlay_lock_token = token
            try:
                yield token
            finally:
                self._active_overlay_lock_token = None

    def _require_overlay_token(self, token: object) -> None:
        if (
            not isinstance(token, _OVERLAY_LOCK_TOKEN_TYPE)
            or token is not self._active_overlay_lock_token
        ):
            raise TypeError("overlay lock token is invalid")
        try:
            current = self.lock_path.lstat()
        except OSError as exc:
            raise ValueError("authority transition lock is unavailable") from exc
        if (
            (current.st_dev, current.st_ino) != token.identity
            or not stat.S_ISREG(current.st_mode)
            or current.st_uid != os.geteuid()
            or current.st_nlink != 1
            or stat.S_IMODE(current.st_mode) != 0o600
        ):
            raise ValueError("authority transition lock was replaced")

    @contextmanager
    def read_locked(self) -> Any:
        """Hold the overlay writer inode shared without mutation."""
        self._require_overlay_domain()
        with _verified_private_lock(
            self.lock_path,
            exclusive=False,
            create=False,
        ):
            yield
    def _require_overlay_domain(self) -> None:
        if os.path.abspath(self.path.parent) != os.path.abspath(self.root):
            raise ValueError("overlay journal path is outside its root")
        if self.root.is_symlink():
            raise ValueError("overlay journal root symlink is not allowed")
        if self.root.exists() and not self.root.is_dir():
            raise ValueError("overlay journal root is invalid")

    def _owned_recovery_path(self, path: Path | None) -> Path:
        self._require_overlay_domain()
        target = self.path if path is None else Path(path)
        if os.path.abspath(target) != os.path.abspath(self.path):
            raise ValueError("overlay recovery cannot mutate a foreign journal domain")
        return target

    def _ensure_overlay_parent(self) -> None:
        self._require_overlay_domain()
        parent = self.path.parent
        if parent.is_symlink():
            raise ValueError("overlay journal parent symlink is not allowed")
        parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        if parent.is_symlink() or not parent.is_dir():
            raise ValueError("overlay journal parent is invalid")
        parent.chmod(0o700)

    @staticmethod
    def _validate_overlay_file(descriptor: int, named: os.stat_result) -> tuple[int, int]:
        opened = os.fstat(descriptor)
        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 ValueError("overlay journal file is unsafe")
        return identity

    @contextmanager
    def _open_overlay_file(self, *, create: bool) -> Any:
        flags = os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC
        if create:
            self._ensure_overlay_parent()
            flags |= os.O_CREAT
        descriptor = os.open(self.path, flags, 0o600)
        try:
            named = self.path.lstat()
            identity = self._validate_overlay_file(descriptor, named)
            yield descriptor, identity
            current = self.path.lstat()
            opened = os.fstat(descriptor)
            if (
                (current.st_dev, current.st_ino) != identity
                or (opened.st_dev, opened.st_ino) != identity
            ):
                raise ValueError("overlay journal file was replaced")
        finally:
            os.close(descriptor)

    @staticmethod
    def _read_descriptor(descriptor: int) -> bytes:
        os.lseek(descriptor, 0, os.SEEK_SET)
        chunks: list[bytes] = []
        while True:
            chunk = os.read(descriptor, 1024 * 1024)
            if not chunk:
                return b"".join(chunks)
            chunks.append(chunk)

    def _read_overlay_rows_locked(
        self,
        token: object,
        *,
        allow_truncated_tail: bool,
    ) -> tuple[list[dict[str, object]], int, int, tuple[int, int] | None]:
        self._require_overlay_token(token)
        try:
            self.path.lstat()
        except FileNotFoundError:
            return [], 0, 0, None
        with self._open_overlay_file(create=False) as (descriptor, identity):
            data = self._read_descriptor(descriptor)
            opened = os.fstat(descriptor)
            current = self.path.lstat()
            if (
                (opened.st_dev, opened.st_ino) != identity
                or (current.st_dev, current.st_ino) != identity
                or opened.st_size != len(data)
            ):
                raise ValueError("overlay journal file changed during read")
            rows, complete_end = _parse_jsonl_bytes(
                data,
                self.path,
                allow_truncated_tail=allow_truncated_tail,
            )
            return rows, complete_end, opened.st_size, identity

    def recover(self, *, path: Path | None = None) -> Mapping[str, object]:
        """Recover a torn overlay tail under the authority transition lock."""
        target = self._owned_recovery_path(path)
        with self._locked() as token:
            return self._recover_locked(token, path=target)

    def _recover_locked(
        self,
        token: object,
        *,
        path: Path | None = None,
    ) -> Mapping[str, object]:
        """Recover the owned overlay journal while its lock token is held."""
        self._require_overlay_token(token)
        target = self._owned_recovery_path(path)
        if target.is_symlink():
            raise ValueError("overlay recovery journal symlink is not allowed")
        try:
            target.lstat()
        except FileNotFoundError:
            return {}
        rows, complete_end, current_size, identity = self._read_overlay_rows_locked(
            token,
            allow_truncated_tail=True,
        )
        self._validate_rows(rows)
        self._require_overlay_token(token)
        if identity is not None:
            current = target.lstat()
            if (current.st_dev, current.st_ino) != identity:
                raise ValueError("overlay journal file was replaced")
        if complete_end < current_size:
            self._require_overlay_token(token)
            with self._open_overlay_file(create=False) as (descriptor, identity):
                opened = os.fstat(descriptor)
                if (opened.st_dev, opened.st_ino) != identity or opened.st_size != current_size:
                    raise ValueError("overlay journal file changed during recovery")
                self._require_overlay_token(token)
                os.ftruncate(descriptor, complete_end)
                os.fsync(descriptor)
                self._require_overlay_token(token)
                os.fchmod(descriptor, 0o600)
                current = self.path.lstat()
                if (current.st_dev, current.st_ino) != identity:
                    raise ValueError("overlay journal file was replaced")
            return {str(target): "truncated_torn_tail"}
        return {str(target): len(rows)}

    def _append(
        self,
        event_type: str,
        payload: Mapping[str, object],
    ) -> Mapping[str, object]:
        self._ensure_overlay_parent()
        with self._locked() as token:
            return self._append_locked(token, event_type, payload)

    def _append_locked(
        self,
        token: object,
        event_type: str,
        payload: Mapping[str, object],
        *,
        rows: Sequence[Mapping[str, object]] | None = None,
    ) -> Mapping[str, object]:
        self._require_overlay_token(token)
        if event_type not in self._EVENT_TYPES:
            raise ValueError("unknown overlay event type")
        if not isinstance(payload, Mapping):
            raise ValueError("overlay payload is invalid")
        if rows is None:
            self._recover_locked(token, path=self.path)
            rows, complete_end, current_size, identity = self._read_overlay_rows_locked(
                token,
                allow_truncated_tail=False,
            )
            if identity is not None:
                current = self.path.lstat()
                if (current.st_dev, current.st_ino) != identity:
                    raise ValueError("overlay journal file was replaced")
            if complete_end != current_size:
                raise ValueError("overlay journal has a torn tail")
        self._validate_rows(rows)
        sequence = (rows[-1]["append_sequence"] + 1) if rows else 1
        body = {
            "schema_version": "1.0",
            "event_type": event_type,
            "append_sequence": sequence,
            "payload": _plain(dict(payload)),
        }
        row = {**body, "row_digest": digest(body)}
        self._validate_row(row)
        self._validate_rows([*rows, row])
        encoded = (canonical_json(row) + "\n").encode("utf-8")
        self._require_overlay_token(token)
        with self._open_overlay_file(create=True) as (descriptor, identity):
            self._require_overlay_token(token)
            os.lseek(descriptor, 0, os.SEEK_END)
            offset = 0
            while offset < len(encoded):
                offset += os.write(descriptor, encoded[offset:])
            self._require_overlay_token(token)
            os.fsync(descriptor)
            os.fchmod(descriptor, 0o600)
            current = self.path.lstat()
            opened = os.fstat(descriptor)
            if (
                (current.st_dev, current.st_ino) != identity
                or (opened.st_dev, opened.st_ino) != identity
            ):
                raise ValueError("overlay journal file was replaced")
        return row

    def _overlay_value(self, overlay: AdaptiveOverlay | Mapping[str, object]) -> AdaptiveOverlay:
        if isinstance(overlay, AdaptiveOverlay):
            candidate = overlay
        else:
            try:
                raw = dict(overlay)
            except (TypeError, ValueError) as exc:
                raise ValueError("overlay is invalid") from exc
            allowed_fields = {
                "revision_id",
                "proposal_digest",
                "digest",
                "registration_digest",
                "effective_from",
                "effective_through",
                "supersedes_revision_id",
                "supersedes",
                "authority_snapshot_id",
                "state",
                "append_sequence",
                "calorie_changing",
                "calories_before",
                "calories_after",
                "committed_at_kst",
            }
            if set(raw) - allowed_fields:
                raise ValueError("overlay schema mismatch")
            try:
                candidate = AdaptiveOverlay(
                    revision_id=raw["revision_id"],
                    proposal_digest=raw.get("proposal_digest", raw.get("digest", "")),
                    registration_digest=raw.get("registration_digest"),
                    effective_from=raw["effective_from"],
                    effective_through=raw.get("effective_through"),
                    supersedes_revision_id=raw.get("supersedes_revision_id", raw.get("supersedes")),
                    authority_snapshot_id=raw.get("authority_snapshot_id", ""),
                    state=raw.get("state", "effective"),
                    append_sequence=raw.get("append_sequence", 0),
                    calorie_changing=raw.get("calorie_changing", False),
                    calories_before=raw.get("calories_before"),
                    calories_after=raw.get("calories_after"),
                    committed_at_kst=raw.get("committed_at_kst"),
                )
            except (KeyError, TypeError, ValueError) as exc:
                raise ValueError("overlay is invalid") from exc
        if candidate.registration_digest is not None:
            _require_digest(candidate.registration_digest, "overlay registration")
        if type(candidate.calorie_changing) is not bool:
            raise ValueError("overlay calorie-change flag is invalid")
        for value in (candidate.calories_before, candidate.calories_after):
            if value is not None and (type(value) is not int or value < 0):
                raise ValueError("overlay calorie-change calories are invalid")
        if candidate.committed_at_kst is not None:
            try:
                _kst_timestamp(candidate.committed_at_kst)
            except (TypeError, ValueError) as exc:
                raise ValueError("overlay committed timestamp is invalid") from exc
        if (
            not isinstance(candidate.revision_id, str)
            or not candidate.revision_id.strip()
            or not isinstance(candidate.proposal_digest, str)
            or not candidate.proposal_digest.strip()
            or not isinstance(candidate.authority_snapshot_id, str)
            or not isinstance(candidate.state, str)
            or candidate.state not in {"effective", "superseded"}
            or isinstance(candidate.append_sequence, bool)
            or not isinstance(candidate.append_sequence, int)
            or candidate.append_sequence < 0
        ):
            raise ValueError("overlay is invalid")
        if candidate.supersedes_revision_id is not None and (
            not isinstance(candidate.supersedes_revision_id, str)
            or not candidate.supersedes_revision_id.strip()
        ):
            raise ValueError("overlay predecessor is invalid")
        try:
            effective_from = _kst_timestamp(candidate.effective_from)
            effective_through = (
                _kst_timestamp(candidate.effective_through)
                if candidate.effective_through is not None
                else None
            )
            if effective_through is not None and _kst_compare(effective_through) <= _kst_compare(effective_from):
                raise ValueError("overlay effective interval is invalid")
        except (TypeError, ValueError) as exc:
            raise ValueError("overlay effective interval is invalid") from exc
        return replace(
            candidate,
            effective_from=effective_from,
            effective_through=effective_through,
        )

    def _active_at_now(self, query: str | date | datetime) -> tuple[AdaptiveOverlay, ...]:
        return tuple(
            overlay
            for overlay in self._replay()
            if overlay.state == "effective"
            and _kst_compare(overlay.effective_from) <= _kst_compare(query)
            and (
                overlay.effective_through is None
                or _kst_compare(overlay.effective_through) > _kst_compare(query)
            )
        )

    def _replay_rows(
        self,
        rows: Sequence[Mapping[str, object]],
        *,
        as_of_sequence: int | None = None,
    ) -> tuple[AdaptiveOverlay, ...]:
        if as_of_sequence is not None and (
            isinstance(as_of_sequence, bool)
            or not isinstance(as_of_sequence, int)
            or as_of_sequence < 0
        ):
            raise ValueError("overlay as-of sequence is invalid")
        overlays: dict[str, list[AdaptiveOverlay]] = {}

        def overlaps(left: AdaptiveOverlay, right: AdaptiveOverlay) -> bool:
            left_end = (
                _kst_compare(left.effective_through)
                if left.effective_through is not None
                else datetime.max.replace(tzinfo=timezone.utc)
            )
            right_end = (
                _kst_compare(right.effective_through)
                if right.effective_through is not None
                else datetime.max.replace(tzinfo=timezone.utc)
            )
            return (
                _kst_compare(left.effective_from) < right_end
                and _kst_compare(right.effective_from) < left_end
            )

        def add_overlay(candidate: AdaptiveOverlay, sequence: int) -> None:
            if candidate.state != "effective":
                raise ValueError("overlay revision transition is invalid")
            if candidate.revision_id in overlays:
                raise ValueError("overlay revision transition is invalid")
            for revisions in overlays.values():
                current = revisions[-1]
                if current.state == "effective" and overlaps(current, candidate):
                    raise ValueError("overlapping effective overlays")
            overlays[candidate.revision_id] = [
                replace(candidate, append_sequence=sequence)
            ]

        for row in rows:
            sequence = row["append_sequence"]
            assert isinstance(sequence, int)
            if as_of_sequence is not None and sequence > as_of_sequence:
                continue
            event_type = row["event_type"]
            assert isinstance(event_type, str)
            payload = row["payload"]
            assert isinstance(payload, Mapping)
            required, optional = self._PAYLOAD_FIELDS[event_type]
            for key in set(payload) & optional:
                if not isinstance(payload[key], str) or len(payload[key]) > 256:
                    raise ValueError("overlay payload metadata is invalid")
            if event_type == "adaptive_overlay_effective":
                required, optional = self._PAYLOAD_FIELDS[event_type]
                if not required.issubset(payload) or set(payload) - required - optional:
                    raise ValueError("overlay activation payload is invalid")
                raw_overlay = payload["overlay"]
                if not isinstance(raw_overlay, Mapping):
                    raise ValueError("overlay payload is invalid")
                candidate = self._overlay_value(raw_overlay)
                if candidate.supersedes_revision_id is not None and (
                    candidate.supersedes_revision_id not in overlays
                ):
                    raise ValueError("overlay predecessor is missing")
                add_overlay(candidate, sequence)
                continue
            if event_type == "adaptive_overlay_replaced":
                required, optional = self._PAYLOAD_FIELDS[event_type]
                if not required.issubset(payload) or set(payload) - required - optional:
                    raise ValueError("overlay replacement payload is invalid")
                old_id = payload["old_revision_id"]
                new_id = payload["new_revision_id"]
                if not isinstance(old_id, str) or not old_id:
                    raise ValueError("overlay replacement parent is missing")
                if not isinstance(new_id, str) or not new_id:
                    raise ValueError("overlay replacement revision is invalid")
                if old_id not in overlays:
                    raise ValueError("overlay replacement parent is missing")
                raw_overlay = payload["overlay"]
                if not isinstance(raw_overlay, Mapping):
                    raise ValueError("overlay payload is invalid")
                candidate = self._overlay_value(raw_overlay)
                if candidate.revision_id != new_id:
                    raise ValueError("overlay replacement revision mismatch")
                if candidate.revision_id == old_id or candidate.revision_id in overlays:
                    raise ValueError("overlay revision transition is invalid")
                if candidate.supersedes_revision_id not in {None, old_id}:
                    raise ValueError("overlay replacement parent mismatch")
                if payload["effective_from"] != candidate.effective_from_kst:
                    raise ValueError("overlay replacement timestamp mismatch")
                if payload["new_digest"] != candidate.digest:
                    raise ValueError("overlay replacement digest mismatch")
                old = overlays[old_id][-1]
                if _kst_compare(candidate.effective_from) <= _kst_compare(old.effective_from):
                    raise ValueError("overlay revision transition is invalid")
                old_end = _kst_compare(candidate.effective_from)
                if old.effective_through is not None:
                    old_end = min(old_end, _kst_compare(old.effective_through))
                if old_end < _kst_compare(old.effective_from):
                    raise ValueError("overlay revision transition is invalid")
                overlays[old_id][-1] = replace(old, effective_through=old_end)
                add_overlay(candidate, sequence)
                continue
            if event_type == "adaptive_overlay_deactivated":
                required, optional = self._PAYLOAD_FIELDS[event_type]
                if not required.issubset(payload) or set(payload) - required - optional:
                    raise ValueError("overlay deactivation payload is invalid")
                revision_id = payload["revision_id"]
                if not isinstance(revision_id, str) or not revision_id:
                    raise ValueError("overlay rollback target is missing")
                if revision_id not in overlays:
                    raise ValueError("overlay rollback target is missing")
                reason = payload["reason"]
                if not isinstance(reason, str) or len(reason) > 256:
                    raise ValueError("overlay deactivation reason is invalid")
                current = overlays[revision_id][-1]
                if current.state != "effective":
                    raise ValueError("overlay revision transition is invalid")
                try:
                    valid_from = _kst_timestamp(payload["valid_from"])
                except (TypeError, ValueError) as exc:
                    raise ValueError("overlay deactivation timestamp is invalid") from exc
                valid_point = _kst_compare(valid_from)
                effective_point = _kst_compare(current.effective_from)
                abort_same_point = (
                    valid_point == effective_point and reason.startswith("abort:")
                )
                if valid_point < effective_point or (
                    valid_point == effective_point and not abort_same_point
                ):
                    raise ValueError("overlay revision transition is invalid")
                if (
                    current.effective_through is not None
                    and valid_point > _kst_compare(current.effective_through)
                ):
                    raise ValueError("overlay rollback target is not active")
                rollback_end = valid_point
                if current.effective_through is not None:
                    rollback_end = min(rollback_end, _kst_compare(current.effective_through))
                if rollback_end < effective_point:
                    raise ValueError("overlay revision transition is invalid")
                overlays[revision_id][-1] = replace(
                    current,
                    effective_through=rollback_end,
                    state="effective",
                )
                restore_id = payload["restore_revision_id"]
                if restore_id is None:
                    continue
                if not isinstance(restore_id, str) or not restore_id:
                    raise ValueError("overlay predecessor is invalid")
                if restore_id not in overlays or restore_id == revision_id:
                    raise ValueError("overlay predecessor is missing")
                restore = overlays[restore_id][-1]
                restored = replace(
                    restore,
                    effective_from=valid_from,
                    effective_through=None,
                    state="effective",
                    append_sequence=sequence,
                )
                if any(
                    existing.state == "effective"
                    and existing.revision_id != restore_id
                    and overlaps(existing, restored)
                    for revisions in overlays.values()
                    for existing in [revisions[-1]]
                ):
                    raise ValueError("overlapping effective overlays")
                overlays[restore_id].append(restored)
                continue
            raise ValueError("unknown overlay event type")
        values = [
            overlay
            for revisions in overlays.values()
            for overlay in revisions
        ]
        return tuple(sorted(values, key=lambda value: (value.append_sequence, value.revision_id)))

    def _replay(self, *, as_of_sequence: int | None = None) -> tuple[AdaptiveOverlay, ...]:
        if not self.path.exists():
            return ()
        rows = _read_jsonl_rows(self.path)
        self._validate_rows(rows)
        return self._replay_rows(rows, as_of_sequence=as_of_sequence)

    def append(self, overlay: AdaptiveOverlay | Mapping[str, object]) -> Mapping[str, object]:
        candidate = self._overlay_value(overlay)
        if candidate.state != "effective":
            raise ValueError("overlay revision transition is invalid")
        with self._locked() as token:
            self._recover_locked(token, path=self.path)
            rows, complete_end, current_size, identity = self._read_overlay_rows_locked(
                token,
                allow_truncated_tail=False,
            )
            if identity is not None:
                current_path = self.path.lstat()
                if (current_path.st_dev, current_path.st_ino) != identity:
                    raise ValueError("overlay journal file was replaced")
            if complete_end != current_size:
                raise ValueError("overlay journal has a torn tail")
            self._validate_rows(rows)
            current = self._replay_rows(rows)
            if any(item.revision_id == candidate.revision_id for item in current):
                raise ValueError("overlay revision transition is invalid")
            active = tuple(
                item
                for item in current
                if item.state == "effective"
                and _kst_compare(item.effective_from) <= _kst_compare(candidate.effective_from)
                and (
                    item.effective_through is None
                    or _kst_compare(item.effective_through) > _kst_compare(candidate.effective_from)
                )
            )
            if any(existing.revision_id != candidate.revision_id for existing in active):
                raise ValueError("overlapping effective overlays")
            return self._append_locked(
                token,
                "adaptive_overlay_effective",
                {"overlay": asdict(candidate)},
                rows=rows,
            )
    append_effective = append

    def replace(
        self,
        old_revision_id: str | None,
        overlay: AdaptiveOverlay | Mapping[str, object],
    ) -> Mapping[str, object]:
        candidate = self._overlay_value(overlay)
        if not isinstance(old_revision_id, str) or not old_revision_id:
            raise ValueError("overlay replacement parent is missing")
        current = self._replay()
        parent = max(
            (item for item in current if item.revision_id == old_revision_id),
            key=lambda item: item.append_sequence,
            default=None,
        )
        if parent is None:
            raise ValueError("overlay replacement parent is missing")
        if any(item.revision_id == candidate.revision_id for item in current):
            raise ValueError("overlay revision transition is invalid")
        if candidate.supersedes_revision_id not in {None, old_revision_id}:
            raise ValueError("overlay replacement parent mismatch")
        if _kst_compare(candidate.effective_from) <= _kst_compare(parent.effective_from):
            raise ValueError("overlay revision transition is invalid")
        payload = {
            "old_revision_id": old_revision_id,
            "new_revision_id": candidate.revision_id,
            "effective_from": candidate.effective_from_kst,
            "new_digest": candidate.digest,
            "overlay": asdict(candidate),
        }
        return self._append("adaptive_overlay_replaced", payload)
    append_replacement = replace

    def rollback(
        self,
        revision_id: str,
        *,
        as_of_kst: str | date | datetime,
        reason: str,
    ) -> Mapping[str, object]:
        current = self._replay()
        candidate = max(
            (item for item in current if item.revision_id == revision_id),
            key=lambda item: item.append_sequence,
            default=None,
        )
        if candidate is None:
            raise ValueError("overlay rollback target is missing")
        if candidate.state != "effective":
            raise ValueError("overlay revision transition is invalid")
        valid_from = _kst_timestamp(as_of_kst)
        if not isinstance(reason, str) or len(reason) > 256:
            raise ValueError("overlay deactivation reason is invalid")
        valid_point = _kst_compare(valid_from)
        effective_point = _kst_compare(candidate.effective_from)
        if valid_point < effective_point or (
            valid_point == effective_point and not reason.startswith("abort:")
        ):
            raise ValueError("overlay revision transition is invalid")
        return self._append(
            "adaptive_overlay_deactivated",
            {
                "revision_id": revision_id,
                "valid_from": valid_from,
                "reason": reason,
                "restore_revision_id": candidate.supersedes_revision_id,
            },
        )
    append_rollback = rollback

    def resolve(
        self,
        *,
        effective_kst: str | date | datetime,
        as_of_sequence: int | None = None,
    ) -> AdaptiveOverlay | None:
        query = _kst_compare(effective_kst)
        candidates = [
            overlay
            for overlay in self._replay(as_of_sequence=as_of_sequence)
            if overlay.state == "effective"
            and _kst_compare(overlay.effective_from) <= query
            and (
                overlay.effective_through is None
                or _kst_compare(overlay.effective_through) > query
            )
        ]
        if not candidates:
            return None
        candidates.sort(key=lambda item: (_kst_compare(item.effective_from), item.append_sequence, item.revision_id))
        return candidates[-1]
    resolve_as_of = resolve
    effective_as_of = resolve


TemporalOverlayStore = OverlayJournal
SourceDayJournal = AdaptiveEventStore
AuthorityMirrorJournal = AdaptiveEventStore
ConfigEpochJournal = AdaptiveEventStore


def append_effective_overlay(
    path: Path,
    overlay: AdaptiveOverlay | Mapping[str, object],
) -> Mapping[str, object]:
    return OverlayJournal(Path(path)).append(overlay)


def replace_effective_overlay(
    path: Path,
    old_revision_id: str | None,
    overlay: AdaptiveOverlay | Mapping[str, object],
) -> Mapping[str, object]:
    return OverlayJournal(Path(path)).replace(old_revision_id, overlay)


def resolve_effective_overlay(
    path: Path,
    *,
    effective_kst: str | date | datetime,
    as_of_sequence: int | None = None,
) -> AdaptiveOverlay | None:
    return OverlayJournal(Path(path)).resolve(
        effective_kst=effective_kst,
        as_of_sequence=as_of_sequence,
    )


def rollback_effective_overlay(
    path: Path,
    revision_id: str,
    *,
    as_of_kst: str | date | datetime,
    reason: str = "operator_rollback",
) -> Mapping[str, object]:
    return OverlayJournal(Path(path)).rollback(
        revision_id,
        as_of_kst=as_of_kst,
        reason=reason,
    )
def _iso_day(value: date | datetime | str) -> str:
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    return date.fromisoformat(str(value)[:10]).isoformat()