"""Owner/topic-bound personal and customer check-in wizard domain.""" from __future__ import annotations import hashlib import re import secrets from datetime import date, datetime, timedelta from collections.abc import Mapping from pathlib import Path from typing import Final from zoneinfo import ZoneInfo from checkin_cli.customer_coaching import CustomerRuntime, RegisteredCustomerBinding from checkin_cli.models import ( ContractCheckin, ContractStatus, Event, EventType, Provenance, Safety, SafetyReason, TrainerSessionPayload, build_schedule_reference_event, build_trainer_session_event, ) from checkin_cli.store import ( CanonicalEventTransaction, EventStore, _reject_registered_persistence_root, ) from checkin_cli.wizard_models import ( ADAPTIVE_FOLLOW_UP_IDS, WizardBranch, WizardContext, WizardFlow, WizardResult, WizardSession, WizardStatus, ) from checkin_cli.wizard_storage import WizardStorage from checkin_cli.weekly_operations_wizard_host_branch_r4 import ( deterministic_weekly_branch, ) from checkin_cli.weekly_operations_wizard_host_events_r4 import ( WIZARD_MAPPING_ADAPTER, ProjectionSource, WizardValue, numeric_float, project_customer_event_state, ) from checkin_cli.weekly_operations_wizard_host_types_r4 import BranchDecision, SafetyClassification URGENT: Final[re.Pattern[str]] = re.compile(r"흉통|가슴\s*통증|실신|호흡\s*곤란|chest\s*pain", re.IGNORECASE) KST: Final[ZoneInfo] = ZoneInfo("Asia/Seoul") _WIZARD_CONSTRUCTION_TOKEN = object() S1_URGENT: Final[re.Pattern[str]] = re.compile( r"흉통|가슴\s*통증|실신|호흡\s*곤란|chest\s*pain|faint(?:ing)?|dyspnea|shortness\s+of\s+breath", re.IGNORECASE, ) S2_PAIN: Final[re.Pattern[str]] = re.compile( r"통증|아프|부상|pain|ache|hurt|injur(?:y|ies)", re.IGNORECASE, ) S3_DISEASE: Final[re.Pattern[str]] = re.compile( r"질환|질병|병원\s*진단|진단|장애|disease|disorder|diagnos(?:e|ed|is)", re.IGNORECASE, ) S4_EATING_RISK: Final[re.Pattern[str]] = re.compile( r"섭식\s*장애|식이\s*장애|폭식|구토|거식|굶|단식|binge|purge|anorex|bulimi|eating\s+disorder|starv", re.IGNORECASE, ) S5_DRUGS: Final[re.Pattern[str]] = re.compile( r"약물|스테로이드|성장\s*호르몬|이뇨제|약\s*물|drug|steroid|diuretic|growth\s+hormone|\bgh\b", re.IGNORECASE, ) S6_EXTREME: Final[re.Pattern[str]] = re.compile( r"극단|탈수|수분\s*(?:조절|컷)|water\s*cut|dehydrat|peak\s*week|나트륨\s*(?:조절|컷)", re.IGNORECASE, ) S1_SIGNAL: Final[str] = "urgent_symptom" SAFETY_SIGNAL_BY_RULE: Final[dict[str, str]] = { "S1": S1_SIGNAL, "S2": "pain", "S3": "disease", "S4": "eating_risk", "S5": "drugs", "S6": "extreme_manipulation", } SAFETY_CLASS_BY_RULE: Final[dict[str, str]] = { "S1": "urgent", "S2": "pain", "S3": "disease", "S4": "eating_risk", "S5": "drugs", "S6": "extreme_manipulation", } SAFETY_PATTERNS: Final[tuple[tuple[str, re.Pattern[str]], ...]] = ( ("S1", S1_URGENT), ("S2", S2_PAIN), ("S3", S3_DISEASE), ("S4", S4_EATING_RISK), ("S5", S5_DRUGS), ("S6", S6_EXTREME), ) MORNING_STEPS: Final[tuple[str, ...]] = ( "bodyweight", "sleep_duration", "sleep_quality", "condition", "pain", "calories", "training_plan", "optional_note", "summary", ) REGISTERED_MORNING_STEPS: Final[tuple[str, ...]] = ( "bodyweight", "sleep_duration", "sleep_quality", "condition", "pain", "calories", "training_plan", "weight_change_percent", "fatigue", "exercise_feasibility", "meal_deviation", "optional_note", "summary", ) WORKOUT_STEPS: Final[tuple[str, ...]] = ("completion", "training_summary", "workout_quality", "pain", "summary") TRAINER_STEPS: Final[tuple[str, ...]] = ( "done", "training_summary", "performance", "intensity", "pain", "operator_note", "summary", ) SCHEDULE_REFERENCE_STEPS: Final[tuple[str, ...]] = ( "session_kst_date", "session_start_kst", "customer_confirmed", "owner_confirmed", "last_change_note", "summary", ) FOLLOW_UP_STEPS: Final[frozenset[str]] = frozenset(ADAPTIVE_FOLLOW_UP_IDS) FOLLOW_UP_CHOICES: Final[dict[str, dict[str, str]]] = { "Q-SLEEP-CAUSE": { "caffeine": "caffeine", "카페인": "caffeine", "overtime": "overtime", "야근": "overtime", "stress": "stress", "스트레스": "stress", "other": "other", "기타": "other", }, "Q-SLEEP-ADJUST": { "yes": "yes", "y": "yes", "예": "yes", "네": "yes", "no": "no", "n": "no", "아니오": "no", "아니요": "no", }, "Q-COND-SYMPTOM": { "fatigue": "fatigue", "피로": "fatigue", "muscle_soreness": "muscle_soreness", "근육통": "muscle_soreness", "cold": "cold", "감기기운": "cold", "감기 기운": "cold", "other": "other", "기타": "other", }, "Q-COND-INTENSITY": { "maintain": "maintain", "stay": "maintain", "유지": "maintain", "reduce": "reduce", "하향": "reduce", "휴식": "rest", "rest": "rest", }, "Q-PERF-REASON": { "time_shortage": "time_shortage", "시간부족": "time_shortage", "시간 부족": "time_shortage", "condition": "condition", "컨디션": "condition", "pain": "pain", "통증": "pain", "other": "other", "기타": "other", }, "Q-PERF-NEXT": { "yes": "yes", "y": "yes", "예": "yes", "네": "yes", "no": "no", "n": "no", "아니오": "no", "아니요": "no", }, } NUTRITION_STEPS: Final[tuple[str, ...]] = ( "bodyweight", "calories", "macros", "meals", "water", "sleep_duration", "sleep_quality", "digestion", "condition", "appetite_stress", "training_summary", "optional_note", "summary", ) ADAPTIVE_MAXIMUM: Final[int] = 2000 ADAPTIVE_TRIGGER_STEPS: Final[dict[WizardFlow, frozenset[str]]] = { WizardFlow.MORNING: frozenset({"sleep_quality", "condition"}), WizardFlow.NUTRITION: frozenset({"sleep_quality", "condition", "training_summary"}), WizardFlow.WORKOUT: frozenset({"completion", "workout_quality"}), WizardFlow.TRAINER_SESSION: frozenset({"intensity"}), } _DONE_TRUE: Final[frozenset[str]] = frozenset({"complete", "completed", "done", "true", "yes"}) _DONE_FALSE: Final[frozenset[str]] = frozenset({"partial", "missed", "rest", "rest_changed", "not_done", "false", "no"}) def _trainer_session_evidence( value: object, ) -> TrainerSessionPayload | Event | Mapping[str, WizardValue] | None: """Parse projected trainer evidence into the accepted branch input.""" if isinstance(value, (TrainerSessionPayload, Event)): return value if not isinstance(value, Mapping): return None try: return WIZARD_MAPPING_ADAPTER.validate_python(value) except ValueError: return None def _value_from(source: object, *names: str) -> object | None: """Read one of the canonical field aliases from a model or mapping.""" for name in names: if isinstance(source, Mapping): value = source.get(name) else: value = getattr(source, name, None) if value is not None: return value return None def _text_value(value: object) -> str: return str(getattr(value, "value", value)) def _normalise_excerpt(value: object) -> str: """Bound a matched excerpt and remove all control characters.""" text = "".join(char for char in str(value) if ord(char) >= 0x20) return " ".join(text.split())[:160] def _reason_attr(reason: object, name: str, alias: str | None = None) -> object | None: value = getattr(reason, name, None) if value is not None: return value return getattr(reason, alias or name, None) def _new_safety_reason( rule_id: str, source_flow: str, matched_field: str, excerpt: str, ) -> SafetyReason: """Construct the required shared typed reason without a compatibility fallback.""" if not excerpt: raise ValueError("safety reason excerpt must be non-empty") return SafetyReason( class_=SAFETY_CLASS_BY_RULE[rule_id], source_flow=source_flow, matched_field=matched_field, excerpt=excerpt, rule_id=rule_id, ) def _answer_field_name(flow: str, key: str) -> str: if key in {"pain", "pain_summary"}: return "pain_summary" if key in {"operator_note", "trainer_operator_note"}: return "operator_note" if key in {"calories", "calories_kcal"}: return "calories_kcal" if key in {"training_plan", "goal", "stated_goal"}: return "stated_goal" if key in {"structured_test", "safety_test"}: return "structured_test" return "free_text" def _source_flow(flow: WizardFlow | str) -> str: match WizardFlow(flow): case WizardFlow.TRAINER_SESSION: return "trainer_session" case ( WizardFlow.MORNING | WizardFlow.WORKOUT | WizardFlow.NUTRITION | WizardFlow.SCHEDULE_REFERENCE ): return "customer_checkin" def _has_extreme_goal(text: str) -> bool: if S6_EXTREME.search(text): return True if not re.search(r"(?:감량|빼|줄|lose|loss|drop|cut)", text, re.IGNORECASE): return False amounts = [float(value) for value in re.findall(r"(\d+(?:\.\d+)?)\s*(?:kg|킬로|키로)", text, re.IGNORECASE)] durations = [int(value) for value in re.findall(r"(\d+)\s*(?:주|weeks?|일|days?)", text, re.IGNORECASE)] return any(amount >= 5 for amount in amounts) and any( duration <= 2 if "주" in text or re.search(r"weeks?", text, re.IGNORECASE) else duration <= 14 for duration in durations ) def classify_safety_signals( flow: WizardFlow | str, answers: Mapping[str, object], ) -> SafetyClassification: """Classify S1–S6 deterministically into shared typed safety reasons.""" source_flow = _source_flow(flow) entries: list[tuple[str, str, str]] = [] for key, value in answers.items(): if value is None: continue text = _normalise_excerpt(value) if not text: continue field = _answer_field_name(source_flow, str(key)) entries.append((str(key), field, text)) reasons: list[SafetyReason] = [] signals: list[str] = [] def add(rule_id: str, field: str, excerpt: str) -> None: reason = _new_safety_reason(rule_id, source_flow, field, excerpt) key = ( str(_reason_attr(reason, "rule_id") or rule_id), str(_reason_attr(reason, "matched_field") or field), str(_reason_attr(reason, "excerpt") or excerpt), ) if any( ( str(_reason_attr(existing, "rule_id") or "") , str(_reason_attr(existing, "matched_field") or ""), str(_reason_attr(existing, "excerpt") or ""), ) == key for existing in reasons ): return if len(reasons) >= 6: return reasons.append(reason) if rule_id not in signals: signals.append(SAFETY_SIGNAL_BY_RULE[rule_id]) for key, field, text in entries: for rule_id, pattern in SAFETY_PATTERNS: if pattern.search(text) or (rule_id == "S6" and _has_extreme_goal(text)): add(rule_id, field, text) if source_flow == "customer_checkin" and field == "calories_kcal": try: calories = float(text) except ValueError: calories = 1000 if calories < 1000: add("S4", "calories_kcal", text) if source_flow == "customer_checkin" and field == "stated_goal" and _has_extreme_goal(text): add("S6", "stated_goal", text) level = "stop_and_escalate" if "urgent_symptom" in signals else "monitor" if signals else None return SafetyClassification(level, tuple(signals), tuple(reasons), bool(signals)) def deterministic_branch( answers: Mapping[str, WizardValue], *, prior_week_same_weekday_weight: float | None = None, trainer_session: TrainerSessionPayload | Event | Mapping[str, object] | None = None, safety: SafetyClassification | None = None, canonical_event_state: ProjectionSource | None = None, **aliases: WizardValue, ) -> BranchDecision: """Apply the fixed sleep/condition/performance/change branch table.""" return deterministic_weekly_branch( answers, classifier=classify_safety_signals, prior_week_same_weekday_weight=prior_week_same_weekday_weight, trainer_session=trainer_session, safety=safety, canonical_event_state=canonical_event_state, aliases=aliases, ) classify_branch = deterministic_branch branch_for_snapshot = deterministic_branch class WizardService: """Persist and finalize drafts without allowing partial trend mutation.""" def __init__( self, home: Path, *, canonical_transaction: CanonicalEventTransaction | None = None, binding: RegisteredCustomerBinding | None = None, runtime: CustomerRuntime | None = None, owner_context: tuple[str, str, str] | None = None, trainer_context: tuple[str, str, str] | None = None, _construction_token: object | None = None, ) -> None: if _construction_token is not _WIZARD_CONSTRUCTION_TOKEN: raise TypeError("WizardService requires an explicit persistence factory") if isinstance(home, CustomerRuntime) or isinstance(home, RegisteredCustomerBinding): raise TypeError("WizardService home must be a filesystem path") if canonical_transaction is None: if binding is not None: raise ValueError("standalone WizardService cannot carry a registered binding") events = EventStore.for_standalone(Path(home)) else: if not isinstance(binding, RegisteredCustomerBinding): raise ValueError("registered WizardService requires a sealed binding") events = EventStore.for_registered(canonical_transaction, binding) self._storage: WizardStorage = WizardStorage(Path(home)) self._events: EventStore = events self._canonical_transaction: CanonicalEventTransaction | None = canonical_transaction self._owner_context: tuple[str, str, str] | None = owner_context self._trainer_context: tuple[str, str, str] | None = trainer_context self._runtime: CustomerRuntime | None = runtime @classmethod def for_standalone(cls, home: Path) -> WizardService: if isinstance(home, CustomerRuntime) or isinstance(home, RegisteredCustomerBinding): raise TypeError("standalone WizardService cannot accept a registered runtime") candidate = Path(home) _reject_registered_persistence_root(candidate, label="standalone WizardService") return cls(candidate, _construction_token=_WIZARD_CONSTRUCTION_TOKEN) @classmethod def for_registered( cls, runtime: CustomerRuntime | Mapping[str, WizardValue] ) -> WizardService: if not isinstance(runtime, CustomerRuntime): raise TypeError("registered WizardService requires CustomerRuntime") binding = runtime.binding if not isinstance(binding, RegisteredCustomerBinding): raise ValueError("registered customer runtime has no sealed binding") transaction = CanonicalEventTransaction.for_customer_runtime(runtime) trainer = _value_from(runtime.spec, "trainer") trainer_context = None if trainer is not None: trainer_context = ( runtime.spec.customer_key, str(_value_from(trainer, "user_id") or ""), str(_value_from(trainer, "topic_id") or ""), ) return cls( runtime.wizard_root, canonical_transaction=transaction, binding=binding, runtime=runtime, trainer_context=trainer_context, _construction_token=_WIZARD_CONSTRUCTION_TOKEN, ) @classmethod def for_registered_trainer_private( cls, runtime: CustomerRuntime | Mapping[str, WizardValue], ) -> WizardService: """Bind trainer drafts to the assigned trainer's private Telegram topic.""" if not isinstance(runtime, CustomerRuntime): raise TypeError("registered WizardService requires CustomerRuntime") binding = runtime.binding trainer = _value_from(runtime.spec, "trainer") if not isinstance(binding, RegisteredCustomerBinding): raise ValueError("registered customer runtime has no sealed binding") if trainer is None: raise ValueError("registered customer has no trainer assignment") trainer_user_id = _value_from(trainer, "user_id") if not isinstance(trainer_user_id, str) or not trainer_user_id: raise ValueError("registered customer trainer assignment is invalid") return cls( runtime.wizard_root, canonical_transaction=CanonicalEventTransaction.for_customer_runtime(runtime), binding=binding, runtime=runtime, trainer_context=(runtime.spec.customer_key, trainer_user_id, "0"), _construction_token=_WIZARD_CONSTRUCTION_TOKEN, ) def start_morning(self, context: WizardContext, kst_day: str) -> WizardResult: """Start or resume the current KST-day morning draft.""" return self._start(WizardFlow.MORNING, context, kst_day) def start_workout(self, context: WizardContext, kst_day: str) -> WizardResult: """Start or resume the current KST-day post-workout draft.""" return self._start(WizardFlow.WORKOUT, context, kst_day) def start_trainer_session(self, context: WizardContext, kst_day: str) -> WizardResult: """Start or resume the scoped post-PT trainer session.""" return self._start(WizardFlow.TRAINER_SESSION, context, kst_day) def start_schedule_reference(self, context: WizardContext, kst_day: str) -> WizardResult: """Capture one dual-confirmed schedule source fact; never books a session.""" return self._start(WizardFlow.SCHEDULE_REFERENCE, context, kst_day) def start_nutrition(self, context: WizardContext, kst_day: str) -> WizardResult: """Start or resume one customer nutrition submission for the KST day.""" return self._start(WizardFlow.NUTRITION, context, kst_day) def start_nutrition_correction(self, context: WizardContext, kst_day: str) -> WizardResult: return self._start_correction(WizardFlow.NUTRITION, context, kst_day) def start_morning_correction(self, context: WizardContext, kst_day: str) -> WizardResult: """Open the sole explicit correction path for a completed morning record.""" return self._start_correction(WizardFlow.MORNING, context, kst_day) def start_trainer_session_correction(self, context: WizardContext, kst_day: str) -> WizardResult: """Open a correction draft preserving the saved trainer record.""" return self._start_correction(WizardFlow.TRAINER_SESSION, context, kst_day) def start_schedule_reference_correction(self, context: WizardContext, kst_day: str) -> WizardResult: """Correct the current schedule source fact without creating a second reference.""" return self._start_correction(WizardFlow.SCHEDULE_REFERENCE, context, kst_day) def _start_correction(self, flow: WizardFlow, context: WizardContext, kst_day: str) -> WizardResult: if not self._is_kst_day(kst_day): return self._unknown_result(WizardStatus.INVALID, "", "invalid_kst_day") with self._storage.locked(): original = self._storage.find_completed(flow, context.owner_id, context.topic_id, kst_day) if original is None or original.finalized_event_id is None: return self._unknown_result(WizardStatus.REJECTED, "", "no_completed_record") if not self._context_matches(original, context): return self._unknown_result(WizardStatus.REJECTED, "", "unauthorized") latest_event_id = self._storage.latest_completed_event( flow, context.owner_id, context.topic_id, kst_day, original.finalized_event_id, ) latest = self._storage.find_finalized_event(latest_event_id) if latest is None: return self._unknown_result(WizardStatus.REJECTED, "", "latest_correction_missing") existing = self._storage.find_open(flow, context.owner_id, context.topic_id, kst_day, latest_event_id) if existing is not None: if not self._context_matches(existing, context): return self._unknown_result(WizardStatus.REJECTED, "", "unauthorized") return self._result(WizardStatus.ADVANCED, existing, "resumed_correction") created = self._create(flow, context, kst_day, latest_event_id) if flow not in { WizardFlow.TRAINER_SESSION, WizardFlow.NUTRITION, WizardFlow.SCHEDULE_REFERENCE, }: return created correction = self._storage.load(created.session_id) if correction is None: return self._unknown_result(WizardStatus.REJECTED, "", "correction_create_failed") correction_answers = dict(latest.answers) if ( flow is WizardFlow.NUTRITION and latest.macro_order == "protein_carbohydrate_fat" and len(correction_answers.get("macros", "").split()) == 3 ): protein, carbohydrate, fat = correction_answers["macros"].split() correction_answers["macros"] = f"{carbohydrate} {protein} {fat}" correction = correction.model_copy( update={ "answers": correction_answers, "step": "summary", "macro_order": ( "carbohydrate_protein_fat" if flow is WizardFlow.NUTRITION else latest.macro_order ), } ) self._storage.save(correction) return self._result(WizardStatus.ADVANCED, correction, "started_correction") def answer(self, context: WizardContext, session_id: str, expected_version: int, action: str, value: str | None = None) -> WizardResult: """Apply exactly one authorized, version-checked state transition.""" with self._storage.locked(): session = self._storage.load(session_id) if session is None or not self._context_matches(session, context): return self._unknown_result(WizardStatus.REJECTED, session_id, "unauthorized") if session.version != expected_version or session.finalized_event_id is not None: return self._result(WizardStatus.REJECTED, session, "stale_or_finalized") if action == "previous": return self._previous(session) if action == "defer": return self._defer(session) if action == "skip": if session.step not in {"optional_note", "operator_note"}: return self._result(WizardStatus.INVALID, session, "skip_not_allowed") return self._field_action(session, "select", "skip") if action == "model_skip" and session.step == "bodyweight": return self._field_action(session, "skip", None) if session.step != "safety_ack" and value is not None: candidate_answers = {**session.answers, session.step: value} raw_classification = classify_safety_signals(session.flow, candidate_answers) if raw_classification.level == "stop_and_escalate": return self._safety_stop(session, raw_classification, candidate_answers) parsed = self._parse_field(session.step, action, value) if parsed is not None: candidate_answers[session.step] = parsed classification = classify_safety_signals(session.flow, candidate_answers) if classification: return self._safety_stop(session, classification, candidate_answers) if session.step == "safety_ack": return self._acknowledge(session, action) if session.step == "summary": return self._summary_action(session, action, value) return self._field_action(session, action, value) def _start(self, flow: WizardFlow, context: WizardContext, kst_day: str) -> WizardResult: if not self._is_kst_day(kst_day): return self._unknown_result(WizardStatus.INVALID, "", "invalid_kst_day") if flow in {WizardFlow.TRAINER_SESSION, WizardFlow.SCHEDULE_REFERENCE} and ( not isinstance(context.customer_key, str) or not 1 <= len(context.customer_key) <= 64 ): return self._unknown_result(WizardStatus.REJECTED, "", "customer_scope_required") if ( flow is WizardFlow.TRAINER_SESSION and self._trainer_context is not None and (context.customer_key, context.owner_id, context.topic_id) != self._trainer_context ): return self._unknown_result(WizardStatus.REJECTED, "", "customer_scope_required") if ( flow is WizardFlow.SCHEDULE_REFERENCE and self._owner_context is not None and (context.customer_key, context.owner_id, context.topic_id) != self._owner_context ): return self._unknown_result(WizardStatus.REJECTED, "", "customer_scope_required") with self._storage.locked(): existing = self._storage.find_open(flow, context.owner_id, context.topic_id, kst_day, None) if existing is not None: if not self._context_matches(existing, context): return self._unknown_result(WizardStatus.REJECTED, "", "unauthorized") return self._result(WizardStatus.ADVANCED, existing, "resumed") if flow in ( WizardFlow.MORNING, WizardFlow.NUTRITION, WizardFlow.TRAINER_SESSION, WizardFlow.SCHEDULE_REFERENCE, ): completed = self._storage.find_completed(flow, context.owner_id, context.topic_id, kst_day) if completed is not None: if not self._context_matches(completed, context): return self._unknown_result(WizardStatus.REJECTED, "", "unauthorized") return self._result(WizardStatus.SAVED, completed, "completed_review") return self._create(flow, context, kst_day, None) def _create(self, flow: WizardFlow, context: WizardContext, kst_day: str, supersedes: str | None) -> WizardResult: if flow in {WizardFlow.TRAINER_SESSION, WizardFlow.SCHEDULE_REFERENCE} and ( not isinstance(context.customer_key, str) or not 1 <= len(context.customer_key) <= 64 ): return self._unknown_result(WizardStatus.REJECTED, "", "customer_scope_required") if ( flow is WizardFlow.TRAINER_SESSION and self._trainer_context is not None and (context.customer_key, context.owner_id, context.topic_id) != self._trainer_context ): return self._unknown_result(WizardStatus.REJECTED, "", "customer_scope_required") if ( flow is WizardFlow.SCHEDULE_REFERENCE and self._owner_context is not None and (context.customer_key, context.owner_id, context.topic_id) != self._owner_context ): return self._unknown_result(WizardStatus.REJECTED, "", "customer_scope_required") session = WizardSession( session_id=secrets.token_hex(16), flow=flow, owner_id=context.owner_id, customer_key=context.customer_key, topic_id=context.topic_id, kst_day=kst_day, version=0, step=self._steps(flow)[0], supersedes=supersedes, macro_order="carbohydrate_protein_fat" if flow is WizardFlow.NUTRITION else "protein_carbohydrate_fat", ) self._storage.save(session) return self._result(WizardStatus.ADVANCED, session, "started") def _field_action(self, session: WizardSession, action: str, value: str | None) -> WizardResult: parsed = self._parse_field(session.step, action, value) if parsed is None: return self._result(WizardStatus.INVALID, session, "invalid_answer") history = self._session_history(session) answers = { key: answer for key, answer in session.answers.items() if key in history } answers[session.step] = parsed event_state = self._canonical_customer_event_state(session) follow_up_ids = ( self._adaptive_followups(session.flow, answers, event_state) if session.step in ADAPTIVE_TRIGGER_STEPS.get(session.flow, frozenset()) else session.follow_up_ids ) routing = session.model_copy( update={"follow_up_ids": follow_up_ids}, ) next_step = self._next_step(routing, answers) advanced = session.model_copy( update={ "version": session.version + 1, "step": next_step, "answers": answers, "step_history": (*history, session.step), "follow_up_ids": follow_up_ids, "branch": self._branch_for_answers(session.flow, answers, event_state), }, ) self._storage.save(advanced) return self._result(WizardStatus.ADVANCED, advanced, "advanced") @staticmethod def _session_history(session: WizardSession) -> tuple[str, ...]: if session.step_history: return session.step_history return tuple( step for step in session.answers if step != session.step ) def _previous(self, session: WizardSession) -> WizardResult: history = self._session_history(session) if not history: return self._result(WizardStatus.INVALID, session, "previous_unavailable") target = history[-1] retained_history = history[:-1] retained_steps = {*retained_history, target} answers = { key: answer for key, answer in session.answers.items() if key in retained_steps } event_state = self._canonical_customer_event_state(session) trigger_steps = ADAPTIVE_TRIGGER_STEPS.get(session.flow, frozenset()) follow_up_ids = ( self._adaptive_followups(session.flow, answers, event_state) if any(step in retained_steps for step in trigger_steps) else () ) previous = session.model_copy( update={ "version": session.version + 1, "step": target, "answers": answers, "step_history": retained_history, "follow_up_ids": follow_up_ids, "branch": self._branch_for_answers( session.flow, answers, event_state, ), }, ) self._storage.save(previous) return self._result(WizardStatus.ADVANCED, previous, "previous") def _defer(self, session: WizardSession) -> WizardResult: deferred = session.model_copy( update={"version": session.version + 1}, ) self._storage.save(deferred) return self._result(WizardStatus.DEFERRED, deferred, "deferred") @staticmethod def _adaptive_followups( flow: WizardFlow, answers: Mapping[str, object], event_state: ProjectionSource | None = None, ) -> tuple[str, ...]: decision = deterministic_branch( answers, trainer_session=answers if flow is WizardFlow.TRAINER_SESSION else None, canonical_event_state=event_state, ) return tuple(item for item in decision.follow_up_ids if item in ADAPTIVE_FOLLOW_UP_IDS) @staticmethod def _branch_for_answers( flow: WizardFlow, answers: Mapping[str, object], event_state: ProjectionSource | None = None, ) -> WizardBranch | None: decision = deterministic_branch( answers, trainer_session=answers if flow is WizardFlow.TRAINER_SESSION else None, canonical_event_state=event_state, ) return decision.branch if decision.branch is not WizardBranch.NORMAL else None def _canonical_customer_event_state(self, session: WizardSession) -> dict[str, WizardValue]: if not session.customer_key: return {} reader = getattr(self._events, "_read_events", None) if not callable(reader): return {} try: return project_customer_event_state(reader(), session.kst_day) except (OSError, TypeError, ValueError): return {} def canonical_customer_event_state(self, kst_day: str) -> dict[str, WizardValue]: """Expose only canonical customer event projections for bridge grounding.""" reader = getattr(self._events, "_read_events", None) if not callable(reader): return {} try: return project_customer_event_state(reader(), kst_day) except (OSError, TypeError, ValueError): return {} def _next_unanswered_regular_step( self, flow: WizardFlow, answers: Mapping[str, object], ) -> str: for step in self._steps(flow): if step != "summary" and step not in answers: return step return "summary" def _safety_stop( self, session: WizardSession, classification: SafetyClassification | None = None, answers: Mapping[str, object] | None = None, ) -> WizardResult: classification = classification or classify_safety_signals(session.flow, session.answers) signals = classification.signals or session.safety_signals or (S1_SIGNAL,) reasons = classification.reasons or session.safety_reasons stopped = session.model_copy( update={ "version": session.version + 1, "step": "safety_ack", "answers": dict(answers or session.answers), "safety_signals": signals, "safety_reasons": reasons, "branch": WizardBranch.SAFETY_HOLD, }, ) self._storage.save(stopped) return self._result(WizardStatus.SAFETY_STOP, stopped, "stop_and_escalate") def _summary_action(self, session: WizardSession, action: str, value: str | None) -> WizardResult: if action == "save": return self._save(session) if action == "edit" and value in self._steps(session.flow): edited = session.model_copy( update={"version": session.version + 1, "step": value} ) self._storage.save(edited) return self._result(WizardStatus.ADVANCED, edited, "editing") return self._result(WizardStatus.INVALID, session, "invalid_summary_action") def _safety_model(self, session: WizardSession) -> Safety: level = "stop_and_escalate" if S1_SIGNAL in session.safety_signals else "monitor" values: dict[str, object] = { "level": level, "signals": session.safety_signals, "coaching_held": True, } if session.safety_reasons: values["reasons"] = session.safety_reasons return Safety(**values) def _acknowledge(self, session: WizardSession, action: str) -> WizardResult: if action != "acknowledge": return self._result(WizardStatus.INVALID, session, "acknowledgement_required") audit = self._event(session, EventType.SAFETY_AUDIT, ContractCheckin(), self._safety_model(session)) result = self._events.append_wizard_event(audit) finalized = session.model_copy(update={"version": session.version + 1, "finalized_event_id": result.event_id}) self._storage.save(finalized) return self._result(WizardStatus.SAVED, finalized, "safety_audited") def _save(self, session: WizardSession) -> WizardResult: required = tuple(step for step in self._steps(session.flow) if step != "summary") + session.follow_up_ids missing = tuple(step for step in required if step not in session.answers) if missing: return self._result(WizardStatus.INVALID, session, "incomplete") classification = classify_safety_signals(session.flow, session.answers) if classification: return self._safety_stop(session, classification) event_state = self._canonical_customer_event_state(session) trainer_evidence = ( session.answers if session.flow is WizardFlow.TRAINER_SESSION else event_state.get("trainer_session") ) trainer_evidence = _trainer_session_evidence(trainer_evidence) decision = deterministic_branch( session.answers, prior_week_same_weekday_weight=numeric_float( event_state.get("prior_week_same_weekday_weight") ), trainer_session=trainer_evidence, safety=classification, ) if session.flow is WizardFlow.TRAINER_SESSION: if not session.customer_key: return self._result(WizardStatus.INVALID, session, "customer_scope_required") event = build_trainer_session_event( session.customer_key, session.kst_day, session_done=self._trainer_done(session.answers["done"]), workout_summary=session.answers["training_summary"], performance_1to5=int(session.answers["performance"]), intensity_vs_plan=session.answers["intensity"], pain_summary=session.answers["pain"], operator_note=session.answers.get("operator_note") or "", supersedes=session.supersedes, ) elif session.flow is WizardFlow.SCHEDULE_REFERENCE: if not session.customer_key: return self._result(WizardStatus.INVALID, session, "customer_scope_required") payload = self._schedule_reference_payload(session) predecessor_digest = None if session.supersedes is not None: if self._canonical_transaction is None: return self._result( WizardStatus.INVALID, session, "schedule_reference_requires_registered_runtime" ) try: predecessor = self._events.load_wizard_event(session.supersedes) predecessor_digest = self._canonical_transaction.schedule_reference_digest(predecessor) except (OSError, TypeError, ValueError): return self._result(WizardStatus.INVALID, session, "schedule_correction_predecessor_missing") event = build_schedule_reference_event( session.customer_key, payload["session_kst_date"], payload["session_start_kst"], customer_confirmed=payload["customer_confirmed"] == "true", owner_confirmed=payload["owner_confirmed"] == "true", last_change_note=payload["last_change_note"], supersedes=session.supersedes, predecessor_digest=predecessor_digest, ) if self._canonical_transaction is None: return self._result( WizardStatus.INVALID, session, "schedule_reference_requires_registered_runtime" ) coordinator = None if self._runtime is not None: from checkin_cli.customer_coaching import RegisteredCustomerDualCoachCoordinator coordinator = RegisteredCustomerDualCoachCoordinator(self._runtime) coordinator.stage_schedule_reference(event, session.customer_key) try: self._canonical_transaction.append_schedule_reference( event, customer_key=session.customer_key ) except ValueError: try: self._events.load_wizard_event(event.event_id) except (OSError, TypeError, ValueError): if coordinator is not None: coordinator.abandon_staged_schedule_reference(event.event_id) return self._result( WizardStatus.INVALID, session, "schedule_reference_rejected", ) return self._result( WizardStatus.INVALID, session, "schedule_reference_recovery_required", ) if coordinator is not None: try: coordinator.reconcile_schedule_reference() except (OSError, TypeError, ValueError): return self._result( WizardStatus.INVALID, session, "schedule_reference_recovery_required", ) finalized = session.model_copy( update={"version": session.version + 1, "finalized_event_id": event.event_id} ) self._storage.save(finalized) return self._result(WizardStatus.SAVED, finalized, "saved") else: if session.supersedes is not None: event_type = EventType.CORRECTION elif session.flow is WizardFlow.MORNING: event_type = EventType.MORNING_CHECKIN elif session.flow is WizardFlow.NUTRITION: event_type = EventType.NUTRITION_CHECKIN elif session.flow is WizardFlow.WORKOUT: event_type = EventType.WORKOUT_RECORD else: raise ValueError(f"unsupported wizard flow: {session.flow}") event = self._event(session, event_type, self._checkin(session), None) result = self._events.append_wizard_event(event) if session.flow is WizardFlow.MORNING and self._runtime is not None: canonical_event = self._events.load_wizard_event(result.event_id) from checkin_cli.customer_coaching import RegisteredCustomerDualCoachCoordinator RegisteredCustomerDualCoachCoordinator(self._runtime).record_terminal_morning_risk( canonical_event, session.answers ) finalized = session.model_copy( update={ "version": session.version + 1, "finalized_event_id": result.event_id, "branch": decision.branch, "follow_up_ids": decision.follow_up_ids, }, ) self._storage.save(finalized) return self._result(WizardStatus.SAVED, finalized, "saved") @staticmethod def _trainer_done(value: str | bool) -> bool: if isinstance(value, bool): return value return str(value).strip().lower() in _DONE_TRUE @staticmethod def _schedule_reference_payload(session: WizardSession) -> dict[str, str]: values = session.answers return { "session_kst_date": values["session_kst_date"], "session_start_kst": values["session_start_kst"], "customer_confirmed": values["customer_confirmed"], "owner_confirmed": values["owner_confirmed"], "last_change_note": values["last_change_note"], } @staticmethod def _parse_field(step: str, action: str, value: str | None) -> str | None: candidate = value if step == "training_summary" and value is not None else value.strip() if value is not None else "" maximums = { "pain": 2000, "training_plan": 4000, "optional_note": 4000, "training_summary": 4000, "meals": 4000, "digestion": 2000, "appetite_stress": 2000, "operator_note": 2000, "exercise_feasibility": 80, "meal_plan_deviation": 2000, "meal_deviation": 80, "last_change_note": 500, } maximum = maximums.get(step) if maximum is not None and len(candidate) > maximum: return None if step in FOLLOW_UP_STEPS and action == "select": choices = FOLLOW_UP_CHOICES.get(step, {}) return choices.get(candidate.casefold()) if step == "bodyweight" and action == "skip": return "" if step in ("bodyweight", "sleep_duration") and action == "value": try: number = float(candidate) except ValueError: return None maximum = 500.0 if step == "bodyweight" else 24.0 return candidate if 0.0 < number <= maximum else None if step == "weight_change_percent" and action == "value": try: from decimal import Decimal number = Decimal(candidate) except (ArithmeticError, ValueError): return None return candidate if number.is_finite() and number.copy_abs() <= Decimal("100") else None if step == "calories" and action == "value": return candidate if candidate.isdigit() and 0 <= int(candidate) <= 30000 else None if step == "macros" and action == "value": parts = candidate.split() if len(parts) != 3 or not all(part.isdigit() for part in parts): return None protein, carbohydrate, fat = (int(part) for part in parts) return candidate if protein <= 1_000 and carbohydrate <= 2_000 and fat <= 1_000 else None if step == "water" and action == "value": try: liters = float(candidate) except ValueError: return None return candidate if 0.0 <= liters <= 30.0 else None if step in ("sleep_quality", "condition", "workout_quality") and action == "select": return candidate if candidate in {"1", "2", "3", "4", "5"} else None if step == "done" and action == "select": normalized = candidate.casefold() return candidate if normalized in _DONE_TRUE or normalized in _DONE_FALSE else None if step == "performance" and action == "select": return candidate if candidate in {"1", "2", "3", "4", "5"} else None if step == "intensity" and action == "select": return candidate if candidate in {"below", "as_planned", "above"} else None if step == "session_kst_date" and action == "value": try: return candidate if date.fromisoformat(candidate).isoformat() == candidate else None except ValueError: return None if step == "session_start_kst" and action == "value": try: return datetime.strptime(candidate, "%H:%M").time().isoformat(timespec="minutes") except ValueError: return None if step in {"customer_confirmed", "owner_confirmed"} and action == "select": return "true" if candidate.casefold() in {"true", "yes", "y", "예", "네"} else None if step == "fatigue" and action == "select": return candidate if candidate in {"low", "moderate", "high"} else None if step == "exercise_feasibility" and action == "select": return candidate if candidate in {"possible", "limited", "impossible"} else None if step == "meal_deviation" and action == "select": return candidate if candidate in {"none", "partial", "material"} else None if step == "meal_plan_deviation" and action in {"select", "value"}: return candidate if action == "select" and candidate == "none" else candidate or None if step == "last_change_note" and action == "value": return candidate or None if step == "pain" and action in {"select", "value"}: return "none" if action == "select" and candidate == "none" else candidate or None if step == "digestion" and action in {"select", "value"}: return "normal" if action == "select" and candidate == "normal" else candidate or None if step == "operator_note" and action in {"value", "select"}: return "" if action == "select" and candidate == "skip" else candidate or None if step == "training_plan" and action == "select": return candidate if candidate in {"rest", "undecided"} else None if step == "training_plan" and action == "value": return candidate or None if step == "optional_note" and action in {"value", "select"}: return "" if action == "select" and candidate == "skip" else candidate or None if step == "completion" and action == "select": return candidate if candidate in {"complete", "partial", "rest_changed", "rest", "missed"} else None if step == "training_summary": if action == "select" and candidate == "trainer_recorded": return "" if action == "value": return candidate if candidate.strip() else None return None if step in {"meals", "appetite_stress"} and action == "value": return candidate or None return None def _steps(self, flow: WizardFlow) -> tuple[str, ...]: if flow is WizardFlow.MORNING: return REGISTERED_MORNING_STEPS if self._runtime is not None else MORNING_STEPS if flow is WizardFlow.NUTRITION: return NUTRITION_STEPS if flow is WizardFlow.WORKOUT: return WORKOUT_STEPS if flow is WizardFlow.TRAINER_SESSION: return TRAINER_STEPS if flow is WizardFlow.SCHEDULE_REFERENCE: return SCHEDULE_REFERENCE_STEPS raise ValueError(f"unsupported wizard flow: {flow}") def _next_step(self, session: WizardSession, answers: Mapping[str, object]) -> str: pending_followups = tuple( item for item in session.follow_up_ids if item not in answers ) if pending_followups: return pending_followups[0] return self._next_unanswered_regular_step(session.flow, answers) def _checkin(self, session: WizardSession) -> ContractCheckin: values = session.answers if session.flow is WizardFlow.MORNING: bodyweight = values.get("bodyweight") return ContractCheckin( body_weight_kg=float(bodyweight) if bodyweight else None, calories_kcal=int(values["calories"]), sleep_hours=float(values["sleep_duration"]), sleep_quality_1to5=int(values["sleep_quality"]), readiness_1to5=int(values["condition"]), pain_summary=values["pain"], training_plan=values["training_plan"], notes=values.get("optional_note") or None, ) if session.flow is WizardFlow.NUTRITION: first, second, fat = (int(part) for part in values["macros"].split()) if session.macro_order == "carbohydrate_protein_fat": carbohydrate, protein = first, second else: protein, carbohydrate = first, second bodyweight = values.get("bodyweight") return ContractCheckin( body_weight_kg=float(bodyweight) if bodyweight else None, calories_kcal=int(values["calories"]), protein_g=protein, carbohydrate_g=carbohydrate, fat_g=fat, sleep_hours=float(values["sleep_duration"]), sleep_quality_1to5=int(values["sleep_quality"]), readiness_1to5=int(values["condition"]), training_summary=values["training_summary"], digestion_summary=values["digestion"], meal_summary=values["meals"], water_liters=float(values["water"]), appetite_stress_summary=values["appetite_stress"], notes=values.get("optional_note") or None, ) return ContractCheckin(workout_completion=values["completion"], training_summary=values["training_summary"], workout_quality_1to5=int(values["workout_quality"]), pain_summary=values["pain"]) @staticmethod def _event( session: WizardSession, event_type: EventType, check_in: ContractCheckin | None, safety: Safety | None, ) -> Event: digest = hashlib.sha256(f"{session.session_id}:{session.version}:{event_type}".encode()).hexdigest() recorded_at = datetime.now(KST).isoformat() return Event( event_id=f"wizard_{digest[:24]}", event_type=event_type, occurred_at_kst=recorded_at, recorded_at_kst=recorded_at, schema_version="2.0", provenance=Provenance( source_type="telegram", source_ref=f"wizard:{session.session_id}", content_sha256=digest, ), status=ContractStatus.UNSAFE if safety is not None else ContractStatus.ACCEPTED, supersedes=session.supersedes, dedupe_key=digest, check_in=check_in, safety=safety, ) def finalized_event(self, session_id: str) -> Event | None: """Return the canonical event linked by a finalized private draft.""" session = self._storage.load(session_id) event_id = session.finalized_event_id if session is not None else None if not event_id: return None return next( (event for event in self._events._read_events() if event.event_id == event_id), None, ) def finalized_branch_decision(self, session_id: str) -> BranchDecision | None: """Return the branch captured at finalization, without re-projecting later events.""" session = self._storage.load(session_id) if session is None or session.finalized_event_id is None or session.branch is None: return None detailed = session.branch in { WizardBranch.ANOMALY, WizardBranch.CHANGE, WizardBranch.SAFETY_HOLD, } return BranchDecision(session.branch, session.follow_up_ids, detailed, ()) @staticmethod def _context_matches(session: WizardSession, context: WizardContext) -> bool: """Require the persisted customer scope as well as the auth boundary.""" return ( session.owner_id == context.owner_id and session.topic_id == context.topic_id and session.customer_key == context.customer_key ) @staticmethod def _is_kst_day(value: str) -> bool: try: date.fromisoformat(value) except ValueError: return False return True def _result(self, status: WizardStatus, session: WizardSession, message: str) -> WizardResult: total = len(tuple( step for step in self._steps(session.flow) if step != "summary" )) + len(session.follow_up_ids) history = self._session_history(session) position = ( total if session.step in {"summary", "safety_ack"} else min(len(history) + 1, total) ) return WizardResult( status, session.session_id, session.version, session.step, message, session.branch, session.follow_up_ids, session.safety_signals, session.safety_reasons, position, total, bool(history) and session.step != "safety_ack", session.step in {"optional_note", "operator_note"}, ) @staticmethod def _unknown_result(status: WizardStatus, session_id: str, message: str) -> WizardResult: """Return a deliberately data-free response for unknown sessions.""" return WizardResult(status, session_id, 0, "unknown", message)