"""Weekly owner lifecycle hosted beside the legacy coordinator.""" from __future__ import annotations import hashlib import json from datetime import date from collections.abc import Mapping from typing import TYPE_CHECKING, Protocol from checkin_cli.customer_coaching import CustomerRegistry, CustomerRuntime from checkin_cli.customer_grounding import CustomerSnapshot from checkin_cli.weekly_operations_grounding import GroundedWeeklyReviewSource from .nutrition_coaching_proposal import CoachReview, targets_from_plan from gateway.platforms.nutrition_weekly_host_source import ( PreparedWeeklySource, WeeklyReviewSource, WeeklySourceAuthority, WeeklySourceRejected, grounded_source, prepare_weekly_source, ) from .nutrition_weekly_owner_card import render_weekly_owner_card from .nutrition_weekly_owner_contract import WeeklyLifecycleResult from .nutrition_weekly_owner_draft import ( WeeklyOwnerDraftRequest, WeeklyOwnerDraftResult, WeeklyOwnerDraftService, ) from .nutrition_weekly_owner_model import WeeklyExplanationModel from .nutrition_weekly_owner_storage import WeeklyOwnerStorageAuthority if TYPE_CHECKING: from .nutrition_coaching import ( DraftAction, DraftGenerationClaim, DraftGenerationError, IncomingAddress, ) class WeeklyDraftSelection(Protocol): @property def customer(self) -> CustomerRuntime: ... @property def snapshot(self) -> CustomerSnapshot: ... class WeeklyCoordinator(Protocol): @property def owner(self) -> IncomingAddress: ... @property def registry(self) -> CustomerRegistry: ... def customer(self, customer_key: str) -> CustomerRuntime | None: ... def weekly_session_id(self, customer_key: str) -> str | None: ... def weekly_request(self, token: str) -> tuple[str, str] | None: ... def weekly_save( self, token: str, key: str, session: str ) -> tuple[str, str]: ... def create_draft( self, token: str, owner: IncomingAddress, text: str, *, expected_revision_binding_digest: str | None = None, coach_review: CoachReview | None = None, coach_artifacts: Mapping[str, str | bool | None] | None = None, generation_claim: DraftGenerationClaim | None = None, owner_card_text: str | None = None, ) -> DraftAction: ... def draft(self, draft_id: str, owner: IncomingAddress) -> DraftAction: ... def claim_draft_generation(self, token: str, owner: IncomingAddress, worker_id: str) -> DraftGenerationClaim | None: ... def resolve_draft( self, token: str, owner: IncomingAddress ) -> WeeklyDraftSelection | None: ... def release_draft_generation( self, claim: DraftGenerationClaim, error: DraftGenerationError ) -> bool: ... def complete_draft_generation( self, claim: DraftGenerationClaim, generation_provider_receipt: str ) -> bool: ... class WeeklyOwnerHost: """Own weekly source dispatch and grounded draft persistence.""" def __init__( self, coordinator: WeeklyCoordinator, storage: WeeklyOwnerStorageAuthority, ) -> None: self._coordinator: WeeklyCoordinator = coordinator self._storage: WeeklyOwnerStorageAuthority = storage def owner_key(self) -> tuple[str, str, str]: user_id, chat_id, topic_id = self._coordinator.owner.key return str(user_id), str(chat_id), str(topic_id) def create_review_draft( self, customer_key: str, owner: IncomingAddress, source: WeeklyReviewSource, ) -> DraftAction: from .nutrition_coaching import DraftAction, DraftLedgerError coordinator = self._coordinator if owner.key != coordinator.registry.owner.key: return DraftAction(False, error="owner_only") customer = coordinator.customer(str(customer_key or "").strip()) if customer is None: return DraftAction(False, error="customer_not_registered") try: prepared = prepare_weekly_source( source, WeeklySourceAuthority( customer.spec.customer_key, owner.key, self._storage ), ) except WeeklySourceRejected as error: return DraftAction(False, error=error.error) session_id = coordinator.weekly_session_id(customer.spec.customer_key) if not session_id: return DraftAction(False, error="weekly_review_snapshot_unavailable") registered_key = customer.spec.customer_key token = hashlib.sha256( f"weekly-review:{registered_key}:{prepared.source_digest}".encode("utf-8") ).hexdigest()[:16] try: existing = coordinator.weekly_request(token) if existing is not None and existing[0] != registered_key: return DraftAction(False, error="weekly_review_request_conflict") if existing is None: _ = coordinator.weekly_save(token, registered_key, session_id) except DraftLedgerError: return DraftAction(False, error="draft_ledger_corrupt") grounded = grounded_source(prepared) if grounded is None: return coordinator.create_draft(token, owner, prepared.text) return self._create_grounded(token, owner, prepared, grounded) def _create_grounded( self, token: str, owner: IncomingAddress, prepared: PreparedWeeklySource, source: GroundedWeeklyReviewSource, ) -> DraftAction: from .nutrition_coaching import DraftAction, DraftGenerationError coordinator = self._coordinator existing = coordinator.draft(token, owner) if existing.accepted: return existing claim = coordinator.claim_draft_generation(token, owner, "weekly-owner-v1") selection = coordinator.resolve_draft(token, owner) if claim is None or selection is None: return DraftAction(False, draft_id=token, error="weekly_review_generation_unavailable") try: day = date.fromisoformat(selection.snapshot.kst_day) plan = selection.customer.plan_week(day).model_dump(exclude={"week"}) targets = targets_from_plan(plan) except (AttributeError, KeyError, TypeError, ValueError): targets = None if targets is None: _ = coordinator.release_draft_generation( claim, DraftGenerationError( "grounding_unavailable", "weekly owner targets are unavailable", False, ), ) return DraftAction(False, draft_id=token, error="weekly_review_generation_unavailable") review = CoachReview( "nutrition-coach-review-v3", targets, targets, "observe", "low", ("weekly-aggregate",), ("next-week-review",), "주간 운영 집계에 대한 소유자 검토입니다.", (), tuple((key, str(value)) for key, value in source.facts.machine_payload().items())[:32], ("변경 여부는 소유자가 검토하고 승인합니다.",), claim.checkin_revision, ) raw = json.dumps({ "schema": "nutricoach-weekly-owner-review-v1", "grounding_digest": source.facts.grounding_digest, "principle_ids": [item.value for item in source.principle_ids], "emphasis": source.emphasis.value, "decision": "observe", "responsibility": "owner_review", }, ensure_ascii=False, sort_keys=True, separators=(",", ":")) receipt = hashlib.sha256(raw.encode()).hexdigest() text = " ".join(prepared.text.split()) artifacts = { "schema_version": "nutrition-coach-artifacts-v2", "revision_binding_digest": claim.checkin_revision, "raw_coach_output": raw, "raw_coach_sha256": receipt, "accepted_coach_output": text, "raw_polish_output": None, "accepted_polish_output": text, "polish_valid": False, } owner_card_text = render_weekly_owner_card(DraftAction( True, token, text, "created", coach_review=review, )) created = coordinator.create_draft( token, owner, text, expected_revision_binding_digest=claim.checkin_revision, coach_review=review, coach_artifacts=artifacts, generation_claim=claim, owner_card_text=owner_card_text, ) if not created.accepted: return created if not coordinator.complete_draft_generation(claim, receipt): return DraftAction(False, draft_id=token, error="generation_receipt_failed") return coordinator.draft(token, owner) def persist_review( self, customer_key: str, owner_key: tuple[str, str, str], source: GroundedWeeklyReviewSource, ) -> WeeklyLifecycleResult: if owner_key != self.owner_key(): return WeeklyLifecycleResult(False, error="owner_only") return _lifecycle_result( self.create_review_draft(customer_key, self._coordinator.owner, source) ) def review_result( self, draft_id: str, owner_key: tuple[str, str, str] ) -> WeeklyLifecycleResult: if owner_key != self.owner_key(): return WeeklyLifecycleResult(False, error="owner_only") return _lifecycle_result(self._coordinator.draft(draft_id, self._coordinator.owner)) def create_owner_draft( self, request: WeeklyOwnerDraftRequest, model: WeeklyExplanationModel ) -> WeeklyOwnerDraftResult: return WeeklyOwnerDraftService(self, model).create(request) def weekly_owner_key(self) -> tuple[str, str, str]: return self.owner_key() def weekly_owner_storage_authority(self) -> WeeklyOwnerStorageAuthority: self._storage.verify() return self._storage def persist_grounded_weekly_review( self, customer_key: str, owner_key: tuple[str, str, str], source: GroundedWeeklyReviewSource, ) -> WeeklyLifecycleResult: return self.persist_review(customer_key, owner_key, source) def grounded_weekly_review_result( self, draft_id: str, owner_key: tuple[str, str, str] ) -> WeeklyLifecycleResult: return self.review_result(draft_id, owner_key) def _lifecycle_result(action: DraftAction) -> WeeklyLifecycleResult: return WeeklyLifecycleResult( action.accepted, action.draft_id, action.text, action.error, action.generation, action.generation_record_digest, action.generation_checkin_revision, action.generation_draft_revision, )