#!/usr/bin/env python3 """Independent no-network deployed-state and timeout-policy verifier for v6.4.""" 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 = "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb" CORE = "3374be765c7e53fc431b0c6e48cacbacb09618f77f13ef87016cb0ad2575d060" PROFILE_WHEEL = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" HERMES_WHEEL = "9d22e89a0b1a14d4eb7f1c2005fc01860bec5f3b4eda6e01bb1390542a90b875" DEPLOYMENT_RECEIPT = "fe65b634396aaae2818719d1b34fdaad148b2547290d3a66b3af7803bf93dfca" CANDIDATE_INVENTORY = "0119beeb46a15ab25a78550da4c432b4cc52e0dc816822d2688dc9009df89d90" PREDECESSOR_SEAL = "2833428cfab41826a5cb54f90bb51c56f40b2468850d02687f7341c791b7f803" SERVICE_STATE_SHA256 = "81c1e581cb84aac6f44044747e0e8e77a2d7498c71d626f1f9b5ec2f9d09a261" 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" SERVICE_STATE = PROFILE / "data/owner-actions/customer-service-state.json" AUTHORITIES = ( REGISTRY, BOOTSTRAP, READINESS, READY_AUTHORITY, FEATURE_EPOCH, GATEWAY, OWNER_CALLBACKS, INGRESS, SERVICE_STATE, ) 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 != 576011 or not Path("/proc/576011").is_dir() or digest(SERVICE_STATE) != SERVICE_STATE_SHA256 or SERVICE_STATE.stat().st_mode & 0o777 != 0o600 or json.loads(SERVICE_STATE.read_text()) != { "payload_digest": "fbb526e57506141f686cd56dd75c6375c78b974452b985c0737f25230060994d", "schema": "customer-service-state-v1", "states": {}, } ): 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=576011"}: raise RuntimeError("gateway service identity drift") return { "service": "active/running", "pid": 576011, "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_v64.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("candidate_inventory") != CANDIDATE_INVENTORY or bindings.get("predecessor_seal") != PREDECESSOR_SEAL or bindings.get("service_state") != SERVICE_STATE_SHA256 or bindings.get("service_pid") != "576011" 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") != ( "after arming, send one fresh exact production alias 오늘 체크인, then press inline " "오늘 체크인 시작; the pre-arm 2026-08-17 12:10 message produced no state/card " "and is not replayed or accepted" ) or document.get("fresh_alias_requirement") != { "required_after_arming": "오늘 체크인", "expected_inline": "오늘 체크인 시작", "pre_arm_message": "2026-08-17 12:10", "pre_arm_result": "no state/card; not replayed or accepted", } ): raise RuntimeError("current fresh Korean start handoff drift") proof = { "schema": "task26-observer-v6.4-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())