"""Filesystem-backed append-only storage and deterministic view rebuilding.""" from __future__ import annotations import hashlib import json import os import re import stat from datetime import date, datetime, timedelta from dataclasses import dataclass from contextlib import contextmanager from collections.abc import Iterable, Iterator, Mapping from pathlib import Path from typing import Final, assert_never from zoneinfo import ZoneInfo import fcntl from pydantic import ValidationError from checkin_cli.customer_coaching import ( CustomerRegistryError, CustomerRuntime, RegisteredCustomerBinding, TwelveWeekPlan, load_customer_registry, ) from checkin_cli.history_imports import baseline_events, baseline_report, historical_view, legacy_history_events, legacy_manifest, parse_baseline from checkin_cli.models import ( CHEST_PAIN_FLAG, CheckinValues, ContractCheckin, ContractStatus, CurrentView, Event, EventType, OperatorTask, PaymentKind, PaymentMethod, Provenance, RecordRequest, RecordResult, Safety, WeeklyView, build_operator_time_event, build_payment_event, build_satisfaction_event, validate_event, ) KST: Final[ZoneInfo] = ZoneInfo("Asia/Seoul") _STORE_CONSTRUCTION_TOKEN = object() class _CanonicalTransactionLockToken: __slots__ = () _CANONICAL_TOKEN_TYPE: Final[type[_CanonicalTransactionLockToken]] = _CanonicalTransactionLockToken def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _canonical_row(value: Mapping[str, object]) -> bytes: return ( json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) .encode("utf-8") ) def terminal_morning_root_kst_day(events: Iterable[Event], event: Event) -> date | None: """Return an accepted terminal morning lineage's authoritative root KST day.""" rows = tuple(events) by_id = {item.event_id: item for item in rows} canonical = by_id.get(event.event_id) if canonical != event or canonical.status is not ContractStatus.ACCEPTED: return None superseded = {item.supersedes for item in rows if item.supersedes is not None} if canonical.event_id in superseded: return None seen: set[str] = set() current = canonical while current.supersedes is not None: if current.event_id in seen: return None seen.add(current.event_id) parent = by_id.get(current.supersedes) if parent is None: return None current = parent if current.status is not ContractStatus.ACCEPTED: return None if current.event_type is not EventType.MORNING_CHECKIN: return None return date.fromisoformat(str(current.occurred_at_kst)[:10]) def current_terminal_morning_response( events: Iterable[Event], kst_day: date, ) -> Event | None: """Return the accepted terminal response whose correction lineage starts at morning.""" if type(kst_day) is not date: raise TypeError("KST day is required") rows = tuple(events) terminal = ( event for event in rows if terminal_morning_root_kst_day(rows, event) == kst_day ) return max(terminal, key=lambda event: event.occurred_at_kst, default=None) @dataclass(frozen=True, slots=True) class CanonicalEventSnapshot: events: tuple[Event, ...] sequence_rows: tuple[Mapping[str, object], ...] @property def sequence(self) -> tuple[Mapping[str, object], ...]: return self.sequence_rows def __iter__(self) -> Iterator[object]: yield self.events yield self.sequence_rows class CanonicalEventTransaction: """One lock-domain transaction for a wizard event and its sequence sidecar.""" def __init__( self, events_path: Path, sequence_path: Path, *, lock_path: Path | None = None, ) -> None: self.events_path = Path(events_path) self.sequence_path = Path(sequence_path) self.lock_path = self.events_path.parent / ".events.lock" if lock_path is not None and Path(lock_path) != self.lock_path: raise ValueError("canonical transaction lock must be wizard/.events.lock") self._validate_constructor_paths() self._active_lock_token: _CanonicalTransactionLockToken | None = None @classmethod def for_customer_runtime(cls, runtime: CustomerRuntime) -> CanonicalEventTransaction: if not isinstance(runtime, CustomerRuntime): raise TypeError("registered canonical transactions require CustomerRuntime") binding = runtime.binding if not isinstance(binding, RegisteredCustomerBinding): raise ValueError("registered customer runtime has no sealed binding") customer_root = runtime.customer_root wizard_root = runtime.wizard_root nutrition_root = runtime.nutrition_plans_root for path in (customer_root, wizard_root, nutrition_root): if _path_has_symlink(path, customer_root): raise ValueError("registered customer canonical roots cannot contain symlinks") expected_root_digest = hashlib.sha256(str(customer_root).encode("utf-8")).hexdigest() if binding.data_root_digest != expected_root_digest: raise ValueError("registered customer data-root binding mismatch") return cls(wizard_root / "events.jsonl", nutrition_root / "canonical-sequence.jsonl") def _validate_constructor_paths(self) -> None: if self.events_path.name != "events.jsonl": raise ValueError("canonical event path must be wizard/events.jsonl") if self.sequence_path.name != "canonical-sequence.jsonl": raise ValueError("canonical sequence path is invalid") if self.events_path.is_symlink() or self.sequence_path.is_symlink() or self.lock_path.is_symlink(): raise ValueError("canonical transaction paths cannot be symlinks") if self.sequence_path.parent.name != "nutrition-plans": raise ValueError("canonical sequence path must be nutrition-plans/canonical-sequence.jsonl") if self.events_path.parent.name != "wizard": raise ValueError("canonical event path must be under wizard") if self.events_path.parent.parent != self.sequence_path.parent.parent: raise ValueError("canonical event and sequence roots must match") for path in (self.events_path, self.sequence_path, self.lock_path): if path.exists() and path.stat().st_mode & 0o077: raise ValueError("canonical transaction paths must be private") for parent in (self.events_path.parent, self.sequence_path.parent): if parent.exists(): if not parent.is_dir() or parent.is_symlink() or stat.S_IMODE(parent.stat().st_mode) & 0o002: raise ValueError("canonical transaction directories must be private") @contextmanager def _locked(self) -> Iterator[object]: if self._active_lock_token is not None: raise RuntimeError("canonical transaction lock is already held") self.lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC descriptor = os.open(self.lock_path, flags, 0o600) try: opened = os.fstat(descriptor) named = self.lock_path.lstat() identity = (opened.st_dev, opened.st_ino) if ( not stat.S_ISREG(opened.st_mode) or opened.st_uid != os.geteuid() or opened.st_nlink != 1 or stat.S_IMODE(opened.st_mode) != 0o600 or identity != (named.st_dev, named.st_ino) ): raise RuntimeError("canonical transaction lock is unsafe") fcntl.flock(descriptor, fcntl.LOCK_EX) current = self.lock_path.lstat() if identity != (current.st_dev, current.st_ino): raise RuntimeError("canonical transaction lock was replaced") token = _CanonicalTransactionLockToken() self._active_lock_token = token try: yield token finally: self._active_lock_token = None fcntl.flock(descriptor, fcntl.LOCK_UN) finally: os.close(descriptor) @contextmanager def _shared_locked(self) -> Iterator[object]: if self._active_lock_token is not None: raise RuntimeError("canonical transaction lock is already held") flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC descriptor = os.open(self.lock_path, flags) try: opened = os.fstat(descriptor) named = self.lock_path.lstat() identity = (opened.st_dev, opened.st_ino) if ( not stat.S_ISREG(opened.st_mode) or opened.st_uid != os.geteuid() or opened.st_nlink != 1 or stat.S_IMODE(opened.st_mode) != 0o600 or identity != (named.st_dev, named.st_ino) ): raise RuntimeError("canonical transaction lock is unsafe") fcntl.flock(descriptor, fcntl.LOCK_SH) current = self.lock_path.lstat() if identity != (current.st_dev, current.st_ino): raise RuntimeError("canonical transaction lock was replaced") token = _CanonicalTransactionLockToken() self._active_lock_token = token try: yield token finally: self._active_lock_token = None fcntl.flock(descriptor, fcntl.LOCK_UN) finally: os.close(descriptor) def read_snapshot_readonly(self) -> CanonicalEventSnapshot: """Read a valid canonical pair without creating or recovering files.""" with self._shared_locked() as token: return self._read_snapshot_locked(token) @contextmanager def read_locked(self) -> Iterator[CanonicalEventSnapshot]: """Yield one validated canonical pair while retaining the shared lock.""" with self._shared_locked() as token: yield self._read_snapshot_locked(token) def _require_token(self, token: object) -> None: if not isinstance(token, _CANONICAL_TOKEN_TYPE) or token is not self._active_lock_token: raise TypeError("canonical transaction lock token is invalid") @contextmanager def locked(self) -> Iterator[object]: """Hold the canonical writer lock for one multi-step authority decision.""" with self._locked() as token: yield token def _current_schedule_reference_locked( self, customer_key: str, token: object ) -> Event | None: self._require_token(token) events, sequences = self._read_pair_locked(token) self._validate_pair(events, sequences) superseded = {event.supersedes for event in events if event.supersedes} candidates = [ event for event in events if event.event_type in {EventType.SCHEDULE_REFERENCE, EventType.SCHEDULE_CORRECTION} and event.event_id not in superseded and event.schedule_reference is not None and event.schedule_reference.customer_key == customer_key ] if len(candidates) > 1: raise ValueError("multiple current schedule references require reconciliation") return candidates[0] if candidates else None def append_one(self, event: Event) -> Event: with self._locked() as token: return self._append_one_locked(event, token) def append_many(self, events: Iterator[Event] | tuple[Event, ...] | list[Event]) -> tuple[Event, ...]: with self._locked() as token: return self._append_many_locked(tuple(events), token) def append( self, event: Event, *, intent_id: str | None = None, ) -> Mapping[str, object]: """Append one canonical event and return its paired sequence row.""" candidate = validate_event(event) with self._locked() as token: existing_events, existing_sequences = self._read_pair_locked(token) self._validate_pair(existing_events, existing_sequences) for index, existing in enumerate(existing_events): if existing.event_id != candidate.event_id: continue if existing.model_dump(exclude_none=True) != candidate.model_dump(exclude_none=True): raise ValueError("conflicting canonical event replay") return { "canonical_event": existing.model_dump(mode="json", exclude_none=True), "sequence": existing_sequences[index], } self._append_pair_locked(candidate, token, intent_id=intent_id) _, sequence_rows = self._read_pair_locked(token) return { "canonical_event": candidate.model_dump(mode="json", exclude_none=True), "sequence": sequence_rows[-1], } def authorize_missing_morning_reminder( self, kst_day: date, authorize: object, ) -> object | None: """Linearize one reminder authorization against an accepted morning check-in. ``authorize`` runs while the canonical writer lock is held and must perform only durable local state changes; provider I/O is deliberately outside this method. Its two arguments are the canonical sequence and digest that were observed for the absence decision. """ if type(kst_day) is not date or not callable(authorize): raise TypeError("reminder authorization arguments are invalid") with self._locked() as token: snapshot = self._read_snapshot_locked(token) received = current_terminal_morning_response(snapshot.events, kst_day) is not None if received: return None sequence = len(snapshot.sequence_rows) digest = _sha256_bytes( _canonical_row( { "sequence": sequence, "rows": [dict(row) for row in snapshot.sequence_rows], } ) ) return authorize(sequence, digest) def current_terminal_morning_response(self, kst_day: date) -> Event | None: """Read the accepted terminal morning response for one KST day.""" with self._shared_locked() as token: snapshot = self._read_snapshot_locked(token) return current_terminal_morning_response(snapshot.events, kst_day) def current_schedule_reference(self, customer_key: str) -> Event | None: """Return the sole current schedule source fact for one customer.""" with self._shared_locked() as token: return self._current_schedule_reference_locked(customer_key, token) @staticmethod def schedule_reference_digest(event: Event) -> str: if event.event_type not in {EventType.SCHEDULE_REFERENCE, EventType.SCHEDULE_CORRECTION}: raise ValueError("event is not a schedule reference") encoded = json.dumps(event.model_dump(mode="json", exclude_none=True), ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() def append_schedule_reference(self, event: Event, *, customer_key: str) -> Mapping[str, object]: """Atomically validate and append a reference or exact correction.""" if event.event_type not in {EventType.SCHEDULE_REFERENCE, EventType.SCHEDULE_CORRECTION}: raise ValueError("schedule reference event is required") if event.schedule_reference is None or event.schedule_reference.customer_key != customer_key: raise ValueError("schedule reference customer is invalid") event = validate_event(event) with self._locked() as token: current = self._current_schedule_reference_locked(customer_key, token) if event.event_type is EventType.SCHEDULE_REFERENCE: if current is not None: raise ValueError("schedule reference already exists; submit a correction") elif ( current is None or event.supersedes != current.event_id or event.schedule_reference.predecessor_digest != self.schedule_reference_digest(current) ): raise ValueError("schedule correction does not exactly supersede the current reference") existing_events, existing_sequences = self._read_pair_locked(token) self._validate_pair(existing_events, existing_sequences) for index, existing in enumerate(existing_events): if existing.event_id != event.event_id: continue if existing.model_dump(exclude_none=True) != event.model_dump(exclude_none=True): raise ValueError("conflicting canonical event replay") return { "canonical_event": existing.model_dump(mode="json", exclude_none=True), "sequence": existing_sequences[index], } self._append_pair_locked(event, token, intent_id=f"schedule-reference:{event.dedupe_key}") _, sequence_rows = self._read_pair_locked(token) return { "canonical_event": event.model_dump(mode="json", exclude_none=True), "sequence": sequence_rows[-1], } def append_schedule_confirmation(self, event: Event, *, customer_key: str) -> Mapping[str, object]: """Atomically validate and append a confirmation pinned to the current reference.""" if event.event_type is not EventType.SCHEDULE_CONFIRMATION or event.schedule_confirmation is None: raise ValueError("schedule confirmation event is required") event = validate_event(event) with self._locked() as token: current = self._current_schedule_reference_locked(customer_key, token) if current is None: raise ValueError("schedule reference is not current") payload = event.schedule_confirmation if payload.reference_event_id != current.event_id or payload.reference_digest != self.schedule_reference_digest(current): raise ValueError("schedule confirmation does not pin the current reference") existing_events, existing_sequences = self._read_pair_locked(token) self._validate_pair(existing_events, existing_sequences) for index, existing in enumerate(existing_events): if existing.event_id != event.event_id: continue if existing.model_dump(exclude_none=True) != event.model_dump(exclude_none=True): raise ValueError("conflicting canonical event replay") return { "canonical_event": existing.model_dump(mode="json", exclude_none=True), "sequence": existing_sequences[index], } self._append_pair_locked(event, token, intent_id=f"schedule-confirm:{event.dedupe_key}") _, sequence_rows = self._read_pair_locked(token) return { "canonical_event": event.model_dump(mode="json", exclude_none=True), "sequence": sequence_rows[-1], } def _append_many_locked(self, events: tuple[Event, ...], token: object) -> tuple[Event, ...]: self._require_token(token) candidates = tuple(validate_event(event) for event in events) existing_events, existing_sequences = self._read_pair_locked(token) self._validate_pair(existing_events, existing_sequences) existing_by_dedupe = {event.dedupe_key: event for event in existing_events} result: list[Event] = [] pending: list[Event] = [] seen = set(existing_by_dedupe) for candidate in candidates: duplicate = existing_by_dedupe.get(candidate.dedupe_key) if duplicate is not None: result.append(duplicate) continue if candidate.dedupe_key in seen: raise ValueError("conflicting canonical event replay") seen.add(candidate.dedupe_key) pending.append(candidate) result.append(candidate) for candidate in pending: self._append_pair_locked(candidate, token) return tuple(result) def _append_one_locked(self, event: Event, token: object) -> Event: return self._append_many_locked((event,), token)[0] def _append_pair_locked( self, event: Event, token: object, *, intent_id: str | None = None, ) -> None: self._require_token(token) self._ensure_append_paths() event_bytes = (event.model_dump_json(exclude_none=True) + "\n").encode("utf-8") existing_events, existing_sequences = self._read_pair_locked(token) sequence = len(existing_events) + 1 row_body: dict[str, object] = { "schema_version": "canonical_sequence_v1", "sequence": sequence, "event_id": event.event_id, "event_digest": _sha256_bytes(event_bytes), } if intent_id is not None: row_body["intent_id"] = str(intent_id) row_body["row_digest"] = _sha256_bytes(_canonical_row(row_body)) sequence_bytes = _canonical_row(row_body) + b"\n" with self.events_path.open("ab") as event_handle: event_handle.write(event_bytes) event_handle.flush() os.fsync(event_handle.fileno()) with self.sequence_path.open("ab") as sequence_handle: sequence_handle.write(sequence_bytes) sequence_handle.flush() os.fsync(sequence_handle.fileno()) self._validate_pair( tuple((*existing_events, event)), tuple((*existing_sequences, row_body)), ) def recover(self) -> Mapping[str, object]: with self._locked() as token: return self._recover_locked(token) def _recover_locked(self, token: object) -> Mapping[str, object]: self._require_token(token) event_rows, event_end, event_size = _read_jsonl_tail(self.events_path, Event) sequence_rows, sequence_end, sequence_size = _read_jsonl_tail_mapping(self.sequence_path) for row in sequence_rows: if row.get("schema_version") != "canonical_sequence_v1": raise ValueError("canonical sequence schema mismatch") if sequence_end < sequence_size and not sequence_rows: raise ValueError("canonical sequence tail is not recoverable") if len(event_rows) != len(sequence_rows): raise ValueError("canonical event and sequence counts differ") if event_end < event_size: _truncate_private(self.events_path, event_end) if sequence_end < sequence_size: _truncate_private(self.sequence_path, sequence_end) self._validate_pair(tuple(event_rows), tuple(sequence_rows)) return { "events": len(event_rows), "sequence": len(sequence_rows), "recovered": event_end < event_size or sequence_end < sequence_size, } def migrate_existing_events(self) -> Mapping[str, object]: """Explicitly create a canonical sequence for a complete events-only ledger.""" with self._locked() as token: self._require_token(token) events, event_end, event_size = _read_jsonl_tail(self.events_path, Event) sequences, sequence_end, sequence_size = _read_jsonl_tail_mapping( self.sequence_path ) if event_end != event_size or sequence_end != sequence_size: raise ValueError("canonical migration requires complete ledgers") if sequences: self._validate_pair(tuple(events), tuple(sequences)) return { "events": len(events), "sequence": len(sequences), "migrated": False, } if sequence_size: raise ValueError("canonical sequence migration source is not empty") rows: list[dict[str, object]] = [] for sequence, event in enumerate(events, start=1): event_bytes = ( event.model_dump_json(exclude_none=True) + "\n" ).encode("utf-8") row: dict[str, object] = { "schema_version": "canonical_sequence_v1", "sequence": sequence, "event_id": event.event_id, "event_digest": _sha256_bytes(event_bytes), } row["row_digest"] = _sha256_bytes(_canonical_row(row)) rows.append(row) self.sequence_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) self.sequence_path.parent.chmod(0o700) temp_path = self.sequence_path.with_name( f".{self.sequence_path.name}.migration-{os.getpid()}" ) payload = b"".join(_canonical_row(row) + b"\n" for row in rows) descriptor = os.open( temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, ) try: with os.fdopen(descriptor, "wb", closefd=True) as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, self.sequence_path) directory = os.open( self.sequence_path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, ) try: os.fsync(directory) finally: os.close(directory) finally: if temp_path.exists(): temp_path.unlink() migrated = self._read_snapshot_locked(token) return { "events": len(migrated.events), "sequence": len(migrated.sequence_rows), "migrated": True, } def read_snapshot(self) -> CanonicalEventSnapshot: with self._locked() as token: return self._read_snapshot_locked(token) def _read_snapshot_locked(self, token: object) -> CanonicalEventSnapshot: self._require_token(token) events, sequences = self._read_pair_locked(token) self._validate_pair(events, sequences) return CanonicalEventSnapshot(events, sequences) def _read_pair_locked(self, token: object) -> tuple[tuple[Event, ...], tuple[Mapping[str, object], ...]]: self._require_token(token) events, event_end, event_size = _read_jsonl_tail(self.events_path, Event) sequences, sequence_end, sequence_size = _read_jsonl_tail_mapping(self.sequence_path) if event_end != event_size: raise ValueError("canonical event file has a torn tail") if sequence_end != sequence_size: raise ValueError("canonical sequence file has a torn tail") return tuple(events), tuple(sequences) @staticmethod def _validate_pair( events: tuple[Event, ...], sequences: tuple[Mapping[str, object], ...], ) -> None: if len(events) != len(sequences): raise ValueError("canonical event and sequence counts differ") for expected, (event, row) in enumerate(zip(events, sequences), start=1): if row.get("schema_version") != "canonical_sequence_v1": raise ValueError("canonical sequence schema mismatch") if row.get("sequence") != expected or row.get("event_id") != event.event_id: raise ValueError("canonical event and sequence relation mismatch") event_bytes = (event.model_dump_json(exclude_none=True) + "\n").encode("utf-8") if row.get("event_digest") != _sha256_bytes(event_bytes): raise ValueError("canonical event digest mismatch") body = dict(row) supplied = body.pop("row_digest", None) if supplied != _sha256_bytes(_canonical_row(body)): raise ValueError("canonical sequence row digest mismatch") def _ensure_append_paths(self) -> None: for parent in (self.events_path.parent, self.sequence_path.parent): parent.mkdir(parents=True, exist_ok=True, mode=0o700) parent.chmod(0o700) for path in (self.events_path, self.sequence_path): path.touch(mode=0o600, exist_ok=True) path.chmod(0o600) def _path_has_symlink(path: Path, root: Path) -> bool: try: candidate = path.absolute() resolved_root = root.absolute() if not candidate.is_relative_to(resolved_root): return True except ValueError: return True current = resolved_root for part in candidate.relative_to(resolved_root).parts: current /= part if current.is_symlink(): return True return False _REGISTERED_REGISTRY_LOCATIONS: Final[tuple[tuple[str, ...], ...]] = ( ("registry.json",), ("customers", "registry.json"), ) def _is_canonical_registered_namespace(path: Path, profile_root: Path) -> bool: try: relative = path.resolve().relative_to(profile_root.resolve()) except (OSError, RuntimeError, ValueError): return False return relative.parts[:2] == ("data", "customers") def _registered_roots_for_path(path: Path) -> tuple[Path, ...]: """Read registry provenance without creating or recovering filesystem state.""" lexical = Path(path).absolute() try: resolved = lexical.resolve() except (OSError, RuntimeError) as exc: raise ValueError("persistence root provenance is unavailable") from exc search_roots = tuple(dict.fromkeys(( lexical, *lexical.parents, resolved, *resolved.parents, ))) registry_paths: list[tuple[Path, Path]] = [] seen_registry_paths: set[Path] = set() for profile_root in search_roots: for location in _REGISTERED_REGISTRY_LOCATIONS: registry_path = profile_root.joinpath(*location) try: registry_stat = registry_path.lstat() except FileNotFoundError: continue except OSError as exc: if _is_canonical_registered_namespace(resolved, profile_root): raise ValueError("registered customer registry is unsafe") from exc continue if ( stat.S_ISLNK(registry_stat.st_mode) or not stat.S_ISREG(registry_stat.st_mode) ): if _is_canonical_registered_namespace(resolved, profile_root): raise ValueError("registered customer registry is unsafe") continue identity = registry_path.absolute() if identity in seen_registry_paths: continue seen_registry_paths.add(identity) registry_paths.append((registry_path, profile_root)) registered_roots: set[Path] = set() for registry_path, profile_root in registry_paths: descriptor: int | None = None try: descriptor = os.open( registry_path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, ) opened = os.fstat(descriptor) named = registry_path.lstat() if ( not stat.S_ISREG(opened.st_mode) or opened.st_nlink != 1 or (opened.st_dev, opened.st_ino) != (named.st_dev, named.st_ino) ): raise ValueError("registered customer registry is unsafe") registry = load_customer_registry(registry_path, profile_root) except (CustomerRegistryError, OSError, ValueError) as exc: if _is_canonical_registered_namespace(resolved, profile_root): raise ValueError("registered customer registry is invalid") from exc continue finally: if descriptor is not None: os.close(descriptor) for runtime in registry.customers: registered_roots.update(( runtime.customer_root, runtime.wizard_root, runtime.nutrition_plans_root, )) return tuple(sorted(registered_roots, key=str)) def _reject_registered_persistence_root(path: Path, *, label: str) -> None: """Reject a path inside registry-owned persistence before any mutation.""" candidate = Path(path) try: resolved = candidate.resolve() except (OSError, RuntimeError) as exc: raise ValueError(f"{label} root provenance is unavailable") from exc registered_roots = _registered_roots_for_path(resolved) if any( resolved == root or resolved.is_relative_to(root) for root in registered_roots ): raise ValueError(f"{label} cannot use a registered persistence root") def _read_jsonl_tail(path: Path, model: object) -> tuple[list[object], int, int]: if not path.exists(): return [], 0, 0 raw = path.read_bytes() rows: list[object] = [] offset = 0 complete_end = 0 for line in raw.splitlines(keepends=True): offset += len(line) if not line.endswith(b"\n"): break payload = line[:-1].removesuffix(b"\r") if not payload: continue rows.append(model.model_validate_json(payload)) complete_end = offset return rows, complete_end, len(raw) def _read_jsonl_tail_mapping(path: Path) -> tuple[list[Mapping[str, object]], int, int]: if not path.exists(): return [], 0, 0 raw = path.read_bytes() rows: list[Mapping[str, object]] = [] offset = 0 complete_end = 0 for line in raw.splitlines(keepends=True): offset += len(line) if not line.endswith(b"\n"): break payload = line[:-1].removesuffix(b"\r") if not payload: continue value = json.loads(payload) if not isinstance(value, Mapping): raise ValueError("canonical sequence row is not an object") rows.append(dict(value)) complete_end = offset return rows, complete_end, len(raw) def _truncate_private(path: Path, size: int) -> None: if not path.exists(): return with path.open("r+b") as handle: handle.truncate(size) handle.flush() os.fsync(handle.fileno()) path.chmod(0o600) WEIGHT: Final[re.Pattern[str]] = re.compile(r"(?:체중|아공체)\s*[:：]?\s*([^\s]+)", re.IGNORECASE) CALORIES: Final[re.Pattern[str]] = re.compile(r"(?:칼로리|총\s*섭취)\s*[:：]?\s*([^\s]+)", re.IGNORECASE) SLEEP: Final[re.Pattern[str]] = re.compile(r"수면\s*[:：]?\s*([^\s]+)", re.IGNORECASE) WORKOUT: Final[re.Pattern[str]] = re.compile(r"운동\s*[:：]?\s*([^\n]+)", re.IGNORECASE) CHEST_PAIN: Final[re.Pattern[str]] = re.compile(r"가슴\s*통증|흉통|chest\s*pain", re.IGNORECASE) class EventStore: """Append contract-conformant events and rebuild derived views only.""" def __init__( self, home: Path, *, canonical_transaction: CanonicalEventTransaction | None = None, binding: RegisteredCustomerBinding | None = None, _construction_token: object | None = None, ) -> None: if _construction_token is not _STORE_CONSTRUCTION_TOKEN: raise TypeError("EventStore requires an explicit persistence factory") if canonical_transaction is not None: if not isinstance(canonical_transaction, CanonicalEventTransaction): raise TypeError("canonical transaction type is invalid") if not isinstance(binding, RegisteredCustomerBinding): raise ValueError("registered EventStore requires a sealed binding") expected_root_digest = hashlib.sha256( str(canonical_transaction.events_path.parent.parent.resolve()).encode("utf-8") ).hexdigest() if binding.data_root_digest != expected_root_digest: raise ValueError("registered EventStore binding does not match its transaction") elif binding is not None: raise ValueError("standalone EventStore cannot carry a registered binding") self._home = Path(home) self._events = ( canonical_transaction.events_path if canonical_transaction is not None else self._home / "events.jsonl" ) self._raw_payloads = self._home / "raw-payloads" self._imports = self._home / "imports" self._views = self._home / "views" self._canonical_transaction = canonical_transaction self._registered_binding = binding self._active_transaction_token: object | None = None @classmethod def for_standalone(cls, home: Path) -> EventStore: if isinstance(home, CustomerRuntime) or isinstance(home, RegisteredCustomerBinding): raise TypeError("standalone EventStore cannot accept a registered runtime") candidate = Path(home) _reject_registered_persistence_root(candidate, label="standalone EventStore") return cls(candidate, _construction_token=_STORE_CONSTRUCTION_TOKEN) @classmethod def for_registered( cls, transaction_or_runtime: CanonicalEventTransaction | CustomerRuntime, binding: RegisteredCustomerBinding | None = None, ) -> EventStore: if isinstance(transaction_or_runtime, CustomerRuntime): if binding is not None: raise ValueError("runtime-bound EventStore cannot accept a separate binding") runtime = transaction_or_runtime transaction = CanonicalEventTransaction.for_customer_runtime(runtime) binding = runtime.binding else: transaction = transaction_or_runtime if not isinstance(transaction, CanonicalEventTransaction): raise TypeError("registered EventStore requires CanonicalEventTransaction") if not isinstance(binding, RegisteredCustomerBinding): raise ValueError("registered EventStore requires a sealed binding") return cls( transaction.events_path.parent, canonical_transaction=transaction, binding=binding, _construction_token=_STORE_CONSTRUCTION_TOKEN, ) def record(self, request: RecordRequest) -> RecordResult: """Archive a new input, validate references, append, then rebuild.""" with self._record_lock(): events = self._read_events() dedupe_key = self._digest(request.message_id) duplicate = next((event for event in events if event.dedupe_key == dedupe_key), None) if duplicate is not None: return RecordResult("duplicate", duplicate.event_id) event_id = f"checkin_{self._digest(request.message_id + request.received_at)[:24]}" if request.supersedes is not None and not self.supersedes_is_valid(event_id, request.supersedes): return RecordResult("needs_clarification", None) payload_ref, payload_hash = self._write_raw_payload(request) flags = (CHEST_PAIN_FLAG,) if CHEST_PAIN.search(request.text) else () values, missing = self._parse_values(request.text) event_type, status, safety = self._disposition(request.supersedes, flags, values) event = Event( event_id=event_id, event_type=event_type, occurred_at_kst=request.received_at, recorded_at_kst=request.received_at, provenance=Provenance( source_type="telegram", source_ref=request.message_id, content_sha256=payload_hash, received_message_id=request.message_id, ), status=status, supersedes=request.supersedes, dedupe_key=dedupe_key, check_in=self._contract_checkin(values), safety=safety, payload_ref=payload_ref, ) self._append(event) self.rebuild() outcome = "urgent_safety" if flags else "recorded" if values is not None else "needs_clarification" return RecordResult(outcome, event_id, flags) def append_wizard_event(self, event: Event) -> RecordResult: """Append one explicit final wizard event exactly once, then rebuild views.""" with self._record_lock(): return self._append_wizard_event_locked(event) def load_wizard_event(self, event_id: str) -> Event: """Reload one exact persisted canonical wizard event.""" if not isinstance(event_id, str) or not event_id: raise TypeError("canonical event ID is required") with self._record_lock(): event = next((item for item in self._read_events() if item.event_id == event_id), None) if event is None: raise ValueError("canonical event is missing") return event def _append_wizard_event_locked(self, event: Event) -> RecordResult: candidate = validate_event(event) events = self._read_events() duplicate = next( (item for item in events if item.dedupe_key == candidate.dedupe_key), None, ) if duplicate is not None: return RecordResult("duplicate", duplicate.event_id) if not self._operator_time_correction_is_valid(candidate, events): return RecordResult("needs_clarification", None) self._append(candidate) self.rebuild() return RecordResult("recorded", candidate.event_id) def record_payment( self, customer_key: str, *, amount_krw: int = 150000, paid_on: date | str, period_start_on: date | str, period_end_on: date | str, method: PaymentMethod | str = PaymentMethod.BANK_TRANSFER, kind: PaymentKind | str, plan: TwelveWeekPlan | None = None, event_id: str | None = None, occurred_at_kst: str | None = None, recorded_at_kst: str | None = None, supersedes: str | None = None, ) -> RecordResult: """Append one canonical, customer-scoped payment fact.""" event = build_payment_event( customer_key, amount_krw=amount_krw, paid_on=paid_on, period_start_on=period_start_on, period_end_on=period_end_on, method=method, kind=kind, event_id=event_id, occurred_at_kst=occurred_at_kst, recorded_at_kst=recorded_at_kst, supersedes=supersedes, ) self._validate_initial_payment_period(event, plan) return self._append_customer_record(customer_key, event) def record_satisfaction( self, customer_key: str, *, score_1to10: int | None = None, collected_on: date | str | None = None, note: str | None = None, score: int | None = None, collection_date: date | str | None = None, iso_week: str | None = None, event_id: str | None = None, occurred_at_kst: str | None = None, recorded_at_kst: str | None = None, supersedes: str | None = None, ) -> RecordResult: """Append one canonical, customer-scoped satisfaction fact.""" event = build_satisfaction_event( customer_key, score_1to10=score_1to10, collected_on=collected_on, note=note, score=score, collection_date=collection_date, iso_week=iso_week, event_id=event_id, occurred_at_kst=occurred_at_kst, recorded_at_kst=recorded_at_kst, supersedes=supersedes, ) return self._append_customer_record(customer_key, event) def record_operator_time( self, customer_key: str, *, entry_id: str, attempt_id: str, minutes: int, task: OperatorTask | str, work_date: date | str | None = None, work_on: date | str | None = None, supersedes_entry_id: str | None = None, event_id: str | None = None, occurred_at_kst: str | None = None, recorded_at_kst: str | None = None, supersedes: str | None = None, ) -> RecordResult: """Append one canonical, customer-scoped operator-time fact.""" event = build_operator_time_event( customer_key, entry_id=entry_id, attempt_id=attempt_id, minutes=minutes, task=task, work_date=work_date, work_on=work_on, supersedes_entry_id=supersedes_entry_id, event_id=event_id, occurred_at_kst=occurred_at_kst, recorded_at_kst=recorded_at_kst, supersedes=supersedes, ) return self._append_customer_record(customer_key, event) def _append_customer_record(self, customer_key: str, event: Event) -> RecordResult: expected_source = f"pilot:{customer_key}:{event.event_type.value}" if event.provenance.source_ref != expected_source: raise ValueError("event provenance is not scoped to customer") with self._record_lock(): return self._append_wizard_event_locked(event) @staticmethod def _validate_initial_payment_period(event: Event, plan: TwelveWeekPlan | None) -> None: if event.event_type is not EventType.PAYMENT_RECORD or event.payment is None: return canonical_plan = getattr(plan, "plan", plan) starts_on = getattr(canonical_plan, "starts_on", None) if not isinstance(starts_on, date): raise ValueError(f"{event.payment.kind.value} payment requires the canonical customer plan") if event.payment.kind is PaymentKind.INITIAL: expected_start = starts_on expected_end = starts_on + timedelta(days=27) else: expected_start = starts_on + timedelta(days=28) expected_end = starts_on + timedelta(days=55) if event.payment.period_start_on != expected_start or event.payment.period_end_on != expected_end: raise ValueError(f"{event.payment.kind.value} payment period must match the canonical customer plan") @staticmethod def _operator_time_correction_is_valid(candidate: Event, events: tuple[Event, ...]) -> bool: if candidate.event_type is not EventType.OPERATOR_TIME_RECORD or candidate.operator_time is None: return True target_id = candidate.operator_time.supersedes_entry_id if target_id is None: return True if candidate.status is not ContractStatus.ACCEPTED: return False if target_id == candidate.operator_time.entry_id: return False candidate_scope = EventStore._event_customer_scope(candidate) target = next( ( event for event in events if event.event_type is EventType.OPERATOR_TIME_RECORD and event.status is ContractStatus.ACCEPTED and event.operator_time is not None and event.operator_time.entry_id == target_id and EventStore._event_customer_scope(event) == candidate_scope ), None, ) if target is None or target.operator_time is None: return False superseded_event_ids = {event.supersedes for event in events if event.supersedes is not None} superseded_entry_ids = { event.operator_time.supersedes_entry_id for event in events if event.operator_time is not None and event.operator_time.supersedes_entry_id is not None } if target.event_id in superseded_event_ids or target.operator_time.entry_id in superseded_entry_ids: return False try: target_recorded = datetime.fromisoformat(target.recorded_at_kst) replacement_recorded = datetime.fromisoformat(candidate.recorded_at_kst) except (TypeError, ValueError): return False try: return target_recorded < replacement_recorded except TypeError: return False @staticmethod def _event_customer_scope(event: Event) -> str: source_ref = event.provenance.source_ref parts = source_ref.split(":") if len(parts) >= 2 and parts[0] in {"pilot", "customer"} and parts[1]: return parts[1] return source_ref def import_history(self, source: Path, range_label: str) -> int: """Append hash-and-anchor provenance without copying history content.""" source_sha256 = hashlib.sha256(source.read_bytes()).hexdigest() with self._record_lock(): existing = {event.event_id for event in self._read_events()} imports = tuple( event for event in legacy_history_events(source, source_sha256, self._now_kst()) if event.event_id not in existing ) if self._canonical_transaction is not None: token = self._active_transaction_token if token is None: raise RuntimeError("registered history import requires its transaction lock") self._canonical_transaction._append_many_locked(imports, token) else: for event in imports: self._append(event) manifest = legacy_manifest(source, source_sha256, range_label) self._write_private(self._imports / "history-import-manifest.json", manifest) self._write_private(self._imports / f"history-import-{source_sha256[:16]}.json", manifest) self.rebuild() return len(imports) def import_baseline(self, manifest_path: Path) -> int: """Append one dated, partial event per verified historical day atomically.""" with self._record_lock(): baseline = parse_baseline(manifest_path) existing = {event.event_id for event in self._read_events()} imports = tuple( event for event in baseline_events(baseline, self._now_kst(), schema_version="2.0") if event.event_id not in existing ) if self._canonical_transaction is not None: token = self._active_transaction_token if token is None: raise RuntimeError("registered baseline import requires its transaction lock") self._canonical_transaction._append_many_locked(imports, token) else: for event in imports: self._append(event) events = self._read_events() self._write_private(self._imports / "historical-baseline.json", baseline_report(baseline)) self.rebuild() self._write_private(self._views / "historical-baseline.json", historical_view(events, baseline)) return len(imports) def supersedes_is_valid(self, candidate_id: str, target_id: str) -> bool: """Require a known target and prevent self or graph-cycle references.""" if candidate_id == target_id: return False links = {event.event_id: event.supersedes for event in self._read_events()} if target_id not in links: return False cursor: str | None = target_id while cursor is not None: if cursor == candidate_id: return False cursor = links.get(cursor) return True def rebuild(self) -> None: """Recreate current and weekly views solely from immutable events.""" events = self._read_events() superseded = {event.supersedes for event in events if event.supersedes is not None} eligible: list[CheckinValues] = [] flags: list[str] = [] clarification_count = 0 unsafe_count = 0 for event in events: match event.status: case ContractStatus.NEEDS_CLARIFICATION: clarification_count += 1 case ContractStatus.UNSAFE: unsafe_count += 1 if event.safety is not None: flags.extend(event.safety.signals) case ContractStatus.ACCEPTED: if event.event_type in (EventType.CHECK_IN_VALIDATED, EventType.CORRECTION, EventType.MORNING_CHECKIN, EventType.NUTRITION_CHECKIN) and event.event_id not in superseded and event.check_in is not None and event.check_in.body_weight_kg is not None and event.check_in.calories_kcal is not None: eligible.append( CheckinValues( weight_kg=event.check_in.body_weight_kg, calories_kcal=event.check_in.calories_kcal, sleep_hours=event.check_in.sleep_hours, workout=event.check_in.training_summary, ) ) case ContractStatus.ARCHIVED: continue case unreachable: assert_never(unreachable) self._write_private(self._views / "current.json", CurrentView(eligible_checkins=tuple(eligible), safety_flags=tuple(flags)).model_dump_json(indent=2) + "\n") self._write_private(self._views / "weekly.json", WeeklyView(eligible_count=len(eligible), clarification_count=clarification_count, urgent_safety_count=unsafe_count).model_dump_json(indent=2) + "\n") def _parse_values(self, text: str) -> tuple[CheckinValues | None, tuple[str, ...]]: weight, calories = self._capture(WEIGHT, text), self._capture(CALORIES, text) missing = tuple(name for name, value in (("weight_kg", weight), ("calories_kcal", calories)) if value is None) if missing: return None, missing try: return CheckinValues(weight_kg=weight, calories_kcal=calories, sleep_hours=self._capture(SLEEP, text), workout=self._capture(WORKOUT, text)), () except ValidationError: return None, ("invalid_measurement",) @staticmethod def _disposition(supersedes: str | None, flags: tuple[str, ...], values: CheckinValues | None) -> tuple[EventType, ContractStatus, Safety | None]: if flags: return EventType.SAFETY_FLAG, ContractStatus.UNSAFE, Safety(level="stop_and_escalate", signals=flags, coaching_held=True) if values is None: return EventType.CHECK_IN_RECEIVED, ContractStatus.NEEDS_CLARIFICATION, None if supersedes is not None: return EventType.CORRECTION, ContractStatus.ACCEPTED, None return EventType.CHECK_IN_VALIDATED, ContractStatus.ACCEPTED, None @staticmethod def _contract_checkin(values: CheckinValues | None) -> ContractCheckin | None: if values is None: return None return ContractCheckin(body_weight_kg=values.weight_kg, calories_kcal=values.calories_kcal, sleep_hours=values.sleep_hours, training_summary=values.workout) @staticmethod def _capture(pattern: re.Pattern[str], text: str) -> str | None: match = pattern.search(text) return match.group(1).removesuffix("kg").removesuffix("kcal").removesuffix("시간") if match else None def _read_events(self) -> tuple[Event, ...]: if not self._events.exists(): return () return tuple(Event.model_validate_json(line) for line in self._events.read_text(encoding="utf-8").splitlines() if line) def _append(self, event: Event) -> None: """Validate against the model and canonical JSON schema before writing.""" candidate = validate_event(event) if self._canonical_transaction is not None: token = self._active_transaction_token if token is None: raise RuntimeError("registered event append requires its transaction lock") self._canonical_transaction._append_one_locked(candidate, token) return self._events.parent.mkdir(parents=True, exist_ok=True, mode=0o700) self._events.touch(mode=0o600, exist_ok=True) self._events.chmod(0o600) with self._events.open("a", encoding="utf-8") as handle: handle.write(candidate.model_dump_json(exclude_none=True) + "\n") handle.flush() os.fsync(handle.fileno()) @contextmanager def _record_lock(self) -> Iterator[None]: """Serialize event deduplication and append across local processes.""" if self._canonical_transaction is not None: with self._canonical_transaction._locked() as token: self._active_transaction_token = token try: yield finally: self._active_transaction_token = None return lock_path = self._home / ".events.lock" lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with lock_path.open("a", encoding="utf-8") as handle: lock_path.chmod(0o600) fcntl.flock(handle.fileno(), fcntl.LOCK_EX) try: yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _write_raw_payload(self, request: RecordRequest) -> tuple[str, str]: payload_hash = self._digest(request.text) payload_ref = str(self._raw_payloads / f"{self._digest(request.message_id)}.json") self._write_private(Path(payload_ref), json.dumps({"message_id": request.message_id, "received_at": request.received_at, "text": request.text}, ensure_ascii=False, sort_keys=True) + "\n") return payload_ref, payload_hash @staticmethod def _digest(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() @staticmethod def _now_kst() -> str: return datetime.now(KST).isoformat() @staticmethod def _write_private(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) path.parent.chmod(0o700) path.write_text(content, encoding="utf-8") path.chmod(0o600)