from __future__ import annotations import json from datetime import date, datetime, timezone from pathlib import Path from decimal import Decimal import pytest from pydantic import ValidationError from checkin_cli.customer_admin import record_satisfaction from checkin_cli.customer_coaching import ( AdaptiveRegistrationInputs, CustomerRegistryError, CustomerRuntime, RegisteredCustomerBinding, RegisteredCustomerDualCoachCoordinator, build_registered_daily_customer_projection, ScheduleConfirmRequest, ) from checkin_cli.customer_schedule import ( initialize_schedule_delivery_fence, mark_customer_task_sending, mark_customer_task_delivered, reconcile_customer_task_delivery, reserve_customer_task_delivery, reserve_missing_checkin_reminder, mark_customer_task_sent_audited, ) from checkin_cli.weekly_operations_schedule_host_models_r4 import CustomerScheduleTask from checkin_cli.operator_console import CoordinatorLifecycleAdapter from checkin_cli.adaptive_nutrition import ( AdaptiveEventStore, CustomerActionContinuity, MacroTarget, build_snapshot, feature_config_digest, propose, ) from checkin_cli.store import CanonicalEventTransaction, EventStore from checkin_cli.models import ( ContractCheckin, ContractStatus, Event, EventType, Provenance, build_schedule_confirmation_event, build_schedule_reference_event, ) import checkin_cli from checkin_cli.wizard import WizardService from checkin_cli.wizard_models import WizardContext def _registry_payload(*, second_enabled: bool = False) -> dict[str, object]: def weeks() -> list[dict[str, object]]: return [ { "week": week, "calories_kcal": 2300, "protein_g": 150, "meal_structure": ["아침", "점심", "저녁"], } for week in range(1, 13) ] return { "version": 1, "owner": {"user_id": "coach", "chat_id": "control", "topic_id": "owner"}, "customers": [ { "customer_key": "client_001", "display_name": "고객 001", "enabled": True, "telegram": {"user_id": "user-a", "chat_id": "chat-a", "topic_id": "topic-a"}, "schedule": {"daily_time": "08:00", "weekly_weekday": 0, "monthly_day": 1}, "profile": { "primary_goal": "체지방 감량과 식사 습관 안정", "dietary_restrictions": ["유당 제한"], "allergies": ["땅콩"], "food_preferences": ["한식"], "supplements": ["비타민 D"], "digestion_context": "유제품 섭취 후 복부 불편", "sleep_goal_hours": 8, "recovery_goal": "주 5일 일정에서도 피로 누적 최소화", "training_context": "운동 코치의 주 4회 웨이트 계획을 따름", }, "ai_processing_consent": {"granted": True, "recorded_on": "2026-07-19", "notice_version": "privacy-v1"}, "plan": {"starts_on": "2026-07-20", "focus": "nutrition_90_training_10", "weeks": weeks()}, }, { "customer_key": "client_002", "display_name": "고객 002", "enabled": second_enabled, "telegram": {"user_id": "user-b", "chat_id": "chat-b", "topic_id": "topic-b"}, "schedule": {"daily_time": "08:30", "weekly_weekday": 1, "monthly_day": 2}, "profile": {"primary_goal": "근육 증가", "sleep_goal_hours": 7.5}, "ai_processing_consent": {"granted": True, "recorded_on": "2026-07-19", "notice_version": "privacy-v1"}, "plan": {"starts_on": "2026-07-20", "focus": "nutrition_90_training_10", "weeks": weeks()}, }, ], } def _save_morning( service: WizardService, context: WizardContext, weight: str, *, correction: bool = False ) -> object: result = ( service.start_morning_correction(context, "2026-07-20") if correction else service.start_morning(context, "2026-07-20") ) for action, value in ( ("value", weight), ("value", "7"), ("select", "4"), ("select", "4"), ("select", "none"), ("value", "2300"), ("select", "rest"), ("value", "0"), ("select", "low"), ("select", "possible"), ("select", "none"), ("select", "skip"), ("save", None), ): result = service.answer(context, result.session_id, result.version, action, value) assert result.message == "saved" return result def _save_schedule_reference( service: WizardService, context: WizardContext, *, correction: bool = False, start_kst: str = "18:00", ) -> object: result = ( service.start_schedule_reference_correction(context, "2026-07-20") if correction else service.start_schedule_reference(context, "2026-07-20") ) for action, value in ( ("value", "2026-07-20"), ("value", start_kst), ("select", "yes"), ("select", "yes"), ("value", "customer and owner confirmed the schedule"), ): result = service.answer(context, result.session_id, result.version, action, value) return service.answer(context, result.session_id, result.version, "save") def _registered_runtime(tmp_path: Path, *, enabled: bool = True) -> tuple[Path, CustomerRuntime]: payload = _registry_payload() customers = payload["customers"] assert isinstance(customers, list) and isinstance(customers[0], dict) customers[0]["enabled"] = enabled registry_path = tmp_path / "registry.json" registry_path.write_text(json.dumps(payload), encoding="utf-8") runtime = checkin_cli.load_customer_registry(registry_path, tmp_path).customers[0] runtime.data_root.mkdir(parents=True, exist_ok=True) return registry_path, runtime def _terminal_morning_event() -> Event: return Event( event_id="morning-risk-001", event_type=EventType.MORNING_CHECKIN, occurred_at_kst="2026-07-20T08:00:00+09:00", recorded_at_kst="2026-07-20T08:00:00+09:00", provenance=Provenance( source_type="manual", source_ref="morning-risk-001", content_sha256="a" * 64, ), status=ContractStatus.ACCEPTED, dedupe_key="b" * 64, check_in=ContractCheckin( body_weight_kg=70, calories_kcal=2300, protein_g=150, carbohydrate_g=250, fat_g=60, water_liters=2.5, sleep_hours=7, sleep_quality_1to5=4, readiness_1to5=4, digestion_summary="normal", appetite_stress_summary="stable", meal_summary="on plan", pain_summary="none", training_plan="rest", ), ) def _persist_terminal_morning(runtime: CustomerRuntime, event: Event) -> Event: store = EventStore.for_registered(runtime) result = store.append_wizard_event(event) return store.load_wizard_event(result.event_id) def _verified_risk_policy() -> object: from checkin_cli.adaptive_nutrition import DualCoachRiskPolicyV1, digest value = { "weight_change_percent": {"normal": "<=2", "elevated": ">2-4", "high": ">4"}, "sleep_hours": {"normal": ">=7", "elevated": "5-<7", "high": "<5"}, "fatigue": ["low", "moderate", "high"], "pain": ["none", "present", "severe"], "exercise_feasibility": ["possible", "limited", "impossible"], "meal_deviation": ["none", "partial", "material"], "score_threshold": 4, "hard_overrides": ["pain_override", "exercise_impossible_override"], "missing_evidence_reason": "risk_evidence_unavailable", } return DualCoachRiskPolicyV1("verified-v1", value, digest(value), "c" * 64) def test_schedule_lifecycle_projects_baseline_then_one_confirmed_revision( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: from checkin_cli import adaptive_nutrition from checkin_cli import customer_admin from checkin_cli.adaptive_nutrition import initialize_adaptive_customer _, runtime = _registered_runtime(tmp_path) initialize_adaptive_customer(runtime.data_root) policy = _verified_risk_policy() monkeypatch.setattr(adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: policy) registration = AdaptiveRegistrationInputs( customer_key="client_001", meal_count=3, budget_band="standard", cooking_access="home", preferences=(), exclusions=(), allergies=(), training_schedule=( { "date": "2026-07-21", "weekday": 1, "time": "18:00", "load_category": "medium", }, ), ) monkeypatch.setattr( customer_admin, "load_approved_adaptive_registration_inputs", lambda *_: registration ) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) reference = build_schedule_reference_event( "client_001", "2026-07-20", "18:00", customer_confirmed=True, owner_confirmed=True, last_change_note="confirmed schedule", ) coordinator.stage_schedule_reference(reference, "client_001") coordinator.canonical_transaction.append_schedule_reference(reference, customer_key="client_001") baseline = coordinator.reconcile_schedule_reference() assert baseline is not None reference_digest = coordinator.canonical_transaction.schedule_reference_digest(reference) correction = build_schedule_reference_event( "client_001", "2026-07-20", "19:00", customer_confirmed=True, owner_confirmed=True, last_change_note="corrected schedule", supersedes=reference.event_id, predecessor_digest=reference_digest, ) coordinator.stage_schedule_reference(correction, "client_001") coordinator.canonical_transaction.append_schedule_reference(correction, customer_key="client_001") corrected_baseline = coordinator.reconcile_schedule_reference() assert corrected_baseline is not None assert coordinator.reconcile_schedule_reference() is None current = coordinator.current_reference("client_001") assert current is not None and current.event_id == correction.event_id assert len(coordinator.canonical_transaction.read_snapshot().events) == 2 correction_digest = coordinator.canonical_transaction.schedule_reference_digest(correction) confirmation = build_schedule_confirmation_event( "client_001", correction.event_id, correction_digest, "coach", "a" * 64, ) receipt = coordinator.confirm(ScheduleConfirmRequest("client_001", confirmation)) rows = coordinator.adaptive_store.read() assert receipt.adaptive_projection["event_type"] == "schedule_strategy_confirmed" assert [row["event_type"] for row in rows] == [ "schedule_strategy_baseline", "schedule_strategy_baseline", "schedule_strategy_confirmed" ] assert receipt.adaptive_projection["payload"]["categories"] == [ "rest", "training", "rest", "rest", "rest", "rest", "rest" ] assert coordinator.confirm(ScheduleConfirmRequest("client_001", confirmation)).adaptive_projection == receipt.adaptive_projection def test_terminal_morning_risk_holds_unavailable_policy_before_answer_parsing( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _, runtime = _registered_runtime(tmp_path) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) event = _persist_terminal_morning(runtime, _terminal_morning_event()) import checkin_cli.adaptive_nutrition as adaptive_nutrition monkeypatch.setattr( adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: (_ for _ in ()).throw(ValueError("custody unavailable")), ) malformed_answers = {"weight_change_percent": object()} first = coordinator.record_terminal_morning_risk(event, malformed_answers) replay = coordinator.record_terminal_morning_risk(event, malformed_answers) assert first == replay assert first["event_type"] == "dual_coach_risk_review" assert first["payload"]["policy_version"] == "unavailable" assert first["payload"]["policy_digest"] == "unavailable" assert first["payload"]["reasons"] == ["risk_evidence_unavailable"] assert first["payload"]["held"] is True assert "evidence" not in first["payload"] assert len(coordinator.adaptive_store.read()) == 1 def test_unavailable_risk_candidate_tracks_live_strategy_and_epoch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _, runtime = _registered_runtime(tmp_path) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) event = _persist_terminal_morning(runtime, _terminal_morning_event()) import checkin_cli.adaptive_nutrition as adaptive_nutrition monkeypatch.setattr( adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: (_ for _ in ()).throw(ValueError("custody unavailable")), ) first = coordinator.record_terminal_morning_risk(event, {}) flags = { "analytics_shadow": True, "operator_candidates": True, "activation": False, "delivery": False, } feature = {"schema_version": "1.0", "epoch": 7, **flags} feature["config_digest"] = feature_config_digest(feature["epoch"], flags) (runtime.nutrition_plans_root / "feature-epoch.json").write_text( json.dumps(feature), encoding="utf-8" ) coordinator.adaptive_store.append( "schedule_strategy_baseline", {"customer_key": runtime.spec.customer_key, "strategy_state": "schedule_unconfirmed"}, dedupe_key="strategy-after-policy-loss", ) second = coordinator.record_terminal_morning_risk(event, {}) assert second != first assert first["payload"]["policy_version"] == second["payload"]["policy_version"] == "unavailable" assert first["payload"]["policy_digest"] == second["payload"]["policy_digest"] == "unavailable" assert second["payload"]["epoch"] == 7 assert second["payload"]["source_strategy_digest"] != first["payload"]["source_strategy_digest"] assert len(coordinator.adaptive_store.read()) == 3 def test_terminal_morning_risk_evaluates_only_verified_policy( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _, runtime = _registered_runtime(tmp_path) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) policy = _verified_risk_policy() event = _persist_terminal_morning(runtime, _terminal_morning_event()) import checkin_cli.adaptive_nutrition as adaptive_nutrition monkeypatch.setattr(adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: policy) row = coordinator.record_terminal_morning_risk(event, { "weight_change_percent": "2", "sleep_duration": "7", "fatigue": "high", "pain": "none", "exercise_feasibility": "possible", "meal_deviation": "material", }) assert row["payload"]["policy_version"] == "verified-v1" assert row["payload"]["policy_digest"] == policy.policy_digest assert row["payload"]["score"] == 4 assert row["payload"]["reasons"] == ["risk_score_threshold"] assert row["payload"]["held"] is True assert len(coordinator.adaptive_store.read()) == 1 def test_registry_creates_disjoint_private_customer_roots(tmp_path: Path) -> None: # Given: two customers with distinct exact Telegram spaces. registry_path = tmp_path / "registry.json" registry_path.write_text(json.dumps(_registry_payload()), encoding="utf-8") loader = checkin_cli.load_customer_registry # When: the private registry is parsed at the profile boundary. registry = loader(registry_path, tmp_path) # Then: each customer resolves to a different profile-contained data root. assert registry is not None first, second = registry.customers assert first.data_root != second.data_root assert first.data_root.is_relative_to(tmp_path) assert second.data_root.is_relative_to(tmp_path) def test_registry_loads_with_no_enabled_customers(tmp_path: Path) -> None: payload = _registry_payload() customers = payload["customers"] assert isinstance(customers, list) and isinstance(customers[0], dict) customers[0]["enabled"] = False path = tmp_path / "registry.json" path.write_text(json.dumps(payload), encoding="utf-8") registry = checkin_cli.load_customer_registry(path, tmp_path) assert registry is not None assert all(runtime.spec.enabled is False for runtime in registry.customers) def test_registry_loads_one_enabled_customer_with_disabled_history(tmp_path: Path) -> None: path = tmp_path / "registry.json" path.write_text(json.dumps(_registry_payload()), encoding="utf-8") registry = checkin_cli.load_customer_registry(path, tmp_path) assert [runtime.spec.enabled for runtime in registry.customers] == [True, False] def test_registry_rejects_two_enabled_customers(tmp_path: Path) -> None: path = tmp_path / "registry.json" path.write_text(json.dumps(_registry_payload(second_enabled=True)), encoding="utf-8") with pytest.raises((ValidationError, ValueError), match="one external customer"): checkin_cli.load_customer_registry(path, tmp_path) def test_registry_rejects_duplicate_telegram_space(tmp_path: Path) -> None: # Given: two customer keys mapped to the same Telegram address. payload = _registry_payload() customers = payload["customers"] assert isinstance(customers, list) assert isinstance(customers[1], dict) assert isinstance(customers[0], dict) customers[1]["telegram"] = customers[0]["telegram"] registry_path = tmp_path / "registry.json" registry_path.write_text(json.dumps(payload), encoding="utf-8") loader = checkin_cli.load_customer_registry # When/Then: ambiguous routing fails closed. with pytest.raises((ValidationError, ValueError)): loader(registry_path, tmp_path) def test_registry_rejects_same_topic_even_when_user_ids_differ(tmp_path: Path) -> None: payload = _registry_payload() customers = payload["customers"] assert isinstance(customers, list) and isinstance(customers[0], dict) and isinstance(customers[1], dict) first = customers[0]["telegram"] assert isinstance(first, dict) customers[1]["telegram"] = {**first, "user_id": "different-user"} path = tmp_path / "registry.json" path.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises((ValidationError, ValueError), match="spaces must be unique"): checkin_cli.load_customer_registry(path, tmp_path) def test_registry_rejects_owner_topic_overlap(tmp_path: Path) -> None: payload = _registry_payload() customers = payload["customers"] owner = payload["owner"] assert isinstance(customers, list) and isinstance(customers[0], dict) and isinstance(owner, dict) customers[0]["telegram"] = {**owner, "user_id": "customer-user"} path = tmp_path / "registry.json" path.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises((ValidationError, ValueError), match="owner and customer Telegram identities must differ"): checkin_cli.load_customer_registry(path, tmp_path) def test_registry_rejects_symlinked_customer_root_and_registry_parent(tmp_path: Path) -> None: payload = _registry_payload() direct = tmp_path / "registry.json" direct.write_text(json.dumps(payload), encoding="utf-8") roots = tmp_path / "data" / "customers" roots.mkdir(parents=True) (roots / "client_002").mkdir() (roots / "client_001").symlink_to(roots / "client_002", target_is_directory=True) with pytest.raises(ValueError, match="symlink"): checkin_cli.load_customer_registry(direct, tmp_path) (roots / "client_001").unlink() real = tmp_path / "real-registry" real.mkdir() (real / "registry.json").write_text(json.dumps(payload), encoding="utf-8") linked = tmp_path / "linked-registry" linked.symlink_to(real, target_is_directory=True) with pytest.raises(ValueError, match="symlink"): checkin_cli.load_customer_registry(linked / "registry.json", tmp_path) def test_registry_preserves_private_longitudinal_customer_profile(tmp_path: Path) -> None: path = tmp_path / "registry.json" path.write_text(json.dumps(_registry_payload()), encoding="utf-8") first = checkin_cli.load_customer_registry(path, tmp_path).customers[0].spec assert first.profile.primary_goal == "체지방 감량과 식사 습관 안정" assert first.profile.allergies == ("땅콩",) assert first.profile.sleep_goal_hours == 8 assert first.ai_processing_consent.granted is True def test_plan_requires_exactly_twelve_ordered_weeks(tmp_path: Path) -> None: # Given: a registry whose plan omits the final week. payload = _registry_payload() customers = payload["customers"] assert isinstance(customers, list) and isinstance(customers[0], dict) plan = customers[0]["plan"] assert isinstance(plan, dict) and isinstance(plan["weeks"], list) plan["weeks"].pop() registry_path = tmp_path / "registry.json" registry_path.write_text(json.dumps(payload), encoding="utf-8") loader = checkin_cli.load_customer_registry # When/Then: an incomplete three-month project is rejected. with pytest.raises((ValidationError, ValueError)): loader(registry_path, tmp_path) def test_two_customers_on_same_day_write_disjoint_event_stores(tmp_path: Path) -> None: # Given: two registered customers completing the same KST-day flow. registry_path = tmp_path / "registry.json" registry_path.write_text(json.dumps(_registry_payload()), encoding="utf-8") registry = checkin_cli.load_customer_registry(registry_path, tmp_path) first, second = registry.customers # When: each customer finalizes a record through a service rooted at its boundary. _save_morning( WizardService.for_registered(first), WizardContext(first.spec.telegram.user_id, first.spec.telegram.topic_id), "70.1", ) _save_morning( WizardService.for_registered(second), WizardContext(second.spec.telegram.user_id, second.spec.telegram.topic_id), "91.4", ) # Then: neither event store contains the other customer's measurement. first_events = (first.data_root / "wizard" / "events.jsonl").read_text(encoding="utf-8") second_events = (second.data_root / "wizard" / "events.jsonl").read_text(encoding="utf-8") assert "70.1" in first_events and "91.4" not in first_events assert "91.4" in second_events and "70.1" not in second_events for runtime in (first, second): sequence = ( runtime.data_root / "nutrition-plans" / "canonical-sequence.jsonl" ) assert sequence.is_file() assert '"sequence":1' in sequence.read_text(encoding="utf-8") def test_explicit_migration_pairs_a_complete_events_only_ledger( tmp_path: Path, ) -> None: _, runtime = _registered_runtime(tmp_path) _save_morning( WizardService.for_registered(runtime), WizardContext(runtime.spec.telegram.user_id, runtime.spec.telegram.topic_id), "70.1", ) transaction = CanonicalEventTransaction.for_customer_runtime(runtime) events_before = transaction.events_path.read_bytes() transaction.sequence_path.write_bytes(b"") receipt = transaction.migrate_existing_events() snapshot = transaction.read_snapshot() assert receipt == {"events": 1, "sequence": 1, "migrated": True} assert transaction.events_path.read_bytes() == events_before assert len(snapshot.events) == len(snapshot.sequence_rows) == 1 assert transaction.migrate_existing_events()["migrated"] is False def test_persistence_factories_keep_registered_and_standalone_explicit(tmp_path: Path) -> None: _, runtime = _registered_runtime(tmp_path) transaction = CanonicalEventTransaction.for_customer_runtime(runtime) standalone_root = tmp_path / "standalone" standalone_store = EventStore.for_standalone(standalone_root) standalone_service = WizardService.for_standalone(standalone_root / "wizard") registered_store = EventStore.for_registered(runtime) paired_store = EventStore.for_registered(transaction, runtime.registered_binding) registered_service = WizardService.for_registered(runtime) assert standalone_store._canonical_transaction is None assert standalone_store._registered_binding is None assert standalone_service._events._canonical_transaction is None assert registered_store._events == transaction.events_path assert paired_store._events == transaction.events_path assert registered_service._events._canonical_transaction.events_path == transaction.events_path with pytest.raises(TypeError, match="cannot accept a registered runtime"): EventStore.for_standalone(runtime) with pytest.raises(TypeError, match="cannot accept a registered runtime"): EventStore.for_standalone(runtime.registered_binding) with pytest.raises(TypeError, match="cannot accept a registered runtime"): WizardService.for_standalone(runtime) with pytest.raises(TypeError, match="cannot accept a registered runtime"): WizardService.for_standalone(runtime.registered_binding) def _tree_snapshot(root: Path) -> tuple[tuple[str, str, bytes | None], ...]: entries: list[tuple[str, str, bytes | None]] = [] for path in sorted(root.rglob("*"), key=str): relative = str(path.relative_to(root)) if path.is_symlink(): entries.append((relative, "symlink", str(path.readlink()).encode())) elif path.is_dir(): entries.append((relative, "directory", None)) else: entries.append((relative, "file", path.read_bytes())) return tuple(entries) def test_registered_root_bypasses_reject_without_tree_mutation(tmp_path: Path) -> None: _, runtime = _registered_runtime(tmp_path) before = _tree_snapshot(tmp_path) for root in (runtime.wizard_root, runtime.nutrition_plans_root): with pytest.raises(ValueError, match="registered persistence root"): EventStore.for_standalone(root) assert _tree_snapshot(tmp_path) == before with pytest.raises(ValueError, match="registered persistence root"): WizardService.for_standalone(root) assert _tree_snapshot(tmp_path) == before with pytest.raises(ValueError, match="registered persistence root"): AdaptiveEventStore(runtime.nutrition_plans_root / "events.jsonl") assert _tree_snapshot(tmp_path) == before with pytest.raises(ValueError, match="registered persistence root"): AdaptiveEventStore( tmp_path / "unpaired-events.jsonl", root=runtime.nutrition_plans_root, ) assert _tree_snapshot(tmp_path) == before def test_standalone_and_registered_adaptive_factories_remain_explicit(tmp_path: Path) -> None: _, runtime = _registered_runtime(tmp_path) standalone = AdaptiveEventStore(tmp_path / "standalone" / "events.jsonl") registered = AdaptiveEventStore.for_registered(runtime) assert standalone._canonical_transaction is None assert standalone.root == tmp_path / "standalone" assert registered._canonical_transaction is not None assert registered.root == runtime.nutrition_plans_root def test_direct_persistence_constructors_require_their_factories(tmp_path: Path) -> None: _, runtime = _registered_runtime(tmp_path) transaction = CanonicalEventTransaction.for_customer_runtime(runtime) with pytest.raises(TypeError, match="explicit persistence factory"): EventStore(tmp_path) with pytest.raises(TypeError, match="explicit persistence factory"): WizardService(tmp_path) with pytest.raises(TypeError, match="explicit persistence factory"): EventStore( runtime.wizard_root, canonical_transaction=transaction, binding=runtime.registered_binding, ) with pytest.raises(TypeError, match="explicit persistence factory"): WizardService( runtime.wizard_root, canonical_transaction=transaction, binding=runtime.registered_binding, ) def test_registered_factories_reject_missing_or_forged_runtime_binding(tmp_path: Path) -> None: _, runtime = _registered_runtime(tmp_path) missing_binding = CustomerRuntime(runtime.spec, runtime.data_root) with pytest.raises(ValueError, match="no sealed binding"): EventStore.for_registered(missing_binding) with pytest.raises(ValueError, match="no sealed binding"): WizardService.for_registered(missing_binding) with pytest.raises(TypeError, match="CanonicalEventTransaction"): EventStore.for_registered(object()) with pytest.raises(TypeError, match="CustomerRuntime"): WizardService.for_registered(object()) with pytest.raises(CustomerRegistryError, match="sealed"): RegisteredCustomerBinding( "0" * 64, "0" * 64, "0" * 64, "1", "0" * 64, "ordinary_v1", "0" * 64, ) transaction = CanonicalEventTransaction.for_customer_runtime(runtime) forged_binding = object.__new__(RegisteredCustomerBinding) for field_name in ( "customer_key_digest", "registry_digest", "registry_version", "activation_digest", "mode", "binding_digest", ): object.__setattr__(forged_binding, field_name, getattr(runtime.registered_binding, field_name)) object.__setattr__(forged_binding, "data_root_digest", "0" * 64) with pytest.raises(ValueError, match="binding"): EventStore.for_registered(transaction, forged_binding) forged_runtime = CustomerRuntime(runtime.spec, runtime.data_root, forged_binding) with pytest.raises(ValueError, match="binding"): EventStore.for_registered(forged_runtime) with pytest.raises(ValueError, match="binding"): WizardService.for_registered(forged_runtime) def test_registered_wizard_finalization_writes_canonical_event_and_sequence_pair( tmp_path: Path, ) -> None: _, runtime = _registered_runtime(tmp_path) context = WizardContext(runtime.spec.telegram.user_id, runtime.spec.telegram.topic_id) _save_morning(WizardService.for_registered(runtime), context, "70.1") events_path = runtime.wizard_root / "events.jsonl" sequence_path = runtime.nutrition_plans_root / "canonical-sequence.jsonl" events = [json.loads(line) for line in events_path.read_text(encoding="utf-8").splitlines()] sequence = [json.loads(line) for line in sequence_path.read_text(encoding="utf-8").splitlines()] assert len(events) == len(sequence) == 1 assert sequence[0]["sequence"] == 1 assert sequence[0]["event_id"] == events[0]["event_id"] assert sequence[0]["event_digest"] assert not (runtime.data_root / "events.jsonl").exists() def test_registered_later_clock_day_morning_correction_uses_root_risk_day(tmp_path: Path) -> None: _, runtime = _registered_runtime(tmp_path) store = EventStore.for_registered(runtime) original = _persist_terminal_morning(runtime, _terminal_morning_event()) correction = original.model_copy( update={ "event_id": "morning-risk-correction-001", "event_type": EventType.CORRECTION, "occurred_at_kst": "2026-07-29T08:00:00+09:00", "recorded_at_kst": "2026-07-29T08:00:00+09:00", "supersedes": original.event_id, "dedupe_key": "c" * 64, } ) canonical_correction = store.load_wizard_event(store.append_wizard_event(correction).event_id) risk_rows = RegisteredCustomerDualCoachCoordinator(runtime).adaptive_store.read() row = RegisteredCustomerDualCoachCoordinator(runtime).record_terminal_morning_risk( canonical_correction, { "weight_change_percent": "0", "sleep_duration": "7", "fatigue": "low", "pain": "none", "exercise_feasibility": "possible", "meal_deviation": "none", }, ) assert str(canonical_correction.occurred_at_kst)[:10] == "2026-07-29" assert row["payload"]["terminal_checkin_id"] == canonical_correction.event_id assert row["payload"]["evaluation_kst_day"] == "2026-07-20" assert len(risk_rows) == 0 def test_registered_morning_replay_converges_after_session_save_fault( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _, runtime = _registered_runtime(tmp_path) context = WizardContext(runtime.spec.telegram.user_id, runtime.spec.telegram.topic_id) service = WizardService.for_registered(runtime) result = service.start_morning(context, "2026-07-20") for action, value in ( ("value", "70.1"), ("value", "7"), ("select", "4"), ("select", "4"), ("select", "none"), ("value", "2300"), ("select", "rest"), ("value", "0"), ("select", "low"), ("select", "possible"), ("select", "none"), ("select", "skip"), ): result = service.answer(context, result.session_id, result.version, action, value) save = service._storage.save monkeypatch.setattr( service._storage, "save", lambda session: (_ for _ in ()).throw(OSError("session save fault")) if session.finalized_event_id else save(session), ) with pytest.raises(OSError, match="session save fault"): service.answer(context, result.session_id, result.version, "save") monkeypatch.setattr(service._storage, "save", save) replay = service.answer(context, result.session_id, result.version, "save") events = EventStore.for_registered(runtime)._read_events() risks = RegisteredCustomerDualCoachCoordinator(runtime).adaptive_store.read() assert replay.message == "saved" assert len(events) == 1 assert len(risks) == 1 assert service.finalized_event(result.session_id) is not None @pytest.mark.parametrize( ("correction", "fault"), ( pytest.param(False, "append_value_error", id="reference-append-value-error"), pytest.param(False, "projection_value_error", id="reference-projection-value-error"), pytest.param(True, "append_value_error", id="correction-append-value-error"), pytest.param(True, "projection_value_error", id="correction-projection-value-error"), ), ) def test_schedule_reference_recovery_preserves_staged_intent_after_durable_canonical_fault( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, correction: bool, fault: str ) -> None: from checkin_cli import adaptive_nutrition, customer_admin from checkin_cli.adaptive_nutrition import AdaptiveEventStore, initialize_adaptive_customer _, runtime = _registered_runtime(tmp_path) initialize_adaptive_customer(runtime.data_root) policy = _verified_risk_policy() monkeypatch.setattr(adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: policy) registration = AdaptiveRegistrationInputs( customer_key="client_001", meal_count=3, budget_band="standard", cooking_access="home", preferences=(), exclusions=(), allergies=(), training_schedule=( { "date": "2026-07-21", "weekday": 1, "time": "18:00", "load_category": "medium", }, ), ) monkeypatch.setattr( customer_admin, "load_approved_adaptive_registration_inputs", lambda *_: registration ) context = WizardContext("coach", "owner", runtime.spec.customer_key) service = WizardService.for_registered(runtime) if correction: assert _save_schedule_reference(service, context).message == "saved" append = service._canonical_transaction.append_schedule_reference project_baseline = AdaptiveEventStore.project_schedule_baseline if fault == "append_value_error": def append_then_raise(event: Event, *, customer_key: str) -> object: append(event, customer_key=customer_key) raise ValueError("response lost after durable canonical append") monkeypatch.setattr(service._canonical_transaction, "append_schedule_reference", append_then_raise) else: monkeypatch.setattr( AdaptiveEventStore, "project_schedule_baseline", lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("baseline projection fault")), ) recovered = _save_schedule_reference( service, context, correction=correction, start_kst="19:00" if correction else "18:00" ) snapshot = service._canonical_transaction.read_snapshot() source = snapshot.events[-1] pending = runtime.nutrition_plans_root / "schedule-reference-pending.json" assert recovered.message == "schedule_reference_recovery_required" assert pending.exists() assert [event.event_id for event in snapshot.events].count(source.event_id) == 1 if fault == "append_value_error": monkeypatch.setattr(service._canonical_transaction, "append_schedule_reference", append) else: monkeypatch.setattr(AdaptiveEventStore, "project_schedule_baseline", project_baseline) fresh = RegisteredCustomerDualCoachCoordinator(runtime) baselines = [ row for row in fresh.adaptive_store.read() if row["event_type"] == "schedule_strategy_baseline" and row["payload"]["source_reference_id"] == source.event_id ] assert fresh.reconcile_schedule_reference() is None assert not pending.exists() assert len(baselines) == 1 assert [event.event_id for event in fresh.canonical_transaction.read_snapshot().events].count(source.event_id) == 1 def test_schedule_reference_precommit_rejection_abandons_staged_intent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: from checkin_cli import adaptive_nutrition from checkin_cli.adaptive_nutrition import initialize_adaptive_customer _, runtime = _registered_runtime(tmp_path) initialize_adaptive_customer(runtime.data_root) monkeypatch.setattr( adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: _verified_risk_policy() ) context = WizardContext("coach", "owner", runtime.spec.customer_key) service = WizardService.for_registered(runtime) monkeypatch.setattr( service._canonical_transaction, "append_schedule_reference", lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("pre-commit rejection")), ) rejected = _save_schedule_reference(service, context) assert rejected.message == "schedule_reference_rejected" assert not (runtime.nutrition_plans_root / "schedule-reference-pending.json").exists() assert not service._canonical_transaction.read_snapshot().events @pytest.mark.parametrize("correction", (False, True), ids=("reference", "correction")) def test_schedule_reference_reconcile_rejects_same_id_with_different_normalized_canonical_event( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, correction: bool ) -> None: from checkin_cli import adaptive_nutrition from checkin_cli.adaptive_nutrition import initialize_adaptive_customer from checkin_cli.customer_coaching import DualCoachCoordinatorError _, runtime = _registered_runtime(tmp_path) initialize_adaptive_customer(runtime.data_root) monkeypatch.setattr( adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: _verified_risk_policy() ) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) supersedes = None predecessor_digest = None if correction: prior = build_schedule_reference_event( "client_001", "2026-07-20", "17:00", customer_confirmed=True, owner_confirmed=True, last_change_note="prior schedule", ) coordinator.canonical_transaction.append_schedule_reference(prior, customer_key="client_001") supersedes = prior.event_id predecessor_digest = coordinator.canonical_transaction.schedule_reference_digest(prior) staged = build_schedule_reference_event( "client_001", "2026-07-20", "18:00", customer_confirmed=True, owner_confirmed=True, last_change_note="candidate schedule", supersedes=supersedes, predecessor_digest=predecessor_digest, event_id=f"schedule-conflict-{'correction' if correction else 'reference'}", ) canonical = build_schedule_reference_event( "client_001", "2026-07-20", "19:00", customer_confirmed=True, owner_confirmed=True, last_change_note="candidate schedule", supersedes=supersedes, predecessor_digest=predecessor_digest, event_id=staged.event_id, ) coordinator.stage_schedule_reference(staged, "client_001") pending = runtime.nutrition_plans_root / "schedule-reference-pending.json" pending_before = pending.read_bytes() coordinator.canonical_transaction.append_schedule_reference(canonical, customer_key="client_001") canonical_before = tuple( event.model_dump(mode="json", exclude_none=True) for event in coordinator.canonical_transaction.read_snapshot().events ) baseline_count = len(coordinator.adaptive_store.read()) with pytest.raises( DualCoachCoordinatorError, match="schedule reference recovery conflicts with canonical event", ): coordinator.reconcile_schedule_reference() assert pending.read_bytes() == pending_before assert len(coordinator.adaptive_store.read()) == baseline_count assert tuple( event.model_dump(mode="json", exclude_none=True) for event in coordinator.canonical_transaction.read_snapshot().events ) == canonical_before def test_registered_admin_coordinator_and_reconcile_paths_construct_explicitly( tmp_path: Path, ) -> None: registry_path, runtime = _registered_runtime(tmp_path, enabled=False) registry = checkin_cli.load_customer_registry(registry_path, tmp_path) satisfaction = record_satisfaction( registry_path, runtime.spec.customer_key, profile_root=tmp_path, score_1to10=9, collected_on="2026-07-20", ) assert satisfaction.outcome == "recorded" initialize_schedule_delivery_fence(tmp_path) task = CustomerScheduleTask(runtime.spec.customer_key, "daily", date(2026, 7, 20)) prepared = reserve_customer_task_delivery( tmp_path, task, "체크인 안내", { "user_id": runtime.spec.telegram.user_id, "chat_id": runtime.spec.telegram.chat_id, "topic_id": runtime.spec.telegram.topic_id, }, template_digest="a" * 64, registry_digest="b" * 64, config_digest="c" * 64, ) sending = mark_customer_task_sending(tmp_path, prepared) reconciled = reconcile_customer_task_delivery( tmp_path, sending, provider_receipt="provider-receipt-0001", message_id="message-id-0001", ) assert reconciled.state == "sent_audited" class _Coordinator: owner = registry.owner def validate_delivery_transport(self, transport: object) -> object: return transport class _Transport: def send_customer(self, customer_key: str, destination: object, text: str) -> str: return "local-receipt" coordinator = _Coordinator() source = {runtime.spec.customer_key: EventStore.for_registered(runtime)} adapter = CoordinatorLifecycleAdapter(coordinator, source, _Transport()) assert isinstance(adapter, CoordinatorLifecycleAdapter) def test_adaptive_registration_inputs_require_dated_matching_training_schedule() -> None: with pytest.raises((ValidationError, CustomerRegistryError)): AdaptiveRegistrationInputs( customer_key="client_001", meal_count=3, budget_band="standard", cooking_access="home", preferences=(), exclusions=(), allergies=(), training_schedule=[{ "date": "2026-07-20", "weekday": 1, "time": "18:00", "load_category": "high", }], ) def test_adaptive_registration_inputs_require_explicit_dietary_lists() -> None: with pytest.raises(ValidationError): AdaptiveRegistrationInputs( customer_key="client_001", meal_count=3, budget_band="standard", cooking_access="home", training_schedule=[{ "date": "2026-07-20", "weekday": 0, "time": "18:00", "load_category": "high", }], ) def test_reminder_review_is_durable_deduped_and_never_delivers( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _, runtime = _registered_runtime(tmp_path) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) class _Policy: version = "1" policy_digest = "8" * 64 document_digest = "9" * 64 import checkin_cli.adaptive_nutrition as adaptive_nutrition monkeypatch.setattr( adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: _Policy() ) reminder = reserve_missing_checkin_reminder( tmp_path, runtime.spec.customer_key, date(2026, 7, 21), {"chat_id": "reminder-chat"}, registry_digest="4" * 64, config_digest="5" * 64, operator_approval="approval-0001", ) reminder = mark_customer_task_sending(tmp_path, reminder) reminder = mark_customer_task_delivered( tmp_path, reminder, "provider-receipt-1", "message-1", ) reminder = mark_customer_task_sent_audited(tmp_path, reminder) kwargs = { "response_window_ends_at": datetime(2026, 7, 21, 9, tzinfo=timezone.utc), "now": datetime(2026, 7, 21, 10, tzinfo=timezone.utc), } first = coordinator.reminder_review_candidate(reminder, **kwargs) replay = coordinator.reminder_review_candidate(reminder, **kwargs) assert first == replay assert first["event_type"] == "missing_checkin_reminder_review" assert len(coordinator.adaptive_store.read()) == 1 def test_reminder_review_ignores_unknown_terminal_without_delivery( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _, runtime = _registered_runtime(tmp_path) coordinator = RegisteredCustomerDualCoachCoordinator(runtime) import checkin_cli.adaptive_nutrition as adaptive_nutrition monkeypatch.setattr( adaptive_nutrition, "load_verified_dual_coach_risk_policy", lambda _: (_ for _ in ()).throw(AssertionError("policy must not load for terminal reminder")), ) unknown = checkin_cli.mark_customer_task_unknown( tmp_path, reserve_missing_checkin_reminder( tmp_path, runtime.spec.customer_key, date(2026, 7, 21), {"chat_id": "reminder-chat"}, registry_digest="4" * 64, config_digest="5" * 64, operator_approval="approval-0001", ), ) assert coordinator.reminder_review_candidate( unknown, response_window_ends_at=datetime(2026, 7, 21, 9, tzinfo=timezone.utc), now=datetime(2026, 7, 21, 10, tzinfo=timezone.utc), ) is None assert coordinator.adaptive_store.read() == [] def test_registered_daily_projection_is_approved_ordered_and_bounded(tmp_path: Path) -> None: from tests.test_adaptive_nutrition import observations, policy _, runtime = _registered_runtime(tmp_path) day = date(2026, 7, 28) proposal = propose( "client_001", build_snapshot( observations(day, rate_delta=Decimal("0.5"), adherence=False), day, date(2026, 7, 1), ), policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 300, 150, 55), protein_g=150, fat_g=60, ) actions = tuple( CustomerActionContinuity( customer_key="client_001", approved_proposal_digest=proposal.digest, revision=proposal.revision, effective_kst_day=day, action_text=text, action_atom=f"action_{index}", criterion_text="다음 체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-29T08:00:00+09:00", ) for index, text in enumerate(("현재 계획 유지", "수분 기록", "운동 후 피로 기록"), 1) ) projection = build_registered_daily_customer_projection( runtime, proposal, actions=actions, next_check="내일 아침 체크인에서 다시 확인합니다.", ) assert projection.judgement == "오늘은 현재 계획을 유지합니다." assert len(projection.actions) == 3 body = projection.render() assert body.index("오늘 상태") < body.index("이전 흐름과 비교") assert body.index("이전 흐름과 비교") < body.index("오늘 판단") assert body.index("오늘 판단") < body.index("판단 이유") assert body.index("판단 이유") < body.index("오늘 할 일") assert body.index("오늘 할 일") < body.index("다음 확인") assert all(value not in body for value in ("revision", "digest", "epoch", "reservation")) with pytest.raises(CustomerRegistryError, match="approved proposal"): build_registered_daily_customer_projection( runtime, proposal, actions=( CustomerActionContinuity( customer_key="client_001", approved_proposal_digest="f" * 64, revision=proposal.revision, effective_kst_day=day, action_text="위조 행동", action_atom="forged", criterion_text="다음 체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-29T08:00:00+09:00", ), ), next_check="내일 다시 확인합니다.", )