"""Real Todo7 authorities for gateway owner-draft tests.""" from __future__ import annotations import hashlib import json from datetime import date, datetime from pathlib import Path from zoneinfo import ZoneInfo from checkin_cli.customer_coaching import CustomerRuntime, load_customer_registry from checkin_cli.models import ContractCheckin, ContractStatus, Event, EventType, Provenance from checkin_cli.store import CanonicalEventTransaction from checkin_cli.weekly_operations import CustomerKey, DayState, ReminderIdentity, WeeklyOperationInput from checkin_cli.weekly_operations_authority import AuthorityId, begin_authority_initialization from checkin_cli.weekly_operations_customer_authority import CanonicalCheckinCustomerAuthority from checkin_cli.weekly_operations_lineage import canonical_prefix_pin from checkin_cli.weekly_operations_owner_binding import BoundWeeklySummaryForOwnerDraft, bind_weekly_summary_for_owner_draft from checkin_cli.weekly_operations_parent import acquire_parent_authority from checkin_cli.weekly_operations_registration_handoff import begin_canonical_authority_registration from checkin_cli.weekly_operations_store import WeeklyOperationsStore from checkin_cli.weekly_operations_summary import build_weekly_operations_summary _CUSTOMER = CustomerKey("pilot_customer_001") _KST = ZoneInfo("Asia/Seoul") def bound_summary_at(root: Path) -> BoundWeeklySummaryForOwnerDraft: """Build the approved synthetic week through exact registered capabilities.""" authority_root = root / "sidecar" authority_root.mkdir(parents=True, mode=0o700) authority_root.chmod(0o700) parent = acquire_parent_authority(authority_root) authority = None with begin_authority_initialization(parent, AuthorityId("4" * 64)) as transaction: authority = transaction.authority _ = transaction.binding transaction.acknowledge_binding() if authority is None: raise AssertionError("sidecar authority unavailable") store = WeeklyOperationsStore.for_authority(authority, _CUSTOMER) events = _events() runtime = _runtime(root / "canonical", events) canonical: CanonicalCheckinCustomerAuthority | None = None with begin_canonical_authority_registration(runtime, authority) as registration: canonical = registration.authority _ = registration.binding registration.acknowledge_binding() if canonical is None: raise AssertionError("canonical authority unavailable") _append_week(canonical, store) summary = build_weekly_operations_summary(canonical, store, date(2026, 8, 17)) return bind_weekly_summary_for_owner_draft(canonical, store, summary) def _runtime(root: Path, events: tuple[Event, ...]) -> CustomerRuntime: root.mkdir(parents=True, mode=0o700) registry = root / "registry.json" weeks = [{"week": week, "calories_kcal": 2300, "protein_g": 150, "meal_structure": ["a", "b", "c"]} for week in range(1, 13)] payload = { "version": 1, "owner": {"user_id": "owner", "chat_id": "owner", "topic_id": "owner"}, "customers": [{ "customer_key": str(_CUSTOMER), "display_name": "fixture", "enabled": False, "telegram": {"user_id": "user", "chat_id": "chat", "topic_id": "topic"}, "schedule": {"daily_time": "08:00", "weekly_weekday": 0, "monthly_day": 1}, "profile": {"primary_goal": "fixture", "sleep_goal_hours": 8}, "ai_processing_consent": {"granted": True, "recorded_on": "2026-08-01", "notice_version": "privacy-v1"}, "plan": {"starts_on": "2026-08-01", "focus": "nutrition_90_training_10", "weeks": weeks}, }], } _ = registry.write_text(json.dumps(payload), encoding="utf-8") runtime = load_customer_registry(registry, root).customers[0] runtime.customer_root.mkdir(parents=True, mode=0o700) runtime.wizard_root.mkdir(parents=True, mode=0o700) runtime.nutrition_plans_root.mkdir(parents=True, mode=0o700) transaction = CanonicalEventTransaction.for_customer_runtime(runtime) _ = transaction.append_many(events) return runtime def _event(event_id: str, occurred_at: str, weight: float) -> Event: return Event( event_id=event_id, event_type=EventType.NUTRITION_CHECKIN, occurred_at_kst=occurred_at, recorded_at_kst=occurred_at, schema_version="2.0", provenance=Provenance( source_type="telegram", source_ref=f"wizard:{event_id}", content_sha256=hashlib.sha256(event_id.encode()).hexdigest(), ), status=ContractStatus.ACCEPTED, supersedes=None, dedupe_key=f"dedupe:{event_id}", check_in=ContractCheckin(body_weight_kg=weight), safety=None, ) def _events() -> tuple[Event, ...]: return ( _event("prior-mon", "2026-08-10T08:00:00+09:00", 80.0), _event("prior-tue", "2026-08-11T08:00:00+09:00", 80.1), _event("prior-fri", "2026-08-14T08:00:00+09:00", 80.2), _event("current-mon", "2026-08-17T08:00:00+09:00", 81.1), _event("current-tue", "2026-08-18T08:00:00+09:00", 81.2), _event("current-thu", "2026-08-20T08:00:00+09:00", 81.3), _event("current-fri", "2026-08-21T23:59:59+09:00", 81.4), _event("current-sun", "2026-08-23T08:00:00+09:00", 81.5), ) def _append_week(canonical: CanonicalCheckinCustomerAuthority, store: WeeklyOperationsStore) -> None: statuses = ( (date(2026, 8, 10), DayState.SUBMITTED, 1, "prior-mon", None), (date(2026, 8, 11), DayState.SUBMITTED, 2, "prior-tue", None), (date(2026, 8, 12), DayState.MISSED, 2, None, None), (date(2026, 8, 13), DayState.MISSED, 2, None, None), (date(2026, 8, 14), DayState.SUBMITTED, 3, "prior-fri", None), (date(2026, 8, 15), DayState.MISSED, 3, None, None), (date(2026, 8, 16), DayState.MISSED, 3, None, None), (date(2026, 8, 17), DayState.SUBMITTED, 4, "current-mon", None), (date(2026, 8, 18), DayState.SUBMITTED, 5, "current-tue", "01"), (date(2026, 8, 19), DayState.MISSED, 5, None, "02"), (date(2026, 8, 20), DayState.SUBMITTED, 6, "current-thu", None), (date(2026, 8, 21), DayState.MISSED, 6, None, "03"), (date(2026, 8, 21), DayState.LATE_SUBMITTED, 7, "current-fri", None), (date(2026, 8, 22), DayState.MISSED, 7, None, "04"), (date(2026, 8, 23), DayState.SUBMITTED, 8, "current-sun", None), ) with canonical.read_locked() as snapshot: digests = {str(row["event_id"]): str(row["event_digest"]) for row in snapshot.sequence_rows} for day, state, sequence, event_id, reminder_id in statuses: with canonical.read_locked() as snapshot: pin = canonical_prefix_pin(snapshot, sequence) source = None if event_id is not None: from checkin_cli.weekly_operations import SourceLineage source = SourceLineage(event_id, digests[event_id]) _ = store.append(WeeklyOperationInput.for_customer( _CUSTOMER, day, state, pin, datetime.combine(day, datetime.min.time(), _KST), source, None if reminder_id is None else ReminderIdentity(f"reservation-{reminder_id}", f"audit-{reminder_id}"), ))