#!/usr/bin/env python3 """Independent no-network deployed-state and timeout-policy verifier for v6.3.""" 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" SESSION = "cb_PCczfFXoI4GjvCxLBs1oRA" SUCCESSOR = "f38d0373a58877806ff64a9cac54ab01e6e6e47f7101e3dcfd5ffbf05f8bb101" CORE = "5620cf756b3f32901091475a5ed409c1757eeaf143ecb1e4a23a452d3d9e3e8e" PROFILE_WHEEL = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" HERMES_WHEEL = "2110071bc2761e6cbd149de4694dd7328148ec7e5b2e3c7c10a2709082d5e161" DEPLOYMENT_RECEIPT = "8cd6edb7775804302703e20153f23127dd709adbcc24fe81bffb89e5f33f7136" INCIDENT_RECEIPT = "1b2e8939e9c44297114906f8b608035102eb673a2d7eaa78f160af3cb9917ab4" CANDIDATE_INVENTORY = "95265a60e4811697858cd379d37cd4251d6b22a926009055951a025dc2ef7cc5" RECOVERY_SEAL = "00d44de03e8dde668da8058091751291b13457c8bd086917dfe85e42b3813df8" POLLING_OFFSET_LOWER_BOUND = "629525117" REGISTRY = PROFILE / "customers/registry.json" BOOTSTRAP = PROFILE / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" READINESS = PROFILE / f"data/customers/{KEY}/nutrition-onboarding/readiness-receipt-v1.json" READY_AUTHORITY = PROFILE / f"data/customers/{KEY}/nutrition-onboarding/ready.json" FEATURE_EPOCH = PROFILE / f"data/customers/{KEY}/nutrition-plans/feature-epoch.json" GATEWAY = PROFILE / "gateway_state.json" OWNER_CALLBACKS = PROFILE / "data/onboarding/telegram-publication-outbox-v1/owner-callbacks.json" INGRESS = PROFILE / "data/telegram-ingress-receipts-v1-d0aacf0f4bdbb7c0.json" DELIVERIES = PROFILE / "data/owner-actions/draft-deliveries.json" AUTHORITIES = ( REGISTRY, BOOTSTRAP, READINESS, READY_AUTHORITY, FEATURE_EPOCH, GATEWAY, OWNER_CALLBACKS, INGRESS, ) 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" inventory = json.loads(inventory_path.read_text()) seal = json.loads((ROOT / "SEAL.json").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]: 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", SESSION) readiness = json.loads(READINESS.read_text()) feature = json.loads(FEATURE_EPOCH.read_text()) gateway = json.loads(GATEWAY.read_text()) ingress = json.loads(INGRESS.read_text()) target = "629525116" pid = gateway.get("pid") if ( customer.get("enabled") is not True or session.get("state") != "ACTIVE" or session.get("generation") != 6 or readiness.get("activation_enabled") is not False or readiness.get("delivery_enabled") is not False or feature.get("activation") is not False or feature.get("delivery") is not False or DELIVERIES.exists() or target in ingress.get("receipts", {}) or target in ingress.get("terminal_receipts", {}) or target in ingress.get("terminal_failures", {}) or gateway.get("gateway_state") != "running" or gateway.get("platforms", {}).get("telegram", {}).get("state") != "connected" or pid != 545777 or not Path("/proc/545777").is_dir() ): raise RuntimeError("deployed lifecycle invariant drift") service = subprocess.run( ["systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", "-p", "ActiveState", "-p", "SubState", "-p", "MainPID"], check=True, capture_output=True, text=True, ).stdout.splitlines() if set(service) != {"ActiveState=active", "SubState=running", "MainPID=545777"}: raise RuntimeError("gateway service identity drift") return { "service": "active/running", "pid": 545777, "telegram": "connected", "customer_enabled": True, "bootstrap": "generation6/ACTIVE", "adaptive_activation": False, "adaptive_delivery": False, "delivery_count": 0, "recovered_update_629525116": "absent/not-replayed", "polling_offset_lower_bound": 629525117, } 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_v63.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", {}) bindings = document.get("bindings", {}) if ( before != after or bindings.get("successor") != SUCCESSOR or bindings.get("core") != CORE or bindings.get("profile_wheel") != PROFILE_WHEEL or bindings.get("wheel") != HERMES_WHEEL or bindings.get("deployment_receipt") != DEPLOYMENT_RECEIPT or bindings.get("incident_receipt") != INCIDENT_RECEIPT or bindings.get("candidate_inventory") != CANDIDATE_INVENTORY or bindings.get("recovery_seal") != RECOVERY_SEAL or bindings.get("polling_offset_lower_bound") != POLLING_OFFSET_LOWER_BOUND or document.get("per_stage_timeout_max") != 21600 or document.get("total_timeout_max") != 21600 or handoff.get("status") != "READY_CUSTOMER_START_CHECKIN" or handoff.get("role") != "customer" or handoff.get("actor") != "8527916639" or handoff.get("route") != ["8527916639", "0"] or handoff.get("action") != "send:오늘 체크인" or handoff.get("detail") != "send the exact production alias 오늘 체크인, then press inline 오늘 체크인 시작" ): raise RuntimeError("current Korean start handoff drift") proof = { "schema": "task26-observer-v6.3-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, "customer_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())