"""Concrete filesystem, installer, migration, and service adapter.""" from __future__ import annotations import hashlib import json import os import shutil import sys from contextlib import AbstractContextManager from datetime import UTC, datetime from pathlib import Path from typing import Self, cast, final from pydantic import JsonValue, TypeAdapter from checkin_cli.customer_coaching import RegistryDocument, load_customer_registry from gateway.platforms.task26_candidate_authority import ( verify_candidate_authority, ) from gateway.platforms.task26_runtime_authority import ( append_external_authority, build_runtime_authority_pin, canonical as authority_canonical, load_task26_production_authority, publish_runtime_authority_pin, recover_external_authority, ) from scripts.nutricoach_v150_live_models import ( CANDIDATE_DIGEST, HERMES_WHEEL, HERMES_WHEEL_SHA256, PROFILE_WHEEL, PROFILE_WHEEL_SHA256, ) from scripts.nutricoach_v150_host_operations import ( apply_migrations, capacity_after, post_fence, protected_paths, publish_runtime_authority_credentials, service_state, snapshot_paths, switch_postimages, verify_postimages, ) from scripts.nutricoach_v150_live_upgrade_boundary import clean_boundary from scripts.nutricoach_v150_sealed_authority import atomic_write, digest from scripts.nutricoach_v150_runtime_ops import ( channel_inbox_enabled, dependency_snapshot_digest, install_fresh_runtime, make_writable, remove_tree, run_installed_weekly_startup_smoke, ) from scripts.nutricoach_v150_sealed_target import ( DisposableService, HostError, HostPaths, Service, SystemdService, ) from scripts.nutricoach_v150_weekly_authority import ( missing_canonical_weekly_files, weekly_authority_path, ) _OBJECT = TypeAdapter(dict[str, JsonValue]) def _text(value: JsonValue | None, label: str) -> str: if not isinstance(value, str): raise HostError(f"sealed_target:{label}") return value def _wheel(value: JsonValue | None, label: str) -> tuple[Path, str]: if not isinstance(value, dict): raise HostError(f"sealed_target:{label}") return ( Path(_text(value.get("path"), f"{label}_path")), _text(value.get("sha256"), f"{label}_sha256"), ) def _credential_sources(dropin: Path) -> dict[str, Path]: sources: dict[str, Path] = {} for line in dropin.read_bytes().splitlines(): if not line.startswith(b"LoadCredential=") or b":" not in line: continue name, raw_path = line.split(b"=", 1)[1].split(b":", 1) path = Path(raw_path.decode()) if not path.is_absolute(): raise HostError("runtime_authority_credential_path") sources[name.decode()] = path return sources @final class ConcreteLiveHost: """Actual target adapter with no provider, Telegram, or network calls.""" def __init__( self, paths: HostPaths, service: Service, *, live: bool, candidate_digest: str = CANDIDATE_DIGEST, hermes_wheel: Path = HERMES_WHEEL, hermes_wheel_sha256: str = HERMES_WHEEL_SHA256, profile_wheel: Path = PROFILE_WHEEL, profile_wheel_sha256: str = PROFILE_WHEEL_SHA256, registry_sha256: str | None = None, protected_inventory_sha256: str | None = None, fault: BaseException | None = None, fault_stage: str = "install_exact_wheels", cleanup_fault: BaseException | None = None, receipt_preserving_migration: bool | None = None, dependency_snapshot: Path | None = None, dependency_snapshot_sha256: str | None = None, expected_runtime_authority_paths: tuple[Path, Path, Path] | None = None, ) -> None: self.paths = paths self.service = service self.live = live self.candidate_digest = candidate_digest self.hermes_wheel = hermes_wheel self.hermes_wheel_sha256 = hermes_wheel_sha256 self.profile_wheel = profile_wheel self.profile_wheel_sha256 = profile_wheel_sha256 self.registry_sha256 = registry_sha256 self.protected_inventory_sha256 = protected_inventory_sha256 self.fault = fault self.fault_stage = fault_stage self.cleanup_fault = cleanup_fault self.receipt_preserving_migration = ( live if receipt_preserving_migration is None else receipt_preserving_migration ) self.dependency_snapshot = dependency_snapshot self.dependency_snapshot_sha256 = dependency_snapshot_sha256 self._expected_runtime_authority_paths = expected_runtime_authority_paths self.stages: list[str] = [] self.rollback_failures: list[str] = [] self.clean_boundary_checks = 0 self.network_events: list[str] = [] self.telegram_events: list[str] = [] self.provider_events: list[str] = [] self._preflight: dict[Path, str] | None = None self._migration_after: bytes | None = None self._created_files: tuple[Path, ...] = () self._created_directories: tuple[Path, ...] = () self._weekly_created_files: tuple[Path, ...] = () self._postimages: dict[Path, str] = {} self._runtime_authority: dict[str, object] | None = None self._successor_authority_pin: bytes | None = None self._expected_created_files: set[Path] | None = None self._expected_created_directories: set[Path] | None = None self._weekly_authority: Path = weekly_authority_path( paths, candidate_digest, ) @classmethod def disposable( cls, root: Path, service: DisposableService, *, fault: BaseException | None = None, fault_stage: str = "install_exact_wheels", cleanup_fault: BaseException | None = None, receipt_preserving_migration: bool = False, ) -> Self: profile = root / "profile" paths = HostPaths( profile, profile / "customers/registry.json", profile / "config.yaml", root / "gateway.service", root / "authority.conf", profile / ".strict-runtime/current/venv", profile / ".strict-runtime/successor/venv", root / "sealed-execution", root / "global-authorization", ) return cls( paths, service, live=False, fault=fault, fault_stage=fault_stage, cleanup_fault=cleanup_fault, receipt_preserving_migration=receipt_preserving_migration, ) @classmethod def live_target(cls, sealed_target: Path) -> Self: binding = _OBJECT.validate_json(sealed_target.read_bytes()) candidate_digest = _text(binding.get("candidate_digest"), "candidate_digest") wheels = binding.get("wheels") if not isinstance(wheels, list) or len(wheels) != 2: raise HostError("sealed_target:wheels") hermes_wheel, hermes_sha256 = _wheel(wheels[0], "hermes_wheel") profile_wheel, profile_sha256 = _wheel(wheels[1], "profile_wheel") profile = Path(_text(binding.get("profile_root"), "profile_root")) service_name = _text(binding.get("service_name"), "service_name") registry_sha256 = _text(binding.get("registry_sha256"), "registry_sha256") protected_inventory_sha256 = _text( binding.get("protected_inventory_sha256"), "protected_inventory_sha256", ) dependency_snapshot = Path( _text(binding.get("dependency_snapshot"), "dependency_snapshot") ) dependency_snapshot_sha256 = _text( binding.get("dependency_snapshot_sha256"), "dependency_snapshot_sha256", ) baseline = binding.get("authority_baseline") if not isinstance(baseline, dict): raise HostError("sealed_target:authority_baseline") expected_runtime_authority_paths = ( Path(_text(baseline.get("authority_root"), "authority_root")), Path(_text(baseline.get("pin_path"), "authority_pin")), Path(_text(baseline.get("candidate_path"), "authority_candidate")), ) paths = HostPaths( profile, profile / "customers/registry.json", profile / "config.yaml", Path(_text(binding.get("unit"), "unit")), Path(_text(binding.get("dropin"), "dropin")), Path(_text(binding.get("current_runtime"), "current_runtime")), Path(_text(binding.get("successor_runtime"), "successor_runtime")), Path(_text(binding.get("execution_root"), "execution_root")), Path(_text(binding.get("global_approval_ledger"), "ledger")), Path(_text(binding.get("protected_inventory"), "inventory")), ) return cls( paths, SystemdService(service_name), live=True, candidate_digest=candidate_digest, hermes_wheel=hermes_wheel, hermes_wheel_sha256=hermes_sha256, profile_wheel=profile_wheel, profile_wheel_sha256=profile_sha256, registry_sha256=registry_sha256, protected_inventory_sha256=protected_inventory_sha256, dependency_snapshot=dependency_snapshot, dependency_snapshot_sha256=dependency_snapshot_sha256, expected_runtime_authority_paths=expected_runtime_authority_paths, ) @property def successor_root(self) -> Path: return self.paths.successor_runtime.parent @property def weekly_authority(self) -> Path: return self._weekly_authority def _stage(self, name: str) -> None: self.stages.append(name) if self.fault is not None and self.fault_stage == name: error, self.fault = self.fault, None raise error def capture_preflight(self) -> None: if ( self.protected_inventory_sha256 is not None and self.paths.protected_inventory is not None and hashlib.sha256(self.paths.protected_inventory.read_bytes()).hexdigest() != self.protected_inventory_sha256 ): raise HostError("protected_inventory_drift") if ( self.registry_sha256 is not None and hashlib.sha256(self.paths.registry.read_bytes()).hexdigest() != self.registry_sha256 ): raise HostError("registry_drift") self._preflight = { path: digest(path.read_bytes()) for path in protected_paths(self.paths) } self._capture_runtime_authority() def _capture_runtime_authority(self) -> None: credentials = _credential_sources(self.paths.dropin) pin_path = credentials.get("task26-authority-pin.json") candidate_path = credentials.get("task26-candidate-digest") if pin_path is None and candidate_path is None: self._runtime_authority = None return if pin_path is None or candidate_path is None: raise HostError("runtime_authority_credentials") candidate = candidate_path.read_text(encoding="utf-8").strip() 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, authorized = load_task26_production_authority( profile_root=self.paths.profile, package_root=self.paths.execution_root, ) 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 if authorized != candidate: raise HostError("runtime_authority_candidate") current = verify_candidate_authority(source.root, candidate) registry = _OBJECT.validate_json( (source.root / "candidate-authority/registry.json").read_bytes() ) events = registry.get("events") if not isinstance(events, list) or not events: raise HostError("runtime_authority_events") latest = events[-1] if not isinstance(latest, dict): raise HostError("runtime_authority_events") historical_pass = latest.get("historical_pass_digest") if not isinstance(historical_pass, str): raise HostError("runtime_authority_historical_pass") self._runtime_authority = { "authority_root": str(source.root), "candidate_digest": candidate, "event_count": len(events), "genesis_sha256": registry.get("genesis_sha256"), "historical_pass_digest": historical_pass, "ledger_head_sha256": current["ledger_head_sha256"], "pin_path": str(pin_path), "candidate_path": str(candidate_path), "registry_head_sha256": current["registry_head_sha256"], "source_id": registry.get("source_id"), } if self._expected_runtime_authority_paths is None: self._expected_runtime_authority_paths = ( source.root, pin_path, candidate_path, ) def write_recovery_manifest(self) -> None: """Persist pre-stop restoration inputs for a fresh recovery process.""" if self._preflight is None: raise HostError("preflight_missing") receipt_root = self.paths.profile / "data/customer-activation-receipts" document = RegistryDocument.model_validate_json( self.paths.registry.read_bytes() ) missing_receipts = ( tuple( receipt_root / f"{customer.customer_key}.json" for customer in document.customers if customer.enabled and not (receipt_root / f"{customer.customer_key}.json").exists() ) if self.receipt_preserving_migration else () ) self._weekly_created_files = missing_canonical_weekly_files(self.paths) self._created_files = ( *self._weekly_created_files, *missing_receipts, ) self._created_directories = ( (receipt_root,) if missing_receipts and not receipt_root.exists() else () ) self._expected_created_files = set(self._created_files) self._expected_created_directories = set(self._created_directories) payload = { "created_directories": [str(path) for path in self._created_directories], "created_files": [str(path) for path in self._created_files], "preflight": { str(path): expected for path, expected in self._preflight.items() }, "runtime_authority": self._runtime_authority, "weekly_authority": str(self.weekly_authority), "schema": "nutricoach-v150-recovery-manifest-v1", } atomic_write( self.paths.execution_root / "recovery-manifest.json", json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + b"\n", 0o400, ) def load_recovery_manifest(self) -> None: """Reconstruct rollback state without relying on the original process.""" path = self.paths.execution_root / "recovery-manifest.json" document = _OBJECT.validate_json(path.read_bytes()) raw_preflight = document.get("preflight") raw_created = document.get("created_files") raw_directories = document.get("created_directories", []) raw_runtime_authority = document.get("runtime_authority") expected_weekly_authority = self.weekly_authority raw_weekly_authority = document.get( "weekly_authority", str(expected_weekly_authority), ) if ( not isinstance(raw_preflight, dict) or not isinstance(raw_created, list) or not isinstance(raw_directories, list) or not isinstance(raw_weekly_authority, str) ): raise HostError("recovery_manifest") preflight: dict[Path, str] = {} for raw_path, raw_digest in raw_preflight.items(): if not isinstance(raw_digest, str): raise HostError("recovery_manifest") preflight[Path(raw_path)] = raw_digest created_values: list[Path] = [] for item in raw_created: if not isinstance(item, str): raise HostError("recovery_manifest") created_values.append(Path(item)) directory_values: list[Path] = [] for item in raw_directories: if not isinstance(item, str): raise HostError("recovery_manifest") directory_values.append(Path(item)) self._preflight = preflight self._created_files = tuple(created_values) self._created_directories = tuple(directory_values) self._weekly_authority = Path(raw_weekly_authority) if raw_runtime_authority is None: self._runtime_authority = None elif isinstance(raw_runtime_authority, dict): self._runtime_authority = { str(key): value for key, value in raw_runtime_authority.items() } else: raise HostError("recovery_manifest") runtime_authority = self._runtime_authority expected_preflight = set(protected_paths(self.paths)) expected_files = self._expected_created_files expected_directories = self._expected_created_directories if expected_files is None or expected_directories is None: expected_files, expected_directories = self._expected_created_paths( expected_preflight, ) authority_paths = self._expected_runtime_authority_paths authority_bound = ( authority_paths is not None and runtime_authority is not None and runtime_authority.get("authority_root") == str(authority_paths[0]) and runtime_authority.get("pin_path") == str(authority_paths[1]) and runtime_authority.get("candidate_path") == str(authority_paths[2]) ) if ( Path(raw_weekly_authority) != expected_weekly_authority or set(preflight) != expected_preflight or set(self._created_files) != expected_files or set(self._created_directories) != expected_directories or not authority_bound ): raise HostError("recovery_manifest_paths") def _expected_created_paths( self, preflight_paths: set[Path], ) -> tuple[set[Path], set[Path]]: receipt_root = self.paths.profile / "data/customer-activation-receipts" allowed: set[Path] = set() receipt_paths: set[Path] = set() registry = load_customer_registry(self.paths.registry, self.paths.profile) for runtime in registry.customers: if not runtime.spec.enabled: continue receipt = receipt_root / f"{runtime.spec.customer_key}.json" receipt_paths.add(receipt) allowed.update(( runtime.wizard_root / "events.jsonl", runtime.wizard_root / ".events.lock", runtime.nutrition_plans_root / "canonical-sequence.jsonl", receipt, )) expected_files = allowed - preflight_paths missing_receipts = expected_files & receipt_paths receipt_root_preexisted = any( path.is_relative_to(receipt_root) for path in preflight_paths ) expected_directories: set[Path] = set() if missing_receipts and not receipt_root_preexisted: expected_directories.add(receipt_root) return expected_files, expected_directories def verify_preflight(self) -> None: if self._preflight is None: raise HostError("preflight_missing") if any( digest(path.read_bytes()) != expected for path, expected in self._preflight.items() ): raise HostError("protected_drift") def stop(self) -> None: self.service.stop() if self.service.running: raise HostError("service_stop") self._stage("stop") def record_snapshot(self) -> None: self._stage("post_stop_snapshot") def stopped_probe(self) -> None: self.verify_preflight() if not (self.paths.current_runtime / "bin/python").is_file(): raise HostError("current_runtime") self._stage("stopped_probe") def install(self) -> None: self.verify_created_absent() if self.successor_root.exists(): raise HostError("successor_exists") if self.live: if ( hashlib.sha256(self.hermes_wheel.read_bytes()).hexdigest() != self.hermes_wheel_sha256 or hashlib.sha256(self.profile_wheel.read_bytes()).hexdigest() != self.profile_wheel_sha256 ): raise HostError("wheel_integrity") if ( self.dependency_snapshot is None or self.dependency_snapshot_sha256 is None or dependency_snapshot_digest(self.dependency_snapshot) != self.dependency_snapshot_sha256 ): raise HostError("dependency_integrity") _ = install_fresh_runtime( self.paths.successor_runtime, self.hermes_wheel, self.profile_wheel, self.dependency_snapshot, ) else: _ = shutil.copytree(self.paths.current_runtime.parent, self.successor_root) make_writable(self.successor_root) receipt = { "hermes": HERMES_WHEEL_SHA256, "profile": PROFILE_WHEEL_SHA256, } atomic_write( self.successor_root / "installed-wheels.json", json.dumps(receipt, sort_keys=True).encode() + b"\n", ) self._stage("install_exact_wheels") def off_smoke(self) -> None: if channel_inbox_enabled(self.paths.config.read_text(encoding="utf-8")): raise HostError("channel_inbox_on") self._stage("off_smoke_channel_inbox_off") def semantic_clean_boundary(self) -> None: if self.live: _ = clean_boundary(self.paths.profile, datetime.now(UTC)) self.clean_boundary_checks += 1 def snapshot_paths(self) -> tuple[Path, ...]: return snapshot_paths(self.paths) def verify_created_absent(self) -> None: for path in ( self.successor_root, self.weekly_authority, *self._created_files, *self._created_directories, ): if path.exists() or path.is_symlink(): raise HostError("created_path") def verify_created_owned(self) -> None: expected_uid = self.paths.profile.stat().st_uid expected_gid = self.paths.profile.stat().st_gid for path in (self.successor_root, self.weekly_authority): info = path.stat(follow_symlinks=False) if ( path.is_symlink() or not path.is_dir() or info.st_uid != expected_uid or info.st_gid != expected_gid ): raise HostError("created_path_ownership") for path in self._created_files: info = path.stat(follow_symlinks=False) if ( path.is_symlink() or not path.is_file() or info.st_uid != expected_uid or info.st_gid != expected_gid or info.st_mode & 0o777 != 0o600 ): raise HostError("created_path_ownership") for path in self._created_directories: info = path.stat(follow_symlinks=False) if ( path.is_symlink() or not path.is_dir() or info.st_uid != expected_uid or info.st_gid != expected_gid or info.st_mode & 0o777 != 0o700 ): raise HostError("created_path_ownership") @staticmethod def install_fresh_runtime(runtime: Path) -> dict[str, str]: return install_fresh_runtime(runtime) def migration_dry_run(self) -> None: if not self._created_files: self._created_files = missing_canonical_weekly_files(self.paths) self._migration_after = capacity_after( self.paths.registry, self.candidate_digest, ) self._stage("capacity_dry_run") def migration_apply(self) -> None: if self._migration_after is None: raise HostError("capacity_dry_run_missing") apply_migrations( self.paths, self._migration_after, self.candidate_digest, self._weekly_created_files, self.weekly_authority, receipt_preserving=self.receipt_preserving_migration, ) self.verify_created_owned() self._stage("weekly_and_capacity_apply") def weekly_startup_smoke(self) -> None: runtime = ( self.paths.successor_runtime if self.live else Path(sys.executable).resolve().parent.parent ) run_installed_weekly_startup_smoke( runtime, self.paths.profile, self.paths.config, candidate_digest=self.candidate_digest, hermes_wheel=self.hermes_wheel, profile_wheel=self.profile_wheel, source_dependencies=not self.live, ) self._stage("weekly_startup_smoke") def promote_runtime_authority(self) -> None: """Append successor QUALIFY and publish exact canonical credentials.""" baseline = self._runtime_authority if baseline is None: return root = Path(str(baseline["authority_root"])) _ = recover_external_authority(root) current = verify_candidate_authority(root, None) registry = _OBJECT.validate_json( (root / "candidate-authority/registry.json").read_bytes() ) events = registry.get("events") if not isinstance(events, list): raise HostError("runtime_authority_events") baseline_count = int(str(baseline["event_count"])) predecessor = str(baseline["candidate_digest"]) if ( len(events) == baseline_count and current["current_qualified_candidate"] == predecessor ): _ = append_external_authority( root, source_id=str(baseline["source_id"]), candidate_digest=self.candidate_digest, action="qualify", historical_pass_digest=str(baseline["historical_pass_digest"]), reason="NutriCoach v1.5 successor qualification", ) current = verify_candidate_authority(root, self.candidate_digest) registry = _OBJECT.validate_json( (root / "candidate-authority/registry.json").read_bytes() ) events = registry.get("events") if ( not isinstance(events, list) or len(events) != baseline_count + 1 or current["registry_head_sha256"] == baseline["registry_head_sha256"] or current["ledger_head_sha256"] == baseline["ledger_head_sha256"] ): raise HostError("runtime_authority_successor_state") pin = build_runtime_authority_pin(root) self._successor_authority_pin = authority_canonical(pin) + b"\n" credentials = publish_runtime_authority_credentials( self.paths, self.candidate_digest, self._successor_authority_pin, ) self._verify_runtime_authority_credentials(credentials, self.candidate_digest) self._stage("promote_runtime_authority") def _verify_runtime_authority_credentials( self, credentials: Path, expected_candidate: str, ) -> None: old_directory = os.environ.get("CREDENTIALS_DIRECTORY") old_pin = os.environ.pop("TASK26_AUTHORITY_PIN", None) old_candidate = os.environ.pop("TASK26_CANDIDATE_DIGEST_FILE", None) os.environ["CREDENTIALS_DIRECTORY"] = str(credentials) try: source, candidate = load_task26_production_authority( profile_root=self.paths.profile, package_root=self.paths.execution_root, ) if candidate != expected_candidate: raise HostError("runtime_authority_candidate") with source.authorize(candidate, "service_activation") as snapshot: from gateway.config import PlatformConfig from gateway.platforms.telegram import TelegramAdapter adapter = TelegramAdapter( PlatformConfig( enabled=True, token="authority-pre-network-smoke", extra={"nutrition_coaching": {"enabled": True}}, ) ) setattr(adapter, "_task26_runtime_required", True) setattr(adapter, "_task26_authority_source", source) setattr(adapter, "_task26_candidate_digest", candidate) setattr( adapter, "_task26_service_authority_snapshot", snapshot, ) authorize = getattr( adapter, "_authorize_task26_service_network_start", None, ) if not callable(authorize): raise HostError("telegram_authority_activation_path") manager = cast( AbstractContextManager[dict[str, object]], authorize(), ) with manager as adapter_snapshot: if adapter_snapshot.get("candidate_digest") != expected_candidate: raise HostError("telegram_authority_activation_path") finally: if old_directory is None: _ = os.environ.pop("CREDENTIALS_DIRECTORY", None) else: os.environ["CREDENTIALS_DIRECTORY"] = old_directory if old_pin is not None: os.environ["TASK26_AUTHORITY_PIN"] = old_pin if old_candidate is not None: os.environ["TASK26_CANDIDATE_DIGEST_FILE"] = old_candidate def restore_runtime_authority(self) -> None: """Compensate successor promotion and verify predecessor credentials.""" baseline = self._runtime_authority if baseline is None: return root = Path(str(baseline["authority_root"])) _ = recover_external_authority(root) current = verify_candidate_authority(root, None) registry = _OBJECT.validate_json( (root / "candidate-authority/registry.json").read_bytes() ) events = registry.get("events") if not isinstance(events, list): raise HostError("runtime_authority_events") baseline_count = int(str(baseline["event_count"])) predecessor = str(baseline["candidate_digest"]) if ( len(events) == baseline_count + 1 and current["current_qualified_candidate"] == self.candidate_digest ): _ = append_external_authority( root, source_id=str(baseline["source_id"]), candidate_digest=predecessor, action="qualify", historical_pass_digest=str(baseline["historical_pass_digest"]), reason="NutriCoach v1.5 predecessor rollback compensation", ) current = verify_candidate_authority(root, predecessor) registry = _OBJECT.validate_json( (root / "candidate-authority/registry.json").read_bytes() ) events = registry.get("events") if not isinstance(events, list) or len(events) not in { baseline_count, baseline_count + 2, }: raise HostError("runtime_authority_rollback_state") pin = build_runtime_authority_pin(root) pin_path = Path(str(baseline["pin_path"])) publish_runtime_authority_pin(pin_path, pin) credentials = pin_path.parent candidate_path = Path(str(baseline["candidate_path"])) 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=self.paths.profile, package_root=self.paths.execution_root, ) if candidate != predecessor: raise HostError("runtime_authority_predecessor") with source.authorize(candidate, "activation"): pass finally: del credentials 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 def switch_systemd(self) -> None: self._postimages = switch_postimages( self.paths, self.candidate_digest, self._successor_authority_pin, ) self._stage("switch_unit_dropin") def reload(self) -> None: self.service.reload() self._stage("reload") def start(self) -> None: self.service.start() self._stage("start") def post_fence(self) -> None: post_fence( self.paths, self.service_state(), self._postimages, self.candidate_digest, ) credentials = self.paths.successor_runtime.parent / "runtime-authority" if self._runtime_authority is not None: self._verify_runtime_authority_credentials( credentials, self.candidate_digest, ) self._stage("post_fence") def remove_created(self) -> None: for path in self._created_files: if path.is_file() and not path.is_symlink(): path.unlink() for path in (self.successor_root, self.weekly_authority): remove_tree(path) for path in reversed(self._created_directories): if path.is_dir() and not path.is_symlink(): path.rmdir() if self.cleanup_fault is not None: error, self.cleanup_fault = self.cleanup_fault, None raise error def service_state(self) -> dict[str, str]: return service_state(self.service, self.paths) def verify_exact_postimages(self) -> bool: return verify_postimages(self._postimages) def ledger_reserved(self) -> bool: return (self.paths.ledger_root / "authorization-reserved.json").exists() def ledger_consumed(self) -> bool: return (self.paths.ledger_root / "authorization-consumed.json").exists()