#!/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 REPO = Path(__file__).resolve().parents[1] DEFAULT_SOURCE = Path("/home/cube/.cache/task26-strict-successor-1786976146/src-p") ANSWERS = ( "1990-01-01", "남성", "180", "80", "보통", "주 3회 운동", "감량", "75", "2026-12-01", "없음", "없음", "없음", "없음", "균형식", "없음", "없음", "아니오", "아니오", "가능", "보통", "3", "없음", ) 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 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 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) -> tuple[Path, Path]: registry_path = profile / "customers/registry.json" config = profile / "config.yaml" config.write_text("platforms: {}\n", encoding="utf-8") config.chmod(0o600) deployment = { "candidate_digest": candidate_digest, "candidate_core_digest": candidate_digest, "candidate_inventory_digest": candidate_digest, "hermes_wheel_sha256": candidate_digest, "profile_wheel_sha256": candidate_digest, } 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", **deployment, "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, installed_runtime: dict[str, object] | None = None, ) -> dict[str, object]: provenance = source_provenance(source) if installed_runtime is None: authority_root = source candidate_digest = str(provenance["tree_digest"]) 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) binding = { "schema": "installed-golden-candidate-binding-v1", "runtime": installed_runtime, "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(), } candidate_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_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_runtime import TelegramNutritionOnboardingRuntime private_write(profile / "source-provenance.json", provenance) 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, "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, }, ) 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) 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)) for index, answer in enumerate(ANSWERS, 1): current = runtime._service(CUSTOMER).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) if not handled: raise RuntimeError(f"handle_text did not consume answer {index}") service = runtime._service(CUSTOMER) 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) 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") coordinator.persist_approved_delivery_card(draft_id, owner, card_route=owner, message_id="700", nonce=nonce, payload_digest=payload_digest, **pins(approved)) auth = DeliveryLaunchAuthorization(*([candidate_digest] * 8)) 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") 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") 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)}") return { "profile": str(profile), "operation_id": receipt.operation_id, "tree_digest": provenance["tree_digest"], "candidate_digest": candidate_digest, "runtime_mode": "source" if installed_runtime is None else "installed", } 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("--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: 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, 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())