"""Run the fresh V15 transaction against faithful disposable authority clones.""" from __future__ import annotations import hashlib import json import os import shutil import sys from pathlib import Path from pydantic import JsonValue, TypeAdapter sys.path.insert(0, str(Path(__file__).resolve().parents[1])) sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "dualcoach/profile")) from gateway.platforms.task26_candidate_authority import ( verify_candidate_authority, ) from gateway.platforms.task26_runtime_authority import ( build_runtime_authority_pin, publish_runtime_authority_pin, ) from scripts.nutricoach_v150_concrete_host import ConcreteLiveHost from scripts.nutricoach_v150_live_upgrade_common import canonical from scripts.nutricoach_v150_sealed_controller import ( SealedControllerError, execute_disposable, ) from scripts.nutricoach_v150_sealed_target import DisposableService, HostPaths _OBJECT = TypeAdapter(dict[str, JsonValue]) LIVE = Path("/home/cube/.hermes/profiles/dualcoachtest") BASE = Path("/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined") PRESEAL = BASE / "live-transaction-preseal-v15-runtime-authority-r71b-maintenance" EVIDENCE = Path( "/home/cube/projects/richard/traning coach/.omo/evidence/" + "nutricoach-v150-combined/task-v15r71-rehearsal" ) UNIT = Path.home() / ".config/systemd/user/hermes-gateway-dualcoachtest.service" DROPIN = UNIT.parent / f"{UNIT.name}.d/task26-authority.conf" FAULTS = ( "stop", "post_stop_snapshot", "stopped_probe", "install_exact_wheels", "off_smoke_channel_inbox_off", "capacity_dry_run", "weekly_and_capacity_apply", "weekly_startup_smoke", "switch_unit_dropin", "reload", "start", "post_fence", ) def _load(path: Path) -> dict[str, JsonValue]: return _OBJECT.validate_json(path.read_bytes()) def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _tree(root: Path) -> str: rows = [ f"{path.relative_to(root)}:{_sha(path)}" for path in root.rglob("*") if path.is_file() and not path.is_symlink() ] return hashlib.sha256("\n".join(sorted(rows)).encode()).hexdigest() def _remove_tree(root: Path) -> None: for directory, _subdirectories, filenames in os.walk(root): Path(directory).chmod(0o700) for filename in filenames: path = Path(directory) / filename if not path.is_symlink(): path.chmod(0o600) shutil.rmtree(root) def _rebind_activation_authority(profile: Path) -> None: journal_path = profile / "data/customer-activation-journal.json" audit_path = profile / "data/customer-activation-audit.jsonl" journal = _load(journal_path) audit = _load(audit_path) for field in ( "registry_path", "data_root", "audit_path", "staff_membership_evidence_path", ): value = journal.get(field) if isinstance(value, str) and Path(value).is_relative_to(LIVE): journal[field] = str(profile / Path(value).relative_to(LIVE)) value = audit.get(field) if isinstance(value, str) and Path(value).is_relative_to(LIVE): audit[field] = str(profile / Path(value).relative_to(LIVE)) journal["audit_record_sha256"] = hashlib.sha256(canonical(audit)).hexdigest() audit_digest = str(journal["audit_record_sha256"]) _ = journal_path.write_text( json.dumps(journal, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) _ = audit_path.write_bytes(canonical(audit) + b"\n") receipt_root = profile / "data/customer-activation-receipts" for receipt_path in receipt_root.glob("*.json"): receipt = _load(receipt_path) for field in ( "registry_path", "data_root", "audit_path", "staff_membership_evidence_path", ): value = receipt.get(field) if isinstance(value, str) and Path(value).is_relative_to(LIVE): receipt[field] = str(profile / Path(value).relative_to(LIVE)) receipt["audit_record_sha256"] = audit_digest _ = receipt_path.write_text( json.dumps(receipt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def _clone(root: Path, fault: str | None) -> ConcreteLiveHost: target = _load(PRESEAL / "sealed-target.json") baseline = target["authority_baseline"] wheels = target["wheels"] if not isinstance(baseline, dict) or not isinstance(wheels, list): raise RuntimeError("target") profile = root / "profile" profile.mkdir(parents=True) _ = shutil.copytree(LIVE / "data", profile / "data") _ = shutil.copytree(LIVE / "customers", profile / "customers") _ = shutil.copy2(LIVE / "config.yaml", profile / "config.yaml") _rebind_activation_authority(profile) current = profile / ".strict-runtime/current/venv" (current / "bin").mkdir(parents=True) _ = (current / "bin/python").write_text("predecessor\n") authority = root / "external-authority" _ = shutil.copytree(Path(str(baseline["authority_root"])), authority) credentials = root / "predecessor-credentials" credentials.mkdir(mode=0o700) pin = credentials / "task26-authority-pin.json" publish_runtime_authority_pin(pin, build_runtime_authority_pin(authority)) candidate_file = credentials / "task26-candidate-digest" _ = candidate_file.write_text(str(baseline["candidate_digest"]) + "\n") candidate_file.chmod(0o600) unit = root / "gateway.service" live_current = Path(str(target["current_runtime"])) _ = unit.write_bytes( UNIT.read_bytes().replace(str(live_current).encode(), str(current).encode()) ) dropin = root / "authority.conf" lines: list[str] = [] for line in DROPIN.read_text(encoding="utf-8").splitlines(): line = line.replace(str(live_current), str(current)) if line.startswith("LoadCredential=") and ":" in line: name, raw_source = line.split("=", 1)[1].split(":", 1) if name == "task26-authority-pin.json": source = pin elif name == "task26-candidate-digest": source = candidate_file else: source = credentials / name _ = shutil.copy2(Path(raw_source), source) source.chmod(0o600) line = f"LoadCredential={name}:{source}" lines.append(line) _ = dropin.write_text("\n".join(lines) + "\n") successor_name = Path(str(target["successor_runtime"])).parent.name paths = HostPaths( profile, profile / "customers/registry.json", profile / "config.yaml", unit, dropin, current, profile / ".strict-runtime" / successor_name / "venv", root / "execution", root / "ledger", ) hermes, package = wheels if not isinstance(hermes, dict) or not isinstance(package, dict): raise RuntimeError("wheels") return ConcreteLiveHost( paths, DisposableService(), live=True, candidate_digest=str(target["candidate_digest"]), hermes_wheel=Path(str(hermes["path"])), hermes_wheel_sha256=str(hermes["sha256"]), profile_wheel=Path(str(package["path"])), profile_wheel_sha256=str(package["sha256"]), dependency_snapshot=Path(str(target["dependency_snapshot"])), dependency_snapshot_sha256=str(target["dependency_snapshot_sha256"]), fault=ValueError(fault) if fault is not None else None, fault_stage=fault or "install_exact_wheels", ) def main() -> int: if EVIDENCE.exists(): raise RuntimeError("rehearsal_exists") EVIDENCE.mkdir(parents=True) target = _load(PRESEAL / "sealed-target.json") approval = str(target["approval_phrase"]) candidate = str(target["candidate_digest"]) raw_baseline = target.get("authority_baseline") if not isinstance(raw_baseline, dict): raise RuntimeError("authority_baseline") baseline = raw_baseline live_before = { "registry": _sha(LIVE / "customers/registry.json"), "config": _sha(LIVE / "config.yaml"), "data": _tree(LIVE / "data"), "unit": _sha(UNIT), "dropin": _sha(DROPIN), } rows: list[JsonValue] = [] success_root = EVIDENCE / "success" success = _clone(success_root, None) receipt = execute_disposable( approval, success_root, success, expected_approval=approval ) source = verify_candidate_authority(Path(str(baseline["authority_root"])), None) if ( not success.service.running or success.network_events or success.telegram_events or success.provider_events ): raise RuntimeError("success") rows.append({"case": "success", "receipt": receipt, "status": "PASS"}) for fault in FAULTS: case_root = EVIDENCE / f"rollback-{fault}" host = _clone(case_root, fault) before = tuple(path.read_bytes() for path in host.paths.mutable) try: _ = execute_disposable( approval, case_root, host, expected_approval=approval ) except ValueError as error: if str(error) != fault: raise else: raise RuntimeError(f"fault_missing:{fault}") if ( tuple(path.read_bytes() for path in host.paths.mutable) != before or not host.service.running or host.successor_root.exists() or host.weekly_authority.exists() or host.network_events or host.telegram_events or host.provider_events ): raise RuntimeError(f"rollback:{fault}") try: _ = execute_disposable( approval, case_root, host, expected_approval=approval ) except SealedControllerError as error: if "already_used" not in str(error): raise else: raise RuntimeError(f"replay:{fault}") rows.append({"case": f"rollback:{fault}", "status": "PASS"}) live_after = { "registry": _sha(LIVE / "customers/registry.json"), "config": _sha(LIVE / "config.yaml"), "data": _tree(LIVE / "data"), "unit": _sha(UNIT), "dropin": _sha(DROPIN), } if live_after != live_before: raise RuntimeError("live_mutated") report = _OBJECT.validate_python({ "schema": "nutricoach-v150-v15-authority-rehearsal-v1", "candidate_digest": candidate, "package_digest": target["package_digest"], "case_count": len(rows), "cases": rows, "external_events": 0, "live_source_unchanged": True, "production_authority_before": source, "status": "PASS", }) for child in EVIDENCE.iterdir(): if child.is_dir(): _remove_tree(child) _ = (EVIDENCE / "report.json").write_bytes(canonical(report) + b"\n") print(json.dumps(report, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())