from __future__ import annotations from collections.abc import Mapping from copy import deepcopy from datetime import date from pathlib import Path import re 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_clarification_policy import canonicalize_answer from checkin_cli.nutrition_onboarding_calculations import ( UnsafeGoalTrajectoryError as UnsafeGoalTrajectoryError, ) 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_finalization import ( preflight_initial_plan, ) 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, canonical_equation_sex_basis, ) from checkin_cli.nutrition_onboarding_owner_risk import ( OwnerRiskAcceptanceBinding as OwnerRiskAcceptanceBinding, ) from checkin_cli.nutrition_onboarding_owner_risk import ( OwnerRiskAcceptanceRequest as OwnerRiskAcceptanceRequest, ) from checkin_cli.nutrition_onboarding_owner_risk import ( owner_risk_acceptance_identity as owner_risk_acceptance_identity, ) from checkin_cli.nutrition_onboarding_owner_risk import ( record_owner_risk_acceptance, ) from checkin_cli.nutrition_onboarding_reconciliation import ( build_reconciliation, load_reconciliation, recompute_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 preflight_owner_approval( self, *, starts_on: date, ) -> None: self._validate_storage() preflight_initial_plan( baseline_candidate_path=self.baseline_path, starts_on=starts_on, ) 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] = canonicalize_answer(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_issue_id: str | None = None, revision_field: str | None = None, expected_answers_digest: str | None = None, expected_reconciliation_digest: 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": binding = ( revision_issue_id, revision_field, expected_answers_digest, expected_reconciliation_digest, ) if any(item is not None for item in binding): if not all(isinstance(item, str) for item in binding): raise ValueError("revision binding is incomplete") 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") digest_pattern = re.compile(r"^[0-9a-f]{64}$") if not all( digest_pattern.fullmatch(cast(str, item)) for item in ( revision_issue_id, expected_answers_digest, expected_reconciliation_digest, ) ): raise ValueError("revision binding digest is invalid") if canonical_digest(source_answers) != expected_answers_digest: raise ValueError("revision answers digest is stale") current = load_reconciliation(document.get("reconciliation")) visible_ids = current.get("visible_issue_ids") raw_issues = current.get("issues") if ( current["digest"] != expected_reconciliation_digest or not isinstance(visible_ids, list) or revision_issue_id not in visible_ids or not isinstance(raw_issues, list) ): raise ValueError("revision reconciliation binding is stale") issue = next( ( cast(dict[str, object], item) for item in raw_issues if isinstance(item, dict) and cast(dict[str, object], item).get("issue_id") == revision_issue_id ), None, ) if issue is None or issue.get("field") != revision_field: raise ValueError("revision issue binding is stale") if candidate.get("input_reconciliation_digest") != expected_reconciliation_digest: raise ValueError("revision reconciliation digest is stale") 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 record_owner_risk_acceptance( self, *, request: OwnerRiskAcceptanceRequest, ) -> OnboardingStatus: self._validate_storage() self._validate_current_authority(request.authority) return record_owner_risk_acceptance( self, request=request, ) 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, reference_date: date | None = None, ) -> 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=cast(Mapping[str, object], answers), answers_digest=answers_digest, advisory=advisory, reference_date=reference_date or date.today(), ) 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, reference_date: date | None = None, ) -> 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=cast(Mapping[str, object], answers), answers_digest=answers_digest, advisory=advisory, reference_date=reference_date or date.today(), ) 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, expected_issue_id: str | None = None, expected_reconciliation_digest: str | None = None, 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") if expected_issue_id is not None or expected_reconciliation_digest is not None: if reconciliation is None: raise ValueError("input reconciliation is unavailable") current = load_reconciliation(reconciliation) issues = current.get("issues") first_issue = ( cast(dict[str, object], issues[0]) if isinstance(issues, list) and issues and isinstance(issues[0], dict) else None ) if ( first_issue is None or first_issue.get("issue_id") != expected_issue_id or first_issue.get("field") != field or current["digest"] != expected_reconciliation_digest ): raise ValueError("reconciliation revision binding is stale") value = canonicalize_answer(field, value) if field == "equation_sex_basis" and canonical_equation_sex_basis(value) is None: raise ValueError( "equation_sex_basis must resolve to male, female, or decline" ) answers[field] = value if reconciliation is not None: document["reconciliation"] = recompute_reconciliation( reconciliation, answers=answers, 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, )