#!/usr/bin/env python3 """Drive one isolated profile through the production source Golden Path. The driver emits no lifecycle phase declarations. Its only durable output is the profile and the native ledgers written by production APIs, plus raw external-I/O and immutable source-provenance inputs used by the verifier. """ from __future__ import annotations import argparse import asyncio import hashlib import importlib.util import json import os import shutil import stat import sys import tempfile import threading from dataclasses import asdict from datetime import UTC, date, datetime, time, timezone from pathlib import Path from types import SimpleNamespace from typing import Any, TypedDict, cast from zoneinfo import ZoneInfo _PROVENANCE_SPEC = importlib.util.spec_from_file_location( "golden_installed_wheel_provenance", Path(__file__).with_name("installed_wheel_provenance.py"), ) if _PROVENANCE_SPEC is None or _PROVENANCE_SPEC.loader is None: raise RuntimeError("installed provenance helper is unavailable") _PROVENANCE_MODULE = importlib.util.module_from_spec(_PROVENANCE_SPEC) _PROVENANCE_SPEC.loader.exec_module(_PROVENANCE_MODULE) collect_installed_runtime = _PROVENANCE_MODULE.collect_installed_runtime loaded_module_receipts = _PROVENANCE_MODULE.loaded_module_receipts verify_packaged_module_parity = _PROVENANCE_MODULE.verify_packaged_module_parity portable_installed_runtime = _PROVENANCE_MODULE.portable_installed_runtime REPO = Path(__file__).resolve().parents[1] DEFAULT_SOURCE = Path("/home/cube/.cache/task26-strict-successor-1786976146/src-p") ANSWERS = ( "1990년 1월 15일입니다.", "남성입니다.", "키는 180cm입니다.", "현재 체중은 80kg입니다.", "보통 수준입니다.", "주 3회 60분 근력 운동을 하고 평일에는 하루 8천 보 정도 걷습니다.", "현재 체중을 유지하고 싶습니다.", "유지가 목표라 목표 체중은 정하지 않겠습니다.", "유지가 목표라 목표 날짜도 정하지 않겠습니다.", "음식 알레르기는 없습니다.", "음식 불내증은 없습니다.", "종교적 또는 윤리적으로 제외하는 음식은 없습니다.", "싫어해서 피하는 음식은 없습니다.", "특별한 식단 선호는 없습니다.", "진단받은 질환은 없습니다.", "복용 중인 약이나 보충제는 없습니다.", "임신 또는 수유에 해당하지 않습니다.", "섭식장애 위험이나 과거력은 없습니다.", "가스레인지, 전자레인지, 냉장고를 사용할 수 있고 기본 조리가 가능합니다.", "식비 예산은 보통입니다.", "하루 세 끼를 먹습니다.", "평일 점심은 12시, 저녁은 운동 후 8시쯤이며 그 외 제약은 없습니다.", ) CUSTOMER = "client_001" OWNER_ID = 100 CUSTOMER_ID = 200 def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def digest(value: object) -> str: return hashlib.sha256(canonical(value)).hexdigest() def load_product_binding(path: Path | None) -> tuple[str, dict[str, object]]: if path is None: raise ValueError("private structured product/candidate binding is required") from gateway.platforms.task26_candidate_derivation import ( qualification_tool_hashes, trust_boundary_digest, validate_product_binding, ) from gateway.platforms.task26_final_state import TRUST_BOUNDARY selected = path.absolute() info = selected.lstat() if ( stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) not in {0o600, 0o400} ): raise ValueError("product/candidate binding file is not private") document = json.loads(selected.read_text(encoding="utf-8")) validated = validate_product_binding( document, expected_tool_hashes=qualification_tool_hashes( scripts_dir=Path(__file__).resolve().parent, package_root=REPO, ), expected_trust_boundary_sha256=trust_boundary_digest(TRUST_BOUNDARY), ) return str(validated["candidate_digest"]), validated def private_write(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) path.parent.chmod(0o700) data = canonical(value) + b"\n" fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600) try: os.write(fd, data) os.fsync(fd) finally: os.close(fd) def copy_verification_tools(profile: Path) -> None: target = profile / "verification-tools" target.mkdir(mode=0o700) target.chmod(0o700) for name in ( "source_golden_path.py", "task26_frozen_bootstrap.py", "task26_ty_surface_gate.py", "verify_source_golden_path.py", "installed_wheel_provenance.py", ): destination = target / name destination.write_bytes(Path(__file__).with_name(name).read_bytes()) destination.chmod(0o600) def source_provenance(source: Path) -> dict[str, object]: files: list[dict[str, object]] = [] for path in sorted(source.rglob("*.py")): if path.is_symlink() or not path.is_file() or any(part in {".venv", "__pycache__"} for part in path.parts): continue relative = path.relative_to(source).as_posix() files.append({"path": relative, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), "size": path.stat().st_size}) return { "schema": "source-golden-provenance-v1", "source_root": str(source), "files": files, "tree_digest": digest(files), } class FakeTelegram: def __init__(self, raw_log: Path) -> None: self.raw_log = raw_log self.message_id = 1000 self._bot = SimpleNamespace(get_chat_member=self.get_chat_member) async def get_chat_member(self, **_kwargs: object) -> object: return SimpleNamespace(status="member") async def _send_nutrition_topic(self, **kwargs: object) -> object: self.message_id += 1 row = { "kind": "onboarding_publication", "sequence": self.message_id, "chat_id": str(kwargs.get("chat_id")), "topic_id": str(kwargs.get("topic_id")), "text_sha256": hashlib.sha256(str(kwargs.get("text", "")).encode()).hexdigest(), } with self.raw_log.open("ab") as stream: stream.write(canonical(row) + b"\n") stream.flush() os.fsync(stream.fileno()) return SimpleNamespace(message_id=self.message_id) class Query: def __init__(self, actor: int, callback_id: str) -> None: self.id = callback_id self.from_user = SimpleNamespace(id=actor) async def answer(self, text: str | None = None, **_kwargs: object) -> None: del text async def edit_message_reply_markup(self, **_kwargs: object) -> None: return None class Message: def __init__(self, actor: int, message_id: int, text: str = "", *, reply_to: int | None = None) -> None: self.text = text self.message_id = message_id self.message_thread_id = 0 self.chat_id = actor self.chat = SimpleNamespace(id=actor, type="private") self.from_user = SimpleNamespace(id=actor) self.date = datetime.now(UTC) self.reply_to_message = SimpleNamespace(message_id=reply_to) if reply_to is not None else None async def reply_text(self, _text: str) -> None: return None class UnavailableMaliciousProvider: async def reconcile(self, _answers: object, *, consent_granted: bool) -> object: from gateway.platforms.nutrition_onboarding_reconciliation import ReconciliationUnavailable if not consent_granted: raise AssertionError("provider called without consent") raise ReconciliationUnavailable("malicious provider returned untrusted lifecycle text") class GenerationPins(TypedDict): expected_generation: int expected_record_digest: str expected_checkin_revision: str expected_draft_revision: str def pins(value: object) -> GenerationPins: generation = getattr(value, "generation", None) record = getattr(value, "record_digest", getattr(value, "generation_record_digest", None)) checkin = getattr(value, "checkin_revision", getattr(value, "generation_checkin_revision", None)) draft = getattr(value, "draft_revision", getattr(value, "generation_draft_revision", None)) if type(generation) is not int or not all(isinstance(item, str) for item in (record, checkin, draft)): raise RuntimeError("generation pins are unavailable") return { "expected_generation": generation, "expected_record_digest": cast(str, record), "expected_checkin_revision": cast(str, checkin), "expected_draft_revision": cast(str, draft), } def membership_evidence( profile: Path, session: object, candidate_digest: str, product_binding: dict[str, object], ) -> tuple[Path, Path]: registry_path = profile / "customers/registry.json" config = profile / "config.yaml" config.write_text("platforms: {}\n", encoding="utf-8") config.chmod(0o600) derivation_inputs = cast( dict[str, object], product_binding["derivation_inputs"] ) deployment = { "schema": "task26-source-deployment-receipt-v2", "candidate_digest": candidate_digest, "candidate_product_binding_sha256": product_binding["binding_sha256"], "candidate_core_digest": candidate_digest, "candidate_inventory_digest": candidate_digest, "hermes_wheel_sha256": derivation_inputs["hermes_wheel_sha256"], "profile_wheel_sha256": derivation_inputs["profile_wheel_sha256"], "product_binding": product_binding, } deployment_path = profile / "data/source-deployment-receipt.json" private_write(deployment_path, deployment) now = datetime.now(UTC).isoformat() row = { "schema": "telegram-staff-membership-evidence-v1", "phase": "pre_activation", **{ key: value for key, value in deployment.items() if key not in {"product_binding", "schema"} }, "deployment_receipt_path": str(deployment_path), "deployment_receipt_sha256": hashlib.sha256(deployment_path.read_bytes()).hexdigest(), "customer_id": CUSTOMER, "customer_user_id": str(CUSTOMER_ID), "bootstrap_session_id": str(getattr(session, "session_id")), "bootstrap_generation": int(getattr(session, "generation")), "registry_sha256": hashlib.sha256(registry_path.read_bytes()).hexdigest(), "config_sha256": hashlib.sha256(config.read_bytes()).hexdigest(), "staff_chat_inventory": [{"chat_id": str(OWNER_ID), "kind": "private_staff_dm", "staff_user_id": str(OWNER_ID), "roles": ["owner"], "sources": ["registry.owner"]}], "staff_chat_inventory_sha256": digest([{"chat_id": str(OWNER_ID), "kind": "private_staff_dm", "staff_user_id": str(OWNER_ID), "roles": ["owner"], "sources": ["registry.owner"]}]), "subscription_epoch_id": "source-golden-private-epoch", "subscription_armed_at_utc": now, "request_started_at_utc": now, "observed_at_utc": now, "activation_journal_sha256": "0" * 64, "membership_results": [], "private_dm_results": [{"chat_id": str(OWNER_ID), "staff_user_id": str(OWNER_ID), "identity_separated": True}], } row["evidence_sha256"] = digest(row) evidence_path = profile / "data/source-membership-evidence.json" private_write(evidence_path, row) return evidence_path, deployment_path def checklist(profile: Path) -> Path: path = profile / "data/source-activation-checklist.json" private_write(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}}) return path async def drive( profile: Path, source: Path, candidate_digest: str, candidate_binding: dict[str, object], installed_runtime: dict[str, object] | None = None, ) -> dict[str, object]: from gateway.platforms.task26_candidate_authority import append_candidate_authority from gateway.platforms.task26_evidence_contract import Task26EvidenceWriter import gateway.platforms.task26_evidence_contract as task26_contract_module provenance = source_provenance(source) if candidate_digest == provenance["tree_digest"]: raise ValueError( "candidate binding cannot substitute the source tree digest" ) if installed_runtime is None: authority_root = source os.environ["DUALCOACH_PROFILE_PACKAGE"] = str(source) if str(source) not in sys.path: sys.path.insert(0, str(source)) else: distributions = cast(dict[str, dict[str, object]], installed_runtime["distributions"]) profile_distribution = distributions["profile"] package_roots = cast(list[str], profile_distribution["package_roots"]) authority_root = Path(package_roots[0]).parent os.environ["DUALCOACH_PROFILE_PACKAGE"] = str(authority_root) from gateway.platforms.task26_candidate_derivation import validate_product_binding validate_product_binding( candidate_binding, actual_hermes_wheel=Path(str(distributions["hermes"]["wheel_path"])), actual_profile_wheel=Path(str(distributions["profile"]["wheel_path"])), ) contract_parity = verify_packaged_module_parity( source=REPO / "gateway/platforms/task26_evidence_contract.py", wheel=Path(str(distributions["hermes"]["wheel_path"])), wheel_member="gateway/platforms/task26_evidence_contract.py", installed=Path(cast(str, task26_contract_module.__file__)), ) binding = { "schema": "installed-golden-candidate-binding-v1", "runtime": installed_runtime, "runtime_portable": portable_installed_runtime(installed_runtime), "task26_contract_parity": contract_parity, "source_provenance": provenance, "config": { "runtime_mode": "installed", "profile_source_metadata_only": str(source), "profile_package_authority": str(authority_root), }, "harness_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "provenance_helper_sha256": hashlib.sha256( Path(__file__).with_name("installed_wheel_provenance.py").read_bytes() ).hexdigest(), "verifier_sha256": hashlib.sha256( Path(__file__).with_name("verify_source_golden_path.py").read_bytes() ).hexdigest(), "task26_contract_module": "gateway.platforms.task26_evidence_contract", "task26_contract_module_sha256": hashlib.sha256( Path(cast(str, task26_contract_module.__file__)).read_bytes() ).hexdigest(), } installed_product_binding_digest = digest(binding) from checkin_cli.customer_cleanup import ( SharedLedgerProjection, archive_customer_cleanup, post_cleanup_authority_inventory, resume_customer_cleanup, ) from checkin_cli.customer_coaching import load_customer_registry from checkin_cli.nutrition_onboarding import NutritionOnboardingService, QUESTION_FIELDS from checkin_cli.nutrition_onboarding_clarification_policy import compile_clarification_policy from checkin_cli.nutrition_onboarding_contract import MessageEvidence, OnboardingAuthority, canonical_digest as onboarding_digest from checkin_cli.nutrition_restriction_kb import seed_restriction_kb from gateway.commit_observer import COMMIT_MASK, ProductionCommitObserver from gateway.platforms.dualcoach_activation_cutover import activate_customer_cutover from gateway.platforms.nutrition_coaching import DeliveryLaunchAuthorization, DraftGenerationWorker, IncomingAddress, NutritionCoachingCoordinator from gateway.platforms.telegram import TelegramAdapter from gateway.platforms.telegram_customer_bootstrap import BootstrapState, CustomerDraft, RoomBootstrapStore, RoomBootstrapTransport, room_bootstrap_state_dir from gateway.platforms.telegram_customer_bootstrap_registration import TelegramCustomerBootstrapRegistration from gateway.platforms.telegram_nutrition_onboarding import encode_callback_hash from gateway.platforms.telegram_nutrition_onboarding_copy import parse_answer from gateway.platforms.telegram_nutrition_onboarding_preview_completion import build_preview_completion from gateway.platforms.telegram_nutrition_onboarding_runtime import TelegramNutritionOnboardingRuntime evidence_contract = Task26EvidenceWriter( profile, candidate_digest=candidate_digest ) copy_verification_tools(profile) private_write(profile / "source-provenance.json", provenance) private_write(profile / "preexecution-product-binding.json", candidate_binding) if installed_runtime is not None: distributions = cast(dict[str, dict[str, object]], installed_runtime["distributions"]) allowed_roots = tuple( Path(root) for distribution in distributions.values() for root in cast(list[str], distribution["package_roots"]) ) imported = { name: Path(cast(str, module.__file__)) for name, module in sys.modules.items() if (name == "checkin_cli" or name.startswith("checkin_cli.") or name == "gateway" or name.startswith("gateway.")) and getattr(module, "__file__", None) } origin_receipts = loaded_module_receipts(imported, allowed_roots=allowed_roots) private_write( profile / "installed-provenance.json", { "schema": "installed-golden-provenance-v1", "binding": binding, "installed_product_binding_digest": installed_product_binding_digest, "candidate_digest": candidate_digest, "loaded_modules": origin_receipts, }, ) registry = profile / "customers/registry.json" private_write(registry, {"version": 1, "owner": {"user_id": str(OWNER_ID), "chat_id": str(OWNER_ID), "topic_id": "0"}, "customers": []}) raw_onboarding = profile / "raw-onboarding-calls.jsonl" raw_customer = profile / "raw-customer-transport-calls.jsonl" private_write( profile / "driver-input.json", { "schema": "source-golden-driver-input-v1" if installed_runtime is None else "installed-golden-driver-input-v1", "runtime_mode": "source" if installed_runtime is None else "installed", "customer_key": CUSTOMER, "candidate_digest": candidate_digest, "source_tree_digest": provenance["tree_digest"], "candidate_product_binding_sha256": candidate_binding["binding_sha256"], }, ) raw_onboarding.touch(mode=0o600) raw_customer.touch(mode=0o600) state_dir = room_bootstrap_state_dir(profile) store = RoomBootstrapStore(state_dir) receipts = profile / "observer-receipts.jsonl" with ProductionCommitObserver((profile,), receipts) as observer: observer.arm(candidate_id=candidate_digest, profile_id=profile.name, watched_paths=(state_dir / "ledger.json",), watcher_id="source-golden-observer", mask=COMMIT_MASK, starting_cursor=0, update_offset=0, armed_at=datetime.now(UTC).isoformat()) prepared = store.prepare_rehearsal_customer_invite(CustomerDraft(CUSTOMER, "Golden Customer", "2026-08-18", "08:00", 0, 1, 2300, 150, ("breakfast", "lunch", "dinner")), bot_username="isolated_source_bot", owner_id=str(OWNER_ID)) observer.observe(sequence=1, timeout=2.0) claimed = store.claim_rehearsal_customer_invite(prepared.customer_link.rsplit("=", 1)[1], user_id=str(CUSTOMER_ID), chat_id=str(CUSTOMER_ID), message_id="1") observer.observe(sequence=2, timeout=2.0) registration = TelegramCustomerBootstrapRegistration(profile, store, package_root=authority_root).handoff_rehearsal_customer(claimed) observer.observe(sequence=3, timeout=2.0) awaiting = store.transition(registration.session.session_id, expected_generation=registration.session.generation, target=BootstrapState.AWAITING_ACTIVATION) observer.observe(sequence=4, timeout=2.0) parsed_answers = tuple( (field, parse_answer(field, answer)) for field, answer in zip(QUESTION_FIELDS, ANSWERS, strict=True) ) preview = await build_preview_completion( reconciler=cast(Any, UnavailableMaliciousProvider()), answers=parsed_answers, allow_clarifications=True, reference_date=date.today(), ) if preview.phase != "confirming" or preview.clarifications: raise RuntimeError("preview handoff introduced clarification issues") adapter = FakeTelegram(raw_onboarding) runtime = TelegramNutritionOnboardingRuntime(adapter=adapter, profile_root=profile, bootstrap_transport=RoomBootstrapTransport(store, owner_id=str(OWNER_ID)), reconciler=cast(Any, UnavailableMaliciousProvider())) await runtime.start_after_consent(session=awaiting, query=Query(CUSTOMER_ID, "consent"), message=Message(CUSTOMER_ID, 10)) service = runtime._service(CUSTOMER) answer_observations: list[dict[str, object]] = [] for index, (field, answer) in enumerate( zip(QUESTION_FIELDS, ANSWERS, strict=True), 1 ): before = service.status() current = service.store.load_session(awaiting.session_id) message = Message(CUSTOMER_ID, 100 + index, answer, reply_to=current.message_id) handled = await runtime.handle_text(SimpleNamespace(update_id=200 + index), message) after = service.status() answer_observations.append( { "index": index - 1, "field": field, "raw": answer, "canonical": parse_answer(field, answer), "before_count": before.answer_count, "after_count": after.answer_count, "handled": handled, "reask_count": 0 if after.answer_count == before.answer_count + 1 else 1, } ) if not handled or after.answer_count != before.answer_count + 1: raise RuntimeError(f"handle_text re-asked or did not consume answer {index}") evidence_contract.record( "onboarding_22_clear", { "question_fields": list(QUESTION_FIELDS), "observations": answer_observations, "production_answers": service.reconciliation_answers( authority=runtime._current_authority(awaiting) ), }, ) unambiguous_answers = dict(parsed_answers) ambiguous_answers = dict(unambiguous_answers) ambiguous_answers["schedule_constraints"] = "평일 점심 또는 저녁 중 하나만 가능합니다" ambiguity_first = compile_clarification_policy( ambiguous_answers, reference_date=date.today() ) ambiguity_second = compile_clarification_policy( ambiguous_answers, reference_date=date.today() ) revised_answers = dict(ambiguous_answers) revised_answers["schedule_constraints"] = unambiguous_answers["schedule_constraints"] revised_policy = compile_clarification_policy( revised_answers, reference_date=date.today() ) branch_root = profile / "task26-native-branches/ambiguity" branch_root.mkdir(parents=True, mode=0o700) branch_root.chmod(0o700) branch_service = NutritionOnboardingService( profile_root=branch_root, customer_key=CUSTOMER, enforce_current_authority=False, enforce_reconciliation=True, ) branch_authority = OnboardingAuthority( customer_key=CUSTOMER, customer_user_id=CUSTOMER_ID, customer_chat_id=CUSTOMER_ID, customer_topic_id=0, owner_user_id=OWNER_ID, owner_chat_id=OWNER_ID, owner_topic_id=0, consent_notice_version="privacy-v1", consent_granted=True, customer_enabled=False, ) def branch_evidence(sequence: int) -> MessageEvidence: return MessageEvidence( actor_user_id=CUSTOMER_ID, chat_id=CUSTOMER_ID, topic_id=0, message_id=10_000 + sequence, update_id=20_000 + sequence, ) branch_service.start_or_resume( authority=branch_authority, evidence=branch_evidence(0), ) for branch_index, field in enumerate(QUESTION_FIELDS, 1): branch_service.submit_answer( field=field, value=ambiguous_answers[field], authority=branch_authority, evidence=branch_evidence(branch_index), ) branch_record = branch_service.record_reconciliation( answers_digest=onboarding_digest(ambiguous_answers), advisory={"status": "unavailable"}, clarifications=[], authority=branch_authority, reference_date=date.today(), ) field_zero_rejection = "" try: branch_service.revise_reconciliation_answer( field=QUESTION_FIELDS[0], value=unambiguous_answers[QUESTION_FIELDS[0]], expected_issue_id=ambiguity_first.issues[0].issue_id, expected_answers_digest=ambiguity_first.answers_digest, expected_reconciliation_digest=str(branch_record["digest"]), authority=branch_authority, evidence=branch_evidence(23), ) except ValueError as exc: field_zero_rejection = str(exc) branch_status = branch_service.revise_reconciliation_answer( field="schedule_constraints", value=unambiguous_answers["schedule_constraints"], expected_issue_id=ambiguity_first.issues[0].issue_id, expected_answers_digest=ambiguity_first.answers_digest, expected_reconciliation_digest=str(branch_record["digest"]), authority=branch_authority, evidence=branch_evidence(24), ) branch_revised_record = branch_service.reconciliation_record( authority=branch_authority ) branch_workflow = branch_service.session_path evidence_contract.record( "deterministic_ambiguity_revision", { "reference_date": date.today().isoformat(), "ambiguous_answers": ambiguous_answers, "first_result": ambiguity_first.model_dump(mode="json"), "second_result": ambiguity_second.model_dump(mode="json"), "revision_binding": { "issue_id": ambiguity_first.issues[0].issue_id, "field": ambiguity_first.issues[0].field, "field_index": QUESTION_FIELDS.index(ambiguity_first.issues[0].field), "answers_digest": ambiguity_first.answers_digest, "reconciliation_digest": ambiguity_first.result_digest, }, "revised_answers": revised_answers, "revised_result": revised_policy.model_dump(mode="json"), "native_branch": { "workflow_path": str(branch_workflow.relative_to(profile)), "workflow_sha256": hashlib.sha256(branch_workflow.read_bytes()).hexdigest(), "field_zero_rejection": field_zero_rejection, "status": branch_status.model_dump(mode="json"), "initial_reconciliation": branch_record, "revised_reconciliation": branch_revised_record, "answers": branch_service.reconciliation_answers( authority=branch_authority ), }, }, ) evidence_contract.record( "preview_isolation", { "preview_phase": preview.phase, "preview_clarifications": [asdict(item) for item in preview.clarifications], "preview_text_sha256": hashlib.sha256(preview.text.encode()).hexdigest(), "production_answer_count": service.status().answer_count, "production_state": service.status().state.value, "production_session_id": awaiting.session_id, }, ) reconciliation = service.reconciliation_record( authority=runtime._current_authority(awaiting) ) if ( reconciliation is None or reconciliation.get("issues") or reconciliation.get("clarifications") or reconciliation.get("holds") ): raise RuntimeError("production handoff introduced clarification issues") current = service.store.load_session(awaiting.session_id) stale = encode_callback_hash(action="attest", generation=current.generation - 1, sid_hash=awaiting.sid_hash) await runtime.handle_callback(Query(CUSTOMER_ID, "stale-attest"), stale, Message(CUSTOMER_ID, int(current.message_id)), update_id=400) if service.status().state.value != "customer_attestation": raise RuntimeError("stale attestation mutated production state") attest = encode_callback_hash(action="attest", generation=current.generation, sid_hash=awaiting.sid_hash) await runtime.handle_callback(Query(CUSTOMER_ID, "attest"), attest, Message(CUSTOMER_ID, int(current.message_id)), update_id=401) if service.status().state.value != "owner_review": raise RuntimeError("attestation callback did not commit owner review") seed_restriction_kb(profile_root=profile, source=authority_root / "checkin_cli/policies/nutrition-restriction-kb-template-v1.json", owner_digest=candidate_digest, commit=True, as_of=date.today()) owner_publication = service.store.load_session(awaiting.session_id) owner_ok = encode_callback_hash(action="owner_ok", generation=owner_publication.generation, sid_hash=awaiting.sid_hash) await runtime.handle_callback(Query(OWNER_ID, "owner-ok"), owner_ok, Message(OWNER_ID, int(owner_publication.message_id)), update_id=402) if service.status().state.value != "ready": raise RuntimeError("owner approval did not finalize readiness") evidence, deployment = membership_evidence( profile, store.get(awaiting.session_id), candidate_digest, candidate_binding, ) cutover = activate_customer_cutover(profile, profile / f"data/customers/{CUSTOMER}", CUSTOMER, checklist(profile), evidence, bootstrap_session_id=awaiting.session_id, expected_generation=store.get(awaiting.session_id).generation, deployment_receipt_path=deployment, package_root=authority_root, kst_date=date.today()) if cutover.state is not BootstrapState.ACTIVE: raise RuntimeError("activation cutover did not commit ACTIVE") registry_doc = load_customer_registry(registry, profile) coordinator = NutritionCoachingCoordinator(profile, registry_doc, registry_path=registry, kst_date_provider=date.today, generation_now_provider=lambda: datetime.now(ZoneInfo("Asia/Seoul"))) address = IncomingAddress(str(CUSTOMER_ID), str(CUSTOMER_ID), "0") opening = coordinator.open_launcher(CUSTOMER) if not opening.accepted or opening.callback_data is None or not coordinator.bind_launcher(CUSTOMER, opening.callback_data, "500"): raise RuntimeError("daily wizard launcher unavailable") callback_type = sys.modules["gateway.platforms.nutrition_coaching"].CallbackInput if not coordinator.handle_callback(callback_type(opening.callback_data, address, "500")).reply.accepted: raise RuntimeError("daily wizard did not start") resolved = coordinator.resolve(address) if resolved is None: raise RuntimeError("activated customer is not routable") result = None for action, value in zip(("value", "value", "value", "value", "value", "value", "select", "select", "select", "value", "value", "select"), ("80", "2300", "280 150 65", "계획대로 3식", "2.5", "7", "4", "normal", "4", "식욕 보통, 스트레스 낮음", "하체 운동", "skip"), strict=True): result = resolved.bridge.apply_model_action(action, value) if not result.accepted: raise RuntimeError("daily wizard answer rejected") if result is None or result.prompt is None: raise RuntimeError("daily wizard save prompt unavailable") save = next(callback for label, callback in result.prompt.buttons if label == "저장") completed = coordinator.handle_callback(callback_type(save, address, "500")) if completed.completion is None: raise RuntimeError("daily wizard finalization failed") draft_id = str(completed.completion.request_token) owner = coordinator.owner async def generate(_system: str, _input: str) -> str: request = coordinator.build_draft_generation_request(draft_id, owner) if request is None: raise RuntimeError("generation request unavailable") grounding = cast(Any, request[2]) targets = grounding.current_targets return json.dumps({"schema_version": "nutrition-coach-response-v2", "customer_key": grounding.customer_key, "revision_binding_digest": grounding.revision_binding_digest, "decision": "maintain", "confidence": "low", "evidence_ids": [item.option_id for item in grounding.evidence[:1]], "interpretation": "현재 계획을 유지합니다.", "recommendation_unit_system": "kcal_and_grams", "recommendation": targets.as_dict(), "next_checkin_focus_ids": [item.option_id for item in grounding.observations[:1]], "customer_draft": "현재 계획을 유지하고 다음 체크인을 확인하겠습니다."}, ensure_ascii=False) async def update_card(_action: object) -> None: return None created = await DraftGenerationWorker(coordinator, owner, worker_id="source-golden-worker", provider_ready=lambda: True, generate=generate, update_card=update_card).run_once(draft_id) if not created.accepted: raise RuntimeError(f"DraftGenerationWorker failed: {created.error}") generation = coordinator.draft_generation(draft_id) approved = coordinator.approve_draft(draft_id, owner, **pins(generation)) if not approved.accepted: raise RuntimeError(f"approve_draft failed: {approved.error}") nonce = TelegramAdapter._nutrition_generation_card_nonce(approved) payload_digest = TelegramAdapter._nutrition_approved_card_payload_digest(approved, "source-golden-card") if nonce is None or payload_digest is None: raise RuntimeError("approved card binding unavailable") original_write = coordinator._write_json_private projection_fault_seen = False def fail_approved_card_projection(path: Path, payload: object) -> None: nonlocal projection_fault_seen if path == coordinator._delivery_ledger_path(): projection_fault_seen = True raise OSError("source-golden-approved-card-projection-fault") original_write(path, payload) setattr(coordinator, "_write_json_private", fail_approved_card_projection) projection_error = "" try: coordinator.persist_approved_delivery_card( draft_id, owner, card_route=owner, message_id="700", nonce=nonce, payload_digest=payload_digest, **pins(approved) ) except OSError as exc: projection_error = str(exc) finally: setattr(coordinator, "_write_json_private", original_write) capability_error = "" derivation_inputs = cast( dict[str, object], candidate_binding["derivation_inputs"] ) hermes_wheel_sha256 = str(derivation_inputs["hermes_wheel_sha256"]) profile_wheel_sha256 = str(derivation_inputs["profile_wheel_sha256"]) auth = DeliveryLaunchAuthorization( candidate_digest=candidate_digest, full_digest=candidate_digest, core_digest=candidate_digest, inventory_digest=candidate_digest, manifest_digest=candidate_digest, wheel_digest=hermes_wheel_sha256, provider_config_digest=candidate_digest, profile_authorization_digest=candidate_digest, hermes_wheel_sha256=hermes_wheel_sha256, profile_wheel_sha256=profile_wheel_sha256, candidate_product_binding_sha256=str(candidate_binding["binding_sha256"]), ) try: coordinator.issue_delivery_capability( draft_id, owner, launch_authorization=auth, capability_id="projection-fault-capability", issued_at=datetime.now(timezone.utc), **pins(approved) ) except Exception as exc: capability_error = f"{type(exc).__name__}: {exc}" evidence_contract.record( "approved_card_projection_gate", { "projection_fault_seen": projection_fault_seen, "projection_error": projection_error, "delivery_rows_after_failure": coordinator._read_deliveries(), "capability_error": capability_error, }, ) coordinator.persist_approved_delivery_card(draft_id, owner, card_route=owner, message_id="700", nonce=nonce, payload_digest=payload_digest, **pins(approved)) coordinator.issue_delivery_capability(draft_id, owner, launch_authorization=auth, capability_id="source-golden-capability", issued_at=datetime.now(timezone.utc), **pins(approved)) prepared_delivery = coordinator.prepare_delivery(draft_id, owner, **pins(approved)) if not prepared_delivery.accepted or not prepared_delivery.transport_required: raise RuntimeError(f"prepare_delivery failed: {prepared_delivery.error}") active_customer = coordinator.customer(CUSTOMER) if active_customer is None: raise RuntimeError("active customer disappeared before delivery") destination = active_customer.spec.telegram claim = coordinator.claim_delivery_transport(draft_id, owner, destination, approved.text, card_route=owner, card_message_id="700", card_nonce=nonce, card_payload_digest=payload_digest, now=datetime.now(timezone.utc), kst_date=date.today()) if claim is None: raise RuntimeError("claim_delivery_transport rejected exact capability") unknown_first = coordinator.prepare_delivery( draft_id, owner, **pins(coordinator.draft_generation(draft_id)) ) unknown_second = coordinator.prepare_delivery( draft_id, owner, **pins(coordinator.draft_generation(draft_id)) ) evidence_contract.record( "unknown_delivery_no_retry", { "first": { "accepted": unknown_first.accepted, "status": unknown_first.status, "transport_required": unknown_first.transport_required, }, "second": { "accepted": unknown_second.accepted, "status": unknown_second.status, "transport_required": unknown_second.transport_required, }, "delivery_ledger": coordinator._read_deliveries(), "transport_calls_before_send": len(raw_customer.read_text(encoding="utf-8").splitlines()), }, ) call = {"kind": "customer_send", "chat_id": str(destination.chat_id), "topic_id": str(destination.topic_id), "text_sha256": hashlib.sha256(str(approved.text).encode()).hexdigest(), "provider_message_id": "9001"} with raw_customer.open("ab") as stream: stream.write(canonical(call) + b"\n"); stream.flush(); os.fsync(stream.fileno()) delivered = coordinator.mark_delivered(draft_id, owner, "9001", **pins(coordinator.draft_generation(draft_id))) audited = coordinator.mark_sent_audited(draft_id, owner, **pins(delivered)) if not audited.accepted or audited.status != "sent_audited": raise RuntimeError(f"sent audit failed: {audited.error}") surface = coordinator.record_customer_surface_receipt( claim, draft_id, destination, str(approved.text), "9001" ) if surface.schema != "telegram-customer-surface-receipt-v2": raise RuntimeError("customer surface receipt v2 was not committed") duplicate = coordinator.claim_delivery_transport( draft_id, owner, destination, approved.text, card_route=owner, card_message_id="700", card_nonce=nonce, card_payload_digest=payload_digest, now=datetime.now(timezone.utc), kst_date=date.today(), ) if duplicate is not None: raise RuntimeError("consumed delivery capability accepted a duplicate claim") successful_delivery = next(iter(coordinator._read_deliveries().values())) successful_capability = cast( dict[str, object], successful_delivery["delivery_capability"] ) successful_bindings = cast( dict[str, object], successful_capability["bindings"] ) evidence_contract.record( "successful_lifecycle", { "attestation_state": "owner_review", "owner_review_state": "ready", "activation_state": cutover.state.value, "checkin_finalized": completed.completion is not None, "approval_status": approved.status, "delivery_status": audited.status, "surface_schema": surface.schema, "transport_calls": len(raw_customer.read_text(encoding="utf-8").splitlines()), "duplicate_rejected": duplicate is None, "capability_identity": { field: successful_bindings[field] for field in ( "candidate_digest", "candidate_product_binding_sha256", "hermes_wheel_sha256", "profile_wheel_sha256", "wheel_digest", ) }, "delivery_ledger_sha256": hashlib.sha256( coordinator._delivery_ledger_path().read_bytes() ).hexdigest(), }, ) paused = coordinator.handle_text(address, "코칭 일시중지") if not paused.reply.accepted: raise RuntimeError("pause did not commit") withdrawn = coordinator.withdraw_customer(address) if not withdrawn.reply.accepted: raise RuntimeError("withdraw did not commit") fault_seen = False def fault(phase: str) -> None: nonlocal fault_seen if phase == "copied_verified" and not fault_seen: fault_seen = True raise RuntimeError("source-golden-cleanup-fault") try: archive_customer_cleanup( profile, CUSTOMER, shared_ledger_projections=(SharedLedgerProjection.DRAFT_GENERATIONS,), fault_injector=fault, ) except RuntimeError as exc: if str(exc) != "source-golden-cleanup-fault": raise receipt = resume_customer_cleanup(profile, CUSTOMER) inventory = post_cleanup_authority_inventory(profile, CUSTOMER) if receipt.phase != "committed" or not inventory.terminal: raise RuntimeError(f"cleanup terminal inventory unavailable: {asdict(inventory)}") cleanup_journal = profile / f"data/customer-cleanup/{CUSTOMER}.journal.jsonl" evidence_contract.record( "cleanup_resume_terminal", { "fault_phase": "copied_verified", "fault_seen": fault_seen, "operation_id": receipt.operation_id, "terminal_phase": receipt.phase, "journal_sha256": hashlib.sha256(cleanup_journal.read_bytes()).hexdigest(), "terminal": inventory.terminal, "active_count": inventory.active_count, "pending_count": inventory.pending_count, "unknown_count": inventory.unknown_count, "orphan_count": inventory.orphan_count, }, ) task26_chain_head = evidence_contract.finish() authority = append_candidate_authority( profile, candidate_digest=candidate_digest, action="qualify", historical_pass_digest=task26_chain_head, reason="same-invocation Task26 successor qualification", ) return { "profile": str(profile), "operation_id": receipt.operation_id, "tree_digest": provenance["tree_digest"], "source_tree_digest": provenance["tree_digest"], "candidate_digest": candidate_digest, "candidate_product_binding_sha256": candidate_binding["binding_sha256"], "hermes_wheel_sha256": cast(dict[str, object], candidate_binding["derivation_inputs"])["hermes_wheel_sha256"], "profile_wheel_sha256": cast(dict[str, object], candidate_binding["derivation_inputs"])["profile_wheel_sha256"], "runtime_mode": "source" if installed_runtime is None else "installed", "task26_chain_head": task26_chain_head, "authority_registry_head": authority["registry_head_sha256"], "authority_ledger_head": authority["ledger_head_sha256"], } def secure_root(requested: Path | None) -> Path: if requested is None: path = Path(tempfile.mkdtemp(prefix="source-golden-path-")) else: path = requested.absolute() path.mkdir(mode=0o700) path.chmod(0o700) return path def main() -> int: parser = argparse.ArgumentParser() mode = parser.add_mutually_exclusive_group() mode.add_argument("--source", type=Path) mode.add_argument("--installed-venv", type=Path) parser.add_argument("--installed-site-packages", type=Path) parser.add_argument("--profile-wheel", type=Path) parser.add_argument("--hermes-wheel", type=Path) parser.add_argument("--product-binding", type=Path) parser.add_argument("--evidence-root", type=Path) args = parser.parse_args() source = (args.source or DEFAULT_SOURCE).absolute() if source.is_symlink() or not (source / "checkin_cli/customer_cleanup.py").is_file(): print(json.dumps({"status": "SOURCE_GOLDEN_PATH_FAIL", "blocker": "profile_source_unavailable", "path": str(source)}, sort_keys=True), file=sys.stderr) return 2 root: Path | None = None try: candidate_digest, candidate_binding = load_product_binding( args.product_binding ) installed_values = ( args.installed_site_packages, args.profile_wheel, args.hermes_wheel, ) if args.installed_venv is None and any(value is not None for value in installed_values): parser.error("installed wheel inputs require --installed-venv") if args.installed_venv is not None and any(value is None for value in installed_values): parser.error("installed mode requires --installed-site-packages, --profile-wheel, and --hermes-wheel") installed_runtime = None if args.installed_venv is not None: installed_runtime = collect_installed_runtime( venv=args.installed_venv, site_packages=cast(Path, args.installed_site_packages), profile_wheel=cast(Path, args.profile_wheel), hermes_wheel=cast(Path, args.hermes_wheel), ) root = secure_root(args.evidence_root) result = asyncio.run( drive( root, source, candidate_digest, candidate_binding, installed_runtime, ) ) print(json.dumps(result, sort_keys=True, separators=(",", ":"))) return 0 except Exception as exc: if root is not None: shutil.rmtree(root, ignore_errors=True) print(json.dumps({"status": "SOURCE_GOLDEN_PATH_FAIL", "blocker": f"{type(exc).__name__}: {exc}"}, ensure_ascii=False, sort_keys=True), file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())