#!/usr/bin/env python3 """Continuous, event-first Task26 lifecycle observer and role-handoff controller.""" from __future__ import annotations import argparse import base64 import ctypes import hashlib import importlib import json import os import selectors import stat import struct import sys import time from dataclasses import dataclass from datetime import date from pathlib import Path from typing import Final SUCCESSOR: Final = "6aaef77de87489c3be493978645781b24b55ab07513ff1f880164741d0bd3c73" CORE: Final = "f759460a42155925ef68d75a03f4c86cffe6d1b6ec774bd61efb73174e10b60c" WHEEL: Final = "33688875a0d1ce20955bd8272257ee34d84cb138f6a23a43e5d16682368ae5e5" PLAN: Final = "d816dbbd6a27d5826b251d279a9831de096ef7cd79b5fd131ceff7253b7369be" GOLDEN: Final = "8416cd7c83cab5540af2e1e01dbcc5b1bdfd49a59d8ef21ea49512c28d749218" RECOVERY: Final = "a2e15845764e7041957ae4518096fe73f87ec35b57021980cdd4df3ea18ac00d" CONSENT: Final = "5e3067fe3c865224fc73ff5a9a65058c87960f05d9af53bddccaf6f32dc83939" SESSION: Final = "cb_PCczfFXoI4GjvCxLBs1oRA" SID: Final = "53b4f95b5b4db9a611128976a7b951bbe22d9619e05b2ff07fadf5b02f9a5380" KEY: Final = "task26_live_2e_r2_20260815_8527916639" CUSTOMER: Final = "8527916639" OWNER: Final = "8693203710" FIELDS: Final = ( "date_of_birth", "equation_sex_basis", "height_cm", "weight_kg", "activity_category", "activity_rationale", "goal_type", "target_weight_kg", "target_date", "allergies", "intolerances", "religious_ethical_exclusions", "disliked_foods", "dietary_preferences", "conditions", "medications", "pregnancy_breastfeeding", "eating_disorder_risk", "cooking_access", "budget_band", "meal_count", "schedule_constraints", ) RUNTIME: Final = { "gateway/platforms/dualcoach_activation_cutover.py": "95637abce4742d583534cf406b038f7c862ad0a72d10a6bab24d2452dcb0f776", "gateway/platforms/dualcoach_admin.py": "8c7065277d5d96ce1012ac0c098d53288bba66c44c0674704be4072fe0a63951", "gateway/platforms/telegram_customer_bootstrap.py": "8ce2fe8770c2b480083e9d44237d80892ccddcaec06acca14db18576a2ee0088", "gateway/platforms/telegram.py": "b41060dea28eb3bbb83217068f5218d5df2c879dba00e73c9ca4e577a6049dad", "gateway/platforms/nutrition_onboarding_reconciliation.py": "8fb6ba6f855e7508c525447cdec482ceb240f8c0a5827b2ca0ce4268c6880bb0", "gateway/platforms/telegram_nutrition_onboarding_copy.py": "8afb45c5cca082fd85cf07ab43dc5d1062c34678fd268a5722e313ce35cffd8b", "gateway/platforms/telegram_nutrition_onboarding_runtime.py": "3c24d49d656c623edfb95fcfbc7fb0f92365fc123420df66f733f9dbe39d704f", "gateway/platforms/telegram_nutrition_onboarding_runtime_publication.py": "f04ea498ad12a257fb1322da4c8cfd2f91fc9403f0b978617fcc00df41729209", } SITE = Path( "/home/cube/projects/richard/hermes-agent/.venv/lib/python3.12/site-packages" ) SCHEMA = "task26-continuous-lifecycle-v6" CHECKLIST: Final = Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-activation-checklist-6aaef77d-task26-live-2e-r2/activation-checklist-receipt.json") CHECKLIST_SHA256: Final = "b093aeb888ce7325cddd7d56d52a7e4a7ec8a180ac80eef21fea9453a8ef0e8e" VALIDATION_RECEIPT: Final = Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-activation-checklist-6aaef77d-task26-live-2e-r2/checklist-validation-receipt.json") VALIDATION_RECEIPT_SHA256: Final = "acb21c8c02260497c197427cfc93417f2be65c8746768ba788bb33400e3e6e78" PROFILE_WHEEL: Final = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" READINESS_RECEIPT_SHA256: Final = "10cf17fba0e0f60c43d0a379721b71a945ef2c4c168fff8f613c01448297edb4" READINESS_DIGEST: Final = "857f37c973156c3dcc574dc1dd566f3e4724ddf29c2d44f552184d56e94a1a99" OWNER_REVIEW_RECEIPT: Final = "c6ecd6cb55fd68989b04d60f5ef13491a821e261e43dc7c231d0d5fefa92262e" AUTHORITY_DIGEST: Final = "8856ca5146cd74a6e9ed05dcd06747a694cd4f8415ee0575d673d04c9e310f09" DOMAIN_PACKAGE_ROOT: Final = Path( "/home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli" ) DOMAIN_MODULE_ROOT: Final = DOMAIN_PACKAGE_ROOT / "checkin_cli" LEGACY_ACTIVITY: Final = ( "활동 수준: 보통. 주 3회, 회당 60분 정도의 중간 강도 근력운동을 희망합니다\n" " . 평소 걷기·이동은 하루 약 6,000~8,000보 수준입니다." ) DETAILED_RATIONALE: Final = ( "주 3회, 회당 60분 정도의 중간 강도 근력운동을 희망합니다\n" " . 평소 걷기·이동은 하루 약 6,000~8,000보 수준입니다." ) EV = struct.Struct("iIII") MASK = 0x100 | 0x80 | 0x8 OVER = 0x4000 IGNORED = 0x8000 SKIP = { "rehearsal-reset-archives", "profile-reset-archives", "post-lifecycle-cleanup-archives", "dualcoach-provider-auth", "global", "recovery-audits", } class LifecycleError(RuntimeError): pass class LifecycleTimeout(LifecycleError): pass @dataclass(frozen=True) class Prompt: status: str role: str actor: str route: tuple[str, str] message_id: str | None action: str command: str = "" detail: str = "" def doc(self) -> dict[str, object]: return { "status": self.status, "role": self.role, "actor": self.actor, "route": list(self.route), "message_id": self.message_id, "action": self.action, "command": self.command, "detail": self.detail, } def sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def load(p: Path, default: object = None) -> object: if not p.exists(): return default if p.is_symlink(): raise LifecycleError(f"symlink authority: {p}") s = p.stat() if ( not stat.S_ISREG(s.st_mode) or s.st_uid != os.getuid() or stat.S_IMODE(s.st_mode) & 0o077 ): raise LifecycleError(f"unsafe authority: {p}") try: return json.loads(p.read_bytes()) except json.JSONDecodeError as e: raise LifecycleError(f"invalid JSON: {p}") from e def rows(v: object) -> list[dict[str, object]]: if isinstance(v, list): return [x for x in v if isinstance(x, dict)] if isinstance(v, dict): for k in ("records", "generations", "items"): if isinstance(v.get(k), list): return [x for x in v[k] if isinstance(x, dict)] return [] def callback(action: str, generation: int) -> str: h = base64.urlsafe_b64encode(bytes.fromhex(SID)).decode().rstrip("=") return f"non2:{action}:{generation}:{h}" def exact_cutover() -> str: return ( "/home/cube/projects/richard/hermes-agent/.venv/bin/dualcoach_admin customer activate --profile-root /home/cube/.hermes/profiles/dualcoachtest --data-root /home/cube/.hermes/profiles/dualcoachtest/data --customer-id " + KEY + " --checklist-evidence '" + str(CHECKLIST) + "' --bootstrap-session " + SESSION + " --expected-generation 5 --package-root /home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli" ) def _one( seq: list[dict[str, object]], field: str, value: str ) -> dict[str, object] | None: x = [r for r in seq if str(r.get(field, "")) == value] return x[0] if len(x) == 1 else None def capture(root: Path) -> dict[str, object]: boot = load(root / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json", {}) br = _one( rows(boot.get("sessions", []) if isinstance(boot, dict) else []), "session_id", SESSION, ) reg = load(root / "customers/registry.json", {}) cr = _one( rows(reg.get("customers", []) if isinstance(reg, dict) else []), "customer_key", KEY, ) nr = root / "data/customers" / KEY / "nutrition-onboarding" wf = load(nr / "ready.json", None) or load(nr / "transient/workflow.json", None) baseline = load(nr / "baseline-candidate.json", None) pubs = load(nr / "session.json", {}) pr = _one( rows(pubs.get("sessions", {}) if isinstance(pubs, dict) else {}), "session_id", SESSION, ) # session.json stores a keyed object, not records if pr is None and isinstance(pubs, dict) and isinstance(pubs.get("sessions"), dict): x = pubs["sessions"].get(SESSION) pr = x if isinstance(x, dict) else None out = load(root / "data/onboarding/telegram-publication-outbox-v1/ledger.json", {}) oc = load( root / "data/onboarding/telegram-publication-outbox-v1/owner-callbacks.json", {} ) wr = root / "data/customers" / KEY / "wizard" bind = load(wr / "telegram-bindings.json", None) wizard = None if isinstance(bind, dict): active = bind.get("active_session_id") bs = rows(bind.get("bindings", [])) b = _one(bs, "session_id", str(active)) if active else None draft = load(wr / "drafts" / f"{active}.json", None) if active else None if isinstance(b, dict): wizard = { "session_id": active, "version": b.get("version"), "step": b.get("step"), "message_id": b.get("message_id"), "route": [str(b.get("chat_id")), str(b.get("topic_id"))], "finalized_event_id": draft.get("finalized_event_id") if isinstance(draft, dict) else None, } gens = load(root / "data/owner-actions/draft-generations.json", []) cards = load(root / "data/owner-actions/draft-generation-cards.json", {}) drafts = load(root / "data/owner-actions/drafts.json", {}) deliveries = load(root / "data/owner-actions/draft-deliveries.json", {}) audit = [] ap = root / "data/customer-activation-audit.jsonl" if ap.exists(): audit = [json.loads(x) for x in ap.read_text().splitlines() if x.strip()] owner = reg.get("owner") if isinstance(reg, dict) else None return { "bootstrap": br, "customer": cr, "owner": owner, "workflow": wf, "baseline": baseline, "publication": pr, "outbox": rows(out), "owner_callbacks": rows(oc), "wizard": wizard, "generations": rows(gens), "cards": list(cards.values()) if isinstance(cards, dict) else rows(cards), "drafts": drafts if isinstance(drafts, dict) else {}, "deliveries": deliveries if isinstance(deliveries, dict) else {}, "activation": audit, } def validate_snapshot(s: dict[str, object]) -> None: b = s.get("bootstrap") c = s.get("customer") owner = s.get("owner") if ( not isinstance(b, dict) or b.get("session_id") != SESSION or b.get("sid_hash") != SID or b.get("state") not in {"AWAITING_ACTIVATION", "ACTIVE"} or b.get("generation") not in {5, 6} ): raise LifecycleError("bootstrap authority drift") if owner != {"user_id": OWNER, "chat_id": OWNER, "topic_id": "0"}: raise LifecycleError("owner member authority drift") if ( not isinstance(c, dict) or c.get("customer_key", KEY) != KEY or ( c.get("telegram") not in ({"user_id": CUSTOMER, "chat_id": CUSTOMER, "topic_id": "0"}, None) and c.get("route") != [CUSTOMER, CUSTOMER, "0"] ) ): raise LifecycleError("customer route drift") seen = set() outbox = s.get("outbox", []) if not isinstance(outbox, list): raise LifecycleError("outbox schema drift") for r in outbox: if not isinstance(r, dict): raise LifecycleError("outbox schema drift") k = (r.get("session_id"), r.get("generation")) if k in seen: raise LifecycleError("duplicate publication generation") seen.add(k) role = r.get("role") route = r.get("route") if ( role == "customer" and route != [CUSTOMER, "0"] or role == "owner" and route != [OWNER, "0"] ): raise LifecycleError("publication role route drift") d = s.get("deliveries", {}) if isinstance(d, dict): terminal = [ r for r in d.values() if isinstance(r, dict) and r.get("customer_key") == KEY and r.get("status") == "sent_audited" ] if len(terminal) > 1: raise LifecycleError("duplicate sent_audited delivery") def validate_transition(a: dict[str, object], b: dict[str, object]) -> None: validate_snapshot(b) aw = a.get("workflow") bw = b.get("workflow") if ( isinstance(aw, dict) and isinstance(bw, dict) and aw.get("state") == "collecting" and bw.get("state") == "collecting" and int(bw.get("cursor", 0)) < int(aw.get("cursor", 0)) ): raise LifecycleError("onboarding cursor regressed") ap = a.get("publication") bp = b.get("publication") if ( isinstance(ap, dict) and isinstance(bp, dict) and int(bp.get("generation", 0)) < int(ap.get("generation", 0)) ): raise LifecycleError("publication generation regressed") av = a.get("wizard") bv = b.get("wizard") if ( isinstance(av, dict) and isinstance(bv, dict) and av.get("session_id") == bv.get("session_id") and int(bv.get("version", 0)) < int(av.get("version", 0)) ): raise LifecycleError("checkin version regressed") def canonical_digest(value: object) -> str: encoded = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() return hashlib.sha256(encoded).hexdigest() def committed_publication( s: dict[str, object], role: str = "customer" ) -> dict[str, object] | None: publication = s.get("publication") if not isinstance(publication, dict) or publication.get("state") != "COMMITTED": return None payload = publication.get("payload") generation = publication.get("generation") message_id = publication.get("message_id") if ( not isinstance(payload, dict) or type(generation) is not int or generation < 0 or type(message_id) is not int or message_id <= 0 ): return None outbox = s.get("outbox") if not isinstance(outbox, list): return None matches = [ record for record in outbox if isinstance(record, dict) and record.get("session_id") == SESSION and record.get("state") == "COMMITTED" and record.get("generation") == generation and record.get("message_id") == message_id and record.get("payload") == payload and record.get("role") == role and record.get("route") == ([CUSTOMER, "0"] if role == "customer" else [OWNER, "0"]) ] committed_records = [ record for record in outbox if isinstance(record, dict) and record.get("session_id") == SESSION and record.get("state") == "COMMITTED" and type(record.get("generation")) is int and type(record.get("message_id")) is int ] latest = max( (int(record["generation"]) for record in committed_records), default=-1, ) prior_message = max( ( int(record["message_id"]) for record in committed_records if int(record["generation"]) < generation ), default=0, ) return ( publication if len(matches) == 1 and generation == latest and message_id > prior_message else None ) def reconciliation_authority( workflow: dict[str, object], ) -> tuple[dict[str, object], str, list[dict[str, object]], int] | None: answers = workflow.get("answers") record = workflow.get("reconciliation") if not isinstance(answers, dict) or not isinstance(record, dict): return None answers_digest = canonical_digest(answers) clarifications = record.get("clarifications") current_index = record.get("current_index") digest = record.get("digest") unsigned = {key: value for key, value in record.items() if key != "digest"} if ( record.get("state") not in {"clarifying", "resolved"} or record.get("answers_digest") != answers_digest or not isinstance(clarifications, list) or type(current_index) is not int or not 0 <= current_index <= len(clarifications) or not isinstance(digest, str) or canonical_digest(unsigned) != digest or (record.get("state") == "clarifying") != (current_index < len(clarifications)) ): return None typed: list[dict[str, object]] = [] seen: set[str] = set() for item in clarifications: if ( not isinstance(item, dict) or set(item) != {"field", "kind", "question_ko"} or item.get("field") not in FIELDS or item.get("field") in seen or item.get("kind") not in {"ambiguity", "contradiction"} or not isinstance(item.get("question_ko"), str) or not item["question_ko"] ): return None seen.add(str(item["field"])) typed.append(item) return answers, answers_digest, typed, current_index def summary_quality(answers: dict[str, object]) -> tuple[bool, str]: goal_value = answers.get("goal_type") goal = ( {"loss": "감량", "maintain": "유지", "gain": "증량"}.get(goal_value) if isinstance(goal_value, str) else None ) activity_value = answers.get("activity_category") activity = ( { "sedentary": "좌식", "light": "가벼움", "moderate": "보통", "very_active": "매우 활동적", "extra_active": "최고 활동", }.get(activity_value) if isinstance(activity_value, str) else None ) if goal is None: return False, "canonical Korean goal is unavailable" if activity is None: return False, "canonical Korean activity is unavailable" for field in ( "allergies", "intolerances", "religious_ethical_exclusions", "disliked_foods", "dietary_preferences", "conditions", "medications", ): value = answers.get(field) if not isinstance(value, dict) or value.get("status") not in { "none", "provided", }: return False, f"structured answer unavailable: {field}" if answers.get("pregnancy_breastfeeding") is not False: return False, "health review input requires review" if answers.get("eating_disorder_risk") is not False: return False, "health review input requires review" return True, f"goal={goal}; activity={activity}; health=no-review" def authoritative_summary(answers: dict[str, object]) -> str: def value(field: str, default: str = "미입력") -> str: raw = answers.get(field) if raw is None or raw == "" or raw == []: return default if isinstance(raw, list): return ", ".join(str(item) for item in raw) or default return str(raw) def structured(field: str) -> str: raw = answers.get(field) if not isinstance(raw, dict): return "미입력" if raw.get("status") == "none": return "없음" items = raw.get("items") if raw.get("status") != "provided" or not isinstance(items, list): return "미입력" return ( ", ".join(item for item in items if isinstance(item, str) and item) or "미입력" ) goal = {"loss": "감량", "maintain": "유지", "gain": "증량"}.get( str(answers.get("goal_type")) ) activity = { "sedentary": "좌식", "light": "가벼움", "moderate": "보통", "very_active": "매우 활동적", "extra_active": "최고 활동", }.get(str(answers.get("activity_category"))) restrictions = " · ".join( f"{label} {structured(field)}" for label, field in ( ("알레르기", "allergies"), ("불내증", "intolerances"), ("종교·윤리 제한", "religious_ethical_exclusions"), ("비선호 식품", "disliked_foods"), ("식품 선호", "dietary_preferences"), ) ) return "\n".join( ( f"- 목표: {goal or '미입력'}", f"- 활동: {activity or '미입력'} · {value('activity_rationale')}", f"- 식사: 하루 {value('meal_count')}끼", f"- 조리·예산: {value('cooking_access')} · {value('budget_band')}", f"- 식사 시간: {value('schedule_constraints')}", f"- 알레르기·불내증·식품 제한: {restrictions}", "- 건강 관련 입력: 별도 검토 입력이 없습니다.", ) ) def candidate_attestation_text( workflow: dict[str, object], answers: dict[str, object], current_index: int ) -> str | None: reconciliation = workflow.get("reconciliation") if not isinstance(reconciliation, dict): return None advisory = reconciliation.get("advisory") if not isinstance(advisory, dict): return None summary = ( authoritative_summary(answers) if current_index else advisory.get("summary_ko") ) if not isinstance(summary, str) or not summary: return None lines = ["입력 내용을 정리했어요", "", summary] for title, key in ( ("확인된 내용", "facts_ko"), ("안전 관련 참고사항", "safety_observations_ko"), ): values = () if current_index else advisory.get(key) if values: if not isinstance(values, (list, tuple)) or not all( isinstance(item, str) for item in values ): return None lines.extend(("", title, *(f"- {item}" for item in values))) lines.extend( ( "", "이 요약은 참고용이며 안전 판정·승인·활성화를 대신하지 않습니다.", "내용이 맞으면 아래 확인 버튼을 눌러 주세요.", ) ) return ( "\n".join(lines) + "\n\n아래 버튼을 누르면 입력 확인이 저장되고 운영자 검토로 넘어갑니다.\n" "이 단계에서 영양 코칭이 시작되거나 활성화되지는 않습니다." ) def candidate_body_bound( workflow: dict[str, object], answers: dict[str, object], current_index: int, payload: dict[str, object], ) -> bool: text = candidate_attestation_text(workflow, answers, current_index) return bool( text is not None and payload.get("body_digest") == hashlib.sha256(text.encode()).hexdigest() ) def shipped_baseline_validation( answers: dict[str, object], ) -> tuple[bool, str | None]: if set(answers) != set(FIELDS) or answers.get("activity_category") != "moderate": return False, None rationale = answers.get("activity_rationale") if not isinstance(rationale, str): return False, None package = str(DOMAIN_PACKAGE_ROOT) if package not in sys.path: sys.path.insert(0, package) try: models = importlib.import_module("checkin_cli.nutrition_onboarding_models") source = Path(str(getattr(models, "__file__", ""))).resolve() if not source.is_relative_to(DOMAIN_MODULE_ROOT.resolve()): return False, None payload = dict(answers) payload.update( schema_version="1.0", customer_key=KEY, session_status="completed", review_decision="pending", publication_status="unpublished", ) baseline = models.build_onboarding_baseline(payload, as_of=date.today()) normalized = getattr(baseline, "activity_rationale", None) except (ImportError, OSError, TypeError, ValueError): return False, None if not isinstance(normalized, str) or not normalized or len(normalized) > 500: return False, None return True, normalized def baseline_valid_answers(answers: dict[str, object]) -> bool: valid, _ = shipped_baseline_validation(answers) return valid def normalized_rewind(workflow: dict[str, object]) -> bool: answers = workflow.get("answers") return bool( isinstance(answers, dict) and set(answers) == set(FIELDS[:-1]) and answers.get("activity_category") == "moderate" and answers.get("activity_rationale") == DETAILED_RATIONALE and "schedule_constraints" not in answers and "reconciliation" not in workflow ) def publication_summary_quality( workflow: dict[str, object], answers: dict[str, object], current_index: int, payload: dict[str, object], ) -> tuple[bool, str]: text = candidate_attestation_text(workflow, answers, current_index) if text is None or not candidate_body_bound( workflow, answers, current_index, payload ): return False, "publication body is not candidate-render-bound" forbidden = ( "{", "items", "status", "미입력", "수정 전", "실제 총활동량은 불명확", "위험 신호", "maintain", "moderate", "sedentary", "very_active", "extra_active", ) if any(token in text for token in forbidden): return False, "publication contains forbidden internal or stale copy" required = ( "- 목표: 유지", "- 활동: 보통", "주 3회", "60분", "6,000~8,000보", "별도 검토 입력이 없습니다.", ) if not all(token in text for token in required): return False, "publication lacks Korean goal/activity/no-review summary" return True, "candidate-render-bound Korean summary passed" def candidate_revision_handoff(s: dict[str, object]) -> dict[str, object] | None: workflow = s.get("workflow") publication = committed_publication(s) if ( not isinstance(workflow, dict) or workflow.get("state") != "customer_attestation" or publication is None ): return None authority = reconciliation_authority(workflow) payload = publication.get("payload") if authority is None or not isinstance(payload, dict): return None answers, answers_digest, _, current_index = authority quality_ok, reason = summary_quality(answers) field = FIELDS[-1] generation = publication.get("generation") message_id = publication.get("message_id") if ( quality_ok or type(generation) is not int or generation < 25 or payload.get("state") != "customer_attestation" or payload.get("answers_digest") != answers_digest or "clarification_field" in payload or not candidate_body_bound(workflow, answers, current_index, payload) ): return None if type(generation) is not int or type(message_id) is not int: return None return { "status": "READY_CUSTOMER_REVISION", "role": "customer", "actor": CUSTOMER, "route": [CUSTOMER, "0"], "message_id": str(message_id), "action": callback("revise", generation), "next_field": field, "next_value": answers.get(field), "detail": "press the existing 수정 button, then resubmit the unchanged final canonical answer; " + reason, } def wait_matching_role_publication(role: str, state: str) -> Prompt: return Prompt( "WAIT_MATCHING_ROLE_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", detail=f"wait for the latest committed {role} publication matching {state}", ) def validate_owner_baseline(s: dict[str, object], workflow: dict[str, object]) -> str: digest = workflow.get("baseline_digest") baseline = s.get("baseline") if ( not isinstance(digest, str) or len(digest) != 64 or not isinstance(baseline, dict) or baseline.get("digest") != digest or workflow.get("answers") != {} ): raise LifecycleError("owner review baseline authority drift") payload = baseline.get("baseline") source_answers = baseline.get("source_answers") reconciliation = workflow.get("reconciliation") if ( not isinstance(payload, dict) or canonical_digest(payload) != digest or not isinstance(source_answers, dict) or not isinstance(reconciliation, dict) or baseline.get("input_reconciliation_digest") != reconciliation.get("digest") or canonical_digest(source_answers) != reconciliation.get("answers_digest") ): raise LifecycleError("owner review baseline digest drift") valid, _ = shipped_baseline_validation(source_answers) if not valid: raise LifecycleError("owner review shipped baseline validation failed") return digest def classify(s: dict[str, object]) -> Prompt: validate_snapshot(s) b = s["bootstrap"] c = s["customer"] wf = s.get("workflow") assert isinstance(b, dict) and isinstance(c, dict) enabled = bool(c.get("enabled")) if b.get("state") == "AWAITING_ACTIVATION": if not isinstance(wf, dict): raise LifecycleError("onboarding workflow unavailable") state = str(wf.get("state")) if state == "collecting": cursor = wf.get("cursor") if type(cursor) is not int or not 0 <= cursor < len(FIELDS): raise LifecycleError("collection authority drift") committed_collection = committed_publication(s) if committed_collection is None: return Prompt( "WAIT_MATCHING_CUSTOMER_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", ) payload = committed_collection.get("payload") generation = committed_collection.get("generation") message_id = committed_collection.get("message_id") if ( not isinstance(payload, dict) or payload.get("state") != "collecting" or type(generation) is not int or type(message_id) is not int ): return Prompt( "WAIT_MATCHING_CUSTOMER_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", ) if cursor == len(FIELDS) - 1: if generation < 26 or not normalized_rewind(wf): raise LifecycleError("normalized schedule rewind authority drift") return Prompt( "READY_CUSTOMER_RESUBMIT", "customer", CUSTOMER, (CUSTOMER, "0"), str(message_id), "submit_answer:schedule_constraints", "없음", "reply to the committed normalized rewind prompt with the exact unchanged value", ) return Prompt( "READY_CUSTOMER_ONBOARDING", "customer", CUSTOMER, (CUSTOMER, "0"), str(message_id), f"submit_answer:{FIELDS[cursor]}", detail=f"answer onboarding question {cursor + 1}/22 by replying to the committed prompt", ) committed = committed_publication(s) if state == "customer_attestation": if committed is None: return Prompt( "WAIT_MATCHING_CUSTOMER_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", detail="workflow state alone never authorizes a callback", ) payload = committed["payload"] assert isinstance(payload, dict) authority = reconciliation_authority(wf) if authority is None: return Prompt( "WAIT_RECONCILIATION_AUTHORITY", "system", "system", ("local", "0"), None, "reconciliation:auto", ) answers, answers_digest, clarifications, current_index = authority if ( payload.get("state") != "customer_attestation" or payload.get("answers_digest") != answers_digest ): return Prompt( "WAIT_MATCHING_CUSTOMER_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", detail="resolved authority precedes its matching committed publication", ) generation = committed.get("generation") message_id = committed.get("message_id") if type(generation) is not int or type(message_id) is not int: raise LifecycleError("committed publication identity drift") gen = generation mid = str(message_id) reconciliation = wf["reconciliation"] assert isinstance(reconciliation, dict) if reconciliation["state"] == "clarifying": field = clarifications[current_index]["field"] if payload.get("clarification_field") != field: return Prompt( "WAIT_MATCHING_CUSTOMER_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", detail="clarification field is not publication-bound", ) return Prompt( "READY_CUSTOMER_CLARIFICATION", "customer", CUSTOMER, (CUSTOMER, "0"), mid, f"submit_clarification:{field}", detail="reply only to the committed field-bound clarification", ) if "clarification_field" in payload: return Prompt( "WAIT_MATCHING_CUSTOMER_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", detail="resolved authority still has a clarification publication", ) body_bound = candidate_body_bound(wf, answers, current_index, payload) if not body_bound: if ( gen == 24 and message_id == 207 and answers.get("activity_category") == LEGACY_ACTIVITY ): return Prompt( "WAIT_DEPLOY", "operator", "operator", ("local", "0"), None, "deploy-final-candidate", detail="deploy 4e9962be; startup must commit changed summary generation 25 before any UI action", ) return Prompt( "WAIT_CANDIDATE_PUBLICATION", "system", "system", ("local", "0"), None, "publication:auto", detail="candidate-bound summary body is not yet committed", ) quality_ok, quality_detail = summary_quality(answers) if not quality_ok: if gen == 25 and answers.get("activity_category") == LEGACY_ACTIVITY: return Prompt( "READY_CUSTOMER_REVISION", "customer", CUSTOMER, (CUSTOMER, "0"), mid, callback("revise", gen), "", "press latest 수정 once; generation 24/message 207 is stale", ) return Prompt( "STOP_CUSTOMER_SUMMARY_QUALITY", "operator", "operator", ("local", "0"), mid, "do-not-attest", detail=quality_detail, ) if gen < 27 or not baseline_valid_answers(answers): return Prompt( "WAIT_CANONICAL_RECONCILIATION", "system", "system", ("local", "0"), None, "reconciliation:auto", detail="normalized answers are not yet baseline-valid in a new publication", ) quality_ok, quality_detail = publication_summary_quality( wf, answers, current_index, payload ) if not quality_ok: return Prompt( "STOP_CUSTOMER_SUMMARY_QUALITY", "operator", "operator", ("local", "0"), mid, "do-not-attest", detail=quality_detail, ) return Prompt( "READY_CUSTOMER_ATTEST", "customer", CUSTOMER, (CUSTOMER, "0"), mid, callback("attest", gen), "", "press the matching committed attestation button once; " + quality_detail, ) committed = committed_publication(s, "owner") if not isinstance(committed, dict): return wait_matching_role_publication("owner", state) payload = committed.get("payload") generation = committed.get("generation") message_id = committed.get("message_id") if ( not isinstance(payload, dict) or payload.get("state") != state or type(generation) is not int or type(message_id) is not int ): return wait_matching_role_publication("owner", state) gen = generation mid = str(message_id) if state == "owner_review": validate_owner_baseline(s, wf) if b.get("owner_id") != OWNER: raise LifecycleError("owner member authority drift") if ( gen < 29 or payload.get("body_digest") != "4903900d9c98c474006e512d14207439262ce405b696f4c6765dd65e1927cbd9" ): return wait_matching_role_publication("owner", state) return Prompt( "READY_OWNER_REVIEW", "owner", OWNER, (OWNER, "0"), mid, callback("owner_ok", gen), "", "press Approve once; stale customer callbacks are retired", ) if state == "safety_hold": return Prompt( "STOP_SAFETY_HOLD", "owner", OWNER, (OWNER, "0"), mid, "clinical-review-required", ) if state in {"ready", "finalizing"}: if state == "finalizing": return Prompt( "WAIT_AUTOMATIC_READINESS", "system", "system", ("local", "0"), None, "automatic-finalization", ) return Prompt( "READY_OPERATOR_ACTIVATION_CUTOVER", "operator", "operator", ("local", "0"), None, "dualcoach_admin:customer:activate", exact_cutover(), "run only after checklist evidence exists; commits registry activation then bootstrap g5->g6 ACTIVE", ) raise LifecycleError("unknown onboarding outcome") if b.get("state") != "ACTIVE" or b.get("generation") != 6: raise LifecycleError("activation cutover is not atomically committed") d = s.get("deliveries", {}) terminal = [] if isinstance(d, dict): terminal = [ r for r in d.values() if isinstance(r, dict) and r.get("customer_key") == KEY and r.get("status") == "sent_audited" ] if terminal: if enabled: return Prompt( "READY_OPERATOR_DISABLE", "operator", "operator", ("local", "0"), None, "customer_admin:disable", detail="disable exact customer, then stop service and await cleanup handoff", ) return Prompt( "READY_CLEANUP_HANDOFF", "operator", "operator", ("local", "0"), None, "sealed-cleanup:dry-run", detail="service must be inactive; use sealed cleanup dry-run -> execute -> verify", ) if not enabled: raise LifecycleError("active lifecycle customer unexpectedly disabled") w = s.get("wizard") if not isinstance(w, dict): return Prompt( "READY_CUSTOMER_START_CHECKIN", "customer", CUSTOMER, (CUSTOMER, "0"), None, "send:checkin", detail="send checkin, then press Begin check-in", ) if w.get("route") != [CUSTOMER, "0"]: raise LifecycleError("checkin route drift") if w.get("step") != "summary": return Prompt( "READY_CUSTOMER_CHECKIN_ANSWER", "customer", CUSTOMER, (CUSTOMER, "0"), str(w.get("message_id")), f"checkin:{w.get('step')}", detail="answer the visible check-in prompt", ) if not w.get("finalized_event_id"): return Prompt( "READY_CUSTOMER_CHECKIN_SUMMARY", "customer", CUSTOMER, (CUSTOMER, "0"), str(w.get("message_id")), "checkin:finalize", detail="confirm the visible summary", ) drafts = s.get("drafts", {}) matched = [] if isinstance(drafts, dict): matched = [ (k, v) for k, v in drafts.items() if isinstance(v, dict) and v.get("customer_key") == KEY and v.get("session_id") == w.get("session_id") ] generations = s.get("generations", []) if not isinstance(generations, list): raise LifecycleError("generation authority drift") relevant_generations = [ g for g in generations if isinstance(g, dict) and g.get("customer_key") == KEY and g.get("session_id") == w.get("session_id") ] if not matched: if relevant_generations: raise LifecycleError("generated result lacks draft authority") return Prompt( "WAIT_AUTOMATIC_GENERATION", "system", "system", ("local", "0"), None, "generation:auto", detail="no human action; wait for generated owner card", ) # choose unique non-superseded/latest actionable draft active = [ x for x in matched if x[1].get("status") not in {"superseded", "sent", "held", "rejected"} ] if len(active) != 1: raise LifecycleError("draft identity is duplicate or unknown") did, dr = active[0] generation_tokens = { g.get("token") for g in relevant_generations if isinstance(g.get("token"), str) } cards_value = s.get("cards", []) if not isinstance(cards_value, list): raise LifecycleError("owner card authority drift") cards = [ x for x in cards_value if isinstance(x, dict) and ( x.get("draft_id") == did or x.get("token") in {did, dr.get("generation_token")} or x.get("token") in generation_tokens ) and x.get("state") == "published" ] if len(cards) != 1: raise LifecycleError("exact owner generation card unavailable") card = cards[0] dest = card.get("destination") if dest != {"user_id": OWNER, "chat_id": OWNER, "topic_id": "0"}: raise LifecycleError("owner card route drift") status = str(dr.get("status")) if status in {"created", "edited"}: return Prompt( "READY_OWNER_DRAFT_REVIEW", "owner", OWNER, (OWNER, "0"), str(card.get("message_id")), f"draft:{did}:review", detail="choose regenerate or edit if needed; otherwise approve once", ) if status == "approved": return Prompt( "READY_OWNER_EXPLICIT_SEND", "owner", OWNER, (OWNER, "0"), str(card.get("message_id")), f"draft:{did}:send", detail="approval never sends; press Send to customer exactly once", ) raise LifecycleError("unknown draft outcome") def event_hash(v: dict[str, object]) -> str: return hashlib.sha256( json.dumps(v, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def append_event(path: Path, event: dict[str, object]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.parent.chmod(0o700) previous = "0" * 64 if path.exists(): lines = path.read_text().splitlines() previous = json.loads(lines[-1])["event_sha256"] if lines else previous row = { "schema": SCHEMA, "sequence": sum(1 for _ in path.open()) if path.exists() else 0, "previous_event_sha256": previous, **event, } row["event_sha256"] = event_hash(row) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND | os.O_CLOEXEC, 0o600) os.write( fd, (json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n").encode() ) os.fsync(fd) os.close(fd) def secure(path: Path, v: dict[str, object]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.parent.chmod(0o700) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) os.write(fd, (json.dumps(v, sort_keys=True, separators=(",", ":")) + "\n").encode()) os.fsync(fd) os.close(fd) def bindings() -> dict[str, str]: return { "successor": SUCCESSOR, "core": CORE, "wheel": WHEEL, "plan": PLAN, "golden_v6": GOLDEN, "recovery_v6": RECOVERY, "consent_receipt": CONSENT, "profile_wheel": PROFILE_WHEEL, "activation_checklist": CHECKLIST_SHA256, "checklist_validation_receipt": VALIDATION_RECEIPT_SHA256, "readiness_receipt": READINESS_RECEIPT_SHA256, "readiness_digest": READINESS_DIGEST, "owner_review_receipt": OWNER_REVIEW_RECEIPT, "authority_digest": AUTHORITY_DIGEST, } def runtime_check() -> None: for p, h in RUNTIME.items(): if sha(SITE / p) != h: raise LifecycleError(f"installed runtime drift: {p}") def initfd() -> int: libc = ctypes.CDLL(None, use_errno=True) fd = libc.inotify_init1(os.O_CLOEXEC | os.O_NONBLOCK) if fd < 0: raise LifecycleError("inotify init failed") return int(fd) class Watcher: def __init__(self, root: Path): self.root = root.resolve() self.fd = initfd() self.watches: dict[int, Path] = {} def add_tree(self) -> None: for d, dirs, _ in os.walk(self.root): dirs[:] = [x for x in dirs if x not in SKIP] self.add(Path(d)) def add(self, p: Path) -> None: if p.is_symlink() or not p.is_dir(): return libc = ctypes.CDLL(None, use_errno=True) w = libc.inotify_add_watch(self.fd, os.fsencode(p), MASK) if w < 0: raise LifecycleError(f"watch failed: {p}") self.watches[int(w)] = p def wait(self, seconds: float) -> None: sel = selectors.DefaultSelector() sel.register(self.fd, selectors.EVENT_READ) try: ready = sel.select(seconds) if not ready: raise LifecycleTimeout("stage monotonic deadline expired") raw = os.read(self.fd, 65536) off = 0 while off + EV.size <= len(raw): w, mask, _, n = EV.unpack_from(raw, off) off += EV.size name = raw[off : off + n].split(b"\0", 1)[0].decode() off += n if mask & OVER: raise LifecycleError("inotify overflow") if mask & IGNORED: continue child = self.watches.get(w, Path()) / name if mask & 0x40000000 and child.is_dir(): self.add(child) finally: sel.close() def close(self) -> None: os.close(self.fd) def ready_doc(s: dict[str, object], prompt: Prompt) -> dict[str, object]: return { "schema": SCHEMA, "status": "READY_CONTINUOUS_LIFECYCLE", "deployment_readiness": ( "READY_FOR_DEPLOY" if prompt.status == "WAIT_DEPLOY" else "ARMED_POST_DEPLOY" ), "bindings": bindings(), "baseline_validation": ( "PASS_OWNER_BASELINE" if prompt.status == "READY_OWNER_REVIEW" else ( "PASS_SHIPPED_CHECKIN_CLI" if prompt.status == "READY_CUSTOMER_ATTEST" else "NOT_APPLICABLE" ) ), "initial_handoff": prompt.doc(), "candidate_summary_revision": candidate_revision_handoff(s), "authority_sha256": event_hash(s), "event_subscription": "recursive-inotify-before-actions", "per_stage_timeout_max": 1800, "total_timeout_max": 21600, "privacy": "no raw updates, tokens, customer answers, or provider content", } def run(a: argparse.Namespace) -> dict[str, object]: runtime_check() w = Watcher(a.profile) w.add_tree() s = capture(a.profile) validate_snapshot(s) p = classify(s) r = ready_doc(s, p) secure(a.ready, r) if a.mode == "arm-only": w.close() return r append_event( a.events, {"kind": "READY", "handoff": p.doc(), "authority_sha256": event_hash(s)}, ) print(json.dumps(r, sort_keys=True), flush=True) start = time.monotonic() stage = start last = p.status try: while True: remaining = min( a.stage_timeout - (time.monotonic() - stage), a.total_timeout - (time.monotonic() - start), ) if remaining <= 0: raise LifecycleTimeout("bounded lifecycle deadline expired") w.wait(remaining) n = capture(a.profile) if event_hash(n) == event_hash(s): continue validate_transition(s, n) q = classify(n) append_event( a.events, { "kind": "TRANSITION", "from": p.status, "to": q.status, "handoff": q.doc(), "authority_sha256": event_hash(n), }, ) if q.status != last: print(json.dumps(q.doc(), sort_keys=True), flush=True) stage = time.monotonic() last = q.status s, p = n, q if q.status == "READY_CLEANUP_HANDOFF": return { "schema": SCHEMA, "status": "PASS_CONTINUOUS_TO_CLEANUP_HANDOFF", "bindings": bindings(), "final_handoff": q.doc(), } finally: w.close() def main() -> int: ap = argparse.ArgumentParser() sp = ap.add_subparsers(dest="mode", required=True) for mode in ("arm-only", "observe"): x = sp.add_parser(mode) x.add_argument("--profile", type=Path, required=True) x.add_argument("--ready", type=Path, required=True) if mode == "observe": x.add_argument("--events", type=Path, required=True) x.add_argument("--stage-timeout", type=float, default=1800) x.add_argument("--total-timeout", type=float, default=21600) a = ap.parse_args() try: if a.mode == "observe" and ( not 0 < a.stage_timeout <= 1800 or not 0 < a.total_timeout <= 21600 ): raise LifecycleError("timeout bound invalid") print(json.dumps(run(a), sort_keys=True)) return 0 except LifecycleTimeout as e: print(f"FAIL: {e}", file=sys.stderr) return 3 except (LifecycleError, OSError, ValueError, KeyError) as e: print(f"FAIL: {e}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())