"""Real owner lifecycle and weekly authority for disposable Telegram E2E.""" from __future__ import annotations import json from collections.abc import Mapping from datetime import date, datetime, time, timedelta from pathlib import Path from typing import final, override from zoneinfo import ZoneInfo from checkin_cli.weekly_operations import ( CustomerKey, DayState, WeeklyOperationInput, ) from checkin_cli.weekly_operations_lineage import canonical_prefix_pin from checkin_cli import load_customer_registry from checkin_cli.weekly_operations_owner_binding import ( bind_weekly_summary_for_owner_draft, ) from checkin_cli.weekly_operations_summary import build_weekly_operations_summary from gateway.config import PlatformConfig from gateway.platforms.nutrition_coaching import ( CallbackInput, DeliveryLaunchAuthorization, IncomingAddress, NutritionCoachingCoordinator, ) from gateway.platforms.nutrition_weekly_owner_model import ( JsonValue, WeeklyExplanationModel, ) from gateway.platforms.nutrition_weekly_owner_request import ( WeeklyOwnerDraftRequest, WeeklyOwnerDraftResult, ) from scripts.nutricoach_v140_authority_fixture import ( CUSTOMER_KEY, create_weekly_authority, ) class OwnerFixtureError(RuntimeError): """The real disposable owner fixture could not be established.""" _OWNER_NOW = datetime.fromisoformat("2026-08-25T23:59:00+09:00") @final class ValidWeeklyModel(WeeklyExplanationModel): """Return one bounded explanation over production-locked fields.""" @override def generate(self, request: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]: grounding_digest = request.get("grounding_digest") if not isinstance(grounding_digest, str): raise OwnerFixtureError("grounding digest is unavailable") return { "grounding_digest": grounding_digest, "principle_ids": ["recorded_trend", "owner_decides"], "emphasis": "summary_first", } def _registry_document() -> dict[str, JsonValue]: weeks: list[JsonValue] = [ { "week": week, "calories_kcal": 2300, "protein_g": 150, "carbohydrate_g": 280, "fat_g": 65, "meal_structure": ["breakfast", "lunch", "dinner"], } for week in range(1, 13) ] return { "version": 1, "owner": {"user_id": "100", "chat_id": "100", "topic_id": "0"}, "customers": [{ "customer_key": CUSTOMER_KEY, "display_name": "fixture", "enabled": True, "telegram": {"user_id": "300", "chat_id": "300", "topic_id": "41"}, "ai_processing_consent": { "granted": True, "recorded_on": "2026-07-01", "notice_version": "privacy-v1", }, "schedule": {"daily_time": "08:00", "weekly_weekday": 0, "monthly_day": 1}, "plan": { "starts_on": "2026-08-01", "focus": "nutrition_90_training_10", "weeks": weeks, }, }], } def create_owner_coordinator(root: Path) -> NutritionCoachingCoordinator: """Create one real customer check-in required by the owner lifecycle.""" root.mkdir(parents=True, mode=0o700) registry_path = root / "registry.json" _ = registry_path.write_text( json.dumps(_registry_document(), ensure_ascii=False), encoding="utf-8", ) coordinator = NutritionCoachingCoordinator( root, load_customer_registry(registry_path, root), generation_now_provider=lambda: _OWNER_NOW, ) address = IncomingAddress("300", "300", "41") opening = coordinator.open_launcher(CUSTOMER_KEY) if opening.callback_data is None: raise OwnerFixtureError("customer launcher callback is unavailable") _ = coordinator.bind_launcher(CUSTOMER_KEY, opening.callback_data, "44") _ = coordinator.handle_callback( CallbackInput(opening.callback_data, address, "44") ) resolved = coordinator.resolve(address) if resolved is None: raise OwnerFixtureError("customer bridge is unavailable") bridge = resolved.bridge actions = ( "value", "value", "value", "value", "value", "value", "select", "select", "select", "value", "value", "select", ) answers = ( "70", "2300", "150 280 65", "planned meals", "2.5", "7", "4", "normal", "4", "appetite stable", "lower body", "skip", ) reply = bridge.apply_model_action(actions[0], answers[0]) for action, value in zip(actions[1:], answers[1:], strict=True): reply = bridge.apply_model_action(action, value) if reply.prompt is None: raise OwnerFixtureError("check-in summary prompt is unavailable") save = next(callback for label, callback in reply.prompt.buttons if label == "저장") completed = coordinator.handle_callback(CallbackInput(save, address, "44")) if completed.completion is None: raise OwnerFixtureError("customer check-in did not complete") return coordinator def create_weekly_owner_draft( root: Path, coordinator: NutritionCoachingCoordinator, ) -> WeeklyOwnerDraftResult: """Create one authority-bound Monday owner draft through production service.""" authority = create_weekly_authority(root / "weekly-authority") now = datetime.fromisoformat("2026-08-24T08:00:00+09:00") snapshot = authority.owner.tick_snapshot(CUSTOMER_KEY, now) starts_on = date(2026, 8, 17) with authority.customer.source.read_locked() as canonical: pin = canonical_prefix_pin(canonical, 0) for offset in range(7): day = starts_on + timedelta(days=offset) _ = authority.store.append(WeeklyOperationInput.for_customer( CustomerKey(CUSTOMER_KEY), day, DayState.MISSED, pin, datetime.combine(day, time(23), ZoneInfo("Asia/Seoul")), )) summary = build_weekly_operations_summary( authority.customer.source, authority.store, starts_on, ) bound = bind_weekly_summary_for_owner_draft( authority.customer.source, authority.store, summary, ) request = WeeklyOwnerDraftRequest( CUSTOMER_KEY, coordinator.owner.key, snapshot.config, snapshot.receipt, snapshot.runtime, bound, coordinator.weekly_owner_storage_authority().binding_digest, lambda: bound, ) return coordinator.create_grounded_weekly_owner_draft( request, ValidWeeklyModel(), ) def owner_platform_config() -> PlatformConfig: """Return the exact canonical owner-DM adapter configuration.""" extra: dict[str, JsonValue] = {"nutrition_coaching": { "enabled": True, "registry_path": "registry.json", "operator_card_route": { "user_id": "100", "chat_id": "100", "topic_id": 0, "version": 1, }, }} return PlatformConfig(enabled=True, token="disabled", extra=extra) def issue_delivery_capability( coordinator: NutritionCoachingCoordinator, draft_id: str, ) -> None: """Issue one bounded capability after the inherited approval persisted its card.""" current = coordinator.queue_draft_generation(draft_id, coordinator.owner) if ( current.generation is None or current.generation_record_digest is None or current.generation_checkin_revision is None ): raise OwnerFixtureError("delivery capability pins are unavailable") _ = coordinator.issue_delivery_capability( draft_id, coordinator.owner, launch_authorization=DeliveryLaunchAuthorization( candidate_digest="1" * 64, full_digest="2" * 64, core_digest="3" * 64, inventory_digest="4" * 64, manifest_digest="5" * 64, wheel_digest="6" * 64, provider_config_digest="7" * 64, profile_authorization_digest="8" * 64, hermes_wheel_sha256="6" * 64, profile_wheel_sha256="9" * 64, candidate_product_binding_sha256="a" * 64, ), capability_id="disposable-weekly-send", issued_at=_OWNER_NOW, expected_generation=current.generation, expected_record_digest=current.generation_record_digest, expected_checkin_revision=current.generation_checkin_revision, expected_draft_revision=current.generation_draft_revision, )