"""One-use r71b maintenance phase machine with safe rollback ordering.""" from __future__ import annotations import json import stat from dataclasses import dataclass from pathlib import Path from enum import StrEnum from types import TracebackType from typing import Final, Literal, Protocol, Self, assert_never, final from pydantic import JsonValue, TypeAdapter, ValidationError from scripts.nutricoach_v150_sealed_authority import atomic_write KNOWN_R70_CRON_JOB_ID: Final = "6e042d5dff68" KNOWN_R70_CRON_JOB_NAME: Final = "NutriCoach schedule dispatcher" KNOWN_R70_CRON_SCHEDULE: Final = "* * * * *" KNOWN_R70_CRON_ERROR: Final = "weekly operations failed: canonical authority identity drift" KNOWN_R70_CRON_ERROR_EVIDENCE_SHA256: Final = ( "968a3ea4e15d149736d47c239d410ed9b88b7dbb20e0703214a2a6b19c3025d0" ) KNOWN_R70_RUNTIME_CANDIDATE_DIGEST: Final = ( "81a7a06ec2e7a1595784c92ab61df0df7e138d13a2620d9a917e94999e7f7a04" ) _APPROVAL_PREFIX: Final = "AUTHORIZE NUTRICOACH V1.5 R71B MAINTENANCE UPGRADE " _CRON_OBJECT = TypeAdapter(dict[str, JsonValue]) class MaintenanceTransactionError(RuntimeError): """A maintenance transaction crossed a sealed fail-closed boundary.""" class MaintenancePhase(StrEnum): """Durable r71b controller phases after ordinary migration succeeds.""" PREFLIGHT = "PREFLIGHT" RESERVED = "RESERVED" MAINTENANCE_ARMED = "MAINTENANCE_ARMED" START_WATCH_ARMED = "START_WATCH_ARMED" STARTED = "STARTED" MAINTENANCE_CONFIRMED = "MAINTENANCE_CONFIRMED" COMMITTED = "COMMITTED" ROLLED_BACK_SAFE_CRON_PAUSED = "ROLLED_BACK_SAFE_CRON_PAUSED" @dataclass(frozen=True, slots=True) class R71bMaintenanceProof: """Sealed no-send oracle facts required for the narrow r70 exception.""" candidate_digest: str cron_job_id: str cron_job_name: str cron_schedule: str cron_error: str cron_error_evidence_sha256: str cron_ledger_sha256: str expected_cron_ledger_sha256: str runtime_candidate_digest: str expected_candidate_digest: str final_package_digest: str class R71bMaintenanceHost(Protocol): """Minimal side-effect boundary for the post-stop maintenance sequence.""" def reserve(self) -> None: ... def arm_maintenance(self) -> None: ... def arm_start_watch(self) -> None: ... def start_successor(self) -> None: ... def confirm_maintenance(self) -> None: ... def commit(self) -> None: ... def pause_predecessor_scheduler(self) -> None: ... def restore_predecessor(self) -> None: ... def recover_confirmed_maintenance(self) -> None: ... @final class _RollbackOnFailure: """Pause a provider-capable predecessor before restoring its bytes.""" def __init__(self, controller: R71bMaintenanceController) -> None: self._controller = controller def __enter__(self) -> Self: return self def __exit__( self, error_type: type[BaseException] | None, error: BaseException | None, traceback: TracebackType | None, ) -> Literal[False]: del error_type, traceback if error is not None and self._controller.phase is not MaintenancePhase.MAINTENANCE_CONFIRMED: self._controller.rollback() return False @final class R71bMaintenanceController: """Mutable phase machine because one execution advances durable state.""" def __init__( self, host: R71bMaintenanceHost, proof: R71bMaintenanceProof | None, ) -> None: self._host = host # A post-confirmation recovery verifies the durable successor receipt, # not the predecessor's intentionally superseded error record. self._proof = proof self.phase = MaintenancePhase.PREFLIGHT def execute(self, approval: str) -> None: """Run the complete ordered r71b transaction for a single atomic host surface.""" self.arm(approval) self.complete_start() def arm(self, approval: str) -> None: """Reserve and install pre-start files before the systemd postimage is switched.""" if self.phase is not MaintenancePhase.PREFLIGHT: raise MaintenanceTransactionError("maintenance_phase") self._require_approval(approval) self._require_known_r70_exception() with _RollbackOnFailure(self): self._host.reserve() self.phase = MaintenancePhase.RESERVED self._host.arm_maintenance() self.phase = MaintenancePhase.MAINTENANCE_ARMED def complete_start(self) -> None: """Arm the cron event watch, start once, then require all postconditions.""" if self.phase is not MaintenancePhase.MAINTENANCE_ARMED: raise MaintenanceTransactionError("maintenance_phase") with _RollbackOnFailure(self): self._host.arm_start_watch() self.phase = MaintenancePhase.START_WATCH_ARMED self._host.start_successor() self.phase = MaintenancePhase.STARTED self._host.confirm_maintenance() self.phase = MaintenancePhase.MAINTENANCE_CONFIRMED self._host.commit() self.phase = MaintenancePhase.COMMITTED def recover(self, phase: MaintenancePhase) -> None: """Roll forward only an independently reverified maintenance confirmation.""" match phase: case MaintenancePhase.MAINTENANCE_CONFIRMED: self._host.recover_confirmed_maintenance() self._host.commit() self.phase = MaintenancePhase.COMMITTED return case MaintenancePhase.COMMITTED: self.phase = MaintenancePhase.COMMITTED return case MaintenancePhase.ROLLED_BACK_SAFE_CRON_PAUSED: self.phase = MaintenancePhase.ROLLED_BACK_SAFE_CRON_PAUSED return case ( MaintenancePhase.PREFLIGHT | MaintenancePhase.RESERVED | MaintenancePhase.MAINTENANCE_ARMED | MaintenancePhase.START_WATCH_ARMED | MaintenancePhase.STARTED ): self.rollback() return assert_never(phase) def rollback(self) -> None: """Leave a restarted predecessor unable to fire a provider-capable tick.""" self._host.pause_predecessor_scheduler() self._host.restore_predecessor() self.phase = MaintenancePhase.ROLLED_BACK_SAFE_CRON_PAUSED def _require_approval(self, approval: str) -> None: proof = self._proof digest = approval.removeprefix(_APPROVAL_PREFIX) if ( proof is None or not approval.startswith(_APPROVAL_PREFIX) or not _is_digest(digest) or digest != proof.final_package_digest ): raise MaintenanceTransactionError("r71b_approval") def _require_known_r70_exception(self) -> None: proof = self._proof if proof is None or ( proof.cron_job_id != KNOWN_R70_CRON_JOB_ID or proof.cron_job_name != KNOWN_R70_CRON_JOB_NAME or proof.cron_schedule != KNOWN_R70_CRON_SCHEDULE or proof.cron_error != KNOWN_R70_CRON_ERROR or proof.cron_error_evidence_sha256 != KNOWN_R70_CRON_ERROR_EVIDENCE_SHA256 or proof.candidate_digest != proof.expected_candidate_digest or not _is_digest(proof.candidate_digest) or proof.cron_ledger_sha256 != proof.expected_cron_ledger_sha256 or not _is_digest(proof.cron_ledger_sha256) or proof.runtime_candidate_digest != KNOWN_R70_RUNTIME_CANDIDATE_DIGEST or not _is_digest(proof.final_package_digest) ): raise MaintenanceTransactionError("r70_exception_drift") def _is_digest(value: str) -> bool: return len(value) == 64 and all(character in "0123456789abcdef" for character in value) def pause_r70_scheduler(path: Path) -> None: """Atomically disable only the known unsafe r70 dispatcher while preserving truth.""" info = path.stat(follow_symlinks=False) if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: raise MaintenanceTransactionError("cron_snapshot") jobs = _load_r70_jobs(path) _pause_exact_r70_job(jobs) atomic_write( path, json.dumps({"jobs": jobs}, sort_keys=True, separators=(",", ":")).encode() + b"\n", stat.S_IMODE(info.st_mode), ) def verify_paused_r70_scheduler(path: Path) -> None: """Require the restored predecessor's only provider-capable cron job stays paused.""" info = path.stat(follow_symlinks=False) if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: raise MaintenanceTransactionError("cron_snapshot") jobs = _load_r70_jobs(path) matches = [job for job in jobs if job.get("id") == KNOWN_R70_CRON_JOB_ID] if len(matches) != 1: raise MaintenanceTransactionError("cron_job_identity") job = matches[0] if ( job.get("name") != KNOWN_R70_CRON_JOB_NAME or job.get("schedule_display") != KNOWN_R70_CRON_SCHEDULE or job.get("inline_card") != "nutrition-coaching-tick" or job.get("deliver") != "local" or job.get("enabled") is not False or job.get("state") != "paused" or job.get("last_error") != KNOWN_R70_CRON_ERROR or job.get("paused_reason") != "r71b maintenance rollback safety fence" ): raise MaintenanceTransactionError("cron_pause_fence") def _load_r70_jobs(path: Path) -> list[dict[str, JsonValue]]: try: document = _CRON_OBJECT.validate_json(path.read_bytes()) except (OSError, ValidationError) as error: raise MaintenanceTransactionError("cron_json") from error rows = document.get("jobs") if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows): raise MaintenanceTransactionError("cron_json") return [row for row in rows if isinstance(row, dict)] def _pause_exact_r70_job(jobs: list[dict[str, JsonValue]]) -> None: matches = [job for job in jobs if job.get("id") == KNOWN_R70_CRON_JOB_ID] if len(matches) != 1: raise MaintenanceTransactionError("cron_job_identity") job = matches[0] if ( job.get("name") != KNOWN_R70_CRON_JOB_NAME or job.get("schedule_display") != KNOWN_R70_CRON_SCHEDULE or job.get("inline_card") != "nutrition-coaching-tick" or job.get("deliver") != "local" or job.get("last_error") != KNOWN_R70_CRON_ERROR ): raise MaintenanceTransactionError("cron_job_drift") if job.get("state") == "paused": if ( job.get("enabled") is False and job.get("paused_reason") == "r71b maintenance rollback safety fence" ): return raise MaintenanceTransactionError("cron_job_drift") if job.get("state") != "scheduled": raise MaintenanceTransactionError("cron_job_drift") job.update({ "enabled": False, "state": "paused", "paused_at": None, "paused_reason": "r71b maintenance rollback safety fence", })