from __future__ import annotations from collections.abc import Mapping from copy import deepcopy from datetime import date from pathlib import Path from typing import cast from checkin_cli.nutrition_onboarding_authority import ( requires_safety_hold, validate_authority, validate_current_registry_authority, validate_route, ) from checkin_cli.nutrition_onboarding_contract import ( QUESTION_FIELDS, MessageEvidence, OnboardingAuthority, OnboardingStatus, canonical_digest, ) from checkin_cli.nutrition_onboarding_contract import ( example_answer as example_answer, ) from checkin_cli.nutrition_onboarding_finalize_service import ( NutritionOnboardingFinalizationMixin, ) from checkin_cli.nutrition_onboarding_fs import ( atomic_write_private_json, read_private_json, validate_profile_path, ) from checkin_cli.nutrition_onboarding_models import ( NUTRITION_ONBOARDING_API_VERSION as NUTRITION_ONBOARDING_API_VERSION, ) from checkin_cli.nutrition_onboarding_models import ( OnboardingState, build_onboarding_baseline, ) from checkin_cli.nutrition_onboarding_reconciliation import ( advance_reconciliation, build_reconciliation, load_reconciliation, require_resolved_reconciliation, ) from checkin_cli.nutrition_onboarding_session import ( build_status, load_mutable_session, save_session, session_cursor, ) from checkin_cli.nutrition_onboarding_store import NutritionOnboardingStore def record_current_registry_consent( profile_root: Path, customer_key: str, *, recorded_on: date, ) -> None: from checkin_cli.customer_admin import set_customer_ai_consent from checkin_cli.customer_coaching import AiProcessingConsent set_customer_ai_consent( Path(profile_root) / "customers" / "registry.json", customer_key, AiProcessingConsent( granted=True, recorded_on=recorded_on, notice_version="privacy-v1", ), ) class NutritionOnboardingService(NutritionOnboardingFinalizationMixin): def __init__( self, *, profile_root: Path, customer_key: str, enforce_current_authority: bool = True, enforce_reconciliation: bool = False, ) -> None: self.profile_root = Path(profile_root) self.customer_key = customer_key self.enforce_current_authority = enforce_current_authority self.enforce_reconciliation = enforce_reconciliation self.root = ( self.profile_root / "data" / "customers" / customer_key / "nutrition-onboarding" ) validate_profile_path(self.root, self.profile_root) self.store = NutritionOnboardingStore( root=self.root, profile_root=self.profile_root, ) self.session_path = self.root / "transient" / "workflow.json" self.baseline_path = self.root / "baseline-candidate.json" self.ready_path = self.root / "ready.json" def _validate_storage(self) -> None: validate_profile_path(self.root, self.profile_root) def _validate_current_authority(self, authority: OnboardingAuthority) -> None: if self.enforce_current_authority: try: validate_current_registry_authority(self.profile_root, authority) except (OSError, ValueError) as exc: raise ValueError( "current onboarding registry authority is unavailable" ) from exc def start_or_resume( self, *, authority: OnboardingAuthority, evidence: MessageEvidence, consent_receipt_digest: str | None = None, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) validate_authority(authority, customer_key=self.customer_key) validate_route(authority, evidence, role="customer") self.store.initialize() consent_digest = consent_receipt_digest or canonical_digest( { "authority": authority.model_dump(mode="json"), "message_id": evidence.message_id, "update_id": evidence.update_id, } ) if len(consent_digest) != 64: raise ValueError("consent receipt digest is invalid") if self.ready_path.exists(): return self.status() if self.session_path.exists(): return self.status() document: dict[str, object] = { "schema_version": "2.0", "review_flow": "owner_v1", "customer_key": self.customer_key, "state": OnboardingState.COLLECTING.value, "cursor": 0, "answers": {}, "authority_digest": canonical_digest(authority.model_dump(mode="json")), "consumed_updates": [evidence.update_id], "owner_reviewed": False, "baseline_digest": None, "privacy_consent_digest": consent_digest, } atomic_write_private_json(self.session_path, document) return build_status(customer_key=self.customer_key, document=document) def submit_answer( self, *, field: str, value: object, authority: OnboardingAuthority, evidence: MessageEvidence, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) document = load_mutable_session( session_path=self.session_path, customer_key=self.customer_key, authority=authority, evidence=evidence, role="customer", ) cursor = session_cursor(document) if cursor >= len(QUESTION_FIELDS) or QUESTION_FIELDS[cursor] != field: raise ValueError(f"expected question: {QUESTION_FIELDS[cursor]}") answers = document["answers"] if not isinstance(answers, dict): raise ValueError("invalid onboarding answers") answers = cast(dict[str, object], answers) answers[field] = value document["cursor"] = cursor + 1 if cursor + 1 == len(QUESTION_FIELDS): document["state"] = OnboardingState.CUSTOMER_ATTESTATION.value save_session(session_path=self.session_path, document=document, evidence=evidence) return build_status(customer_key=self.customer_key, document=document) def attest_baseline( self, *, authority: OnboardingAuthority, evidence: MessageEvidence, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) document = load_mutable_session( session_path=self.session_path, customer_key=self.customer_key, authority=authority, evidence=evidence, role="customer", ) if document["state"] != OnboardingState.CUSTOMER_ATTESTATION.value: raise ValueError("customer attestation is not available") raw_answers = document.get("answers") if not isinstance(raw_answers, dict): raise ValueError("invalid onboarding answers") answers = deepcopy(cast(dict[str, object], raw_answers)) reconciliation = None if self.enforce_reconciliation: reconciliation = require_resolved_reconciliation( document.get("reconciliation"), answers_digest=canonical_digest(answers), ) source_answers = deepcopy(answers) answers.update( { "schema_version": "1.0", "customer_key": self.customer_key, "session_status": "completed", "review_decision": "pending", "publication_status": "unpublished", } ) baseline = build_onboarding_baseline(answers, as_of=date.today()) payload = baseline.model_dump(mode="json") baseline_digest = canonical_digest(payload) atomic_write_private_json( self.baseline_path, { "schema_version": "1.0", "digest": baseline_digest, "baseline": payload, "source_answers": source_answers, **( {"input_reconciliation_digest": reconciliation["digest"]} if reconciliation is not None else {} ), }, ) document["answers"] = {} document["baseline_digest"] = baseline_digest document["state"] = ( OnboardingState.SAFETY_HOLD.value if requires_safety_hold(baseline) else OnboardingState.OWNER_REVIEW.value ) save_session(session_path=self.session_path, document=document, evidence=evidence) return build_status(customer_key=self.customer_key, document=document) def _review_receipt( self, *, role: str, decision: str, baseline_digest: object, evidence: MessageEvidence, ) -> str: if not isinstance(baseline_digest, str) or len(baseline_digest) != 64: raise ValueError("review baseline digest is invalid") return canonical_digest( { "schema_version": "1.0", "customer_key": self.customer_key, "role": role, "decision": decision, "baseline_digest": baseline_digest, "actor_user_id": evidence.actor_user_id, "chat_id": evidence.chat_id, "topic_id": evidence.topic_id, "message_id": evidence.message_id, "update_id": evidence.update_id, } ) def _reopen_revision( self, document: dict[str, object], *, field: str | None, ) -> None: if field not in QUESTION_FIELDS: raise ValueError("revision field must be a canonical onboarding field") candidate = read_private_json(self.baseline_path) source_answers = candidate.get("source_answers") if not isinstance(source_answers, dict): raise ValueError("revision source answers are unavailable") typed_answers = cast(Mapping[str, object], source_answers) cursor = QUESTION_FIELDS.index(field) retained = { name: deepcopy(typed_answers[name]) for name in QUESTION_FIELDS[:cursor] if name in typed_answers } if tuple(retained) != QUESTION_FIELDS[:cursor]: raise ValueError("revision source answers are incomplete") document.update( state=OnboardingState.COLLECTING.value, cursor=cursor, answers=retained, baseline_digest=None, owner_reviewed=False, owner_review_receipt=None, ) document.pop("reconciliation", None) self.baseline_path.unlink() (self.root / "clinical-review.json").unlink(missing_ok=True) def review_as_owner( self, *, decision: str, authority: OnboardingAuthority, evidence: MessageEvidence, revision_field: str | None = None, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) document = load_mutable_session( session_path=self.session_path, customer_key=self.customer_key, authority=authority, evidence=evidence, role="owner", ) if document["state"] != OnboardingState.OWNER_REVIEW.value: raise ValueError("owner review is not available") if decision == "approved": document.update( state=OnboardingState.FINALIZING.value, owner_reviewed=True, owner_review_receipt=self._review_receipt( role="owner", decision=decision, baseline_digest=document.get("baseline_digest"), evidence=evidence, ), ) elif decision == "revise": self._reopen_revision(document, field=revision_field) elif decision == "rejected": document.update( state=OnboardingState.REJECTED.value, owner_reviewed=False, owner_review_receipt=None, ) elif decision == "safety_hold": document.update( state=OnboardingState.SAFETY_HOLD.value, owner_reviewed=False, owner_review_receipt=None, ) else: raise ValueError("owner review decision is invalid") save_session(session_path=self.session_path, document=document, evidence=evidence) return build_status(customer_key=self.customer_key, document=document) def record_clinical_review( self, *, cleared: bool, external_reference: str, authority: OnboardingAuthority, evidence: MessageEvidence, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) document = load_mutable_session( session_path=self.session_path, customer_key=self.customer_key, authority=authority, evidence=evidence, role="owner", ) if document["state"] != OnboardingState.SAFETY_HOLD.value: raise ValueError("clinical review is not available") reference = external_reference.strip() if not reference or len(reference) > 200: raise ValueError("opaque external clinical reference is required") receipt = { "schema_version": "1.0", "cleared_for_nonmedical_coaching": cleared, "external_reference": reference, "baseline_digest": document["baseline_digest"], } receipt["digest"] = canonical_digest(receipt) atomic_write_private_json(self.root / "clinical-review.json", receipt) if cleared: document["state"] = OnboardingState.OWNER_REVIEW.value save_session(session_path=self.session_path, document=document, evidence=evidence) return build_status(customer_key=self.customer_key, document=document) def cancel( self, *, authority: OnboardingAuthority, evidence: MessageEvidence, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) document = load_mutable_session( session_path=self.session_path, customer_key=self.customer_key, authority=authority, evidence=evidence, role="customer", ) document.update( state=OnboardingState.CANCELLED.value, cursor=0, answers={}, baseline_digest=None, owner_reviewed=False, ) self.baseline_path.unlink(missing_ok=True) save_session(session_path=self.session_path, document=document, evidence=evidence) return build_status(customer_key=self.customer_key, document=document) def status(self) -> OnboardingStatus: self._validate_storage() return build_status( customer_key=self.customer_key, document=read_private_json( self.session_path if self.session_path.exists() else self.ready_path ), ) def reconciliation_answers( self, *, authority: OnboardingAuthority, ) -> dict[str, object]: self._validate_storage() self._validate_current_authority(authority) validate_authority(authority, customer_key=self.customer_key) document = read_private_json(self.session_path) expected = canonical_digest(authority.model_dump(mode="json")) if document.get("authority_digest") != expected: raise ValueError("onboarding authority is stale") if ( document.get("state") != OnboardingState.CUSTOMER_ATTESTATION.value ): raise ValueError("customer attestation is not available") answers = document.get("answers") if not isinstance(answers, dict): raise ValueError("invalid onboarding answers") return deepcopy(cast(dict[str, object], answers)) def record_reconciliation( self, *, answers_digest: str, advisory: dict[str, object], clarifications: list[dict[str, object]], authority: OnboardingAuthority, ) -> dict[str, object]: self._validate_storage() self._validate_current_authority(authority) validate_authority(authority, customer_key=self.customer_key) document = read_private_json(self.session_path) expected = canonical_digest(authority.model_dump(mode="json")) if document.get("authority_digest") != expected: raise ValueError("onboarding authority is stale") if document.get("state") != OnboardingState.CUSTOMER_ATTESTATION.value: raise ValueError("customer attestation is not available") answers = document.get("answers") if not isinstance(answers, dict): raise ValueError("invalid onboarding answers") if canonical_digest(answers) != answers_digest: raise ValueError("reconciliation answers digest is stale") existing = document.get("reconciliation") if existing is not None: record = load_reconciliation(existing) if record["answers_digest"] == answers_digest: return deepcopy(record) raise ValueError("conflicting reconciliation record") record = build_reconciliation( answers_digest=answers_digest, advisory=advisory, clarifications=clarifications, ) document["reconciliation"] = record atomic_write_private_json(self.session_path, document) return deepcopy(record) def reconciliation_record( self, *, authority: OnboardingAuthority, ) -> dict[str, object] | None: self._validate_storage() self._validate_current_authority(authority) validate_authority(authority, customer_key=self.customer_key) document = read_private_json(self.session_path) expected = canonical_digest(authority.model_dump(mode="json")) if document.get("authority_digest") != expected: raise ValueError("onboarding authority is stale") value = document.get("reconciliation") return None if value is None else deepcopy(load_reconciliation(value)) def replace_stale_reconciliation( self, *, expected_stale_answers_digest: str, answers_digest: str, advisory: dict[str, object], clarifications: list[dict[str, object]], authority: OnboardingAuthority, ) -> dict[str, object]: self._validate_storage() self._validate_current_authority(authority) validate_authority(authority, customer_key=self.customer_key) with self.store.locked(): document = read_private_json(self.session_path) expected = canonical_digest(authority.model_dump(mode="json")) if document.get("authority_digest") != expected: raise ValueError("onboarding authority is stale") if document.get("state") != OnboardingState.CUSTOMER_ATTESTATION.value: raise ValueError("customer attestation is not available") answers = document.get("answers") if not isinstance(answers, dict): raise ValueError("invalid onboarding answers") if canonical_digest(answers) != answers_digest: raise ValueError("reconciliation answers digest is stale") existing = document.get("reconciliation") if existing is None: raise ValueError("stale reconciliation record is unavailable") current = load_reconciliation(existing) if current["answers_digest"] == answers_digest: return deepcopy(current) if current["answers_digest"] != expected_stale_answers_digest: raise ValueError("stale reconciliation record changed") record = build_reconciliation( answers_digest=answers_digest, advisory=advisory, clarifications=clarifications, ) document["reconciliation"] = record atomic_write_private_json(self.session_path, document) return deepcopy(record) def revise_reconciliation_answer( self, *, field: str, value: object, expected_answers_digest: str, authority: OnboardingAuthority, evidence: MessageEvidence, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(authority) if field not in QUESTION_FIELDS: raise ValueError("reconciliation field is not canonical") document = load_mutable_session( session_path=self.session_path, customer_key=self.customer_key, authority=authority, evidence=evidence, role="customer", ) if document["state"] != OnboardingState.CUSTOMER_ATTESTATION.value: raise ValueError("customer attestation is not available") answers = document.get("answers") if not isinstance(answers, dict): raise ValueError("invalid onboarding answers") answers = cast(dict[str, object], answers) if canonical_digest(answers) != expected_answers_digest: raise ValueError("reconciliation answers digest is stale") reconciliation = document.get("reconciliation") if self.enforce_reconciliation and reconciliation is None: raise ValueError("input reconciliation is unavailable") answers[field] = value if reconciliation is not None: document["reconciliation"] = advance_reconciliation( reconciliation, field=field, answers_digest=canonical_digest(answers), ) save_session( session_path=self.session_path, document=document, evidence=evidence, ) return build_status( customer_key=self.customer_key, document=document, )