"""Validated customer registry and twelve-week nutrition plan boundary.""" from __future__ import annotations import hashlib import json import os from dataclasses import dataclass, field from datetime import date, time from pathlib import Path from collections.abc import Mapping from typing import Annotated, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator from .weekly_operations_domain_trainer import CustomerTrainingScheduleEntry, TrainerAssignment class CustomerRegistryError(ValueError): pass class DualCoachCoordinatorError(ValueError): """Raised when a registered-customer dual-coach transition is unsafe.""" @dataclass(frozen=True, slots=True) class ScheduleConfirmRequest: """Typed gateway input for one pinned schedule confirmation.""" customer_key: str event: object @dataclass(frozen=True, slots=True) class ScheduleConfirmReceipt: canonical_event: Mapping[str, object] canonical_sequence: Mapping[str, object] adaptive_projection: Mapping[str, object] @dataclass(frozen=True, slots=True) class CustomerDailyProjection: """Narrow, customer-safe daily source; it has no delivery authority.""" today_state: str prior_comparison: str judgement: str reason: str actions: tuple[str, ...] next_check: str def __post_init__(self) -> None: if not self.today_state or not self.prior_comparison or not self.judgement or not self.reason: raise ValueError("daily customer projection is incomplete") if not 1 <= len(self.actions) <= 3 or any(not item.strip() for item in self.actions): raise ValueError("daily customer actions are invalid") if not self.next_check: raise ValueError("daily customer next check is required") def render(self) -> str: return "\n".join(( f"오늘 상태: {self.today_state}", f"이전 흐름과 비교: {self.prior_comparison}", f"오늘 판단: {self.judgement}", f"판단 이유: {self.reason}", "오늘 할 일: " + " / ".join(self.actions), f"다음 확인: {self.next_check}", )) def build_registered_daily_customer_projection( runtime: CustomerRuntime, proposal: object, *, actions: tuple[object, ...], next_check: str, ) -> CustomerDailyProjection: """Build the gateway's read-only daily source from an approved proposal.""" from checkin_cli.adaptive_nutrition import ( CustomerActionContinuity, Decision, NutritionProposal, validate_explanation, ) if not isinstance(runtime, CustomerRuntime) or not isinstance(runtime.binding, RegisteredCustomerBinding): raise CustomerRegistryError("daily customer projection requires a registered customer") if not isinstance(proposal, NutritionProposal) or proposal.customer_key != runtime.spec.customer_key: raise CustomerRegistryError("daily customer projection customer mismatch") if not actions or len(actions) > 3: raise CustomerRegistryError("daily customer actions must contain one to three items") selected = tuple(actions) if any(not isinstance(item, CustomerActionContinuity) for item in selected): raise CustomerRegistryError("daily customer actions require approved continuity records") if any( item.customer_key != proposal.customer_key or item.approved_proposal_digest != proposal.digest or item.revision != proposal.revision for item in selected ): raise CustomerRegistryError("daily customer actions do not match the approved proposal") if not isinstance(next_check, str) or not next_check.strip() or len(next_check) > 160: raise CustomerRegistryError("daily customer next check is invalid") snapshot = proposal.snapshot today_state = ( "최근 기록을 바탕으로 오늘 계획을 확인했습니다." if snapshot.current_samples else "오늘 기록을 받으면 계획을 더 정확히 확인할 수 있습니다." ) if snapshot.weekly_rate_percent is None: comparison = "이전 흐름과 비교할 기록이 아직 충분하지 않습니다." else: comparison = f"최근 변화 흐름은 주간 {snapshot.weekly_rate_percent}%입니다." judgement = ( "오늘은 현재 계획을 유지합니다." if proposal.decision is Decision.MAINTAIN else "오늘 계획을 확인하고 필요한 조정을 운영자가 검토합니다." ) reason = ( "최근 흐름이 현재 계획을 바꿀 근거를 만들지 않아 유지가 적절합니다." if proposal.decision is Decision.MAINTAIN else validate_explanation(proposal.explanation, proposal) ) return CustomerDailyProjection( today_state, comparison, judgement, reason, tuple(item.action_text for item in selected), next_check, ) class RegisteredCustomerDualCoachCoordinator: """The only registered-customer authority for dual-coach confirmation. Canonical authority is committed first under its lock. The projection is a separately durable, idempotent reconciliation step: disagreement raises and never causes a second canonical mutation. """ def __init__(self, runtime: CustomerRuntime) -> None: if not isinstance(runtime, CustomerRuntime): raise TypeError("dual-coach coordinator requires CustomerRuntime") if not isinstance(runtime.binding, RegisteredCustomerBinding): raise DualCoachCoordinatorError("registered customer runtime has no sealed binding") self.runtime = runtime from checkin_cli.adaptive_nutrition import AdaptiveEventStore self.canonical_transaction = self._canonical_transaction() self.adaptive_store = AdaptiveEventStore( runtime.nutrition_plans_root / "events.jsonl", canonical_transaction=self.canonical_transaction, root=runtime.nutrition_plans_root, ) self.reconcile_schedule_reference() self.reconcile_pending() def _canonical_transaction(self) -> object: from checkin_cli.store import CanonicalEventTransaction return CanonicalEventTransaction.for_customer_runtime(self.runtime) def current_reference(self, customer_key: str) -> object | None: self._require_customer(customer_key) return self.canonical_transaction.current_schedule_reference(customer_key) @property def _reference_pending_path(self) -> Path: return self.runtime.nutrition_plans_root / "schedule-reference-pending.json" def _strategy_pins(self) -> tuple[object, int, str]: from checkin_cli.adaptive_nutrition import ( feature_config_digest, load_verified_dual_coach_risk_policy, ) policy = load_verified_dual_coach_risk_policy(self.runtime) feature_path = self.runtime.nutrition_plans_root / "feature-epoch.json" try: document = json.loads(feature_path.read_text(encoding="utf-8")) epoch = document["epoch"] config_digest = document["config_digest"] except (KeyError, OSError, TypeError, ValueError) as exc: raise DualCoachCoordinatorError("adaptive feature epoch is unavailable") from exc flags = { name: document.get(name) for name in ("analytics_shadow", "operator_candidates", "activation", "delivery") } if ( type(epoch) is not int or epoch < 0 or not isinstance(config_digest, str) or config_digest != feature_config_digest(epoch, flags) ): raise DualCoachCoordinatorError("adaptive feature epoch is invalid") return policy, epoch, self.runtime.registered_binding.binding_digest def _unavailable_risk_context(self) -> tuple[str, int]: """Pin held reviews to live strategy and epoch without reading risk policy.""" from checkin_cli.adaptive_nutrition import feature_config_digest, digest feature_path = self.runtime.nutrition_plans_root / "feature-epoch.json" try: document = json.loads(feature_path.read_text(encoding="utf-8")) epoch = document["epoch"] config_digest = document["config_digest"] flags = { name: document.get(name) for name in ("analytics_shadow", "operator_candidates", "activation", "delivery") } if ( type(epoch) is not int or epoch < 0 or not isinstance(config_digest, str) or config_digest != feature_config_digest(epoch, flags) ): raise ValueError("feature epoch is invalid") except (KeyError, OSError, TypeError, ValueError): epoch = 0 strategy = next( ( row for row in reversed(self.adaptive_store.read()) if row.get("event_type") in {"schedule_strategy_baseline", "schedule_strategy_confirmed"} and isinstance(row.get("payload"), Mapping) and row["payload"].get("customer_key") == self.runtime.spec.customer_key ), None, ) if strategy is None: source_strategy_digest = digest({ "registered_binding": self.runtime.registered_binding.binding_digest, "strategy_state": "unprojected", }) else: source_strategy_digest = digest({ "strategy_event_id": strategy.get("event_id"), "strategy_payload": strategy["payload"], }) return source_strategy_digest, epoch def stage_schedule_reference(self, event: object, customer_key: str) -> None: """Durably reserve baseline reconciliation before the canonical append.""" from checkin_cli.models import Event, EventType self._require_customer(customer_key) if ( not isinstance(event, Event) or event.event_type not in {EventType.SCHEDULE_REFERENCE, EventType.SCHEDULE_CORRECTION} or event.schedule_reference is None ): raise DualCoachCoordinatorError("dual-coach schedule reference event is required") policy, epoch, parent_digest = self._strategy_pins() payload = { "customer_key": customer_key, "event": event.model_dump(mode="json", exclude_none=True), "event_digest": self._event_digest(event), "policy_version": policy.version, "policy_digest": policy.policy_digest, "policy_document_digest": policy.document_digest, "epoch": epoch, "parent_digest": parent_digest, } path = self._reference_pending_path path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") try: with temporary.open("x", encoding="utf-8") as handle: handle.write(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) handle.flush() os.fsync(handle.fileno()) temporary.chmod(0o600) os.replace(temporary, path) path.chmod(0o600) finally: if temporary.exists(): temporary.unlink() def abandon_staged_schedule_reference(self, event_id: str) -> None: path = self._reference_pending_path if not path.exists(): return try: pending = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise DualCoachCoordinatorError("schedule reference recovery state is invalid") from exc if pending.get("event", {}).get("event_id") != event_id: raise DualCoachCoordinatorError("another schedule reference requires recovery") path.unlink() def reconcile_schedule_reference(self) -> Mapping[str, object] | None: """Append the restrictive baseline once after its canonical source exists.""" path = self._reference_pending_path if not path.exists(): return None try: pending = json.loads(path.read_text(encoding="utf-8")) from checkin_cli.models import Event event = Event.model_validate(pending["event"]) customer_key = pending["customer_key"] except (KeyError, OSError, TypeError, ValueError) as exc: raise DualCoachCoordinatorError("schedule reference recovery state is invalid") from exc if customer_key != self.runtime.spec.customer_key or pending.get("event_digest") != self._event_digest(event): raise DualCoachCoordinatorError("schedule reference recovery state mismatches customer authority") snapshot = self.canonical_transaction.read_snapshot() canonical = next((item for item in snapshot.events if item.event_id == event.event_id), None) if canonical is None: return None if ( canonical.model_dump(mode="json", exclude_none=True) != event.model_dump(mode="json", exclude_none=True) ): raise DualCoachCoordinatorError( "schedule reference recovery conflicts with canonical event" ) policy, epoch, parent_digest = self._strategy_pins() if ( pending.get("policy_version") != policy.version or pending.get("policy_digest") != policy.policy_digest or pending.get("policy_document_digest") != policy.document_digest or pending.get("epoch") != epoch or pending.get("parent_digest") != parent_digest ): raise DualCoachCoordinatorError("schedule reference recovery pins mismatch authority") reference_digest = self.canonical_transaction.schedule_reference_digest(canonical) baseline = self.adaptive_store.project_schedule_baseline( customer_key=customer_key, source_reference_id=canonical.event_id, source_reference_digest=reference_digest, policy_version=policy.version, policy_digest=policy.policy_digest, policy_document_digest=policy.document_digest, epoch=epoch, parent_digest=parent_digest, ) path.unlink() return baseline @property def _pending_path(self) -> Path: return self.runtime.nutrition_plans_root / "schedule-confirmation-pending.json" def _write_pending(self, event: object, customer_key: str, policy: object) -> None: from checkin_cli.models import Event if not isinstance(event, Event): raise TypeError("pending confirmation event must be typed") payload = { "customer_key": customer_key, "event": event.model_dump(mode="json", exclude_none=True), "event_digest": self._event_digest(event), "policy_version": policy.version, "policy_digest": policy.policy_digest, "policy_document_digest": policy.document_digest, } path = self._pending_path path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") try: with temporary.open("x", encoding="utf-8") as handle: handle.write(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) handle.flush() os.fsync(handle.fileno()) temporary.chmod(0o600) os.replace(temporary, path) path.chmod(0o600) finally: if temporary.exists(): temporary.unlink() def _clear_pending(self, event_id: str) -> None: path = self._pending_path if not path.exists(): return try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise DualCoachCoordinatorError("schedule confirmation recovery state is invalid") from exc if payload.get("event", {}).get("event_id") != event_id: raise DualCoachCoordinatorError("another schedule confirmation requires recovery") path.unlink() def reconcile_pending(self) -> ScheduleConfirmReceipt | None: """Finish the sole durable post-canonical projection, or stop on mismatch.""" path = self._pending_path if not path.exists(): return None try: pending = json.loads(path.read_text(encoding="utf-8")) customer_key = pending["customer_key"] from checkin_cli.models import Event event = Event.model_validate(pending["event"]) except (KeyError, OSError, ValueError, TypeError) as exc: raise DualCoachCoordinatorError("schedule confirmation recovery state is invalid") from exc if customer_key != self.runtime.spec.customer_key or pending.get("event_digest") != self._event_digest(event): raise DualCoachCoordinatorError("schedule confirmation recovery state mismatches customer authority") snapshot = self.canonical_transaction.read_snapshot() canonical = next((item for item in snapshot.events if item.event_id == event.event_id), None) if canonical is None: return None if canonical.model_dump(exclude_none=True) != event.model_dump(exclude_none=True): raise DualCoachCoordinatorError("canonical confirmation mismatches recovery state") from checkin_cli.adaptive_nutrition import load_verified_dual_coach_risk_policy policy = load_verified_dual_coach_risk_policy(self.runtime) if ( pending.get("policy_version") != policy.version or pending.get("policy_digest") != policy.policy_digest or pending.get("policy_document_digest") != policy.document_digest ): raise DualCoachCoordinatorError("schedule confirmation recovery policy mismatches authority") projection = self._project_confirmation(event, customer_key, policy) sequence = next(row for row in snapshot.sequence_rows if row["event_id"] == event.event_id) self._clear_pending(event.event_id) return ScheduleConfirmReceipt( canonical_event=canonical.model_dump(mode="json", exclude_none=True), canonical_sequence=sequence, adaptive_projection=projection, ) def _project_confirmation(self, event: object, customer_key: str, policy: object) -> Mapping[str, object]: from checkin_cli.models import Event if not isinstance(event, Event): raise TypeError("dual-coach confirmation event must be typed") payload = event.schedule_confirmation if payload is None: raise DualCoachCoordinatorError("dual-coach confirmation payload is missing") reference = self.current_reference(customer_key) if reference is None or reference.event_id != payload.reference_event_id: raise DualCoachCoordinatorError("dual-coach confirmation reference is not current") reference_digest = self.canonical_transaction.schedule_reference_digest(reference) if reference_digest != payload.reference_digest: raise DualCoachCoordinatorError("dual-coach confirmation reference digest is invalid") pinned_policy, epoch, _ = self._strategy_pins() if ( pinned_policy.version != policy.version or pinned_policy.policy_digest != policy.policy_digest or pinned_policy.document_digest != policy.document_digest ): raise DualCoachCoordinatorError("dual-coach confirmation policy mismatches authority") baseline_rows = [ row for row in self.adaptive_store.read() if row["event_type"] == "schedule_strategy_baseline" and row["payload"].get("source_reference_id") == reference.event_id and row["payload"].get("source_reference_digest") == reference_digest ] if len(baseline_rows) != 1: raise DualCoachCoordinatorError("dual-coach confirmation baseline is unavailable") from checkin_cli.customer_admin import load_approved_adaptive_registration_inputs registration = load_approved_adaptive_registration_inputs( self.runtime.customer_root.parents[2], customer_key ) start = reference.schedule_reference.session_kst_date training_days = { item.date for item in registration.training_schedule if item.load_category != "rest" } categories = tuple( "training" if start.fromordinal(start.toordinal() + offset) in training_days else "rest" for offset in range(7) ) mapping = self.adaptive_store.append_source_day_mapping( root_event_id=reference.event_id, customer_key=customer_key, mapped_flow="schedule_confirmation", observation_kst_day=start, session_id=event.event_id, writer_epoch=epoch, root_preimage_digest=reference_digest, ) return self.adaptive_store.project_confirmed_schedule_strategy( customer_key=customer_key, source_reference_id=reference.event_id, source_reference_digest=reference_digest, confirmation_id=event.event_id, source_day_mapping_digest=mapping["row_digest"], policy_version=policy.version, policy_digest=policy.policy_digest, policy_document_digest=policy.document_digest, epoch=epoch, parent_digest=baseline_rows[0]["event_id"], categories=categories, last_change_note=reference.schedule_reference.last_change_note, ) def record_terminal_morning_risk(self, event: object, answers: Mapping[str, object]) -> Mapping[str, object]: """Verify policy custody before evaluating the six terminal inputs.""" from decimal import Decimal from checkin_cli.adaptive_nutrition import ( DualCoachRiskEvidence, _evaluate_dual_coach_risk, held_dual_coach_risk_candidate, load_verified_dual_coach_risk_policy, ) from checkin_cli.models import Event from checkin_cli.store import terminal_morning_root_kst_day if not isinstance(event, Event): raise DualCoachCoordinatorError("terminal morning event is required") snapshot = self.canonical_transaction.read_snapshot() evaluation_kst_day = terminal_morning_root_kst_day(snapshot.events, event) if evaluation_kst_day is None: raise DualCoachCoordinatorError("terminal morning event is required") customer_key = self.runtime.spec.customer_key terminal_checkin_digest = self._event_digest(event) try: policy = load_verified_dual_coach_risk_policy(self.runtime) except (OSError, TypeError, ValueError): source_strategy_digest, epoch = self._unavailable_risk_context() candidate = held_dual_coach_risk_candidate( customer_key=customer_key, evaluation_kst_day=evaluation_kst_day, terminal_checkin_id=event.event_id, terminal_checkin_digest=terminal_checkin_digest, source_strategy_digest=source_strategy_digest, epoch=epoch, ) preimage = { "customer_key": candidate.customer_key, "evaluation_kst_day": candidate.evaluation_kst_day.isoformat(), "reason_code": candidate.reason_code, "policy_version": candidate.policy_version, "policy_digest": candidate.policy_digest, "policy_document_digest": candidate.policy_document_digest, "terminal_checkin_id": candidate.terminal_checkin_id, "terminal_checkin_digest": candidate.terminal_checkin_digest, "source_strategy_digest": candidate.source_strategy_digest, "epoch": candidate.epoch, } return self.adaptive_store.append( "dual_coach_risk_review", { **preimage, "candidate_preimage": preimage, "score": None, "reasons": [candidate.reason_code], "held": True, }, dedupe_key=candidate.dedupe_key, ) try: evidence = DualCoachRiskEvidence( weight_change_percent=Decimal(str(answers["weight_change_percent"])), sleep_hours=Decimal(str(answers["sleep_duration"])), fatigue=str(answers["fatigue"]), pain=str(answers["pain"]), exercise_feasibility=str(answers["exercise_feasibility"]), meal_deviation=str(answers["meal_deviation"]), ) evaluation = _evaluate_dual_coach_risk(policy, evidence) except (ArithmeticError, KeyError, ValueError) as exc: raise DualCoachCoordinatorError("terminal morning risk evidence is invalid") from exc payload = { "customer_key": customer_key, "terminal_checkin_id": event.event_id, "terminal_checkin_digest": terminal_checkin_digest, "evaluation_kst_day": evaluation_kst_day.isoformat(), "evidence": { "weight_change_percent": str(evidence.weight_change_percent), "sleep_hours": str(evidence.sleep_hours), "fatigue": evidence.fatigue, "pain": evidence.pain, "exercise_feasibility": evidence.exercise_feasibility, "meal_deviation": evidence.meal_deviation, }, "policy_version": evaluation.policy_version, "policy_digest": evaluation.policy_digest, "policy_document_digest": evaluation.policy_document_digest, "score": evaluation.score, "reasons": list(evaluation.reasons), "held": evaluation.held, } return self.adaptive_store.append( "dual_coach_risk_review", payload, dedupe_key=f"dual-coach-risk:{event.event_id}" ) def confirm(self, request: ScheduleConfirmRequest) -> ScheduleConfirmReceipt: if not isinstance(request, ScheduleConfirmRequest): raise TypeError("dual-coach confirmation requires ScheduleConfirmRequest") self._require_customer(request.customer_key) from checkin_cli.models import Event, EventType from checkin_cli.adaptive_nutrition import load_verified_dual_coach_risk_policy if not isinstance(request.event, Event): raise TypeError("dual-coach confirmation event must be typed") event = request.event if event.event_type is not EventType.SCHEDULE_CONFIRMATION: raise DualCoachCoordinatorError("dual-coach confirmation event is required") # Revalidate policy before the canonical mutation. A durable intent is # written first so a process failure after canonical append is recoverable. policy = load_verified_dual_coach_risk_policy(self.runtime) payload = event.schedule_confirmation reference = self.current_reference(request.customer_key) if ( payload is None or reference is None or reference.event_id != payload.reference_event_id or self.canonical_transaction.schedule_reference_digest(reference) != payload.reference_digest ): raise DualCoachCoordinatorError("dual-coach confirmation reference is not current") self.stage_schedule_reference(reference, request.customer_key) self.reconcile_schedule_reference() self._write_pending(event, request.customer_key, policy) try: canonical = self.canonical_transaction.append_schedule_confirmation( event, customer_key=request.customer_key ) projection = self._project_confirmation(event, request.customer_key, policy) except ValueError as exc: raise DualCoachCoordinatorError( "canonical confirmation requires adaptive reconciliation" ) from exc self._clear_pending(event.event_id) return ScheduleConfirmReceipt( canonical_event=canonical["canonical_event"], canonical_sequence=canonical["sequence"], adaptive_projection=projection, ) schedule_confirm_handler = confirm def reserve_missing_checkin_reminder( self, missing_window: date, destination: object, *, registry_digest: str, config_digest: str, operator_approval: str, canonical_sequence: int | None = None, canonical_digest: str | None = None, ) -> object: """Reserve the static reminder only after current policy re-verification.""" from checkin_cli.adaptive_nutrition import load_verified_dual_coach_risk_policy from checkin_cli.customer_schedule import reserve_missing_checkin_reminder # The policy loader also revalidates current registration and activation # pins. Its result is deliberately not cached across lifecycle writes. load_verified_dual_coach_risk_policy(self.runtime) return reserve_missing_checkin_reminder( self.runtime.customer_root.parents[2], self.runtime.spec.customer_key, missing_window, destination, registry_digest=registry_digest, config_digest=config_digest, operator_approval=operator_approval, canonical_sequence=canonical_sequence, canonical_digest=canonical_digest, ) def reminder_review_candidate( self, reminder: object, **kwargs: object ) -> object | None: """Persist a deduped non-response review signal; this never sends.""" from checkin_cli.adaptive_nutrition import load_verified_dual_coach_risk_policy from checkin_cli.weekly_operations_schedule_host_review_r4 import ( non_response_review_candidate, ) candidate = non_response_review_candidate(reminder, **kwargs) if candidate is None: return None policy = load_verified_dual_coach_risk_policy(self.runtime) return self.adaptive_store.append( "missing_checkin_reminder_review", { "customer_key": candidate.customer_key, "missing_window": candidate.missing_window.isoformat(), "reminder_reservation_id": candidate.reminder_reservation_id, "response_window_ends_at": candidate.response_window_ends_at.isoformat(), "correlation_id": candidate.correlation_id, "policy_version": policy.version, "policy_digest": policy.policy_digest, "policy_document_digest": policy.document_digest, }, dedupe_key=f"missing-checkin-reminder-review:{candidate.correlation_id}", ) def _require_customer(self, customer_key: str) -> None: if not isinstance(customer_key, str) or customer_key != self.runtime.spec.customer_key: raise DualCoachCoordinatorError("customer key is not registered for this coordinator") @staticmethod def _event_digest(event: object) -> str: if not hasattr(event, "model_dump"): raise TypeError("dual-coach event is invalid") return _canonical_digest(event.model_dump(mode="json", exclude_none=True)) CONSENT_VERSION = "privacy-v1" _REGISTERED_BINDING_TOKEN = object() _DIGEST_ZERO = "0" * 64 def _canonical_digest(value: object) -> str: """Hash a stable JSON representation used by sealed registry bindings.""" payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _path_digest(path: Path) -> str: return hashlib.sha256(str(path.resolve()).encode("utf-8")).hexdigest() @dataclass(frozen=True, slots=True) class RegisteredCustomerBinding: """Sealed durable identity for a registry-backed customer runtime.""" customer_key_digest: str data_root_digest: str registry_digest: str registry_version: str activation_digest: str mode: Literal["ordinary_v1", "diagnostic_isolated_v1"] binding_digest: str _token: object = field(default=None, repr=False, compare=False) def __post_init__(self) -> None: values = ( self.customer_key_digest, self.data_root_digest, self.registry_digest, self.activation_digest, self.binding_digest, ) if self._token is not _REGISTERED_BINDING_TOKEN: raise CustomerRegistryError("registered customer binding is sealed") if any(not isinstance(value, str) or len(value) != 64 for value in values): raise CustomerRegistryError("registered customer binding digest is invalid") if any(any(char not in "0123456789abcdef" for char in value) for value in values): raise CustomerRegistryError("registered customer binding digest is invalid") if not self.registry_version or self.mode not in {"ordinary_v1", "diagnostic_isolated_v1"}: raise CustomerRegistryError("registered customer binding metadata is invalid") expected = _canonical_digest( { "customer_key_digest": self.customer_key_digest, "data_root_digest": self.data_root_digest, "registry_digest": self.registry_digest, "registry_version": self.registry_version, "activation_digest": self.activation_digest, "mode": self.mode, } ) if self.binding_digest != expected: raise CustomerRegistryError("registered customer binding digest mismatch") def __getstate__(self) -> object: raise TypeError("registered customer bindings are not serializable") def __reduce__(self) -> object: raise TypeError("registered customer bindings are not serializable") def _make_registered_binding( *, customer_key: str, data_root: Path, registry_digest: str, registry_version: str, activation_digest: str, mode: Literal["ordinary_v1", "diagnostic_isolated_v1"], ) -> RegisteredCustomerBinding: customer_key_digest = _canonical_digest({"customer_key": customer_key}) data_root_digest = _path_digest(data_root) binding_digest = _canonical_digest( { "customer_key_digest": customer_key_digest, "data_root_digest": data_root_digest, "registry_digest": registry_digest, "registry_version": registry_version, "activation_digest": activation_digest, "mode": mode, } ) return RegisteredCustomerBinding( customer_key_digest, data_root_digest, registry_digest, registry_version, activation_digest, mode, binding_digest, _REGISTERED_BINDING_TOKEN, ) ProfileItem = Annotated[str, Field(min_length=1, max_length=200)] class TelegramAddress(BaseModel): model_config = ConfigDict(frozen=True) user_id: str = Field(min_length=1, max_length=64) chat_id: str = Field(min_length=1, max_length=64) topic_id: str = Field(min_length=1, max_length=64) @property def key(self) -> tuple[str, str, str]: return (self.user_id, self.chat_id, self.topic_id) @property def space_key(self) -> tuple[str, str]: return (self.chat_id, self.topic_id) class CustomerSchedule(BaseModel): model_config = ConfigDict(frozen=True) daily_time: time weekly_weekday: int = Field(ge=0, le=6) monthly_day: int = Field(ge=1, le=28) class AdaptiveRegistrationInputs(BaseModel): """Versioned, owner-approved customer inputs for adaptive nutrition. The model deliberately keeps the approval envelope alongside the customer values. Callers may construct a value-only instance for validation; the administrative approval boundary fills and verifies the envelope before a document is persisted. """ model_config = ConfigDict(frozen=True, extra="forbid") schema_version: Literal["1.0"] = "1.0" customer_key: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]{2,63}$") version: str = Field(default="v1", min_length=1, max_length=80) meal_count: int = Field(ge=1, le=8) budget_band: str = Field(min_length=1, max_length=80) cooking_access: str = Field(min_length=1, max_length=80) preferences: tuple[ProfileItem, ...] = Field(max_length=40) exclusions: tuple[ProfileItem, ...] = Field(max_length=40) allergies: tuple[ProfileItem, ...] = Field(max_length=40) training_schedule: tuple[CustomerTrainingScheduleEntry, ...] = Field( min_length=1, max_length=366, ) # Approval envelope. These values are absent on a value-only draft and # required by the customer-admin load/approval boundary. digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$") supersedes_digest: str = Field(default="0" * 64, pattern=r"^[0-9a-f]{64}$") approved: bool = False approved_by: TelegramAddress | None = None approved_at_kst: str | None = None activation_receipt_id: str | None = None registry_digest: str | None = None owner_digest: str | None = None activation_receipt_digest: str | None = None authority_digest: str | None = None authority: dict[str, object] | None = None base_policy_version: str | None = None base_policy_digest: str | None = None meal_constraints_version: str | None = None meal_constraints_digest: str | None = None derived_constraints: dict[str, object] | None = None derived_constraints_digest: str | None = Field( default=None, pattern=r"^[0-9a-f]{64}$", ) catalog_version: str | None = None catalog_digest: str | None = None artifact_digests: dict[str, str] | None = None artifact_documents: dict[str, dict[str, object]] | None = None @model_validator(mode="after") def require_unique_training_days(self) -> AdaptiveRegistrationInputs: days = tuple(item.date for item in self.training_schedule) if len(set(days)) != len(days): raise CustomerRegistryError("training schedule dates must be unique") return self def value_payload(self) -> dict[str, object]: """Return the canonical digest preimage without approval metadata.""" return { "schema_version": self.schema_version, "customer_key": self.customer_key, "version": self.version, "meal_count": self.meal_count, "budget_band": self.budget_band, "cooking_access": self.cooking_access, "preferences": list(self.preferences), "exclusions": list(self.exclusions), "allergies": list(self.allergies), "training_schedule": [ { "date": item.date.isoformat(), "weekday": item.weekday, "time": item.time.isoformat(), "load_category": item.load_category, } for item in self.training_schedule ], } class CustomerProfile(BaseModel): model_config = ConfigDict(frozen=True) primary_goal: str = Field(default="미정", min_length=1, max_length=500) starting_context: str | None = Field(default=None, max_length=2_000) dietary_restrictions: tuple[ProfileItem, ...] = Field(default=(), max_length=20) allergies: tuple[ProfileItem, ...] = Field(default=(), max_length=20) food_preferences: tuple[ProfileItem, ...] = Field(default=(), max_length=20) disliked_foods: tuple[ProfileItem, ...] = Field(default=(), max_length=20) supplements: tuple[ProfileItem, ...] = Field(default=(), max_length=30) cooking_access: str | None = Field(default=None, max_length=500) budget_band: str | None = Field(default=None, max_length=500) meal_count: int | None = Field(default=None, ge=2, le=6) schedule_constraints: str | None = Field(default=None, max_length=500) digestion_context: str | None = Field(default=None, max_length=2_000) sleep_goal_hours: float | None = Field(default=None, ge=0, le=24) recovery_goal: str | None = Field(default=None, max_length=2_000) training_context: str | None = Field(default=None, max_length=2_000) coach_notes: str | None = Field(default=None, max_length=4_000) class AiProcessingConsent(BaseModel): model_config = ConfigDict(frozen=True) granted: bool = False recorded_on: date | None = None notice_version: str | None = Field(default=None, max_length=80) @model_validator(mode="after") def require_auditable_grant(self) -> AiProcessingConsent: if self.granted and (self.recorded_on is None or not self.notice_version): raise CustomerRegistryError("AI processing consent requires date and notice version") return self class PlanWeek(BaseModel): model_config = ConfigDict(frozen=True) week: int = Field(ge=1, le=12) calories_kcal: int = Field(ge=800, le=10_000) protein_g: int = Field(ge=20, le=500) meal_structure: tuple[ProfileItem, ...] = Field(min_length=1, max_length=8) carbohydrate_g: int | None = Field(default=None, ge=0, le=2_000) fat_g: int | None = Field(default=None, ge=0, le=1_000) water_liters: float | None = Field(default=None, ge=0, le=30) nutrition_focus: str | None = Field(default=None, max_length=1_000) recovery_focus: str | None = Field(default=None, max_length=1_000) class TwelveWeekPlan(BaseModel): model_config = ConfigDict(frozen=True) starts_on: date focus: Literal["nutrition_90_training_10"] weeks: tuple[PlanWeek, ...] = Field(min_length=12, max_length=12) @model_validator(mode="after") def require_complete_sequence(self) -> TwelveWeekPlan: if tuple(item.week for item in self.weeks) != tuple(range(1, 13)): raise CustomerRegistryError("plan weeks must be exactly 1 through 12") return self class CustomerSpec(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") customer_key: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]{2,63}$") display_name: str = Field(min_length=1, max_length=80) enabled: bool telegram: TelegramAddress trainer: TrainerAssignment | None = None schedule: CustomerSchedule profile: CustomerProfile = Field(default_factory=CustomerProfile) ai_processing_consent: AiProcessingConsent = Field(default_factory=AiProcessingConsent) plan: TwelveWeekPlan def validate_local_activation_prerequisites(self) -> None: """Validate only the customer-owned requirements for activation.""" consent = self.ai_processing_consent if not consent.granted: raise CustomerRegistryError("customer AI processing consent is not granted") if consent.recorded_on is None or consent.notice_version != CONSENT_VERSION: raise CustomerRegistryError( f"customer AI processing consent must use current version {CONSENT_VERSION}" ) try: plan = TwelveWeekPlan.model_validate(self.plan.model_dump()) except (TypeError, ValueError) as exc: raise CustomerRegistryError("customer inline plan is invalid") from exc if tuple(item.week for item in plan.weeks) != tuple(range(1, 13)): raise CustomerRegistryError("customer inline plan must contain exactly 12 weeks") class RegistryDocument(BaseModel): model_config = ConfigDict(frozen=True) version: Literal[1] registry_mode: Literal["ordinary_v1", "diagnostic_isolated_v1"] = "ordinary_v1" diagnostic_session_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$") owner: TelegramAddress customers: tuple[CustomerSpec, ...] @model_validator(mode="after") def require_unique_boundaries(self) -> RegistryDocument: if self.registry_mode == "diagnostic_isolated_v1": if self.diagnostic_session_digest is None: raise CustomerRegistryError( "diagnostic registry requires a diagnostic session binding" ) elif self.diagnostic_session_digest is not None: raise CustomerRegistryError( "ordinary registry cannot carry a diagnostic session binding" ) keys = tuple(item.customer_key for item in self.customers) addresses = tuple(item.telegram.key for item in self.customers) spaces = tuple(item.telegram.space_key for item in self.customers) if len(set(keys)) != len(keys): raise CustomerRegistryError("customer keys must be unique") if len(set(addresses)) != len(addresses): raise CustomerRegistryError("customer Telegram addresses must be unique") if len(set(spaces)) != len(spaces): raise CustomerRegistryError("customer Telegram spaces must be unique") enabled = tuple(item for item in self.customers if item.enabled) if len(enabled) > 1: raise CustomerRegistryError("pilot supports only one external customer") for customer in enabled: self.validate_enabled_customer(customer) return self def validate_pilot_customer_boundary( self, candidate: CustomerSpec | None = None, ) -> None: """Enforce the one-external-customer pilot boundary. Disabled entries remain parseable for historical/test data and do not count toward the one-enabled-customer limit. """ customers = self.customers + ((candidate,) if candidate is not None else ()) if sum(item.enabled for item in customers) > 1: raise CustomerRegistryError("pilot supports only one external customer") def validate_enabled_customer(self, spec: CustomerSpec) -> None: """Validate owner-inclusive identity and local activation requirements.""" if not spec.enabled: return spec.validate_local_activation_prerequisites() if self.owner.key == spec.telegram.key or self.owner.space_key == spec.telegram.space_key: raise CustomerRegistryError("owner and customer Telegram identities must differ") @dataclass(frozen=True, slots=True) class CustomerRuntime: spec: CustomerSpec data_root: Path binding: RegisteredCustomerBinding | None = None def __post_init__(self) -> None: root = self.data_root if not isinstance(root, Path): raise CustomerRegistryError("customer data root must be a path") if root.is_symlink(): raise CustomerRegistryError("customer data root symlinks are not allowed") @property def customer_root(self) -> Path: return self.data_root.resolve() @property def wizard_root(self) -> Path: return (self.customer_root / "wizard").resolve() @property def nutrition_plans_root(self) -> Path: return (self.customer_root / "nutrition-plans").resolve() @property def registered_binding(self) -> RegisteredCustomerBinding: if self.binding is None: raise CustomerRegistryError("customer runtime has no registered binding") return self.binding @property def mode(self) -> str: return self.binding.mode if self.binding is not None else "standalone" def plan_week(self, day: date) -> PlanWeek: elapsed = max(0, (day - self.spec.plan.starts_on).days) return self.spec.plan.weeks[min(11, elapsed // 7)] @dataclass(frozen=True, slots=True) class CustomerRegistry: owner: TelegramAddress customers: tuple[CustomerRuntime, ...] def load_customer_registry(path: Path, profile_root: Path) -> CustomerRegistry: """Parse a private ordinary registry and derive profile-contained roots.""" return _load_customer_registry(path, profile_root, allow_diagnostic=False) def load_diagnostic_runtime_customer_registry( path: Path, profile_root: Path, *, session_digest: str | None = None, ) -> CustomerRegistry: """Load the explicitly marked diagnostic registry, never via ordinary paths.""" if ( not isinstance(session_digest, str) or len(session_digest) != 64 or any(char not in "0123456789abcdef" for char in session_digest) ): raise CustomerRegistryError("diagnostic session binding is required") try: registry = _load_customer_registry(path, profile_root, allow_diagnostic=True) except CustomerRegistryError: raise except ValueError as exc: raise CustomerRegistryError("diagnostic registry is invalid") from exc raw = path.read_bytes() document = RegistryDocument.model_validate_json(raw) if document.registry_mode != "diagnostic_isolated_v1": raise CustomerRegistryError("ordinary registry is not valid for diagnostic loading") if document.diagnostic_session_digest != session_digest: raise CustomerRegistryError("diagnostic registry session binding mismatch") return registry def _load_customer_registry( path: Path, profile_root: Path, *, allow_diagnostic: bool, ) -> CustomerRegistry: if _has_symlink(path, profile_root): raise CustomerRegistryError("registry symlinks are not allowed") resolved_root = profile_root.resolve() resolved_registry = path.resolve() if not resolved_registry.is_relative_to(resolved_root): raise CustomerRegistryError("registry escapes the profile") raw = path.read_bytes() document = RegistryDocument.model_validate_json(raw) if allow_diagnostic and document.registry_mode != "diagnostic_isolated_v1": raise CustomerRegistryError("ordinary registry is not valid for diagnostic loading") if not allow_diagnostic and document.registry_mode == "diagnostic_isolated_v1": raise CustomerRegistryError("diagnostic registry is not valid for ordinary loading") return _registry_from_document(document, raw, profile_root, resolved_root) def load_customer_registry_bytes(raw: bytes, profile_root: Path) -> CustomerRegistry: """Parse retained descriptor bytes under an already pinned profile root.""" document = RegistryDocument.model_validate_json(raw) if document.registry_mode == "diagnostic_isolated_v1": raise CustomerRegistryError("diagnostic registry is not valid for ordinary loading") return _registry_from_document( document, raw, profile_root, profile_root.resolve(strict=True) ) def _registry_from_document( document: RegistryDocument, raw: bytes, profile_root: Path, resolved_root: Path, ) -> CustomerRegistry: registry_digest = hashlib.sha256(raw).hexdigest() runtimes: list[CustomerRuntime] = [] resolved_data_roots: set[Path] = set() for spec in document.customers: candidate = profile_root.absolute() / "data" / "customers" / spec.customer_key if _has_symlink(candidate, profile_root): raise CustomerRegistryError("customer data root symlinks are not allowed") if candidate.exists() and not candidate.is_dir(): raise CustomerRegistryError("customer data root must be a directory") data_root = candidate.resolve() if not data_root.is_relative_to(resolved_root): raise CustomerRegistryError("customer data root escapes the profile") if data_root in resolved_data_roots: raise CustomerRegistryError("customer data roots must be unique") resolved_data_roots.add(data_root) activation_payload = { "activation_receipt_id": getattr(spec, "activation_receipt_id", None), "activation_receipt_digest": getattr(spec, "activation_receipt_digest", None), "authority_digest": getattr(spec, "authority_digest", None), "customer_key": spec.customer_key, } receipt_digest = getattr(spec, "activation_receipt_digest", None) activation_digest = receipt_digest or _canonical_digest(activation_payload) binding = _make_registered_binding( customer_key=spec.customer_key, data_root=data_root, registry_digest=registry_digest, registry_version=str(document.version), activation_digest=activation_digest, mode=document.registry_mode, ) runtimes.append(CustomerRuntime(spec, data_root, binding)) return CustomerRegistry(document.owner, tuple(runtimes)) def _has_symlink(path: Path, profile_root: Path) -> bool: root = profile_root.absolute() target = path.absolute() if root.is_symlink() or not target.is_relative_to(root): return True current = root for part in target.relative_to(root).parts: current /= part if current.is_symlink(): return True return False