#!/usr/bin/env python3 """Deployed-candidate entry point for the append-only Task26 observer v6.4.""" from __future__ import annotations import hashlib import importlib.util import json import subprocess import sys import zipfile from pathlib import Path from types import ModuleType from urllib.parse import unquote, urlparse ROOT = Path(__file__).resolve().parent SOURCE = Path("/home/cube/projects/richard/hermes-agent") INSTALLED = SOURCE / ".venv/lib/python3.12/site-packages" HERMES_DIRECT_URL = INSTALLED / "hermes_agent-0.17.0.dist-info/direct_url.json" PROFILE_DIRECT_URL = INSTALLED / "physique_checkin_cli-0.1.0.dist-info/direct_url.json" CANDIDATE = Path( "/home/cube/projects/richard/traning coach/.omo/evidence/task26/" "task26-customer-service-state-bootstrap-successor-" "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb" ) PRESERVED_HERMES_WHEEL = CANDIDATE / "artifacts/hermes_agent-0.17.0-py3-none-any.whl" DEPLOYMENT_RECEIPT = Path( "/home/cube/projects/richard/traning coach/.omo/evidence/task26/" "task26-certification-deployment-4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb-st_01a00b39/" "receipts/deployment-receipt.json" ) CANDIDATE_INVENTORY = CANDIDATE / "hash-inventory.json" SERVICE_STATE = Path( "/home/cube/.hermes/profiles/dualcoachtest/data/owner-actions/" "customer-service-state.json" ) RECOVERY_SEAL = Path( "/home/cube/projects/richard/traning coach/.omo/evidence/task26/" "task26-telegram-poll-recovery-st_01a00d2c/FINAL-SEAL.json" ) RECOVERY_RESULT = RECOVERY_SEAL.parent / "run-4/recovery-result.json" PROFILE_PACKAGE_ROOT = Path( "/home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli" ) PROFILE_PACKAGE = PROFILE_PACKAGE_ROOT / "checkin_cli" PROFILE_WHEEL_PATH = Path( "/home/cube/projects/richard/traning coach/.omo/evidence/task26/" "task26-inode-test-fix-successor-6aaef77de87489c3be493978645781b24b55ab07513ff1f880164741d0bd3c73/" "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl" ) SUCCESSOR = "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb" CORE = "3374be765c7e53fc431b0c6e48cacbacb09618f77f13ef87016cb0ad2575d060" PROFILE_WHEEL = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" HERMES_WHEEL = "9d22e89a0b1a14d4eb7f1c2005fc01860bec5f3b4eda6e01bb1390542a90b875" DEPLOYMENT_RECEIPT_SHA256 = "fe65b634396aaae2818719d1b34fdaad148b2547290d3a66b3af7803bf93dfca" CANDIDATE_INVENTORY_DIGEST = "0119beeb46a15ab25a78550da4c432b4cc52e0dc816822d2688dc9009df89d90" CANDIDATE_INVENTORY_SHA256 = "47e52b683f20e3b15755e0f9747ac85e543cea8a1a1dc8d67b7718f5933854ae" SERVICE_STATE_SHA256 = "81c1e581cb84aac6f44044747e0e8e77a2d7498c71d626f1f9b5ec2f9d09a261" PREDECESSOR_SEAL_SHA256 = "2833428cfab41826a5cb54f90bb51c56f40b2468850d02687f7341c791b7f803" RECOVERY_SEAL_SHA256 = "00d44de03e8dde668da8058091751291b13457c8bd086917dfe85e42b3813df8" RECOVERY_RESULT_SHA256 = "92d9eee2b556282079e2e8fd673c09bcdf7bc00644fca9791cd20bee06450268" SOURCE_PARITY = { "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_callback.py": "bf2f312245b59ebf33dd1e3e449cfffed12809d18f820953cc28cecef25d34d2", "gateway/platforms/telegram_nutrition_onboarding_runtime_publication.py": "f04ea498ad12a257fb1322da4c8cfd2f91fc9403f0b978617fcc00df41729209", "gateway/platforms/telegram.py": "5d3103b744b6e92ddbc576a77df433c2127d6df17260a3c78d2000d7b90d2f97", "gateway/platforms/nutrition_coaching.py": "b0747c4f366d5a78fce7c4ae14b46bd7731b2ecc1591735c6ad75933a3a72abf", "gateway/platforms/nutrition_service_state.py": "996d4b629eb21f59b9021a6b96ab2d82c0c89029f21e8de7cbb6143c59359f88", } def digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def verify_direct_url(path: Path, expected_path: Path, expected_hash: str | None) -> None: direct = json.loads(path.read_text()) installed = Path(unquote(urlparse(str(direct.get("url", ""))).path)).resolve() if installed != expected_path.resolve(): raise RuntimeError(f"installed direct-url binding drift: {path.name}") if expected_hash is not None: archive_hash = direct.get("archive_info", {}).get("hash") if archive_hash != f"sha256={expected_hash}": raise RuntimeError(f"installed archive hash drift: {path.name}") elif direct.get("dir_info", {}).get("editable") is not True: raise RuntimeError("profile package is not exact editable binding") def verify_profile_wheel_parity() -> int: if digest(PROFILE_WHEEL_PATH) != PROFILE_WHEEL: raise RuntimeError("candidate profile wheel drift") count = 0 with zipfile.ZipFile(PROFILE_WHEEL_PATH) as archive: for member in archive.namelist(): if not member.startswith("checkin_cli/") or not member.endswith(".py"): continue relative = member.removeprefix("checkin_cli/") installed = PROFILE_PACKAGE / relative if not installed.is_file() or hashlib.sha256(archive.read(member)).hexdigest() != digest(installed): raise RuntimeError(f"installed/profile module drift: {relative}") count += 1 if count != 44: raise RuntimeError("profile module inventory drift") return count def verify_binding() -> None: verify_direct_url(HERMES_DIRECT_URL, PRESERVED_HERMES_WHEEL, HERMES_WHEEL) verify_direct_url(PROFILE_DIRECT_URL, PROFILE_PACKAGE_ROOT, None) if digest(PRESERVED_HERMES_WHEEL) != HERMES_WHEEL: raise RuntimeError("preserved Hermes wheel drift") result = subprocess.run( [sys.executable, str(CANDIDATE / "verify_candidate.py"), str(CANDIDATE)], check=True, capture_output=True, text=True, ) verification = json.loads(result.stdout) manifest = json.loads((CANDIDATE / "candidate-manifest.json").read_text()) seal = json.loads((CANDIDATE / "candidate-seal.json").read_text()) deployment = json.loads(DEPLOYMENT_RECEIPT.read_text()) recovery = json.loads(RECOVERY_RESULT.read_text()) if ( verification.get("status") != "PASS" or verification.get("full_candidate_digest") != SUCCESSOR or manifest.get("full_candidate_digest") != SUCCESSOR or manifest.get("core_candidate_digest") != CORE or manifest.get("hermes_wheel_sha256") != HERMES_WHEEL or manifest.get("profile_wheel_sha256") != PROFILE_WHEEL or manifest.get("inventory_count") != 20 or manifest.get("inventory_digest") != CANDIDATE_INVENTORY_DIGEST or seal.get("full_candidate_digest") != SUCCESSOR or seal.get("core_candidate_digest") != CORE or seal.get("verdict") != "PASS" or digest(DEPLOYMENT_RECEIPT) != DEPLOYMENT_RECEIPT_SHA256 or deployment.get("status") != "PASS_DEPLOYED_HEALTHY" or deployment.get("binding", {}).get("observer_v63_seal_sha256") != PREDECESSOR_SEAL_SHA256 or deployment.get("runtime", {}).get("main_pid") != 576011 or deployment.get("runtime", {}).get("telegram_state") != "connected" or deployment.get("transport_gate") != { "customer_key": "task26_live_2e_r2_20260815_8527916639", "customer_transport_allowed": True, "kst_date": "2026-08-17", "route": {"chat_id": "8527916639", "topic_id": "0", "user_id": "8527916639"}, } or deployment.get("service_state") != { "mode": "0600", "path": str(SERVICE_STATE), "payload_digest": "fbb526e57506141f686cd56dd75c6375c78b974452b985c0737f25230060994d", "schema": "customer-service-state-v1", "sha256": SERVICE_STATE_SHA256, "states": {}, } or digest(CANDIDATE_INVENTORY) != CANDIDATE_INVENTORY_SHA256 or digest(RECOVERY_SEAL) != RECOVERY_SEAL_SHA256 or digest(RECOVERY_RESULT) != RECOVERY_RESULT_SHA256 or recovery.get("status") != "PASS_RECOVERED" or recovery.get("replay", {}).get("offset_lower_bound") < 629525117 ): raise RuntimeError("candidate, deployment, incident, or inventory binding drift") service_state = json.loads(SERVICE_STATE.read_text()) if ( digest(SERVICE_STATE) != SERVICE_STATE_SHA256 or SERVICE_STATE.stat().st_mode & 0o777 != 0o600 or service_state != { "payload_digest": "fbb526e57506141f686cd56dd75c6375c78b974452b985c0737f25230060994d", "schema": "customer-service-state-v1", "states": {}, } ): raise RuntimeError("canonical empty unpaused service-state drift") verify_profile_wheel_parity() for relative, expected in SOURCE_PARITY.items(): if digest(SOURCE / relative) != expected: raise RuntimeError(f"candidate source parity drift: {relative}") if digest(INSTALLED / relative) != expected: raise RuntimeError(f"installed Hermes runtime drift: {relative}") def load_observer() -> ModuleType: spec = importlib.util.spec_from_file_location( "task26_continuous_lifecycle_v64", ROOT / "continuous_lifecycle.py" ) if spec is None or spec.loader is None: raise RuntimeError("observer import unavailable") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) if ( module.SUCCESSOR != SUCCESSOR or module.CORE != CORE or module.WHEEL != HERMES_WHEEL or module.PROFILE_WHEEL != PROFILE_WHEEL ): raise RuntimeError("observer digest binding drift") setattr(module, "SITE", INSTALLED) return module def main() -> int: try: verify_binding() return int(load_observer().main()) except (OSError, ValueError, RuntimeError, KeyError, subprocess.SubprocessError, zipfile.BadZipFile) as exc: print(f"FAIL: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())