#!/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 json import os import selectors import stat import struct import sys import time from dataclasses import dataclass from pathlib import Path from typing import Final SUCCESSOR: Final = "30bcd6633875050aa4f56f49a8bb26cd616ffa2409fc04caef523a8a62d5d1cc" CORE: Final = "a113a57564a11710dffd238688dd73f8037eabaed0b48532b4bc90b2996e69fc" WHEEL: Final = "2fbd6c9ad9d4b5ee979e7671d44e32fd4769e8737e26d7459fe73c81a525c656" 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", } SITE = Path( "/home/cube/projects/richard/hermes-agent/.venv/lib/python3.12/site-packages" ) SCHEMA = "task26-continuous-lifecycle-v1" 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 '' --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) 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()] return { "bootstrap": br, "customer": cr, "workflow": wf, "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") 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 ( 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 classify(s: dict[str, object]) -> Prompt: validate_snapshot(s) b = s["bootstrap"] c = s["customer"] wf = s.get("workflow") p = s.get("publication") 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": cur = int(wf.get("cursor", 0)) if not 0 <= cur < len(FIELDS) or not isinstance(p, dict): raise LifecycleError("collection authority drift") return Prompt( "READY_CUSTOMER_ONBOARDING", "customer", CUSTOMER, (CUSTOMER, "0"), str(p.get("message_id")), f"submit_answer:{FIELDS[cur]}", detail=f"answer onboarding question {cur + 1}/22 by replying to the committed prompt", ) if not isinstance(p, dict): raise LifecycleError("onboarding publication unavailable") gen = int(p.get("generation", 0)) mid = str(p.get("message_id")) if state == "customer_attestation": return Prompt( "READY_CUSTOMER_ATTEST", "customer", CUSTOMER, (CUSTOMER, "0"), mid, callback("attest", gen), "", "press the attestation button once", ) if state == "owner_review": return Prompt( "READY_OWNER_ONBOARDING_APPROVE", "owner", OWNER, (OWNER, "0"), mid, callback("owner_ok", gen), "", "press Approve once; readiness finalization is automatic", ) 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, } 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", "bindings": bindings(), "initial_handoff": prompt.doc(), "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())