"""Freeze the fresh V15 runtime-authority deployment package.""" from __future__ import annotations import hashlib import json import os import shlex import shutil import stat import subprocess import sys import zipfile from pathlib import Path from pydantic import JsonValue, TypeAdapter sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from gateway.platforms.task26_runtime_authority import ( build_runtime_authority_pin, load_task26_production_authority, ) from scripts.nutricoach_v150_live_upgrade_common import canonical from scripts.nutricoach_v150_runtime_ops import dependency_snapshot_digest from scripts.nutricoach_v150_r71b_maintenance_transaction import KNOWN_R70_CRON_JOB_ID from scripts.nutricoach_v150_r71b_preparation import ( collect_authoritative_r71b_preflight, prepare_r71b_maintenance, ) from scripts.nutricoach_v150_r71b_preseal_arguments import parse_preseal_arguments _OBJECT = TypeAdapter(dict[str, JsonValue]) SOURCE = Path("/home/cube/projects/richard/.worktrees/nutricoach-v150-combined") BASE = Path("/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined") PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest") UNIT = Path.home() / ".config/systemd/user/hermes-gateway-dualcoachtest.service" DROPIN = UNIT.parent / f"{UNIT.name}.d/task26-authority.conf" PRESEAL = BASE / "live-transaction-preseal-v15-runtime-authority-r71b-maintenance" PREFLIGHT = BASE / "preflight-v15-runtime-authority-r71b-maintenance" AUTHORITY_ID = "nutricoach-v150-v15-runtime-authority-r71b-maintenance" PRIOR_PRESEAL = BASE / "live-transaction-preseal-v15-runtime-authority-r63" PRIOR_PREFLIGHT = BASE / "preflight-v15-runtime-authority-r63" def _load(path: Path) -> dict[str, JsonValue]: return _OBJECT.validate_json(path.read_bytes()) def _write(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) document = _OBJECT.validate_python(value) _ = path.write_bytes(canonical(document) + b"\n") def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _record_sha(path: Path) -> str: with zipfile.ZipFile(path) as wheel: record = next(name for name in wheel.namelist() if name.endswith("/RECORD")) return hashlib.sha256(wheel.read(record)).hexdigest() def _service_state() -> dict[str, JsonValue]: fields = ("ActiveState", "SubState", "MainPID", "ExecStart", "NRestarts") result = subprocess.run( ( "/usr/bin/systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", *(f"--property={field}" for field in fields), ), check=True, capture_output=True, text=True, ) rows = dict( line.split("=", 1) for line in result.stdout.splitlines() if "=" in line ) return {field: rows.get(field, "") for field in fields} def _current_runtime() -> Path: line = next( ( value for value in UNIT.read_text(encoding="utf-8").splitlines() if value.startswith("ExecStart=") ), "", ) command = shlex.split(line.removeprefix("ExecStart=")) if not command: raise RuntimeError("current_runtime") runtime = Path(command[0]).parent.parent expected_parent = PROFILE / ".strict-runtime" if ( runtime.name != "venv" or runtime.parent.parent != expected_parent or not (runtime / "bin/python").is_file() ): raise RuntimeError("current_runtime") return runtime def _credential_paths() -> tuple[Path, Path]: values: dict[str, Path] = {} for line in DROPIN.read_text(encoding="utf-8").splitlines(): if not line.startswith("LoadCredential=") or ":" not in line: continue name, raw_path = line.split("=", 1)[1].split(":", 1) values[name] = Path(raw_path) return ( values["task26-authority-pin.json"], values["task26-candidate-digest"], ) def _authority_baseline() -> dict[str, JsonValue]: pin_path, candidate_path = _credential_paths() old_pin = os.environ.get("TASK26_AUTHORITY_PIN") old_candidate = os.environ.get("TASK26_CANDIDATE_DIGEST_FILE") os.environ["TASK26_AUTHORITY_PIN"] = str(pin_path) os.environ["TASK26_CANDIDATE_DIGEST_FILE"] = str(candidate_path) try: source, candidate = load_task26_production_authority( profile_root=PROFILE, package_root=BASE, ) finally: if old_pin is None: _ = os.environ.pop("TASK26_AUTHORITY_PIN", None) else: os.environ["TASK26_AUTHORITY_PIN"] = old_pin if old_candidate is None: _ = os.environ.pop("TASK26_CANDIDATE_DIGEST_FILE", None) else: os.environ["TASK26_CANDIDATE_DIGEST_FILE"] = old_candidate registry = source.root / "candidate-authority/registry.json" ledger = source.root / "candidate-authority/qualification-ledger.json" pin = build_runtime_authority_pin(source.root) registry_document = _load(registry) events = registry_document.get("events") if not isinstance(events, list) or not events: raise RuntimeError("authority_events") latest = events[-1] if not isinstance(latest, dict): raise RuntimeError("authority_event") historical_pass = latest.get("historical_pass_digest") if not isinstance(historical_pass, str): raise RuntimeError("authority_historical_pass") return _OBJECT.validate_python({ "authority_root": str(source.root), "candidate_digest": candidate, "candidate_path": str(candidate_path), "candidate_file_sha256": _sha(candidate_path), "event_count": pin["event_count"], "genesis_sha256": pin["genesis_sha256"], "historical_pass_digest": historical_pass, "ledger_file_sha256": _sha(ledger), "ledger_head_sha256": pin["ledger_head_sha256"], "pin_path": str(pin_path), "pin_file_sha256": _sha(pin_path), "registry_file_sha256": _sha(registry), "registry_head_sha256": pin["registry_head_sha256"], "source_id": pin["source_id"], }) def _controller_files(candidate: dict[str, JsonValue]) -> dict[str, JsonValue]: rows = candidate.get("source_inventory") if not isinstance(rows, list): raise RuntimeError("candidate_source_inventory") files: dict[str, JsonValue] = {} for row in rows: if not isinstance(row, dict) or not isinstance(row.get("path"), str): raise RuntimeError("candidate_source_entry") relative = Path(str(row["path"])) product = Path(*relative.parts[2:]) if (SOURCE / product).is_file(): files[product.as_posix()] = _sha(SOURCE / product) for relative in ( "scripts/execute_nutricoach_v150_sealed_live.py", "scripts/nutricoach_v150_controller_worker.py", "scripts/nutricoach_v150_detached_bootstrap.py", "scripts/prepare_nutricoach_v150_v15_preseal.py", "scripts/nutricoach_v150_r71b_preseal_arguments.py", "scripts/nutricoach_v150_r71b_preparation.py", "scripts/nutricoach_v150_r71b_preseal_verifier.py", "scripts/verify_nutricoach_v150_preseal_v15.py", ): files[relative] = _sha(SOURCE / relative) return dict(sorted(files.items())) def main() -> int: arguments = parse_preseal_arguments(sys.argv[1:]) candidate_root = arguments.candidate_root hermes_wheel = arguments.hermes_wheel profile_wheel = arguments.profile_wheel if PRESEAL.exists() or PREFLIGHT.exists(): raise RuntimeError("v15_paths_exist") candidate = _load(candidate_root / "manifest.json") digest = candidate.get("candidate_digest") if not isinstance(digest, str): raise RuntimeError("candidate_digest") for wheel in (hermes_wheel, profile_wheel): info = wheel.stat(follow_symlinks=False) if ( wheel.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) & 0o222 ): raise RuntimeError("wheel_integrity") PREFLIGHT.mkdir(parents=True) PRESEAL.mkdir(parents=True) protected = PREFLIGHT / "snapshot-before.json" _ = shutil.copy2(PRIOR_PREFLIGHT / "snapshot-before.json", protected) dependency_root = PRESEAL / "dependencies/site-packages" current_runtime = _current_runtime() predecessor_site = next( (current_runtime / "lib").glob("python*/site-packages") ) for name in ( "croniter", "croniter-6.0.0.dist-info", "python_telegram_bot-22.6.dist-info", "telegram", ): _ = shutil.copytree( predecessor_site / name, dependency_root / name, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", "*.pyo"), ) controller_files = _controller_files(candidate) controller_root = PRESEAL / "controller-source" for relative in controller_files: destination = controller_root / relative destination.parent.mkdir(parents=True, exist_ok=True) _ = shutil.copy2(SOURCE / relative, destination) _write( PRESEAL / "controller-source-manifest.json", { "schema": "nutricoach-v150-controller-source-manifest-v15", "files": controller_files, }, ) baseline = _authority_baseline() target_root = PROFILE / f".strict-runtime/{digest[:8]}-v150" wheels = [ {"path": str(path), "sha256": _sha(path), "record_sha256": _record_sha(path)} for path in (hermes_wheel, profile_wheel) ] target_binding = { "authority_id": AUTHORITY_ID, "profile_root": str(PROFILE), "current_runtime": str(current_runtime), "successor_runtime": str(target_root / "venv"), "service_name": "hermes-gateway-dualcoachtest.service", "unit": str(UNIT), "dropin": str(DROPIN), "execution_root": str(BASE / "live-executions-v15" / AUTHORITY_ID), "global_approval_ledger": str(BASE / "live-authorization-v15" / AUTHORITY_ID), "protected_inventory": str(protected), "protected_inventory_sha256": _sha(protected), "dependency_snapshot": str(dependency_root), "dependency_snapshot_sha256": dependency_snapshot_digest(dependency_root), "registry_sha256": _sha(PROFILE / "customers/registry.json"), } collected = collect_authoritative_r71b_preflight( candidate_root, PROFILE, arguments.window ) service = _service_state() controller_derivation = hashlib.sha256(canonical(controller_files)).hexdigest() binding_seed = _OBJECT.validate_python({ "schema": "nutricoach-v150-r71b-package-binding-inputs-v1", "product_generation": "r71", "package_namespace": "r71b-maintenance", "authority_id": AUTHORITY_ID, "paths": {"preseal_root": str(PRESEAL)}, "candidate": { "candidate_digest": digest, "manifest": str(candidate_root / "manifest.json"), "manifest_sha256": _sha(candidate_root / "manifest.json"), }, "wheels": wheels, "controller": { "files": controller_files, "derivation_sha256": controller_derivation, }, "protected_inventory": { "path": str(protected), "sha256": _sha(protected), "explicit_maintenance_volatile_paths": [], }, "dependency_snapshot": { "path": str(dependency_root), "dependency_snapshot_sha256": dependency_snapshot_digest(dependency_root), }, "authority_baseline": baseline, "target_binding_base": target_binding, "service_preparation": service, "diagnosed_blocker_evidence": [ collected.evidence["observer_evidence"], collected.evidence["fixed_collector_evidence"], ], **collected.evidence, }) prepared = prepare_r71b_maintenance( collected.preflight, arguments.window, collected.facts, binding_seed ) maintenance = PRESEAL / "maintenance" _write( maintenance / "maintenance-scope.json", prepared.scope.model_dump(mode="json", by_alias=True), ) _write(maintenance / "no-send-oracle.json", prepared.oracle) binding_inputs = _OBJECT.validate_json(prepared.artifacts.binding_inputs_bytes) _write(maintenance / "package-binding-inputs.json", binding_inputs) authority = PRESEAL / "maintenance-authority/nutricoach-topic59-maintenance-r71b.json" authority.parent.mkdir(parents=True, exist_ok=True) _ = authority.write_bytes(prepared.artifacts.authority_bytes) _write(PREFLIGHT / "package.json", prepared.artifacts.permission_package) package_digest = prepared.artifacts.final_package_digest phrase = str(prepared.artifacts.permission_package["approval_phrase"]) target = { **target_binding, "schema": "nutricoach-v150-sealed-live-target-r71b", "approval_phrase": phrase, "package_digest": package_digest, "candidate_digest": digest, "candidate_manifest": str(candidate_root / "manifest.json"), "candidate_manifest_sha256": _sha(candidate_root / "manifest.json"), "controller_derivation_sha256": controller_derivation, "permission_package": str(PREFLIGHT / "package.json"), "permission_package_sha256": _sha(PREFLIGHT / "package.json"), "authority_baseline": baseline, "wheels": wheels, "service_preparation": service, "package_binding_digest": prepared.artifacts.package_binding_digest, "maintenance_authority_path": str(authority), "maintenance_authority_sha256": prepared.artifacts.authority_sha256, "maintenance_hold_id": prepared.artifacts.hold.hold_id, "maintenance_hold_sha256": prepared.artifacts.hold_sha256, "maintenance_scope_path": str(maintenance / "maintenance-scope.json"), "maintenance_scope_sha256": _sha(maintenance / "maintenance-scope.json"), "maintenance_ledger_path": str(PROFILE / "data/weekly-operations-topic59.jsonl"), "maintenance_ledger_sha256": collected.preflight.publication_ledger_sha256, "maintenance_output_directory": str(PROFILE / "cron/output" / KNOWN_R70_CRON_JOB_ID), "r70_error_evidence_path": collected.evidence["r70_error_evidence_path"], "r70_error_evidence_sha256": collected.evidence["r70_error_evidence_sha256"], "maintenance_window": { "kst_day": arguments.window.kst_day.isoformat(), "not_before": arguments.window.not_before.isoformat(), "expires_at": arguments.window.expires_at.isoformat(), }, } _write(PRESEAL / "sealed-target.json", target) prior = _load(PRIOR_PRESEAL / "package-supersession.json") rows = prior.get("superseded") if not isinstance(rows, list): raise RuntimeError("supersession") rows.append({ "package_digest": prior["active_package_digest"], "approval_phrase_reusable": False, "reason": "r43 wheel UX audit found internal normal and no_workout markers in customer copy", }) _write( PRESEAL / "package-supersession.json", { "schema": "nutricoach-v150-package-supersession-v15", "active_package_digest": package_digest, "active_package_sha256": _sha(PREFLIGHT / "package.json"), "active_approval_phrase": phrase, "superseded": rows, }, ) entries = { path.relative_to(PRESEAL).as_posix(): _sha(path) for path in PRESEAL.rglob("*") if path.is_file() and path.name != "package-manifest.json" } _write( PRESEAL / "package-manifest.json", { "schema": "nutricoach-v150-detached-preseal-package-v15", "status": "FROZEN_AWAITING_AUDIT", "self_exclusion": "package-manifest.json only", "new_package_digest": package_digest, "new_approval_phrase": phrase, "approval_consumed": False, "entries": dict(sorted(entries.items())), }, ) for path in PRESEAL.rglob("*"): if path.is_file() and not path.is_symlink(): path.chmod(0o444) for path in sorted( (item for item in PRESEAL.rglob("*") if item.is_dir()), reverse=True ): path.chmod(0o555) PRESEAL.chmod(0o555) if any( stat.S_IMODE(path.stat(follow_symlinks=False).st_mode) & 0o222 for path in (PRESEAL, *PRESEAL.rglob("*")) ): raise RuntimeError("immutable_mode") print(json.dumps({"candidate_digest": digest, "package_digest": package_digest})) return 0 if __name__ == "__main__": raise SystemExit(main())