"""Close the consumed V10 failure after exact lifecycle-drift verification.""" from __future__ import annotations import argparse import json import os import stat import subprocess import sys from pathlib import Path from typing import Final, cast sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.nutricoach_v150_live_upgrade_common import ( JsonValue, canonical, load_object, object_at, sha256_file, ) from scripts.nutricoach_v150_phase_journal import PhaseJournal from scripts.nutricoach_v150_sealed_authority import atomic_write BASE: Final = Path("/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined") PRESEAL: Final = BASE / "live-transaction-preseal-v10-user-bus-environment" TARGET: Final = PRESEAL / "sealed-target.json" EXPECTED_CANDIDATE: Final = ( "066a794d44861d2cd0fe8c1ea14c0050e00219b38a2d2026b389772783056dad" ) LIFECYCLE_PATHS: Final = frozenset({ "auth.json", "data/onboarding/telegram-staff-membership-v1/events.jsonl", "gateway.pid", "gateway_state.json", "logs/gateway-exit-diag.log", "logs/gateway-shutdown-diag.log", }) APPEND_ONLY_PATHS: Final = frozenset({ "data/onboarding/telegram-staff-membership-v1/events.jsonl", "logs/gateway-exit-diag.log", "logs/gateway-shutdown-diag.log", }) class RecoveryDenied(RuntimeError): """The V10 terminal rollback boundary is not exact.""" def _text(value: JsonValue | None, label: str) -> str: if not isinstance(value, str): raise RecoveryDenied(label) return value def verify_inventory( profile: Path, snapshot: dict[str, JsonValue] ) -> tuple[int, tuple[str, ...]]: profiles = object_at(snapshot.get("profiles"), "profiles") rows = profiles.get("stable") if not isinstance(rows, list): raise RecoveryDenied("stable_rows") checked = 0 lifecycle: list[str] = [] for raw in rows: if not isinstance(raw, dict) or raw.get("profile") != profile.name: continue relative = _text(raw.get("path"), "stable_path") path = profile / relative kind = raw.get("kind") if kind == "symlink": if not path.is_symlink(): raise RecoveryDenied(f"stable_missing:{relative}") if os.readlink(path) != raw.get("target"): raise RecoveryDenied(f"stable_symlink:{relative}") mode = path.lstat().st_mode if stat.S_IMODE(mode) != raw.get("mode"): raise RecoveryDenied(f"stable_mode:{relative}") checked += 1 continue if kind != "file" or not path.is_file() or path.is_symlink(): raise RecoveryDenied(f"stable_missing:{relative}") mode = path.stat(follow_symlinks=False).st_mode if stat.S_IMODE(mode) != raw.get("mode"): raise RecoveryDenied(f"stable_mode:{relative}") checked += 1 if sha256_file(path) == raw.get("sha256"): continue if relative not in LIFECYCLE_PATHS: raise RecoveryDenied(f"unexpected_drift:{relative}") sealed_size = raw.get("size") if ( relative in APPEND_ONLY_PATHS and isinstance(sealed_size, int) and path.stat().st_size < sealed_size ): raise RecoveryDenied(f"append_truncated:{relative}") if relative.endswith(".json") or relative.endswith(".pid"): _ = load_object(path) lifecycle.append(relative) if checked == 0 or frozenset(lifecycle) != LIFECYCLE_PATHS: raise RecoveryDenied("lifecycle_drift_set") return checked, tuple(sorted(lifecycle)) def _service() -> dict[str, str]: fields = ( "ActiveState", "SubState", "MainPID", "ExecMainStartTimestampMonotonic", ) command = ["systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service"] command.extend(f"--property={field}" for field in fields) result = subprocess.run(command, check=True, capture_output=True, text=True) return dict( line.split("=", 1) for line in result.stdout.splitlines() if "=" in line ) def verify() -> tuple[dict[str, JsonValue], Path, Path]: binding = load_object(TARGET) profile = Path(_text(binding.get("profile_root"), "profile_root")) execution = Path(_text(binding.get("execution_root"), "execution_root")) ledger = Path(_text(binding.get("global_approval_ledger"), "ledger")) current = Path(_text(binding.get("current_runtime"), "current_runtime")) successor = Path(_text(binding.get("successor_runtime"), "successor_runtime")) inventory = Path(_text(binding.get("protected_inventory"), "inventory")) phase_path = execution / "phase.json" if load_object(ledger / "authorization-consumed.json") != { "candidate_digest": EXPECTED_CANDIDATE, "outcome": "FAILED", "status": "CONSUMED", }: raise RecoveryDenied("failed_ledger") if load_object(phase_path).get("phase") != "RECOVERY_REQUIRED": raise RecoveryDenied("phase") snapshot = load_object(inventory) checked, lifecycle = verify_inventory(profile, snapshot) contract = object_at(snapshot.get("contract_hashes_only"), "contract") rows = contract.get("inventory") if not isinstance(rows, list): raise RecoveryDenied("contract_inventory") for raw in rows: if not isinstance(raw, dict): raise RecoveryDenied("contract_row") path = Path(_text(raw.get("path"), "contract_path")) if sha256_file(path) != raw.get("sha256"): raise RecoveryDenied(f"contract_drift:{path}") raw_manifest = cast( JsonValue, json.loads( (execution / "post-stop-snapshots/manifest.json").read_text( encoding="utf-8" ) ), ) if not isinstance(raw_manifest, list): raise RecoveryDenied("snapshot_manifest") manifest = [object_at(raw, "snapshot_row") for raw in raw_manifest] for row in manifest: live = Path(_text(row.get("live"), "snapshot_live")) if sha256_file(live) != row.get("sha256"): raise RecoveryDenied(f"mutable_drift:{live}") service = _service() if service.get("ActiveState") != "active" or service.get("SubState") != "running": raise RecoveryDenied("service") pid = int(_text(service.get("MainPID"), "service_pid")) pid_state = load_object(profile / "gateway.pid") gateway_state = load_object(profile / "gateway_state.json") if pid_state.get("pid") != pid or gateway_state.get("pid") != pid: raise RecoveryDenied("pid_identity") if not (current / "bin/python").is_file() or successor.parent.exists(): raise RecoveryDenied("runtime_restoration") service_json: dict[str, JsonValue] = {key: value for key, value in service.items()} receipt: dict[str, JsonValue] = { "candidate_digest": EXPECTED_CANDIDATE, "contract_rows": len(rows), "lifecycle_drift": list(lifecycle), "mutable_snapshots_exact": len(manifest), "service": service_json, "stable_rows_exact_or_lifecycle": checked, "status": "V10_TERMINAL_ROLLBACK_VERIFIED", "successor_absent": True, } return receipt, execution, phase_path def main() -> int: parser = argparse.ArgumentParser(description=__doc__) action = parser.add_mutually_exclusive_group(required=True) _ = action.add_argument("--dry-run", action="store_true") _ = action.add_argument("--apply", action="store_true") _ = parser.parse_args() receipt, execution, phase_path = verify() if "--apply" in sys.argv: recovery = execution / "recovery-receipt.json" if recovery.exists(): raise RecoveryDenied("recovery_receipt_exists") atomic_write(recovery, canonical(receipt) + b"\n", 0o400) PhaseJournal(phase_path).advance("ROLLED_BACK") print(canonical(receipt).decode()) return 0 if __name__ == "__main__": raise SystemExit(main())