"""Private disposable NutriCoach profile and runtime-authority fixtures.""" from __future__ import annotations import hashlib import json import os import shutil from dataclasses import dataclass from datetime import date from collections.abc import Mapping from pathlib import Path from typing import Final from gateway.platforms.task26_runtime_authority import ( append_external_authority, digest as authority_digest, ) from gateway.platforms.telegram_customer_bootstrap import ( CustomerDraft, RoomBootstrapStore, room_bootstrap_state_dir, ) from scripts.dualcoach_v111_evidence import JsonValue, json_object, parse_json, write_json _OWNER_ID: Final = 100 _OWNER_CHAT_ID: Final = -100_700 _OWNER_TOPIC_ID: Final = 73 _CUSTOMER_ID: Final = 200 _CUSTOMER_KEY: Final = "disposable_001" _BOT_TOKEN: Final = "7000000000:DISPOSABLE_ONLY" _BOT_ID: Final = 7_000_000_002 _BOT_USERNAME: Final = "disposable_nutricoach_bot" @dataclass(frozen=True, slots=True) class DisposableRuntimeInputs: candidate_digest: str authority_root: Path model_base_url: str @dataclass(frozen=True, slots=True) class DisposableProfile: root: Path registry_path: Path config_path: Path package_root: Path customer_start_token: str authority_pin: Path authority_candidate: Path authority_root: Path def owner_route() -> tuple[int, int, int]: return _OWNER_ID, _OWNER_CHAT_ID, _OWNER_TOPIC_ID def customer_identity() -> tuple[str, int]: return _CUSTOMER_KEY, _CUSTOMER_ID def prepare_profile( root: Path, repository: Path, inputs: DisposableRuntimeInputs, ) -> DisposableProfile: """Create a private profile with one unclaimed production bootstrap invite.""" _ = root.mkdir(mode=0o700, parents=True, exist_ok=True) _ = root.chmod(0o700) registry = root / "customers" / "registry.json" _ = registry.parent.mkdir(mode=0o700, parents=True) _ = registry.parent.chmod(0o700) write_json( registry, { "version": 1, "owner": { "user_id": str(_OWNER_ID), "chat_id": str(_OWNER_CHAT_ID), "topic_id": str(_OWNER_TOPIC_ID), }, "customers": [], }, ) package_root = root / "workspace" / "checkin_cli" shutil.copytree(repository / "dualcoach" / "profile", package_root) _private_tree(package_root) config = root / "config.yaml" config.write_text( "physique_coach:\n" " draft_provider: disposable-loopback\n" " draft_model: disposable-coach-v1\n" " draft_timeout: 5\n" "providers:\n" " disposable-loopback:\n" f" base_url: {inputs.model_base_url}\n" " default_model: disposable-coach-v1\n" "platforms:\n" " telegram:\n" " enabled: true\n" " extra:\n" " group_topics:\n" f" - chat_id: {_OWNER_CHAT_ID}\n" f" topic_id: {_OWNER_TOPIC_ID}\n" " nutrition_coaching:\n" " enabled: true\n" " registry_path: customers/registry.json\n" " operator_review:\n" f" user_id: '{_OWNER_ID}'\n" f" chat_id: '{_OWNER_CHAT_ID}'\n" " topic_id: '59'\n" " room_bootstrap:\n" " enabled: true\n" " nutrition_onboarding: true\n" " production_preflight:\n" f" expected_bot_id: {_BOT_ID}\n" f" expected_bot_username: {_BOT_USERNAME}\n" f" candidate_package_identity: {inputs.candidate_digest}\n", encoding="utf-8", ) config.chmod(0o600) environment = root / ".env" environment.write_text(f"TELEGRAM_BOT_TOKEN={_BOT_TOKEN}\n", encoding="utf-8") environment.chmod(0o600) kb_source = ( repository / "dualcoach/releases/v1.1.0/installed-golden-bundle/data/global/" "nutrition-safety/restriction-kb-v1.json" ) kb_target = root / "data/global/nutrition-safety/restriction-kb-v1.json" _ = kb_target.parent.mkdir(mode=0o700, parents=True) _ = kb_target.parent.chmod(0o700) shutil.copyfile(kb_source, kb_target) kb_target.chmod(0o600) prepared = RoomBootstrapStore(room_bootstrap_state_dir(root)).prepare_rehearsal_customer_invite( CustomerDraft( customer_key=_CUSTOMER_KEY, display_name="Disposable Customer", starts_on=activation_day().isoformat(), daily_time="08:00", weekly_weekday=0, monthly_day=1, calories_kcal=2200, protein_g=140, meals=("breakfast", "lunch", "dinner"), primary_goal="maintain", ), bot_username=_BOT_USERNAME, owner_id=str(_OWNER_ID), ) token = prepared.customer_link.rsplit("=", 1)[1] pin, candidate = write_runtime_authority( inputs.authority_root, inputs.candidate_digest, ) return DisposableProfile( root=root, registry_path=registry, config_path=config, package_root=package_root, customer_start_token=token, authority_pin=pin, authority_candidate=candidate, authority_root=inputs.authority_root, ) def write_deployment_receipt(path: Path, package_root: Path) -> dict[str, JsonValue]: """Bind the disposable run to the exact profile package source tree.""" digest = hashlib.sha256() for source in sorted((package_root / "checkin_cli").glob("*.py")): digest.update(source.name.encode("utf-8")) digest.update(source.read_bytes()) candidate = digest.hexdigest() payload: dict[str, JsonValue] = { "candidate_digest": candidate, "candidate_core_digest": candidate, "candidate_inventory_digest": candidate, "hermes_wheel_sha256": hashlib.sha256(b"repository-source").hexdigest(), "profile_wheel_sha256": hashlib.sha256(b"profile-source").hexdigest(), } write_json(path, payload) return payload def write_delivery_launch_authorization( path: Path, profile: DisposableProfile, product: Mapping[str, str], ) -> Path: """Create one private operator launch authority bound to disposable state.""" from gateway.platforms.dualcoach_tasks21_25_controller import ( LAUNCH_AUTHORIZATION_SCHEMA, _profile_authorization_digest as profile_authorization_digest, ) from gateway.platforms.task26_runtime_authority import ( FileCandidateAuthoritySource, ) candidate = product["candidate_digest"] write_json(profile.root / "auth.json", {}) source = FileCandidateAuthoritySource( profile.authority_pin, forbidden_roots=(profile.root, path.parent), ) with source.authorize(candidate, "service_activation") as snapshot: runtime_snapshot = json_object( parse_json(json.dumps(snapshot, sort_keys=True, separators=(",", ":"))), "runtime authority snapshot", ) write_json( path, { "schema": LAUNCH_AUTHORIZATION_SCHEMA, "status": "AUTHORIZED", "candidate_digest": candidate, "full_digest": candidate, "core_digest": product["candidate_core_digest"], "inventory_digest": product["candidate_inventory_digest"], "manifest_digest": hashlib.sha256( b"disposable-candidate-manifest" ).hexdigest(), "wheel_digest": product["hermes_wheel_sha256"], "candidate_product_binding_sha256": product[ "candidate_product_binding_sha256" ], "hermes_wheel_sha256": product["hermes_wheel_sha256"], "profile_wheel_sha256": product["profile_wheel_sha256"], "provider_config_digest": hashlib.sha256( profile.config_path.read_bytes() ).hexdigest(), "profile_authorization_digest": profile_authorization_digest(profile.root), "profile": str(profile.root), "service_action": "disposable_operator_authorization", "telegram_action": "none", "runtime_authority_snapshot": runtime_snapshot, }, ) return path def write_runtime_authority(root: Path, candidate_digest: str) -> tuple[Path, Path]: """Create an external two-event authority chain and its private credentials.""" _ = root.mkdir(mode=0o700, parents=True, exist_ok=True) _ = root.chmod(0o700) prior_candidate = hashlib.sha256(b"dualcoach-v111-disposable-prior").hexdigest() historical_pass = hashlib.sha256(b"dualcoach-v111-disposable-pass").hexdigest() _ = append_external_authority( root, source_id="dualcoach-v111-disposable", candidate_digest=prior_candidate, action="qualify", historical_pass_digest=historical_pass, reason="disposable prior qualification", ) current = append_external_authority( root, source_id="dualcoach-v111-disposable", candidate_digest=candidate_digest, action="qualify", historical_pass_digest=historical_pass, reason="disposable current qualification", ) pin = root / "task26-authority-pin.json" unsigned = { "schema": "task26-authority-pin-v1", "authority_root": str(root.resolve()), "source_id": current["source_id"], "genesis_sha256": current["genesis_sha256"], "registry_head_sha256": current["registry_head_sha256"], "ledger_head_sha256": current["ledger_head_sha256"], "event_count": 2, } payload = {**unsigned, "pin_sha256": authority_digest(unsigned)} _ = pin.write_text( json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8", ) _ = pin.chmod(0o600) candidate = root / "task26-candidate-digest" _ = candidate.write_text(candidate_digest + "\n", encoding="utf-8") _ = candidate.chmod(0o600) return pin, candidate def write_activation_checklist(path: Path) -> None: write_json( path, { "checklist": { "token_rotated": True, "missend_test_passed": True, "provider_terms_checked": { "checked": True, "version": "privacy-v1", }, "withdrawal_deletion_doc": True, "retention_backup_doc": True, "manual_fallback_doc": True, } }, ) def activation_day() -> date: return date.today() def _private_tree(root: Path) -> None: for directory, _subdirs, files in os.walk(root): Path(directory).chmod(0o700) for filename in files: (Path(directory) / filename).chmod(0o600)