"""Explicit operator cutover from committed customer activation to ACTIVE bootstrap.""" from __future__ import annotations import importlib import hashlib import hmac import json import os import stat import sys from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, date, datetime, time, timedelta from pathlib import Path from typing import Callable, ContextManager, Protocol, cast from .telegram_customer_bootstrap import ( BootstrapConflict, BootstrapError, BootstrapState, GenerationConflict, Role, RoomBootstrapSession, RoomBootstrapStore, room_bootstrap_state_dir, ) class ActivationCutoverError(BootstrapError): pass class _RuntimeAuthority(Protocol): def authorize( self, candidate: str, stage: str, predecessor: object | None = None ) -> ContextManager[dict[str, object]]: ... class _CustomerAdmin(Protocol): __file__: str | None def activate_customer( self, profile_root: Path, data_root: Path, customer_id: str, checklist_evidence_path: Path, staff_membership_evidence_path: Path, *, kst_date: date | None, ) -> object: ... def validate_committed_activation( self, profile_root: Path, registry_path: Path | None = None, customer_id: str | None = None, ) -> bool: ... @dataclass(frozen=True, slots=True) class ActivationCutoverResult: customer_id: str session_id: str generation: int state: BootstrapState reconciled: bool runtime_authority_snapshot: dict[str, object] | None = None def to_dict(self) -> dict[str, object]: return { "customer_id": self.customer_id, "session_id": self.session_id, "generation": self.generation, "state": self.state.value, "reconciled": self.reconciled, "runtime_authority_snapshot": self.runtime_authority_snapshot, } def activate_customer_cutover( profile_root: Path | str, data_root: Path | str, customer_id: str, checklist_evidence_path: Path | str, staff_membership_evidence_path: Path | str, *, bootstrap_session_id: str, expected_generation: int, deployment_receipt_path: Path | str, package_root: Path | str | None = None, kst_date: date | None = None, task26_authority_source: _RuntimeAuthority | None = None, task26_candidate_digest: str | None = None, task26_authority_predecessor: object | None = None, task26_runtime_required: bool = False, ) -> ActivationCutoverResult: """Hold external Task26 authority through activation and bootstrap mutation.""" if task26_authority_source is None: if task26_runtime_required: raise ActivationCutoverError("Task26 external runtime authority is required") return _activate_customer_cutover_authorized( profile_root, data_root, customer_id, checklist_evidence_path, staff_membership_evidence_path, bootstrap_session_id=bootstrap_session_id, expected_generation=expected_generation, deployment_receipt_path=deployment_receipt_path, package_root=package_root, kst_date=kst_date, ) if task26_candidate_digest is None: raise ActivationCutoverError("Task26 candidate identity is unavailable") try: with task26_authority_source.authorize( task26_candidate_digest, "activation", task26_authority_predecessor, ) as snapshot: result = _activate_customer_cutover_authorized( profile_root, data_root, customer_id, checklist_evidence_path, staff_membership_evidence_path, bootstrap_session_id=bootstrap_session_id, expected_generation=expected_generation, deployment_receipt_path=deployment_receipt_path, package_root=package_root, kst_date=kst_date, ) return ActivationCutoverResult( result.customer_id, result.session_id, result.generation, result.state, result.reconciled, snapshot, ) except ValueError as exc: raise ActivationCutoverError("Task26 activation authority rejected") from exc def _activate_customer_cutover_authorized( profile_root: Path | str, data_root: Path | str, customer_id: str, checklist_evidence_path: Path | str, staff_membership_evidence_path: Path | str, *, bootstrap_session_id: str, expected_generation: int, deployment_receipt_path: Path | str, package_root: Path | str | None = None, kst_date: date | None = None, ) -> ActivationCutoverResult: """Commit customer activation, then its exact bootstrap lifecycle.""" root = _profile_root(profile_root) state_dir = room_bootstrap_state_dir(root) if not state_dir.is_dir() or not (state_dir / "ledger.json").is_file(): raise ActivationCutoverError("customer bootstrap authority is unavailable") store = RoomBootstrapStore(state_dir) session = _preflight_bootstrap( store, bootstrap_session_id, expected_generation=expected_generation, customer_id=customer_id, ) validate_first_claim_owner_review(root, session) admin = _profile_admin(root, package_root) registry_path = _admin_registry_path(admin, root) document = _admin_read(admin, registry_path) customer = _preflight_registry(document, session, customer_id) enabled = getattr(customer, "enabled", None) if enabled is True: _validate_committed(admin, root, registry_path, customer_id) reconciled = session.state is not BootstrapState.ACTIVE elif enabled is False: if session.state is BootstrapState.ACTIVE: raise ActivationCutoverError( "active bootstrap has no committed customer activation" ) _preflight_membership_evidence( root, registry_path, Path(staff_membership_evidence_path), Path(deployment_receipt_path), session, customer_id=customer_id, ) _ = admin.activate_customer( root, Path(data_root), customer_id, Path(checklist_evidence_path), Path(staff_membership_evidence_path), kst_date=kst_date, ) _validate_committed(admin, root, registry_path, customer_id) reconciled = False else: raise ActivationCutoverError("customer activation state is invalid") _finalize_membership_activation_binding( root, Path(staff_membership_evidence_path), customer_id=customer_id, ) committed = _activate_bootstrap( store, session, customer_id=customer_id, ) committed_document = _admin_read(admin, registry_path) committed_customer = _preflight_registry( committed_document, committed, customer_id, ) _reserve_activation_completion_notice( root, committed_customer, committed, membership_evidence_path=Path(staff_membership_evidence_path), ) return ActivationCutoverResult( customer_id, committed.session_id, committed.generation, committed.state, reconciled, ) def _preflight_bootstrap( store: RoomBootstrapStore, session_id: str, *, expected_generation: int, customer_id: str, ) -> RoomBootstrapSession: session = store.get(session_id) allowed_generation = ( {session.generation, session.generation - 1} if session.state is BootstrapState.ACTIVE else {session.generation} ) customer = session.role_claim(Role.CUSTOMER) if ( expected_generation not in allowed_generation or session.customer_key != customer_id or session.state not in {BootstrapState.AWAITING_ACTIVATION, BootstrapState.ACTIVE} or customer is None or len(session.role_claims) != 1 or customer.user_id != customer.chat_id or customer.topic_id != "0" or customer.user_id != session.customer_draft.customer_user_id or customer.user_id == session.owner_id or session.consent_handoff is None or session.consent_recovery_reconciled is not True ): raise GenerationConflict( "customer activation cutover consent authority is stale" ) competing = tuple( item for item in store.list_sessions() if item.session_id != session.session_id and item.customer_key == customer_id and item.state in {BootstrapState.AWAITING_ACTIVATION, BootstrapState.ACTIVE} ) if competing: raise BootstrapConflict("customer has multiple activation lifecycles") return session def _preflight_registry( document: object, session: RoomBootstrapSession, customer_id: str, ) -> object: owner = getattr(document, "owner", None) raw_customers = getattr(document, "customers", None) if not isinstance(raw_customers, (tuple, list)): raise ActivationCutoverError("customer registry identity is unavailable") customers = cast(Sequence[object], raw_customers) matches = tuple( item for item in customers if getattr(item, "customer_key", None) == customer_id ) claim = session.role_claim(Role.CUSTOMER) if len(matches) != 1 or claim is None: raise ActivationCutoverError("customer registry identity is unavailable") telegram = getattr(matches[0], "telegram", None) if ( str(getattr(owner, "user_id", "")) != session.owner_id or str(getattr(telegram, "user_id", "")) != claim.user_id or str(getattr(telegram, "chat_id", "")) != claim.chat_id or str(getattr(telegram, "topic_id", "")) != claim.topic_id ): raise ActivationCutoverError("customer registry identity is stale") return matches[0] def validate_first_claim_owner_review( profile_root: Path, session: RoomBootstrapSession, ) -> None: """Require finalized owner-reviewed nutrition evidence for first claims.""" if not session.first_claim: return from .telegram_nutrition_onboarding_publication_outbox import ( GatewayOnboardingPublicationOutbox, ) onboarding_root = ( profile_root / "data" / "customers" / session.customer_key / "nutrition-onboarding" ) try: baseline = _private_json( onboarding_root / "baseline-v1.json", "first-claim baseline", ) readiness = _private_json( onboarding_root / "readiness-receipt-v1.json", "first-claim readiness", ) pointer = _private_json( onboarding_root / "readiness-current.json", "first-claim readiness pointer", ) ready = _private_json( onboarding_root / "ready.json", "first-claim ready state", ) candidate = _private_json( onboarding_root / "baseline-candidate.json", "first-claim scrubbed baseline", ) except ActivationCutoverError as exc: raise ActivationCutoverError( "first-claim owner review is unavailable" ) from exc owner_receipt = baseline.get("owner_review_receipt") try: callback = GatewayOnboardingPublicationOutbox( profile_root, initialize=False, ).owner_callback(session.session_id) except ValueError as exc: raise ActivationCutoverError( "first-claim owner review is unavailable" ) from exc if ( baseline.get("customer_key") != session.customer_key or baseline.get("schema_version") != "2.1" or "eating_disorder_risk" not in baseline or not isinstance(owner_receipt, str) or len(owner_receipt) != 64 or readiness.get("owner_review_receipt") != owner_receipt or readiness.get("delivery_enabled") is not False or readiness.get("activation_enabled") is not False or ready.get("state") != "ready" or ready.get("baseline_digest") != baseline.get("digest") or ready.get("readiness_pointer_digest") != pointer.get("digest") or pointer.get("readiness_receipt_digest") != readiness.get("digest") or "source_answers" in candidate or not isinstance(candidate.get("source_answers_digest"), str) or len(str(candidate.get("source_answers_digest"))) != 64 or (onboarding_root / "transient/workflow.json").exists() or not hmac.compare_digest( owner_receipt, str(readiness.get("owner_review_receipt", "")), ) or callback is None or callback.session_id != session.session_id or callback.customer_key != session.customer_key or callback.action != "Approve" or str(callback.actor_user_id) != session.owner_id or str(callback.authority[0]) != session.owner_id or callback.route != (str(callback.authority[1]), str(callback.authority[2])) ): raise ActivationCutoverError( "first-claim owner review is unavailable" ) def _private_json(path: Path, label: str) -> dict[str, object]: try: metadata = path.lstat() value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise ActivationCutoverError(f"{label} is unavailable") from exc if ( path.is_symlink() or not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_nlink != 1 or not isinstance(value, dict) ): raise ActivationCutoverError(f"{label} is invalid") return value def _preflight_membership_evidence( profile_root: Path, registry_path: Path, evidence_path: Path, deployment_path: Path, session: RoomBootstrapSession, *, customer_id: str, ) -> None: """Reject stale or mismatched evidence before the profile can write.""" from .telegram_staff_membership_gate import sha256_json from .task26_candidate_derivation import validate_deployment_receipt evidence = _private_json(evidence_path, "staff membership evidence") deployment = _private_json(deployment_path, "deployment receipt") if deployment.get("schema") == "task26-source-deployment-receipt-v2": product_binding = _private_json( profile_root / "preexecution-product-binding.json", "preexecution product binding", ) if deployment.get("product_binding") != product_binding: raise ActivationCutoverError("deployment product binding is stale") try: validate_deployment_receipt(deployment, product_binding) except ValueError as exc: raise ActivationCutoverError("deployment wheel binding is invalid") from exc body = { key: value for key, value in evidence.items() if key != "evidence_sha256" } claim = session.role_claim(Role.CUSTOMER) config_path = profile_root / "config.yaml" required = { "schema": "telegram-staff-membership-evidence-v1", "phase": "pre_activation", "customer_id": customer_id, "customer_user_id": claim.user_id if claim is not None else None, "bootstrap_session_id": session.session_id, "bootstrap_generation": session.generation, "registry_sha256": hashlib.sha256(registry_path.read_bytes()).hexdigest(), "config_sha256": hashlib.sha256(config_path.read_bytes()).hexdigest(), "deployment_receipt_sha256": hashlib.sha256( deployment_path.read_bytes() ).hexdigest(), } if any(evidence.get(key) != value for key, value in required.items()): raise ActivationCutoverError("staff membership evidence binding is stale") if evidence.get("evidence_sha256") != sha256_json(body): raise ActivationCutoverError("staff membership evidence digest is invalid") for field in ( "candidate_digest", "candidate_core_digest", "candidate_inventory_digest", "candidate_product_binding_sha256", "hermes_wheel_sha256", "profile_wheel_sha256", ): if evidence.get(field) != deployment.get(field): raise ActivationCutoverError("staff membership deployment binding is stale") try: armed = datetime.fromisoformat(str(evidence["subscription_armed_at_utc"])) started = datetime.fromisoformat(str(evidence["request_started_at_utc"])) observed = datetime.fromisoformat(str(evidence["observed_at_utc"])) except (KeyError, ValueError) as exc: raise ActivationCutoverError("staff membership evidence timestamp is invalid") from exc current = datetime.now(UTC) if ( any(value.tzinfo is None for value in (armed, started, observed)) or not armed <= started <= observed <= current or current - observed > timedelta(minutes=5) ): raise ActivationCutoverError("staff membership evidence is stale") def _finalize_membership_activation_binding( profile_root: Path, evidence_path: Path, *, customer_id: str, ) -> None: """Bind the committed activation journal before advancing bootstrap state.""" from .telegram_staff_membership_gate import MembershipJournal evidence = _private_json(evidence_path, "staff membership evidence") evidence_digest = hashlib.sha256(evidence_path.read_bytes()).hexdigest() journal = MembershipJournal( profile_root / "data/onboarding/telegram-staff-membership-v1/events.jsonl" ) matches = [ row for row in journal.verify() if row.get("event") == "activation_binding" and row.get("customer_id") == customer_id ] replay_expected = { "membership_evidence_sha256": evidence_digest, "subscription_epoch_id": evidence.get("subscription_epoch_id"), } if matches: if all( matches[-1].get(key) == value for key, value in replay_expected.items() ): return raise ActivationCutoverError("membership activation binding conflicts") activation_path = profile_root / "data/customer-activation-journal.json" activation = _private_json(activation_path, "customer activation journal") if activation.get("state") != "committed" or activation.get( "customer_id" ) != customer_id: raise ActivationCutoverError("committed customer activation is unavailable") activation_digest = hashlib.sha256(activation_path.read_bytes()).hexdigest() journal.append( { "event": "activation_binding", "customer_id": customer_id, "observed_at_utc": datetime.now(UTC).isoformat(), "activation_journal_sha256": activation_digest, **replay_expected, } ) def _activate_bootstrap( store: RoomBootstrapStore, session: RoomBootstrapSession, *, customer_id: str, ) -> RoomBootstrapSession: customer = session.role_claim(Role.CUSTOMER) if customer is None: raise ActivationCutoverError("customer bootstrap identity is unavailable") return store.activate_bound_customer( session.session_id, expected_generation=session.generation, customer_key=customer_id, customer_user_id=customer.user_id, owner_id=session.owner_id, ) def _reserve_activation_completion_notice( profile_root: Path, customer: object, session: RoomBootstrapSession, *, membership_evidence_path: Path, ) -> None: """Reserve the customer completion notice from committed activation facts.""" from .telegram_activation_notice import ActivationNoticeStore def value(container: object, field: str) -> object: if isinstance(container, dict): return container.get(field) return getattr(container, field, None) plan = getattr(customer, "plan", None) schedule = getattr(customer, "schedule", None) destination = getattr(customer, "telegram", None) starts_on = value(plan, "starts_on") daily_time = value(schedule, "daily_time") if isinstance(starts_on, date): starts_on = starts_on.isoformat() if isinstance(daily_time, time): daily_time = daily_time.isoformat(timespec="minutes") destination_pin = { field: value(destination, field) for field in ("user_id", "chat_id", "topic_id") } if ( not isinstance(starts_on, str) or not isinstance(daily_time, str) or any(item is None for item in destination_pin.values()) ): raise ActivationCutoverError("activation notice authority is unavailable") _ = _private_json(membership_evidence_path, "staff membership evidence") from .telegram_staff_membership_gate import MembershipJournal binding_rows = [ row for row in MembershipJournal( profile_root / "data/onboarding/telegram-staff-membership-v1/events.jsonl" ).verify() if row.get("event") == "activation_binding" and row.get("customer_id") == session.customer_key ] membership_digest = hashlib.sha256( membership_evidence_path.read_bytes() ).hexdigest() if ( not binding_rows or binding_rows[-1].get("membership_evidence_sha256") != membership_digest or not isinstance( binding_rows[-1].get("activation_journal_sha256"), str, ) ): raise ActivationCutoverError("activation notice binding is unavailable") authority_digest = hashlib.sha256( json.dumps( { "schema": "customer-activation-notice-authority-v1", "customer_id": session.customer_key, "bootstrap_session_id": session.session_id, "bootstrap_generation": session.generation, "activation_journal_sha256": binding_rows[-1][ "activation_journal_sha256" ], "membership_evidence_sha256": membership_digest, }, sort_keys=True, separators=(",", ":"), ).encode() ).hexdigest() _ = ActivationNoticeStore(profile_root).reserve( customer_key=session.customer_key, starts_on=starts_on, daily_time=daily_time, destination=destination_pin, authority_digest=authority_digest, ) def _admin_registry_path(admin: _CustomerAdmin, root: Path) -> Path: resolve = cast( Callable[[Path], Path], getattr(admin, "_resolve_registry_path"), ) return resolve(root) def _admin_read(admin: _CustomerAdmin, path: Path) -> object: read = cast(Callable[[Path], object], getattr(admin, "_read")) return read(path) def _validate_committed( admin: _CustomerAdmin, root: Path, registry_path: Path, customer_id: str, ) -> None: if admin.validate_committed_activation(root, registry_path, customer_id) is not True: raise ActivationCutoverError("committed customer activation is unavailable") def _profile_root(value: Path | str) -> Path: root = Path(value) if root.is_symlink() or not root.is_dir(): raise ActivationCutoverError("activation profile root is unavailable") return root.resolve() def _profile_admin( root: Path, package_root: Path | str | None ) -> _CustomerAdmin: configured = Path( package_root or os.environ.get("DUALCOACH_PROFILE_PACKAGE") or root / "workspace" / "checkin_cli" ).resolve() package = configured / "checkin_cli" source_path = package / "customer_admin.py" if not package.is_dir() or not source_path.is_file() or source_path.is_symlink(): raise ActivationCutoverError("profile-local checkin_cli is unavailable") root_text = str(configured) if root_text not in sys.path: sys.path.insert(0, root_text) admin = importlib.import_module("checkin_cli.customer_admin") source = Path(str(admin.__file__)).resolve() if source != source_path.resolve(): raise ActivationCutoverError("loaded checkin_cli is not profile-local") return cast(_CustomerAdmin, cast(object, admin)) __all__ = [ "ActivationCutoverError", "ActivationCutoverResult", "activate_customer_cutover", ]