"""One-use authorization ledger and reloadable hash-linked phase journal.""" from __future__ import annotations import hashlib import json from dataclasses import dataclass from pathlib import Path from typing import ClassVar, Final from pydantic import BaseModel, ConfigDict, Field from .contract import ZERO_DIGEST, Phase from .durable import publish_atomic, reserve_exclusive from .faults import NO_FAULTS, FaultInjector from .fdio import SecureIoError _SCHEMA: Final = "nutricoach-day-status-continuity-phase-v1" class JournalError(RuntimeError): """Durable transaction history is divergent or out of order.""" class PhaseRecord(BaseModel): """One immutable phase link.""" model_config: ClassVar[ConfigDict] = ConfigDict(frozen=True, extra="forbid") schema_version: str = Field(alias="schema") phase: Phase package_digest: str = Field(pattern=r"^[0-9a-f]{64}$") prior_phase_digest: str = Field(pattern=r"^[0-9a-f]{64}$") source_row_digest: str = Field(pattern=r"^[0-9a-f]{64}$") expected_target_frame_digest: str = Field(pattern=r"^[0-9a-f]{64}$") observer_baseline_head: str | None = None observer_manual_head: str | None = None observer_timer_head: str | None = None def canonical_bytes(self) -> bytes: """Encode the phase record canonically.""" value = self.model_dump(mode="json", by_alias=True) return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + b"\n" @property def digest(self) -> str: """Return the durable link digest.""" return hashlib.sha256(self.canonical_bytes()).hexdigest() @dataclass(frozen=True, slots=True) class PhaseContext: """Fixed link fields shared by every phase.""" package_digest: str source_row_digest: str expected_frame_digest: str class PhaseJournal: """Advance and reload only the exact design phase order.""" def __init__( self, root: Path, context: PhaseContext, faults: FaultInjector = NO_FAULTS, ) -> None: self._root: Path = root self._context: PhaseContext = context self._faults: FaultInjector = faults self._records: list[PhaseRecord] = [] @property def current(self) -> Phase | None: """Return the final durable phase.""" return self._records[-1].phase if self._records else None @property def records(self) -> tuple[PhaseRecord, ...]: """Return the validated durable prefix.""" return tuple(self._records) def reload(self) -> None: """Rehydrate and authenticate every phase file from disk.""" records: list[PhaseRecord] = [] names = sorted(self._root.glob("phase-*.json")) if len(names) > len(tuple(Phase)): raise JournalError("extra phase file") prior = ZERO_DIGEST for index, path in enumerate(names): record = PhaseRecord.model_validate_json(path.read_bytes()) expected_name = f"phase-{index:02d}-{record.phase.value}.json" if ( path.name != expected_name or record.phase is not tuple(Phase)[index] or record.prior_phase_digest != prior or record.package_digest != self._context.package_digest or record.source_row_digest != self._context.source_row_digest or record.expected_target_frame_digest != self._context.expected_frame_digest or path.stat(follow_symlinks=False).st_mode & 0o7777 != 0o400 ): raise JournalError("phase chain") prior = record.digest records.append(record) self._records = records def advance( self, phase: Phase, *, baseline: str | None = None, manual: str | None = None, timer: str | None = None, ) -> PhaseRecord: """Durably publish the next exact phase and verify readback.""" index = len(self._records) phases = tuple(Phase) if index >= len(phases) or phases[index] is not phase: raise JournalError("phase order") record = PhaseRecord( schema=_SCHEMA, phase=phase, package_digest=self._context.package_digest, prior_phase_digest=self._records[-1].digest if self._records else ZERO_DIGEST, source_row_digest=self._context.source_row_digest, expected_target_frame_digest=self._context.expected_frame_digest, observer_baseline_head=baseline, observer_manual_head=manual, observer_timer_head=timer, ) name = f"phase-{index:02d}-{phase.value}.json" publish_atomic( self._root, name, record.canonical_bytes(), faults=self._faults, checkpoint=f"phase-{index:02d}-{phase.value}", ) parsed = PhaseRecord.model_validate_json((self._root / name).read_bytes()) if parsed != record: raise JournalError("phase readback") self._records.append(record) return record class OneUseLedger: """Reserve one launch and consume it without deleting its reservation.""" def __init__( self, root: Path, package_digest: str, faults: FaultInjector = NO_FAULTS, ) -> None: self._root: Path = root self._package_digest: str = package_digest self._faults: FaultInjector = faults @property def pending(self) -> bool: """Return whether exactly one unconsumed reservation exists.""" return (self._root / "authorization-reserved.json").is_file() and not ( self._root / "authorization-consumed.json" ).exists() def reserve(self) -> None: """Create the sole immutable reservation.""" reserve_exclusive( self._root, "authorization-reserved.json", self._receipt("RESERVED"), faults=self._faults, ) def consume(self, outcome: str) -> None: """Publish one terminal outcome; a second consumption is refused.""" if outcome not in {"SUCCEEDED", "FAILED"}: raise JournalError("authorization outcome") try: reserve_exclusive( self._root, "authorization-consumed.json", self._receipt(f"CONSUMED/{outcome}"), faults=self._faults, ) except SecureIoError as error: raise JournalError("authorization already consumed") from error def consumed_outcome(self) -> str | None: """Return the authenticated terminal outcome when present.""" path = self._root / "authorization-consumed.json" if not path.is_file(): return None for outcome in ("SUCCEEDED", "FAILED"): if path.read_bytes() == self._receipt(f"CONSUMED/{outcome}"): return outcome raise JournalError("consumed receipt") def require_pending(self) -> None: """Authenticate the existing reservation for recovery only.""" path = self._root / "authorization-reserved.json" if not self.pending or path.read_bytes() != self._receipt("RESERVED"): raise JournalError("no matching pending reservation") def _receipt(self, status: str) -> bytes: return ( json.dumps( {"package_digest": self._package_digest, "status": status}, sort_keys=True, separators=(",", ":"), ).encode() + b"\n" )