"""Canonical profile-contract models and internal derived-view values.""" from __future__ import annotations import hashlib import json import re from dataclasses import dataclass from datetime import date, datetime, time, timedelta from enum import StrEnum from collections.abc import Mapping from typing import Any, Final, Literal, NotRequired, TypedDict, Unpack from zoneinfo import ZoneInfo from pydantic import ( AliasChoices, BaseModel, ConfigDict, Field, model_validator, ValidationError, ) from checkin_cli.weekly_operations_domain_adherence import ( ADHERENCE_SIGNAL_SCHEMA_VERSION as ADHERENCE_SIGNAL_SCHEMA_VERSION, AdherenceSignal as AdherenceSignal, CanonicalAdherenceSignal as CanonicalAdherenceSignal, adherence_signal_from_target_actual as adherence_signal_from_target_actual, build_canonical_adherence_signal as build_canonical_adherence_signal, derive_adherence_signal as derive_adherence_signal, derive_canonical_adherence_signal as derive_canonical_adherence_signal, ) from checkin_cli.weekly_operations_domain_trainer import ( TrainerIntensity as TrainerIntensity, TrainerSessionPayload as TrainerSessionPayload, ) _KST = ZoneInfo("Asia/Seoul") _RFC3339_KST = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?\+09:00$") _STRICT_FROZEN = ConfigDict(frozen=True, extra="forbid") class EventType(StrEnum): """Allowed append-only event types from the profile contract.""" RAW_PAYLOAD_ARCHIVED = "raw_payload_archived" HISTORY_IMPORTED = "history_imported" CHECK_IN_RECEIVED = "check_in_received" CHECK_IN_VALIDATED = "check_in_validated" CORRECTION = "correction" SAFETY_FLAG = "safety_flag" COACHING_RESPONSE = "coaching_response" DECISION = "decision" MORNING_CHECKIN = "morning_checkin" NUTRITION_CHECKIN = "nutrition_checkin" WORKOUT_RECORD = "workout_record" SAFETY_AUDIT = "safety_audit" TRAINER_SESSION_RECORD = "trainer_session_record" PAYMENT_RECORD = "payment_record" SATISFACTION_RECORD = "satisfaction_record" OPERATOR_TIME_RECORD = "operator_time_record" DRAFT_CREATED = "draft_created" DRAFT_EDITED = "draft_edited" DRAFT_APPROVED = "draft_approved" DRAFT_SENT = "draft_sent" SCHEDULE_REFERENCE = "schedule_reference" SCHEDULE_CORRECTION = "schedule_correction" SCHEDULE_CONFIRMATION = "schedule_confirmation" _PILOT_EVENT_TYPES: Final[frozenset[EventType]] = frozenset( { EventType.TRAINER_SESSION_RECORD, EventType.PAYMENT_RECORD, EventType.SATISFACTION_RECORD, EventType.OPERATOR_TIME_RECORD, EventType.DRAFT_CREATED, EventType.DRAFT_EDITED, EventType.DRAFT_APPROVED, EventType.DRAFT_SENT, } ) # Safety audits use the pre-pilot event shape but are pilot safety holds. _PILOT_SAFETY_EVENT_TYPES: Final[frozenset[EventType]] = _PILOT_EVENT_TYPES | frozenset({EventType.SAFETY_AUDIT}) class ContractStatus(StrEnum): """Allowed status values from the profile contract.""" ACCEPTED = "accepted" NEEDS_CLARIFICATION = "needs_clarification" UNSAFE = "unsafe" DUPLICATE = "duplicate" SUPERSEDED = "superseded" ARCHIVED = "archived" class SafetyLevel(StrEnum): NONE = "none" MONITOR = "monitor" STOP_AND_ESCALATE = "stop_and_escalate" class SafetyClass(StrEnum): URGENT = "urgent" PAIN = "pain" DISEASE = "disease" EATING_RISK = "eating_risk" DRUGS = "drugs" EXTREME_MANIPULATION = "extreme_manipulation" class SafetySourceFlow(StrEnum): CUSTOMER_CHECKIN = "customer_checkin" TRAINER_SESSION = "trainer_session" class SafetyRule(StrEnum): S1 = "S1" S2 = "S2" S3 = "S3" S4 = "S4" S5 = "S5" S6 = "S6" class SafetyMatchedField(StrEnum): FREE_TEXT = "free_text" PAIN_SUMMARY = "pain_summary" OPERATOR_NOTE = "operator_note" CALORIES_KCAL = "calories_kcal" STATED_GOAL = "stated_goal" STRUCTURED_TEST = "structured_test" BODY_WEIGHT_KG = "body_weight_kg" SLEEP_HOURS = "sleep_hours" SLEEP_QUALITY_1TO5 = "sleep_quality_1to5" READINESS_1TO5 = "readiness_1to5" TRAINING_PLAN = "training_plan" TRAINING_SUMMARY = "training_summary" WORKOUT_COMPLETION = "workout_completion" WORKOUT_QUALITY_1TO5 = "workout_quality_1to5" DIGESTION_SUMMARY = "digestion_summary" MEAL_SUMMARY = "meal_summary" WATER_LITERS = "water_liters" APPETITE_STRESS_SUMMARY = "appetite_stress_summary" NOTES = "notes" class PaymentKind(StrEnum): INITIAL = "initial" RENEWAL = "renewal" class PaymentMethod(StrEnum): BANK_TRANSFER = "bank_transfer" class OperatorTask(StrEnum): DRAFT_REVIEW = "draft_review" CUSTOMER_MESSAGE = "customer_message" ONBOARDING = "onboarding" REPORTING = "reporting" OTHER = "other" class DraftActor(StrEnum): AI = "ai" RICHARD = "richard" CHEST_PAIN_FLAG: Final[str] = "chest_pain" @dataclass(frozen=True, slots=True) class RecordRequest: """Untrusted message boundary supplied by the local CLI.""" message_id: str received_at: str text: str supersedes: str | None @dataclass(frozen=True, slots=True) class RecordResult: """Observable result returned to the CLI caller.""" outcome: str event_id: str | None safety_flags: tuple[str, ...] = () class CheckinValues(BaseModel): """Validated values used by the local current-state view.""" model_config = _STRICT_FROZEN weight_kg: float = Field(gt=0, le=500) calories_kcal: int = Field(ge=0, le=30000) sleep_hours: float | None = Field(default=None, ge=0, le=24) workout: str | None = Field(default=None, max_length=4000) class Provenance(BaseModel): """Source identity and immutable content digest required by the schema.""" model_config = _STRICT_FROZEN source_type: str source_ref: str = Field(min_length=1, max_length=1024) content_sha256: str = Field(pattern=r"^[a-f0-9]{64}$") section_anchor: str | None = Field(default=None, max_length=512) received_message_id: str | None = Field(default=None, max_length=256) class SafetyReason(BaseModel): """One bounded, typed explanation for a safety hold.""" model_config = ConfigDict( frozen=True, extra="forbid", populate_by_name=True, serialize_by_alias=True, ) class_: SafetyClass = Field( alias="class", validation_alias=AliasChoices("class_", "class"), serialization_alias="class", ) source_flow: SafetySourceFlow matched_field: SafetyMatchedField excerpt: str = Field(min_length=1, max_length=160) rule_id: SafetyRule @model_validator(mode="before") @classmethod def normalize_excerpt(cls, values: Any) -> Any: if not isinstance(values, Mapping) or "excerpt" not in values: return values normalized = "".join(character for character in str(values["excerpt"]) if ord(character) >= 0x20 and ord(character) != 0x7F) normalized = " ".join(normalized.split()) result = dict(values) result["excerpt"] = normalized return result @property def class_name(self) -> str: """Return the JSON-facing safety class without using the reserved word.""" return self.class_.value class Safety(BaseModel): """Safety disposition required for unsafe events.""" model_config = _STRICT_FROZEN level: SafetyLevel signals: tuple[str, ...] = Field(max_length=20) coaching_held: bool = False reasons: tuple[SafetyReason, ...] = Field(default=(), max_length=6) @model_validator(mode="after") def validate_reasons(self) -> Safety: keys = tuple((reason.rule_id, reason.matched_field, reason.excerpt) for reason in self.reasons) if len(set(keys)) != len(keys): raise ValueError("safety reasons must be unique by rule_id, matched_field, and excerpt") if self.reasons and (not self.coaching_held or self.level is SafetyLevel.NONE): raise ValueError("safety reasons require held non-none safety") return self class ContractCheckin(BaseModel): """Schema-named representation of live or partial historical measurements.""" model_config = _STRICT_FROZEN body_weight_kg: float | None = Field(default=None, gt=0, le=500) calories_kcal: int | None = Field(default=None, ge=0, le=30000) protein_g: int | None = Field(default=None, ge=0, le=1_000) carbohydrate_g: int | None = Field(default=None, ge=0, le=2_000) fat_g: int | None = Field(default=None, ge=0, le=1_000) sleep_hours: float | None = Field(default=None, ge=0, le=24) sleep_quality_1to5: int | None = Field(default=None, ge=1, le=5) readiness_1to5: int | None = Field(default=None, ge=1, le=5) training_summary: str | None = Field(default=None, max_length=4000) training_plan: str | None = Field(default=None, max_length=4000) workout_completion: str | None = Field(default=None, max_length=80) workout_quality_1to5: int | None = Field(default=None, ge=1, le=5) digestion_summary: str | None = Field(default=None, max_length=2000) meal_summary: str | None = Field(default=None, max_length=4000) water_liters: float | None = Field(default=None, ge=0, le=30) appetite_stress_summary: str | None = Field(default=None, max_length=2000) pain_summary: str | None = Field(default=None, max_length=2000) notes: str | None = Field(default=None, max_length=4000) class ImportManifest(BaseModel): """Typed source-day identity required for schema-2 history imports.""" model_config = _STRICT_FROZEN observation_kst_day: date class PaymentPayload(BaseModel): """Typed payment record; pilot payment terms are intentionally exact.""" model_config = _STRICT_FROZEN amount_krw: Literal[150000] paid_on: date period_start_on: date period_end_on: date method: Literal["bank_transfer"] kind: PaymentKind class SatisfactionPayload(BaseModel): """Typed customer satisfaction record.""" model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) score_1to10: int = Field(ge=1, le=10, validation_alias=AliasChoices("score_1to10", "score")) collected_on: date = Field(validation_alias=AliasChoices("collected_on", "collection_date")) note: str | None = Field(default=None, max_length=2000) @property def collection_date(self) -> date: return self.collected_on @property def score(self) -> int: return self.score_1to10 class OperatorTimePayload(BaseModel): """Typed immutable operator work entry and retry/correction identity.""" model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) entry_id: str = Field(min_length=16, max_length=80) attempt_id: str = Field(min_length=16, max_length=80) minutes: int = Field(ge=1, le=600) task: OperatorTask work_date: date = Field(validation_alias=AliasChoices("work_date", "work_on")) supersedes_entry_id: str | None = Field(default=None, min_length=16, max_length=80) @property def work_on(self) -> date: return self.work_date class DraftPayload(BaseModel): """Typed audit payload shared by the four draft lifecycle events.""" model_config = ConfigDict(frozen=True, extra="forbid") draft_id: str = Field(min_length=1, max_length=128) actor: DraftActor text: str = Field(min_length=1, max_length=8000) session_id: str | None = Field(default=None, min_length=1, max_length=128) edited_from_draft_id: str | None = Field(default=None, min_length=1, max_length=128) approved_from_draft_id: str | None = Field(default=None, min_length=1, max_length=128) approved_message_id: str | None = Field(default=None, min_length=1, max_length=128) # Explicit lifecycle names make the eight event discriminators discoverable while # retaining one canonical strict draft shape on Event. class DraftCreatedPayload(DraftPayload): pass class DraftEditedPayload(DraftPayload): pass class DraftApprovedPayload(DraftPayload): pass class DraftSentPayload(DraftPayload): pass class ScheduleReferencePayload(BaseModel): """Minimal, dual-confirmed source fact for the current training schedule.""" model_config = _STRICT_FROZEN customer_key: str = Field(min_length=1, max_length=64) session_kst_date: date session_start_kst: time customer_confirmed: bool owner_confirmed: bool last_change_note: str = Field(min_length=1, max_length=500) predecessor_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$") @model_validator(mode="after") def require_dual_confirmation(self) -> ScheduleReferencePayload: if not self.customer_confirmed or not self.owner_confirmed: raise ValueError("schedule reference requires both confirmations") return self class ScheduleConfirmationPayload(BaseModel): """Operator confirmation pinned to one exact current schedule reference.""" model_config = _STRICT_FROZEN reference_event_id: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]{7,127}$") reference_digest: str = Field(pattern=r"^[0-9a-f]{64}$") actor_id: str = Field(min_length=1, max_length=128) pin_digest: str = Field(pattern=r"^[0-9a-f]{64}$") def schedule_reference_dedupe_key(customer_key: str, session_kst_date: date | datetime | str) -> str: return f"schedule-reference:{_customer_key(customer_key)}:{_iso_date(session_kst_date)}" def schedule_confirmation_dedupe_key(customer_key: str, reference_event_id: str, reference_digest: str) -> str: if not re.fullmatch(r"[0-9a-f]{64}", reference_digest): raise ValueError("reference_digest must be a SHA-256 digest") return f"schedule-confirmation:{_customer_key(customer_key)}:{reference_event_id}:{reference_digest[:16]}" def build_schedule_reference_event( customer_key: str, session_kst_date: date | datetime | str, session_start_kst: time | str, *, customer_confirmed: bool, owner_confirmed: bool, last_change_note: str, supersedes: str | None = None, predecessor_digest: str | None = None, **kwargs: Any, ) -> Event: parsed_start = time.fromisoformat(session_start_kst) if isinstance(session_start_kst, str) else session_start_kst payload = ScheduleReferencePayload( customer_key=_customer_key(customer_key), session_kst_date=date.fromisoformat(_iso_date(session_kst_date)), session_start_kst=parsed_start, customer_confirmed=customer_confirmed, owner_confirmed=owner_confirmed, last_change_note=last_change_note, predecessor_digest=predecessor_digest, ) event_type = EventType.SCHEDULE_CORRECTION if supersedes else EventType.SCHEDULE_REFERENCE key = schedule_reference_dedupe_key(customer_key, session_kst_date) if supersedes: key = f"{key}:correction:{_payload_digest(payload)[:16]}" return _pilot_event( event_type, "schedule_reference", payload, customer_key, key, occurred_at_kst=kwargs.get("occurred_at_kst") or f"{_iso_date(session_kst_date)}T{parsed_start.isoformat()}+09:00", recorded_at_kst=kwargs.get("recorded_at_kst"), event_id=kwargs.get("event_id"), provenance=kwargs.get("provenance"), status=kwargs.get("status", ContractStatus.ACCEPTED), supersedes=supersedes, ) def build_schedule_confirmation_event( customer_key: str, reference_event_id: str, reference_digest: str, actor_id: str, pin_digest: str, **kwargs: Any, ) -> Event: payload = ScheduleConfirmationPayload( reference_event_id=reference_event_id, reference_digest=reference_digest, actor_id=actor_id, pin_digest=pin_digest, ) return _pilot_event( EventType.SCHEDULE_CONFIRMATION, "schedule_confirmation", payload, customer_key, schedule_confirmation_dedupe_key(customer_key, reference_event_id, reference_digest), occurred_at_kst=kwargs.get("occurred_at_kst"), recorded_at_kst=kwargs.get("recorded_at_kst"), event_id=kwargs.get("event_id"), provenance=kwargs.get("provenance"), status=kwargs.get("status", ContractStatus.ACCEPTED), ) def _value(value: Any, *names: str) -> Any: if isinstance(value, BaseModel): for name in names: candidate = getattr(value, name, None) if candidate is not None: return candidate return None if isinstance(value, Mapping): for name in names: if name in value and value[name] is not None: return value[name] return None def _iso_date(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() def _iso_week(value: date | datetime | str) -> str: parsed = date.fromisoformat(_iso_date(value)) iso = parsed.isocalendar() return f"{iso.year:04d}-W{iso.week:02d}" def _customer_key(value: str) -> str: if not isinstance(value, str) or not value or len(value) > 64: raise ValueError("customer_key must be a non-empty string no longer than 64 characters") return value def trainer_session_dedupe_key(customer_key: str, kst_date: date | datetime | str) -> str: return f"trainer-session:{_customer_key(customer_key)}:{_iso_date(kst_date)}" def payment_dedupe_key(customer_key: str, kind: PaymentKind | str, period_start_on: date | datetime | str) -> str: return f"payment:{_customer_key(customer_key)}:{PaymentKind(kind).value}:{_iso_date(period_start_on)}" def satisfaction_dedupe_key(customer_key: str, iso_week: date | datetime | str) -> str: week = _iso_week(iso_week) if not isinstance(iso_week, str) or not iso_week.startswith("20") else iso_week return f"satisfaction:{_customer_key(customer_key)}:{week}" def operator_time_dedupe_key(customer_key: str, entry_id: str) -> str: return f"operator-time:{_customer_key(customer_key)}:{entry_id}" def draft_dedupe_key(event_type: EventType | str, customer_key: str, draft_id: str) -> str: event_value = EventType(event_type).value if event_value not in { EventType.DRAFT_CREATED.value, EventType.DRAFT_EDITED.value, EventType.DRAFT_APPROVED.value, EventType.DRAFT_SENT.value, }: raise ValueError("draft dedupe keys require a draft lifecycle event type") return f"{event_value}:{_customer_key(customer_key)}:{draft_id}" # Common alternative spellings are intentionally aliases, not separate key logic. trainer_session_key = trainer_session_dedupe_key payment_key = payment_dedupe_key satisfaction_key = satisfaction_dedupe_key operator_time_key = operator_time_dedupe_key def _event_id(event_type: EventType | str, dedupe_key: str) -> str: digest = hashlib.sha256(f"{EventType(event_type).value}:{dedupe_key}".encode()).hexdigest()[:24] return f"pilot_{digest}" def _is_kst_rfc3339(value: object) -> bool: if not isinstance(value, str) or _RFC3339_KST.fullmatch(value) is None: return False try: parsed = datetime.fromisoformat(value) except ValueError: return False return parsed.utcoffset() == timedelta(hours=9) def _now_kst() -> str: return datetime.now(_KST).isoformat() def _at_kst(day: date | datetime | str) -> str: if isinstance(day, datetime): if day.tzinfo is None: return day.replace(tzinfo=_KST).isoformat() return day.astimezone(_KST).isoformat() return f"{_iso_date(day)}T00:00:00+09:00" def _payload_digest(payload: BaseModel) -> str: encoded = json.dumps(payload.model_dump(mode="json"), ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode()).hexdigest() def _pilot_event( event_type: EventType, payload_name: str, payload: BaseModel, customer_key: str, dedupe_key: str, *, event_id: str | None = None, occurred_at_kst: str | None = None, recorded_at_kst: str | None = None, provenance: Provenance | None = None, status: ContractStatus = ContractStatus.ACCEPTED, schema_version: Literal["1.0", "2.0"] = "1.0", supersedes: str | None = None, safety: Safety | None = None, payload_ref: str | None = None, ) -> Event: customer_key = _customer_key(customer_key) recorded = recorded_at_kst or _now_kst() source = provenance or Provenance( source_type="manual", source_ref=f"pilot:{customer_key}:{event_type.value}", content_sha256=_payload_digest(payload), ) return Event.model_validate( { "event_id": event_id or _event_id(event_type, dedupe_key), "event_type": event_type, "occurred_at_kst": occurred_at_kst or recorded, "recorded_at_kst": recorded, "schema_version": schema_version, "provenance": source, "status": status, "supersedes": supersedes, "dedupe_key": dedupe_key, payload_name: payload, "safety": safety, "payload_ref": payload_ref, } ) class TrainerSessionEventOptions(TypedDict): occurred_at_kst: NotRequired[str | None] recorded_at_kst: NotRequired[str | None] event_id: NotRequired[str | None] provenance: NotRequired[Provenance | None] status: NotRequired[ContractStatus] schema_version: NotRequired[Literal["1.0", "2.0"]] supersedes: NotRequired[str | None] safety: NotRequired[Safety | None] payload_ref: NotRequired[str | None] def build_trainer_session_event( customer_key: str, kst_date: date | datetime | str, *, session_done: bool | None = None, performance_1to5: int | None = None, intensity_vs_plan: TrainerIntensity | str | None = None, pain_summary: str | None = None, workout_summary: str = "", operator_note: str = "", done: bool | None = None, performance: int | None = None, intensity: TrainerIntensity | str | None = None, pain: str | None = None, **kwargs: Unpack[TrainerSessionEventOptions], ) -> Event: payload = TrainerSessionPayload.model_validate( { "session_done": session_done if session_done is not None else done, "performance_1to5": ( performance_1to5 if performance_1to5 is not None else performance ), "intensity_vs_plan": ( intensity_vs_plan if intensity_vs_plan is not None else intensity ), "pain_summary": pain_summary if pain_summary is not None else pain, "workout_summary": workout_summary, "operator_note": operator_note, } ) supersedes = kwargs.get("supersedes") key = trainer_session_dedupe_key(customer_key, kst_date) if supersedes: key = f"{key}:correction:{_payload_digest(payload)[:16]}" return _pilot_event( EventType.TRAINER_SESSION_RECORD, "trainer_session", payload, customer_key, key, occurred_at_kst=kwargs.get("occurred_at_kst") or _at_kst(kst_date), recorded_at_kst=kwargs.get("recorded_at_kst"), event_id=kwargs.get("event_id"), provenance=kwargs.get("provenance"), status=kwargs.get("status", ContractStatus.ACCEPTED), schema_version=kwargs.get("schema_version", "1.0"), supersedes=supersedes, safety=kwargs.get("safety"), payload_ref=kwargs.get("payload_ref"), ) def build_payment_event( customer_key: str, *, amount_krw: int = 150000, paid_on: date | str, period_start_on: date | str, period_end_on: date | str, method: PaymentMethod | str = PaymentMethod.BANK_TRANSFER, kind: PaymentKind | str, **kwargs: Any, ) -> Event: payload = PaymentPayload( amount_krw=amount_krw, paid_on=paid_on, period_start_on=period_start_on, period_end_on=period_end_on, method=method, kind=kind, ) key = payment_dedupe_key(customer_key, payload.kind, payload.period_start_on) return _pilot_event( EventType.PAYMENT_RECORD, "payment", payload, customer_key, key, occurred_at_kst=kwargs.get("occurred_at_kst") or _at_kst(payload.paid_on), recorded_at_kst=kwargs.get("recorded_at_kst"), event_id=kwargs.get("event_id"), provenance=kwargs.get("provenance"), status=kwargs.get("status", ContractStatus.ACCEPTED), schema_version=kwargs.get("schema_version", "1.0"), supersedes=kwargs.get("supersedes"), safety=kwargs.get("safety"), payload_ref=kwargs.get("payload_ref"), ) def build_satisfaction_event( customer_key: str, *, score_1to10: int | None = None, collected_on: date | str | None = None, note: str | None = None, score: int | None = None, collection_date: date | str | None = None, iso_week: str | None = None, **kwargs: Any, ) -> Event: payload = SatisfactionPayload( score_1to10=score_1to10 if score_1to10 is not None else score, collected_on=collected_on if collected_on is not None else collection_date, note=note, ) expected_iso_week = _iso_week(payload.collected_on) if iso_week is not None and iso_week != expected_iso_week: raise ValueError("iso_week must match satisfaction collected_on") key = satisfaction_dedupe_key(customer_key, expected_iso_week) return _pilot_event( EventType.SATISFACTION_RECORD, "satisfaction", payload, customer_key, key, occurred_at_kst=kwargs.get("occurred_at_kst") or _at_kst(payload.collected_on), recorded_at_kst=kwargs.get("recorded_at_kst"), event_id=kwargs.get("event_id"), provenance=kwargs.get("provenance"), status=kwargs.get("status", ContractStatus.ACCEPTED), schema_version=kwargs.get("schema_version", "1.0"), supersedes=kwargs.get("supersedes"), safety=kwargs.get("safety"), payload_ref=kwargs.get("payload_ref"), ) def build_operator_time_event( customer_key: str, *, entry_id: str, attempt_id: str, minutes: int, task: OperatorTask | str, work_date: date | str | None = None, work_on: date | str | None = None, supersedes_entry_id: str | None = None, **kwargs: Any, ) -> Event: payload = OperatorTimePayload( entry_id=entry_id, attempt_id=attempt_id, minutes=minutes, task=task, work_date=work_date if work_date is not None else work_on, supersedes_entry_id=supersedes_entry_id, ) key = operator_time_dedupe_key(customer_key, payload.entry_id) return _pilot_event( EventType.OPERATOR_TIME_RECORD, "operator_time", payload, customer_key, key, occurred_at_kst=kwargs.get("occurred_at_kst") or _at_kst(payload.work_date), recorded_at_kst=kwargs.get("recorded_at_kst"), event_id=kwargs.get("event_id"), provenance=kwargs.get("provenance"), status=kwargs.get("status", ContractStatus.ACCEPTED), schema_version=kwargs.get("schema_version", "1.0"), supersedes=kwargs.get("supersedes"), safety=kwargs.get("safety"), payload_ref=kwargs.get("payload_ref"), ) def _build_draft_event(event_type: EventType, customer_key: str, **values: Any) -> Event: payload = DraftPayload( draft_id=values.pop("draft_id"), actor=values.pop("actor"), text=values.pop("text"), session_id=values.pop("session_id", None), edited_from_draft_id=values.pop("edited_from_draft_id", None), approved_from_draft_id=values.pop("approved_from_draft_id", None), approved_message_id=values.pop("approved_message_id", None), ) key = draft_dedupe_key(event_type, customer_key, payload.draft_id) return _pilot_event( event_type, "draft", payload, customer_key, key, occurred_at_kst=values.get("occurred_at_kst"), recorded_at_kst=values.get("recorded_at_kst"), event_id=values.get("event_id"), provenance=values.get("provenance"), status=values.get("status", ContractStatus.ACCEPTED), schema_version=values.get("schema_version", "1.0"), supersedes=values.get("supersedes"), safety=values.get("safety"), payload_ref=values.get("payload_ref"), ) def build_draft_created_event(customer_key: str, draft_id: str, actor: DraftActor | str, text: str, **kwargs: Any) -> Event: return _build_draft_event(EventType.DRAFT_CREATED, customer_key, draft_id=draft_id, actor=actor, text=text, **kwargs) def build_draft_edited_event(customer_key: str, draft_id: str, actor: DraftActor | str, text: str, **kwargs: Any) -> Event: return _build_draft_event(EventType.DRAFT_EDITED, customer_key, draft_id=draft_id, actor=actor, text=text, **kwargs) def build_draft_approved_event(customer_key: str, draft_id: str, actor: DraftActor | str, text: str, **kwargs: Any) -> Event: return _build_draft_event(EventType.DRAFT_APPROVED, customer_key, draft_id=draft_id, actor=actor, text=text, **kwargs) def build_draft_sent_event(customer_key: str, draft_id: str, actor: DraftActor | str, text: str, **kwargs: Any) -> Event: return _build_draft_event(EventType.DRAFT_SENT, customer_key, draft_id=draft_id, actor=actor, text=text, **kwargs) class Event(BaseModel): """One schema-conformant immutable record.""" model_config = _STRICT_FROZEN event_id: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]{7,127}$") event_type: EventType occurred_at_kst: str recorded_at_kst: str schema_version: Literal["1.0", "2.0"] = "1.0" provenance: Provenance status: ContractStatus supersedes: str | None = Field(default=None, pattern=r"^[a-z0-9][a-z0-9_-]{7,127}$") dedupe_key: str = Field(min_length=16, max_length=256) check_in: ContractCheckin | None = None safety: Safety | None = None trainer_session: TrainerSessionPayload | None = None payment: PaymentPayload | None = None satisfaction: SatisfactionPayload | None = None operator_time: OperatorTimePayload | None = None draft: DraftPayload | None = None schedule_reference: ScheduleReferencePayload | None = None schedule_confirmation: ScheduleConfirmationPayload | None = None import_manifest: ImportManifest | None = None payload_ref: str | None = Field(default=None, max_length=1024) @model_validator(mode="before") @classmethod def fill_dedupe_key(cls, values: Any) -> Any: if not isinstance(values, Mapping): return values data = dict(values) if data.get("dedupe_key") not in (None, ""): return data event_type = str(getattr(data.get("event_type"), "value", data.get("event_type", ""))) customer = _infer_customer_key(data.get("provenance")) payload_name = { EventType.TRAINER_SESSION_RECORD.value: "trainer_session", EventType.PAYMENT_RECORD.value: "payment", EventType.SATISFACTION_RECORD.value: "satisfaction", EventType.OPERATOR_TIME_RECORD.value: "operator_time", EventType.DRAFT_CREATED.value: "draft", EventType.DRAFT_EDITED.value: "draft", EventType.DRAFT_APPROVED.value: "draft", EventType.DRAFT_SENT.value: "draft", EventType.SCHEDULE_REFERENCE.value: "schedule_reference", EventType.SCHEDULE_CORRECTION.value: "schedule_reference", EventType.SCHEDULE_CONFIRMATION.value: "schedule_confirmation", }.get(event_type) payload = data.get(payload_name) if payload_name else None inferred: str | None = None if payload_name == "trainer_session" and payload is not None: inferred = trainer_session_dedupe_key( customer, _value(payload, "kst_date", "work_date") or _occurred_date({"occurred_at": data.get("occurred_at")}), ) elif payload_name == "payment" and payload is not None: inferred = payment_dedupe_key(customer, _value(payload, "kind"), _value(payload, "period_start_on")) elif payload_name == "satisfaction" and payload is not None: inferred = satisfaction_dedupe_key(customer, _value(payload, "collected_on", "collection_date")) elif payload_name == "operator_time" and payload is not None: inferred = operator_time_dedupe_key(customer, str(_value(payload, "entry_id"))) elif payload_name == "draft" and payload is not None: inferred = draft_dedupe_key(event_type, customer, str(_value(payload, "draft_id"))) elif payload_name == "schedule_reference" and payload is not None: inferred = schedule_reference_dedupe_key(customer, _value(payload, "session_kst_date")) elif payload_name == "schedule_confirmation" and payload is not None: inferred = schedule_confirmation_dedupe_key( customer, str(_value(payload, "reference_event_id")), str(_value(payload, "reference_digest")) ) if inferred is None: event_id = str(data.get("event_id", "")) inferred = f"event:{hashlib.sha256(event_id.encode()).hexdigest()}" data["dedupe_key"] = inferred return data @model_validator(mode="after") def validate_discriminator(self) -> Event: payloads = { "trainer_session": {EventType.TRAINER_SESSION_RECORD.value}, "payment": {EventType.PAYMENT_RECORD.value}, "satisfaction": {EventType.SATISFACTION_RECORD.value}, "operator_time": {EventType.OPERATOR_TIME_RECORD.value}, "draft": { EventType.DRAFT_CREATED.value, EventType.DRAFT_EDITED.value, EventType.DRAFT_APPROVED.value, EventType.DRAFT_SENT.value, }, "schedule_reference": {EventType.SCHEDULE_REFERENCE.value, EventType.SCHEDULE_CORRECTION.value}, "schedule_confirmation": {EventType.SCHEDULE_CONFIRMATION.value}, } event_value = self.event_type.value for field_name, allowed_types in payloads.items(): present = getattr(self, field_name) is not None if event_value in allowed_types and not present: raise ValueError(f"{event_value} requires {field_name}") if present and event_value not in allowed_types: raise ValueError(f"{field_name} is not valid for {event_value}") if self.event_type is EventType.SCHEDULE_CORRECTION: if self.supersedes is None or self.schedule_reference is None or self.schedule_reference.predecessor_digest is None: raise ValueError("schedule correction requires supersession and predecessor digest") if self.event_type is EventType.SCHEDULE_CONFIRMATION and self.supersedes is not None: raise ValueError("schedule confirmation cannot supersede a reference") if self.status is ContractStatus.UNSAFE and self.safety is None: raise ValueError("unsafe events require safety") if self.event_type is EventType.HISTORY_IMPORTED: if self.status is not ContractStatus.ARCHIVED: raise ValueError("history imports must be archived") if self.schema_version == "1.0": if self.import_manifest is not None: raise ValueError("schema-1 history imports must not set import_manifest") elif ( self.import_manifest is None or self.import_manifest.observation_kst_day != datetime.fromisoformat(self.occurred_at_kst).date() ): raise ValueError( "schema-2 history imports require a manifest day matching occurred_at_kst" ) elif self.import_manifest is not None: raise ValueError("import_manifest is valid only for history_imported") if ( self.event_type in _PILOT_SAFETY_EVENT_TYPES and self.safety is not None and (self.status is ContractStatus.UNSAFE or self.safety.coaching_held) and not self.safety.reasons ): raise ValueError("pilot unsafe/held events require at least one typed SafetyReason") if self.event_type in _PILOT_EVENT_TYPES and self.payload_ref is not None: raise ValueError("pilot events must not set payload_ref (payload sidecars)") if self.event_type in _PILOT_EVENT_TYPES and self.schema_version != "1.0": raise ValueError("pilot events must use schema version 1.0") for field_name in ("occurred_at_kst", "recorded_at_kst"): if not _is_kst_rfc3339(getattr(self, field_name)): raise ValueError(f"{field_name} must be an RFC3339 timestamp with KST offset +09:00") return self @property def trainer_session_record(self) -> TrainerSessionPayload | None: return self.trainer_session @property def payment_record(self) -> PaymentPayload | None: return self.payment @property def satisfaction_record(self) -> SatisfactionPayload | None: return self.satisfaction @property def operator_time_record(self) -> OperatorTimePayload | None: return self.operator_time @property def draft_created(self) -> DraftPayload | None: return self.draft if self.event_type is EventType.DRAFT_CREATED else None @property def draft_edited(self) -> DraftPayload | None: return self.draft if self.event_type is EventType.DRAFT_EDITED else None @property def draft_approved(self) -> DraftPayload | None: return self.draft if self.event_type is EventType.DRAFT_APPROVED else None @property def draft_sent(self) -> DraftPayload | None: return self.draft if self.event_type is EventType.DRAFT_SENT else None def _infer_customer_key(provenance: Any) -> str: source_ref = _value(provenance, "source_ref") if isinstance(source_ref, str) and source_ref: parts = source_ref.split(":") if len(parts) >= 2 and parts[0] in {"pilot", "customer"} and parts[1]: return parts[1] return "unknown_customer" def _occurred_date(values: Mapping[str, Any]) -> str: occurred = values.get("occurred_at_kst") if isinstance(occurred, str) and len(occurred) >= 10: return occurred[:10] return date.today().isoformat() def validate_event(event: Event | Mapping[str, Any]) -> Event: """Validate an event against the package-owned strict canonical model.""" try: return event if isinstance(event, Event) else Event.model_validate(event) except ValidationError as exc: raise ValueError("event does not conform to canonical schema: " + str(exc)) from exc class CurrentView(BaseModel): """Rebuildable current state containing only eligible check-ins.""" model_config = _STRICT_FROZEN eligible_checkins: tuple[CheckinValues, ...] safety_flags: tuple[str, ...] class WeeklyView(BaseModel): """Small rebuildable weekly operational summary.""" model_config = _STRICT_FROZEN eligible_count: int = Field(ge=0) clarification_count: int = Field(ge=0) urgent_safety_count: int = Field(ge=0)