#!/usr/bin/env python3 """Independent no-network current-state and sealed-inventory verifier for observer v6.""" from __future__ import annotations import hashlib import json import os import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parent PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest") KEY = "task26_live_2e_r2_20260815_8527916639" SUCCESSOR = "6aaef77de87489c3be493978645781b24b55ab07513ff1f880164741d0bd3c73" CORE = "f759460a42155925ef68d75a03f4c86cffe6d1b6ec774bd61efb73174e10b60c" PROFILE_WHEEL = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" HERMES_WHEEL = "33688875a0d1ce20955bd8272257ee34d84cb138f6a23a43e5d16682368ae5e5" CHECKLIST = Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-activation-checklist-6aaef77d-task26-live-2e-r2/activation-checklist-receipt.json") CHECKLIST_SHA256 = "b093aeb888ce7325cddd7d56d52a7e4a7ec8a180ac80eef21fea9453a8ef0e8e" VALIDATION = CHECKLIST.with_name("checklist-validation-receipt.json") VALIDATION_SHA256 = "acb21c8c02260497c197427cfc93417f2be65c8746768ba788bb33400e3e6e78" READINESS = PROFILE / f"data/customers/{KEY}/nutrition-onboarding/readiness-receipt-v1.json" READINESS_SHA256 = "10cf17fba0e0f60c43d0a379721b71a945ef2c4c168fff8f613c01448297edb4" READY_AUTHORITY = PROFILE / f"data/customers/{KEY}/nutrition-onboarding/ready.json" REGISTRY = PROFILE / "customers/registry.json" BOOTSTRAP = PROFILE / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" GATEWAY = PROFILE / "gateway_state.json" OWNER_CALLBACKS = PROFILE / "data/onboarding/telegram-publication-outbox-v1/owner-callbacks.json" DELIVERIES = PROFILE / "data/owner-actions/draft-deliveries.json" AUTHORITIES = (REGISTRY, BOOTSTRAP, READINESS, READY_AUTHORITY, GATEWAY, OWNER_CALLBACKS) def digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def one(rows: list[object], field: str, value: str) -> dict[str, object]: matches = [row for row in rows if isinstance(row, dict) and row.get(field) == value] if len(matches) != 1: raise RuntimeError(f"non-unique authority: {field}") return matches[0] def verify_inventory() -> tuple[int, str]: inventory_path = ROOT / "hash-inventory.json" seal_path = ROOT / "SEAL.json" inventory = json.loads(inventory_path.read_text()) seal = json.loads(seal_path.read_text()) files = inventory.get("files") if not isinstance(files, list): raise RuntimeError("inventory schema drift") for row in files: if not isinstance(row, dict) or digest(ROOT / str(row.get("path"))) != row.get("sha256"): raise RuntimeError(f"sealed artifact drift: {row}") inventory_sha = digest(inventory_path) if seal.get("inventory_sha256") != inventory_sha or seal.get("sealed_file_count") != len(files): raise RuntimeError("seal binding drift") return len(files), inventory_sha def verify_current() -> dict[str, object]: if digest(CHECKLIST) != CHECKLIST_SHA256 or digest(VALIDATION) != VALIDATION_SHA256: raise RuntimeError("activation checklist binding drift") if digest(READINESS) != READINESS_SHA256: raise RuntimeError("readiness receipt binding drift") registry = json.loads(REGISTRY.read_text()) customer = one(registry.get("customers", []), "customer_key", KEY) bootstrap = json.loads(BOOTSTRAP.read_text()) session = one(bootstrap.get("sessions", []), "session_id", "cb_PCczfFXoI4GjvCxLBs1oRA") readiness = json.loads(READINESS.read_text()) ready = json.loads(READY_AUTHORITY.read_text()) gateway = json.loads(GATEWAY.read_text()) callbacks = json.loads(OWNER_CALLBACKS.read_text()).get("records", []) callback = one(callbacks, "customer_key", KEY) pid = gateway.get("pid") if ( customer.get("enabled") is not False or readiness.get("activation_enabled") is not False or readiness.get("delivery_enabled") is not False or readiness.get("owner_review_receipt") != "c6ecd6cb55fd68989b04d60f5ef13491a821e261e43dc7c231d0d5fefa92262e" or ready.get("state") != "ready" or ready.get("owner_reviewed") is not True or ready.get("authority_digest") != "8856ca5146cd74a6e9ed05dcd06747a694cd4f8415ee0575d673d04c9e310f09" or session.get("state") != "AWAITING_ACTIVATION" or session.get("generation") != 5 or callback.get("action") != "Approve" or gateway.get("gateway_state") != "running" or gateway.get("platforms", {}).get("telegram", {}).get("state") != "connected" or type(pid) is not int or not Path(f"/proc/{pid}").is_dir() or DELIVERIES.exists() ): raise RuntimeError("current lifecycle invariant drift") service = subprocess.run( ["systemctl", "--user", "is-active", "hermes-gateway-dualcoachtest.service"], check=True, capture_output=True, text=True, ) if service.stdout.strip() != "active": raise RuntimeError("gateway service not active") return { "service": "active/running", "telegram": "connected", "customer_enabled": False, "adaptive_activation": False, "adaptive_delivery": False, "bootstrap": "generation5/AWAITING_ACTIVATION", "owner_review": "approved/finalized-ready", "delivery_count": 0, "authority_digest": ready["authority_digest"], } def exclusive_json(path: Path, document: object) -> None: fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600) try: os.write(fd, (json.dumps(document, sort_keys=True, separators=(",", ":")) + "\n").encode()) os.fsync(fd) finally: os.close(fd) def main() -> int: count, inventory_sha = verify_inventory() before = {str(path): digest(path) for path in AUTHORITIES} current = verify_current() arm = ROOT / "independent-arm-only.json" result = subprocess.run( [sys.executable, str(ROOT / "observer_v6.py"), "arm-only", "--profile", str(PROFILE), "--ready", str(arm)], check=True, capture_output=True, text=True, ) document = json.loads(result.stdout) after = {str(path): digest(path) for path in AUTHORITIES} handoff = document.get("initial_handoff", {}) command = handoff.get("command") if ( before != after or document.get("bindings", {}).get("successor") != SUCCESSOR or document.get("bindings", {}).get("core") != CORE or document.get("bindings", {}).get("profile_wheel") != PROFILE_WHEEL or document.get("bindings", {}).get("wheel") != HERMES_WHEEL or handoff.get("status") != "READY_OPERATOR_ACTIVATION_CUTOVER" or handoff.get("role") != "operator" or handoff.get("action") != "dualcoach_admin:customer:activate" or not isinstance(command, str) or str(CHECKLIST) not in command or "<" in command or ">" in command ): raise RuntimeError("current cutover handoff drift") proof = { "schema": "task26-observer-v6-independent-proof-v1", "status": "PASS_INDEPENDENT_NO_NETWORK_ARM_ONLY", "sealed_file_count": count, "inventory_sha256": inventory_sha, "authority_unchanged": True, "authority_sha256": after, "arm_only_sha256": digest(arm), "current": current, "handoff": handoff, "network_calls": 0, "profile_mutations": 0, "service_actions": 0, "ui_actions": 0, } exclusive_json(ROOT / "independent-proof.json", proof) print(json.dumps(proof, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())