#!/usr/bin/env python3 """Deployed-candidate entry point for the append-only Task26 observer v6.3.""" 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-live-dm-checkin-fix-successor-" "f38d0373a58877806ff64a9cac54ab01e6e6e47f7101e3dcfd5ffbf05f8bb101" ) 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-f38d0373a58877806ff64a9cac54ab01e6e6e47f7101e3dcfd5ffbf05f8bb101-st_01a00b39/" "receipts/deployment-receipt.json" ) INCIDENT_RECEIPT = CANDIDATE / "bindings/launch-alias-incident-correction.json" CANDIDATE_INVENTORY = CANDIDATE / "hash-inventory.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 = "f38d0373a58877806ff64a9cac54ab01e6e6e47f7101e3dcfd5ffbf05f8bb101" CORE = "5620cf756b3f32901091475a5ed409c1757eeaf143ecb1e4a23a452d3d9e3e8e" PROFILE_WHEEL = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" HERMES_WHEEL = "2110071bc2761e6cbd149de4694dd7328148ec7e5b2e3c7c10a2709082d5e161" DEPLOYMENT_RECEIPT_SHA256 = "8cd6edb7775804302703e20153f23127dd709adbcc24fe81bffb89e5f33f7136" INCIDENT_RECEIPT_SHA256 = "1b2e8939e9c44297114906f8b608035102eb673a2d7eaa78f160af3cb9917ab4" CANDIDATE_INVENTORY_SHA256 = "95265a60e4811697858cd379d37cd4251d6b22a926009055951a025dc2ef7cc5" 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", } 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") != 23 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 digest(INCIDENT_RECEIPT) != INCIDENT_RECEIPT_SHA256 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") 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_v63", 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())