"""Deterministic KST schedule projection and crash-safe delivery reservations.""" from __future__ import annotations import fcntl import hashlib import json import os import re import uuid import stat from contextvars import ContextVar from contextlib import contextmanager from collections.abc import Generator, Iterator, Mapping from dataclasses import dataclass from datetime import date, datetime, timezone from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from checkin_cli.customer_coaching import CustomerRegistry from checkin_cli.weekly_operations_schedule_host_authority_r4 import ( schedule_authority_validator, schedule_descriptor_identity, schedule_provider_admission, ) from checkin_cli.weekly_operations_schedule_host_models_r4 import ( ApprovedReminderScheduleEvidence, CustomerScheduleError, CustomerScheduleTask, ScheduledDeliveryReceipt, WeeklyOperationsSchedulePolicy, WeeklyReminderReservationAuthority, ) from checkin_cli.weekly_operations_schedule_host_projection_r4 import ( build_due_customer_tasks as project_due_customer_tasks, ) from checkin_cli.weekly_operations_schedule_host_rows_r4 import ( receipt_from_row, weekly_authority_fields, weekly_authority_from_receipt, ) KST = ZoneInfo("Asia/Seoul") _SCHEDULE_SCHEMA_VERSION = 1 _SCHEDULE_SCHEMA_NAME = "scheduled_deliveries" _SCHEDULE_LEDGER_NAME = "scheduled-deliveries.jsonl" _SCHEDULE_FENCE_NAME = "scheduled-deliveries-fence.json" _SCHEDULE_LOCK_NAME = ".scheduled-deliveries.lock" _SCHEDULE_CLAIMS_NAME = "customer-schedule-claims" _SCHEDULE_FENCE_STATES = frozenset({"preparing", "ready", "recovery_required"}) _SCHEDULE_STATES = frozenset( { "prepared", "sending", "delivered", "unknown", "known_failure", "sent_audited", "abandoned", } ) _SCHEDULE_TERMINAL_STATES = frozenset( {"unknown", "known_failure", "sent_audited", "abandoned"} ) _SCHEDULE_ZERO_DIGEST = "0" * 64 _TERMINAL_MORNING_RESPONSE_BEFORE_PROVIDER = "terminal_morning_response_before_provider" _SCHEDULE_ATTEMPT_LOCK_PREFIX = ".scheduled-delivery-attempt-" _SCHEDULE_INITIAL_UNKNOWN_REASONS = frozenset( {"legacy_claim_unknown", "tombstone_ledger_missing_recovered"} ) _SCHEDULE_PROVIDER_LEASES: dict[tuple[str, str], Any] = {} _SCHEDULE_READ_MAX_INVENTORY = 4096 _SCHEDULE_READ_LOCK_STATE: ContextVar[ tuple[Path, Path, Path] | None ] = ContextVar("schedule_read_lock_state", default=None) _VALID_CUSTOMER_KEY = re.compile(r"[a-z0-9][a-z0-9_-]{2,63}") _VALID_TASK_KINDS = frozenset({"daily", "weekly", "monthly", "reminder", "cutoff"}) _VALID_DIGEST = re.compile(r"[0-9a-f]{64}") _VALID_OPAQUE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,159}") _STATIC_OPERATIONAL_REMINDER_TEMPLATES = { "missing-checkin-v1": "체크인이 확인되지 않았습니다. 오늘 아침 체크인을 제출해 주세요.", } @dataclass(frozen=True, slots=True) class ScheduleDeliveryReadSnapshot: """Bounded, deterministic inventory of a validated schedule tree.""" schema_version: str fence_epoch: str | None row_count: int tombstone_count: int ledger_digest: str tombstone_inventory_digest: str fence_digest: str revision_token: str def to_dict(self) -> dict[str, object]: return { "schema_version": self.schema_version, "fence_epoch": self.fence_epoch, "row_count": self.row_count, "tombstone_count": self.tombstone_count, "ledger_digest": self.ledger_digest, "tombstone_inventory_digest": self.tombstone_inventory_digest, "fence_digest": self.fence_digest, "revision_token": self.revision_token, } @dataclass(frozen=True, slots=True) class ScheduleFenceReceipt: state: str schema_version: int ledger_schema_version: int def _validate_task(task: CustomerScheduleTask) -> None: if ( not isinstance(task.customer_key, str) or _VALID_CUSTOMER_KEY.fullmatch(task.customer_key) is None ): raise CustomerScheduleError("invalid customer key") if not isinstance(task.kind, str) or task.kind not in _VALID_TASK_KINDS: raise CustomerScheduleError("invalid task kind") if type(task.kst_day) is not date: raise CustomerScheduleError("invalid task date") def build_due_customer_tasks( registry: CustomerRegistry, now: datetime, *, missing_morning_checkins: Mapping[str, date] | None = None, reminder_evidence: ApprovedReminderScheduleEvidence | None = None, weekly_operations_policy: WeeklyOperationsSchedulePolicy | None = None, ) -> tuple[CustomerScheduleTask, ...]: """Return enabled due work, including only evidenced missing-check-in reminders.""" if reminder_evidence is not None: _require_digest(reminder_evidence.policy_digest, "reminder policy") _require_digest(reminder_evidence.config_digest, "reminder config") return project_due_customer_tasks( registry, now, missing_morning_checkins=missing_morning_checkins, reminder_evidence_present=reminder_evidence is not None, weekly_operations_policy=weekly_operations_policy, ) def _canonical_json(value: object) -> str: if hasattr(value, "model_dump"): try: value = value.model_dump(mode="json") # type: ignore[union-attr] except (TypeError, ValueError): raise CustomerScheduleError("schedule pin is not canonical JSON") try: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ) except (TypeError, ValueError) as exc: raise CustomerScheduleError("schedule pin is not canonical JSON") from exc def _digest(value: object) -> str: return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() def _bytes_digest(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _require_digest(value: object, label: str, *, default: str | None = None) -> str: if value is None and default is not None: return default if not isinstance(value, str) or _VALID_DIGEST.fullmatch(value) is None: raise CustomerScheduleError(f"{label} digest is invalid") return value def _require_opaque(value: object, label: str, *, optional: bool = False) -> str | None: if value is None and optional: return None if isinstance(value, int) and not isinstance(value, bool): value = str(value) if not isinstance(value, str) or _VALID_OPAQUE.fullmatch(value) is None: raise CustomerScheduleError(f"{label} is invalid") return value def _schedule_key(task: CustomerScheduleTask) -> str: return f"{task.customer_key}:{task.kst_day.isoformat()}:{task.kind}" def _data_root(profile_root: Path, *, create: bool = True) -> Path: root = Path(profile_root) if root.is_symlink(): admitted = schedule_descriptor_identity() try: observed = root.stat() except OSError as exc: raise CustomerScheduleError("profile root symlink is unavailable") from exc if ( admitted is None or not str(root).startswith("/proc/self/fd/") or admitted != (observed.st_dev, observed.st_ino) ): raise CustomerScheduleError("profile root symlinks are not allowed") if not root.exists(): if not create: raise CustomerScheduleError("profile root is unavailable") try: root.mkdir(parents=True, mode=0o700) except OSError as exc: raise CustomerScheduleError("profile root is unavailable") from exc if not root.is_dir(): raise CustomerScheduleError("profile root must be a directory") root = root.resolve() if root.is_symlink(): raise CustomerScheduleError("profile root symlinks are not allowed") data = root / "data" if data.is_symlink(): raise CustomerScheduleError("schedule data symlinks are not allowed") if create: try: data.mkdir(parents=True, exist_ok=True, mode=0o700) data.chmod(0o700) except OSError as exc: raise CustomerScheduleError("schedule data root is unavailable") from exc if not data.exists() or not data.is_dir(): raise CustomerScheduleError("schedule data root is unavailable") return data def _schedule_paths(profile_root: Path, *, create: bool = True) -> tuple[Path, Path, Path, Path]: data = _data_root(profile_root, create=create) return ( data / _SCHEDULE_LEDGER_NAME, data / _SCHEDULE_FENCE_NAME, data / _SCHEDULE_LOCK_NAME, data / _SCHEDULE_CLAIMS_NAME, ) def _ensure_regular(path: Path, label: str) -> None: if path.is_symlink(): raise CustomerScheduleError(f"{label} symlinks are not allowed") if path.exists() and not path.is_file(): raise CustomerScheduleError(f"{label} must be a regular file") def _read_private_regular_bytes( path: Path, label: str, *, allow_missing: bool = False, max_bytes: int = 8 * 1024 * 1024, ) -> bytes | None: """Read an existing private regular file without changing filesystem state.""" _ensure_regular(path, label) if not path.exists(): if allow_missing: return None raise CustomerScheduleError(f"{label} is unavailable") descriptor: int | None = None try: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) descriptor = os.open(path, flags) before = os.fstat(descriptor) named = os.stat(path, follow_symlinks=False) if ( not stat.S_ISREG(before.st_mode) or before.st_uid != os.geteuid() or before.st_nlink != 1 or stat.S_IMODE(before.st_mode) != 0o600 or (before.st_dev, before.st_ino) != (named.st_dev, named.st_ino) or before.st_size > max_bytes ): raise CustomerScheduleError(f"{label} is unsafe") chunks: list[bytes] = [] remaining = before.st_size while remaining: chunk = os.read(descriptor, remaining) if not chunk: break chunks.append(chunk) remaining -= len(chunk) after = os.fstat(descriptor) if ( (after.st_dev, after.st_ino, after.st_size) != (before.st_dev, before.st_ino, before.st_size) ): raise CustomerScheduleError(f"{label} changed during read") return b"".join(chunks) except CustomerScheduleError: raise except (FileNotFoundError, OSError) as exc: raise CustomerScheduleError(f"{label} is unavailable") from exc finally: if descriptor is not None: os.close(descriptor) def _schedule_attempt_lock_path(profile_root: Path, reservation_id: str) -> Path: data = _data_root(profile_root) digest = hashlib.sha256(reservation_id.encode("utf-8")).hexdigest() path = data / f"{_SCHEDULE_ATTEMPT_LOCK_PREFIX}{digest}.lock" _ensure_regular(path, "scheduled-delivery attempt lock") return path def _lease_key(profile_root: Path, reservation_id: str) -> tuple[str, str]: return (str(_data_root(profile_root).resolve()), reservation_id) def _acquire_provider_lease(profile_root: Path, reservation_id: str) -> Any | None: key = _lease_key(profile_root, reservation_id) if key in _SCHEDULE_PROVIDER_LEASES: return None path = _schedule_attempt_lock_path(profile_root, reservation_id) handle: Any | None = None try: descriptor = os.open( path, os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600, ) handle = os.fdopen(descriptor, "a+b") path.chmod(0o600) fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: if handle is not None: handle.close() return None except OSError as exc: if handle is not None: handle.close() raise CustomerScheduleError( "scheduled-delivery provider lease is unavailable" ) from exc return handle def _release_provider_lease(profile_root: Path, reservation_id: str) -> None: handle = _SCHEDULE_PROVIDER_LEASES.pop(_lease_key(profile_root, reservation_id), None) if handle is None: return try: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except (OSError, ValueError): pass try: handle.close() except (OSError, ValueError): pass def _provider_lease_active(profile_root: Path, reservation_id: str) -> bool: key = _lease_key(profile_root, reservation_id) if key in _SCHEDULE_PROVIDER_LEASES: return True handle = _acquire_provider_lease(profile_root, reservation_id) if handle is None: return True try: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except (OSError, ValueError): pass try: handle.close() except (OSError, ValueError): pass return False def _fsync_directory(path: Path) -> None: try: descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) except OSError as exc: raise CustomerScheduleError("schedule directory could not be synchronized") from exc try: os.fsync(descriptor) except OSError as exc: raise CustomerScheduleError("schedule directory could not be synchronized") from exc finally: os.close(descriptor) def _atomic_json(path: Path, value: Mapping[str, object]) -> None: _ensure_regular(path, "schedule fence") parent = path.parent if parent.is_symlink(): raise CustomerScheduleError("schedule directory symlinks are not allowed") parent.mkdir(parents=True, exist_ok=True, mode=0o700) parent.chmod(0o700) temporary = parent / f".{path.name}.{uuid.uuid4().hex}.tmp" content = (_canonical_json(dict(value)) + "\n").encode("utf-8") try: with temporary.open("wb") as handle: temporary.chmod(0o600) handle.write(content) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) path.chmod(0o600) _fsync_directory(parent) except OSError as exc: try: temporary.unlink() except OSError: pass raise CustomerScheduleError("schedule fence could not be persisted") from exc def _write_fence(path: Path, state: str) -> ScheduleFenceReceipt: if state not in _SCHEDULE_FENCE_STATES: raise CustomerScheduleError("schedule startup fence state is invalid") payload = { "schema_version": _SCHEDULE_SCHEMA_VERSION, "ledger_schema_version": _SCHEDULE_SCHEMA_VERSION, "state": state, "updated_at": datetime.now(timezone.utc).isoformat(), } _atomic_json(path, payload) return ScheduleFenceReceipt(state, _SCHEDULE_SCHEMA_VERSION, _SCHEDULE_SCHEMA_VERSION) def _read_fence_document( path: Path, ) -> tuple[ScheduleFenceReceipt, dict[str, object]] | None: raw = _read_private_regular_bytes( path, "schedule fence", allow_missing=True, ) if raw is None: return None try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CustomerScheduleError("schedule startup fence is corrupt") from exc if ( not isinstance(payload, dict) or set(payload) != { "schema_version", "ledger_schema_version", "state", "updated_at", } or type(payload.get("schema_version")) is not int or type(payload.get("ledger_schema_version")) is not int or payload.get("schema_version") != _SCHEDULE_SCHEMA_VERSION or payload.get("ledger_schema_version") != _SCHEDULE_SCHEMA_VERSION or payload.get("state") not in _SCHEDULE_FENCE_STATES or not isinstance(payload.get("updated_at"), str) ): raise CustomerScheduleError("schedule startup fence is unsupported") try: datetime.fromisoformat(str(payload["updated_at"])) except ValueError as exc: raise CustomerScheduleError("schedule startup fence timestamp is invalid") from exc return ( ScheduleFenceReceipt( str(payload["state"]), int(payload["schema_version"]), int(payload["ledger_schema_version"]), ), payload, ) def _read_fence(path: Path) -> ScheduleFenceReceipt | None: document = _read_fence_document(path) return None if document is None else document[0] @contextmanager def _schedule_lock( profile_root: Path, ) -> Generator[tuple[Path, Path, Path], None, None]: ledger, fence, lock_path, claims_root = _schedule_paths(profile_root) try: flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC descriptor = os.open(lock_path, flags, 0o600) owns_lock = False try: opened = os.fstat(descriptor) named = 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 CustomerScheduleError("schedule lock is unsafe") admission = schedule_provider_admission() owns_lock = admission is None if admission is None: fcntl.flock(descriptor, fcntl.LOCK_EX) elif admission[0] != identity: raise CustomerScheduleError("weekly provider admission lock disagrees") if identity != (lock_path.lstat().st_dev, lock_path.lstat().st_ino): raise CustomerScheduleError("schedule lock was replaced") validator = admission[1] if admission is not None else schedule_authority_validator() if validator is not None: validator() yield ledger, fence, claims_root finally: if owns_lock: fcntl.flock(descriptor, fcntl.LOCK_UN) os.close(descriptor) except CustomerScheduleError: raise except OSError as exc: raise CustomerScheduleError("schedule lock is unavailable") from exc def _read_jsonl(path: Path) -> list[dict[str, object]]: raw = _read_private_regular_bytes( path, "scheduled-deliveries ledger", allow_missing=True, ) if raw is None: return [] rows: list[dict[str, object]] = [] offset = 0 for index, line in enumerate(raw.splitlines(keepends=True), start=1): if not line.endswith(b"\n"): raise CustomerScheduleError( f"scheduled-deliveries ledger has a torn tail at row {index}" ) value = line[:-1] if value.endswith(b"\r"): value = value[:-1] if not value: raise CustomerScheduleError("scheduled-deliveries ledger contains a blank row") try: parsed = json.loads(value.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CustomerScheduleError( f"scheduled-deliveries ledger is corrupt at row {index}" ) from exc if not isinstance(parsed, dict): raise CustomerScheduleError("scheduled-deliveries ledger row is invalid") rows.append(parsed) offset += len(line) if offset != len(raw): raise CustomerScheduleError("scheduled-deliveries ledger has a torn tail") return rows def _claim_path(claims_root: Path, task: CustomerScheduleTask) -> Path: _validate_task(task) customer = claims_root / task.customer_key day = customer / task.kst_day.isoformat() for candidate in (customer, day): if candidate.is_symlink(): raise CustomerScheduleError("schedule claim path symlinks are not allowed") return day / f"{task.kind}.claim" def _ensure_schedule_directory(path: Path, label: str) -> None: if path.is_symlink() or not path.exists() or not path.is_dir(): raise CustomerScheduleError(f"{label} must be a directory") try: metadata = path.stat() except OSError as exc: raise CustomerScheduleError(f"{label} is unavailable") from exc if metadata.st_uid != os.geteuid() or stat.S_IMODE(metadata.st_mode) & 0o002: raise CustomerScheduleError(f"{label} is unsafe") def _iter_claims(claims_root: Path) -> tuple[tuple[CustomerScheduleTask, Path], ...]: if not claims_root.exists(): return () _ensure_schedule_directory(claims_root, "schedule claims root") result: list[tuple[CustomerScheduleTask, Path]] = [] try: customers = tuple(claims_root.iterdir()) except OSError as exc: raise CustomerScheduleError("schedule claims could not be inventoried") from exc for customer in customers: _ensure_schedule_directory(customer, "schedule customer directory") if _VALID_CUSTOMER_KEY.fullmatch(customer.name) is None: raise CustomerScheduleError("schedule claim customer key is invalid") try: days = tuple(customer.iterdir()) except OSError as exc: raise CustomerScheduleError("schedule claims could not be inventoried") from exc for day in days: _ensure_schedule_directory(day, "schedule claim date directory") try: kst_day = date.fromisoformat(day.name) except ValueError as exc: raise CustomerScheduleError("schedule claim date is invalid") from exc try: files = tuple(day.iterdir()) except OSError as exc: raise CustomerScheduleError("schedule claims could not be inventoried") from exc for claim in files: if claim.is_symlink() or not claim.is_file(): raise CustomerScheduleError("schedule claim contains an unsafe entry") if claim.suffix != ".claim" or claim.stem not in _VALID_TASK_KINDS: raise CustomerScheduleError("schedule claim kind is invalid") result.append( (CustomerScheduleTask(customer.name, claim.stem, kst_day), claim) ) return tuple(sorted(result, key=lambda item: _schedule_key(item[0]))) def _legacy_tombstone(reservation_id: str) -> bytes: return ( b"scheduled-delivery-tombstone-v1\n" + reservation_id.encode("ascii") + b"\n" ) def _tombstone_reservation(raw: bytes) -> str | None: lines = raw.splitlines() if len(lines) == 2 and lines[0] == b"scheduled-delivery-tombstone-v1": value = lines[1].decode("ascii", errors="ignore") if _VALID_OPAQUE.fullmatch(value): return value return None def _write_tombstone(path: Path, reservation_id: str) -> str: _ensure_regular(path, "schedule claim") parent = path.parent if parent.is_symlink(): raise CustomerScheduleError("schedule claim directory symlinks are not allowed") parent.mkdir(parents=True, exist_ok=True, mode=0o700) parent.chmod(0o700) if path.exists(): try: raw = path.read_bytes() except OSError as exc: raise CustomerScheduleError("schedule claim is unavailable") from exc existing = _tombstone_reservation(raw) if existing is not None: if existing != reservation_id: raise CustomerScheduleError("schedule claim tombstone conflicts") return _bytes_digest(raw) raise CustomerScheduleError("legacy schedule claim already exists") raw = _legacy_tombstone(reservation_id) try: descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "wb") as handle: handle.write(raw) handle.flush() os.fsync(handle.fileno()) path.chmod(0o600) _fsync_directory(parent) except FileExistsError as exc: raise CustomerScheduleError("schedule claim appeared during reservation") from exc except OSError as exc: raise CustomerScheduleError("schedule claim tombstone could not be persisted") from exc return _bytes_digest(raw) def _replace_tombstone( path: Path, reservation_id: str, *, previous_reservation_id: str ) -> str: """Durably fence and replace a failed reminder's claim with its new attempt.""" _ensure_regular(path, "schedule claim") parent = path.parent if parent.is_symlink(): raise CustomerScheduleError("schedule claim directory symlinks are not allowed") try: existing = path.read_bytes() except OSError as exc: raise CustomerScheduleError("schedule claim is unavailable") from exc if _tombstone_reservation(existing) != previous_reservation_id: raise CustomerScheduleError("schedule claim tombstone conflicts") raw = _legacy_tombstone(reservation_id) temporary = parent / f".{path.name}.{uuid.uuid4().hex}.tmp" try: descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "wb") as handle: handle.write(raw) handle.flush() os.fsync(handle.fileno()) temporary.chmod(0o600) os.replace(temporary, path) path.chmod(0o600) _fsync_directory(parent) except OSError as exc: try: temporary.unlink(missing_ok=True) except OSError: pass raise CustomerScheduleError("schedule claim tombstone could not be replaced") from exc return _bytes_digest(raw) def _row_body(row: Mapping[str, object]) -> dict[str, object]: return {key: value for key, value in row.items() if key != "row_digest"} def _row_digest(row: Mapping[str, object]) -> str: return _digest(_row_body(row)) def _validate_schedule_row_shape(row: Mapping[str, object], expected_sequence: int) -> None: expected = { "schema_version", "kind", "append_sequence", "reservation_id", "state", "schedule_key", "customer_key", "task_kind", "kst_day", "body", "body_digest", "template_digest", "destination", "destination_digest", "registry_digest", "config_digest", "legacy_claim_digest", "provider_receipt", "message_id", "reason", "previous_row_digest", "created_at", "row_digest", } weekly_fields = { "weekly_authority_digest", "weekly_customer_identity_digest", "weekly_candidate_digest", "weekly_canonical_registry_digest", "weekly_canonical_binding_digest", "weekly_sidecar_authority_digest", "weekly_sidecar_history_digest", "weekly_ledger_authority_digest", "weekly_owner_digest", "weekly_consent_digest", "weekly_feature_epoch", "weekly_route_digest", "weekly_canonical_sequence", "weekly_canonical_digest", } keys = frozenset(row) if keys != frozenset(expected) and keys != frozenset(expected | weekly_fields): raise CustomerScheduleError("scheduled-deliveries ledger schema mismatch") if row.get("schema_version") != _SCHEDULE_SCHEMA_VERSION: raise CustomerScheduleError("scheduled-deliveries ledger schema version is unsupported") if row.get("kind") != _SCHEDULE_SCHEMA_NAME: raise CustomerScheduleError("scheduled-deliveries ledger kind is invalid") if type(row.get("append_sequence")) is not int or row.get("append_sequence") != expected_sequence: raise CustomerScheduleError("scheduled-deliveries ledger sequence is not contiguous") if row.get("row_digest") != _row_digest(row): raise CustomerScheduleError("scheduled-deliveries ledger row digest mismatch") reservation_id = row.get("reservation_id") if not isinstance(reservation_id, str) or _VALID_OPAQUE.fullmatch(reservation_id) is None: raise CustomerScheduleError("scheduled-deliveries reservation is invalid") if row.get("state") not in _SCHEDULE_STATES: raise CustomerScheduleError("scheduled-deliveries state is invalid") customer_key = row.get("customer_key") task_kind = row.get("task_kind") day = row.get("kst_day") if ( not isinstance(customer_key, str) or _VALID_CUSTOMER_KEY.fullmatch(customer_key) is None or task_kind not in _VALID_TASK_KINDS or not isinstance(day, str) ): raise CustomerScheduleError("scheduled-deliveries task pin is invalid") try: date.fromisoformat(day) except ValueError as exc: raise CustomerScheduleError("scheduled-deliveries date pin is invalid") from exc if row.get("schedule_key") != f"{customer_key}:{day}:{task_kind}": raise CustomerScheduleError("scheduled-deliveries schedule key is invalid") body = row.get("body") if body is not None and not isinstance(body, str): raise CustomerScheduleError("scheduled-deliveries body pin is invalid") if body is not None and _digest(body) != row.get("body_digest"): raise CustomerScheduleError("scheduled-deliveries body digest mismatch") if body is None and row.get("body_digest") != _SCHEDULE_ZERO_DIGEST: raise CustomerScheduleError("scheduled-deliveries legacy body digest is invalid") _require_digest(row.get("body_digest"), "body") _require_digest(row.get("template_digest"), "template") destination = row.get("destination") if destination is not None and not isinstance(destination, (str, Mapping, list, tuple, int)): raise CustomerScheduleError("scheduled-deliveries destination pin is invalid") if destination is None: if row.get("destination_digest") != _SCHEDULE_ZERO_DIGEST: raise CustomerScheduleError("scheduled-deliveries legacy destination digest is invalid") elif _digest(destination) != row.get("destination_digest"): raise CustomerScheduleError("scheduled-deliveries destination digest mismatch") _require_digest(row.get("destination_digest"), "destination") _require_digest(row.get("registry_digest"), "registry") _require_digest(row.get("config_digest"), "config") if "weekly_authority_digest" in row: for field_name in weekly_fields - {"weekly_feature_epoch", "weekly_canonical_sequence"}: _ = _require_digest(row.get(field_name), field_name) if not isinstance(row.get("weekly_feature_epoch"), str): raise CustomerScheduleError("weekly reminder epoch pin is invalid") sequence = row.get("weekly_canonical_sequence") if type(sequence) is not int or sequence < 0: raise CustomerScheduleError("weekly reminder canonical sequence is invalid") claim_digest = row.get("legacy_claim_digest") if not isinstance(claim_digest, str): raise CustomerScheduleError("scheduled-deliveries legacy claim digest is missing") _require_digest(claim_digest, "legacy claim") for field_name in ("provider_receipt", "message_id", "reason", "previous_row_digest", "created_at"): value = row.get(field_name) if field_name == "previous_row_digest": if value is not None: _require_digest(value, "previous row") continue if field_name == "created_at": if not isinstance(value, str): raise CustomerScheduleError("scheduled-deliveries timestamp is invalid") try: datetime.fromisoformat(value) except ValueError as exc: raise CustomerScheduleError("scheduled-deliveries timestamp is invalid") from exc continue if value is not None: if not isinstance(value, str) or not value: raise CustomerScheduleError(f"scheduled-deliveries {field_name} is invalid") if field_name in {"provider_receipt", "message_id", "reason"}: _require_opaque(value, f"scheduled-deliveries {field_name}") def _validate_schedule_rows(rows: list[dict[str, object]]) -> None: previous_digest: str | None = None current_by_reservation: dict[str, dict[str, object]] = {} reservation_by_schedule: dict[str, str] = {} current_by_schedule: dict[str, dict[str, object]] = {} for sequence, row in enumerate(rows, start=1): _validate_schedule_row_shape(row, sequence) if row.get("previous_row_digest") != previous_digest: raise CustomerScheduleError("scheduled-deliveries predecessor digest mismatch") previous_digest = str(row["row_digest"]) reservation_id = str(row["reservation_id"]) schedule_key = str(row["schedule_key"]) state = str(row["state"]) existing_reservation = reservation_by_schedule.get(schedule_key) prior_for_schedule = current_by_schedule.get(schedule_key) if existing_reservation is not None and existing_reservation != reservation_id: if not ( row.get("task_kind") == "reminder" and "weekly_authority_digest" not in row and prior_for_schedule is not None and prior_for_schedule.get("state") == "known_failure" and state == "prepared" ): raise CustomerScheduleError("duplicate scheduled-delivery task") reservation_by_schedule[schedule_key] = reservation_id prior = current_by_reservation.get(reservation_id) if prior is None: if state == "prepared": pass elif ( state == "unknown" and row.get("reason") in _SCHEDULE_INITIAL_UNKNOWN_REASONS ): pass else: raise CustomerScheduleError( "scheduled-deliveries first state must be prepared or legacy unknown" ) else: if prior.get("state") in _SCHEDULE_TERMINAL_STATES: raise CustomerScheduleError("scheduled-deliveries transition follows terminal state") allowed = { "prepared": {"sending", "unknown", "abandoned"}, "sending": {"delivered", "unknown", "known_failure", "abandoned"}, "delivered": {"sent_audited"}, }.get(str(prior["state"]), set()) if state not in allowed: raise CustomerScheduleError("scheduled-deliveries transition is invalid") for field in ( "reservation_id", "schedule_key", "customer_key", "task_kind", "kst_day", "body", "body_digest", "template_digest", "destination", "destination_digest", "registry_digest", "config_digest", "weekly_authority_digest", "weekly_customer_identity_digest", "weekly_candidate_digest", "weekly_canonical_registry_digest", "weekly_canonical_binding_digest", "weekly_sidecar_authority_digest", "weekly_sidecar_history_digest", "weekly_ledger_authority_digest", "weekly_owner_digest", "weekly_consent_digest", "weekly_feature_epoch", "weekly_route_digest", "weekly_canonical_sequence", "weekly_canonical_digest", "legacy_claim_digest", ): if row.get(field) != prior.get(field): raise CustomerScheduleError("scheduled-delivery immutable pin changed") current_by_reservation[reservation_id] = dict(row) current_by_schedule[schedule_key] = dict(row) def _append_row(path: Path, row: Mapping[str, object]) -> dict[str, object]: _ensure_regular(path, "scheduled-deliveries ledger") parent = path.parent if parent.is_symlink(): raise CustomerScheduleError("schedule directory symlinks are not allowed") parent.mkdir(parents=True, exist_ok=True, mode=0o700) parent.chmod(0o700) try: with path.open("a", encoding="utf-8") as handle: handle.write(_canonical_json(dict(row)) + "\n") handle.flush() os.fsync(handle.fileno()) path.chmod(0o600) except (OSError, TypeError, ValueError) as exc: if isinstance(exc, CustomerScheduleError): raise raise CustomerScheduleError("scheduled-deliveries ledger could not be appended") from exc return dict(row) def _current_rows(rows: list[dict[str, object]]) -> dict[str, dict[str, object]]: current: dict[str, dict[str, object]] = {} for row in rows: current[str(row["reservation_id"])] = row return current def _claims_by_schedule( claims_root: Path, ) -> dict[str, tuple[CustomerScheduleTask, Path, bytes, str | None]]: result: dict[str, tuple[CustomerScheduleTask, Path, bytes, str | None]] = {} for task, path in _iter_claims(claims_root): raw = _read_private_regular_bytes( path, "schedule claim", max_bytes=1024 * 1024, ) if raw is None: raise CustomerScheduleError("schedule claim is unavailable") schedule = _schedule_key(task) if schedule in result: raise CustomerScheduleError("duplicate schedule claim") result[schedule] = (task, path, raw, _tombstone_reservation(raw)) if len(result) > _SCHEDULE_READ_MAX_INVENTORY: raise CustomerScheduleError("schedule claim inventory is too large") return result def _legacy_unknown_row( sequence: int, previous_digest: str | None, task: CustomerScheduleTask, claim_digest: str, ) -> dict[str, object]: reservation_pin = {"schedule_key": _schedule_key(task), "claim_digest": claim_digest} reservation_id = f"legacy-{_digest(reservation_pin)[:32]}" body_digest = _SCHEDULE_ZERO_DIGEST destination_digest = _SCHEDULE_ZERO_DIGEST body: dict[str, object] = { "schema_version": _SCHEDULE_SCHEMA_VERSION, "kind": _SCHEDULE_SCHEMA_NAME, "append_sequence": sequence, "reservation_id": reservation_id, "state": "unknown", "schedule_key": _schedule_key(task), "customer_key": task.customer_key, "task_kind": task.kind, "kst_day": task.kst_day.isoformat(), "body": None, "body_digest": body_digest, "template_digest": _SCHEDULE_ZERO_DIGEST, "destination": None, "destination_digest": destination_digest, "registry_digest": _SCHEDULE_ZERO_DIGEST, "config_digest": _SCHEDULE_ZERO_DIGEST, "legacy_claim_digest": claim_digest, "provider_receipt": None, "message_id": None, "reason": "legacy_claim_unknown", "previous_row_digest": previous_digest, "created_at": datetime.now(timezone.utc).isoformat(), } return {**body, "row_digest": _row_digest(body)} def _tombstone_recovery_unknown_row( sequence: int, previous_digest: str | None, task: CustomerScheduleTask, reservation_id: str, claim_digest: str, ) -> dict[str, object]: return _build_row( sequence=sequence, previous_digest=previous_digest, state="unknown", reservation_id=reservation_id, task=task, body=None, template_digest=_SCHEDULE_ZERO_DIGEST, destination=None, registry_digest=_SCHEDULE_ZERO_DIGEST, config_digest=_SCHEDULE_ZERO_DIGEST, legacy_claim_digest=claim_digest, reason="tombstone_ledger_missing_recovered", ) def _check_pairs( rows: list[dict[str, object]], claims_root: Path, *, claims: dict[str, tuple[CustomerScheduleTask, Path, bytes, str | None]] | None = None, ) -> dict[str, tuple[CustomerScheduleTask, Path, bytes, str | None]]: claims = _claims_by_schedule(claims_root) if claims is None else claims current = _current_rows(rows) current_by_schedule = { str(row["schedule_key"]): row for row in current.values() } unpaired_claims = dict(claims) # A known-failure reminder replacement installs a new tombstone for its # new reservation. Prior rows retain their immutable claim digest in the # append-only ledger; only the current reservation pairs with the claim. for row in current_by_schedule.values(): pair = claims.get(str(row["schedule_key"])) if pair is None: raise CustomerScheduleError("scheduled-deliveries ledger row has no tombstone") claim_digest = str(row["legacy_claim_digest"]) if _bytes_digest(pair[2]) != claim_digest: raise CustomerScheduleError("scheduled-deliveries tombstone digest mismatch") tombstone_reservation = pair[3] if tombstone_reservation is None and pair[2].startswith( b"scheduled-delivery-tombstone-v1\n" ): raise CustomerScheduleError("scheduled-deliveries tombstone is invalid") if tombstone_reservation is None and ( row.get("state") != "unknown" or row.get("reason") != "legacy_claim_unknown" ): raise CustomerScheduleError("legacy schedule claim requires cutover") if tombstone_reservation is not None and tombstone_reservation != row["reservation_id"]: raise CustomerScheduleError("scheduled-deliveries tombstone reservation mismatch") unpaired_claims.pop(str(row["schedule_key"]), None) # Anything left is a tombstone or legacy claim without a ledger row. # Both are a recovery fence while the ready state is active. return unpaired_claims def _set_recovery_fence(fence_path: Path) -> None: try: _write_fence(fence_path, "recovery_required") except CustomerScheduleError: pass def _migrate_legacy_claims_locked( ledger_path: Path, fence_path: Path, claims_root: Path, rows: list[dict[str, object]], ) -> list[dict[str, object]]: claims = _claims_by_schedule(claims_root) current = _current_rows(rows) schedules = {str(row["schedule_key"]): row for row in current.values()} # A durable tombstone without its ledger row is recovered as terminal # unknown. It is never interpreted as unsent work or deleted. for schedule, (_, _, raw, tombstone) in claims.items(): if raw.startswith(b"scheduled-delivery-tombstone-v1\n") and tombstone is None: _set_recovery_fence(fence_path) raise CustomerScheduleError("schedule tombstone is invalid") changed = False for schedule, (task, path, raw, tombstone) in claims.items(): if schedule in schedules: existing = schedules[schedule] if ( tombstone is None and not raw.startswith(b"scheduled-delivery-tombstone-v1\n") ): if _bytes_digest(raw) != str(existing["legacy_claim_digest"]): _set_recovery_fence(fence_path) raise CustomerScheduleError("scheduled-deliveries tombstone digest mismatch") if existing.get("state") in {"prepared", "sending"}: try: row = _build_row( sequence=len(rows) + 1, previous_digest=str(rows[-1]["row_digest"]) if rows else None, state="unknown", reservation_id=str(existing["reservation_id"]), task=task, body=existing.get("body") if isinstance(existing.get("body"), str) else None, template_digest=str(existing["template_digest"]), destination=existing.get("destination"), registry_digest=str(existing["registry_digest"]), config_digest=str(existing["config_digest"]), legacy_claim_digest=str(existing["legacy_claim_digest"]), provider_receipt=existing.get("provider_receipt") if isinstance(existing.get("provider_receipt"), str) else None, message_id=existing.get("message_id") if isinstance(existing.get("message_id"), str) else None, reason="legacy_claim_unknown", ) _append_row(ledger_path, row) except CustomerScheduleError: _set_recovery_fence(fence_path) raise rows.append(row) schedules[schedule] = row changed = True continue claim_digest = _bytes_digest(raw) if tombstone is not None: row = _tombstone_recovery_unknown_row( len(rows) + 1, str(rows[-1]["row_digest"]) if rows else None, task, str(tombstone), claim_digest, ) else: row = _legacy_unknown_row( len(rows) + 1, str(rows[-1]["row_digest"]) if rows else None, task, claim_digest, ) try: _append_row(ledger_path, row) except CustomerScheduleError: _set_recovery_fence(fence_path) raise rows.append(row) schedules[schedule] = row changed = True if changed: _validate_schedule_rows(rows) # Keep the original claim bytes untouched; this is the cutover evidence. return rows def _ensure_ready_locked( ledger_path: Path, fence_path: Path, claims_root: Path, ) -> list[dict[str, object]]: rows = _read_jsonl(ledger_path) fence = _read_fence(fence_path) claims_exist = claims_root.exists() if fence is None: if rows or claims_exist: _set_recovery_fence(fence_path) raise CustomerScheduleError("schedule startup fence is missing") _write_fence(fence_path, "preparing") _write_fence(fence_path, "ready") fence = _read_fence(fence_path) if fence is None or fence.state != "ready": raise CustomerScheduleError("schedule startup fence is not ready") try: _validate_schedule_rows(rows) except CustomerScheduleError: _set_recovery_fence(fence_path) raise try: unpaired = _check_pairs(rows, claims_root) except CustomerScheduleError: _set_recovery_fence(fence_path) raise if unpaired: _set_recovery_fence(fence_path) raise CustomerScheduleError("schedule tombstone has no ledger pair") return rows def initialize_schedule_delivery_fence(profile_root: Path) -> ScheduleFenceReceipt: """Create the durable preparing/ready startup fence for a fresh profile.""" with _schedule_lock(profile_root) as (ledger, fence, claims): rows = _read_jsonl(ledger) current = _read_fence(fence) if current is None: _write_fence(fence, "preparing") current = _read_fence(fence) elif current.state not in {"ready", "recovery_required"}: raise CustomerScheduleError("schedule startup fence is not ready") if current.state == "ready": try: _validate_schedule_rows(rows) unpaired = _check_pairs(rows, claims) except CustomerScheduleError: _set_recovery_fence(fence) raise if unpaired: _write_fence(fence, "preparing") try: rows = _migrate_legacy_claims_locked(ledger, fence, claims, rows) unpaired = _check_pairs(rows, claims) except CustomerScheduleError: _set_recovery_fence(fence) raise if unpaired: _set_recovery_fence(fence) raise CustomerScheduleError("schedule tombstone has no ledger pair") _write_fence(fence, "ready") result = _read_fence(fence) assert result is not None return result return current _write_fence(fence, "preparing") try: _validate_schedule_rows(rows) rows = _migrate_legacy_claims_locked(ledger, fence, claims, rows) unpaired = _check_pairs(rows, claims) except CustomerScheduleError: _set_recovery_fence(fence) raise if unpaired: _set_recovery_fence(fence) raise CustomerScheduleError("schedule tombstone has no ledger pair") _write_fence(fence, "ready") result = _read_fence(fence) assert result is not None return result def prepare_schedule_delivery_cutover(profile_root: Path) -> ScheduleFenceReceipt: """Durably enter the preparing fence before a legacy-claim cutover.""" with _schedule_lock(profile_root) as (ledger, fence, claims): current = _read_fence(fence) if current is not None and current.state == "ready": rows = _read_jsonl(ledger) schedules = {str(row.get("schedule_key")) for row in rows} claim_pairs = _claims_by_schedule(claims) legacy_claims = tuple( pair for pair in claim_pairs.values() if pair[3] is None and not pair[2].startswith(b"scheduled-delivery-tombstone-v1\n") ) orphan_tombstones = tuple( pair for schedule, pair in claim_pairs.items() if pair[2].startswith(b"scheduled-delivery-tombstone-v1\n") and schedule not in schedules ) if not legacy_claims and not orphan_tombstones: raise CustomerScheduleError("schedule cutover is already ready") elif current is not None and current.state not in {"preparing", "recovery_required"}: raise CustomerScheduleError("schedule startup fence is not ready") _write_fence(fence, "preparing") result = _read_fence(fence) assert result is not None return result def finalize_schedule_delivery_cutover(profile_root: Path) -> ScheduleFenceReceipt: """Migrate immutable legacy claims and atomically publish the ready fence.""" with _schedule_lock(profile_root) as (ledger, fence, claims): current = _read_fence(fence) if current is None or current.state != "preparing": raise CustomerScheduleError("schedule cutover is not preparing") try: rows = _read_jsonl(ledger) _validate_schedule_rows(rows) rows = _migrate_legacy_claims_locked(ledger, fence, claims, rows) unpaired = _check_pairs(rows, claims) except CustomerScheduleError: _set_recovery_fence(fence) raise if unpaired: _set_recovery_fence(fence) raise CustomerScheduleError("schedule cutover has an unpaired tombstone") _write_fence(fence, "ready") result = _read_fence(fence) assert result is not None return result def _build_row( *, sequence: int, previous_digest: str | None, state: str, reservation_id: str, task: CustomerScheduleTask, body: str | None, template_digest: str, destination: object, registry_digest: str, config_digest: str, legacy_claim_digest: str, provider_receipt: str | None = None, message_id: str | None = None, reason: str | None = None, weekly_authority: WeeklyReminderReservationAuthority | None = None, ) -> dict[str, object]: body_digest = _digest(body) if body is not None else _SCHEDULE_ZERO_DIGEST destination_digest = _digest(destination) if destination is not None else _SCHEDULE_ZERO_DIGEST payload: dict[str, object] = { "schema_version": _SCHEDULE_SCHEMA_VERSION, "kind": _SCHEDULE_SCHEMA_NAME, "append_sequence": sequence, "reservation_id": reservation_id, "state": state, "schedule_key": _schedule_key(task), "customer_key": task.customer_key, "task_kind": task.kind, "kst_day": task.kst_day.isoformat(), "body": body, "body_digest": body_digest, "template_digest": template_digest, "destination": destination, "destination_digest": destination_digest, "registry_digest": registry_digest, "config_digest": config_digest, "legacy_claim_digest": legacy_claim_digest, "provider_receipt": provider_receipt, "message_id": message_id, "reason": reason, "previous_row_digest": previous_digest, "created_at": datetime.now(timezone.utc).isoformat(), } if weekly_authority is not None: payload.update(weekly_authority_fields(weekly_authority)) return {**payload, "row_digest": _row_digest(payload)} def _resolve_reference( rows: list[dict[str, object]], reference: object, ) -> tuple[dict[str, object], CustomerScheduleTask]: current = _current_rows(rows) if isinstance(reference, ScheduledDeliveryReceipt): reservation_id = reference.reservation_id elif isinstance(reference, CustomerScheduleTask): _validate_task(reference) matches = [ row for row in current.values() if row.get("schedule_key") == _schedule_key(reference) ] if len(matches) != 1: raise CustomerScheduleError("scheduled-delivery task reservation is missing") reservation_id = str(matches[0]["reservation_id"]) elif isinstance(reference, str): reservation_id = reference else: raise CustomerScheduleError("scheduled-delivery reference is invalid") row = current.get(reservation_id) if row is None: raise CustomerScheduleError("scheduled-delivery reservation is missing") task = CustomerScheduleTask( str(row["customer_key"]), str(row["task_kind"]), date.fromisoformat(str(row["kst_day"])), ) return row, task def reserve_customer_task_delivery( profile_root: Path, task: CustomerScheduleTask, body: str, destination: object, template_digest: str | None = None, registry_digest: str | None = None, config_digest: str | None = None, reservation_id: str | None = None, ) -> ScheduledDeliveryReceipt: """Persist tombstone then prepared delivery pins before any provider call.""" _validate_task(task) if task.kind == "reminder": raise CustomerScheduleError( "reminder reservations require the approved static reminder lifecycle" ) if not isinstance(body, str): raise CustomerScheduleError("scheduled-delivery body must be text") if not body: raise CustomerScheduleError("scheduled-delivery body must not be empty") body_digest = _digest(body) destination_digest = _digest(destination) template_pin = _require_digest( template_digest, "template", default=body_digest, ) registry_pin = _require_digest(registry_digest, "registry") config_pin = _require_digest(config_digest, "config") requested_reservation = reservation_id or uuid.uuid4().hex requested_reservation = _require_opaque( requested_reservation, "scheduled-delivery reservation", ) with _schedule_lock(profile_root) as (ledger, fence, claims_root): rows = _ensure_ready_locked(ledger, fence, claims_root) schedule_key = _schedule_key(task) current = _current_rows(rows) existing = next( (row for row in current.values() if row.get("schedule_key") == schedule_key), None, ) if existing is not None: immutable = ( existing.get("body_digest") == body_digest and existing.get("template_digest") == template_pin and existing.get("destination_digest") == destination_digest and existing.get("registry_digest") == registry_pin and existing.get("config_digest") == config_pin ) if not immutable: raise CustomerScheduleError("duplicate scheduled-delivery pins conflict") return receipt_from_row(existing) claim = _claim_path(claims_root, task) try: claim_digest = _write_tombstone(claim, requested_reservation) except CustomerScheduleError: _set_recovery_fence(fence) raise row = _build_row( sequence=len(rows) + 1, previous_digest=str(rows[-1]["row_digest"]) if rows else None, state="prepared", reservation_id=requested_reservation, task=task, body=body, template_digest=template_pin, destination=destination, registry_digest=registry_pin, config_digest=config_pin, legacy_claim_digest=claim_digest, ) try: _append_row(ledger, row) except CustomerScheduleError: _set_recovery_fence(fence) raise rows.append(row) _validate_schedule_rows(rows) _check_pairs(rows, claims_root) return receipt_from_row(row) def reserve_missing_checkin_reminder( profile_root: Path, customer_key: str, missing_window: date, destination: object, *, registry_digest: str, config_digest: str, operator_approval: str, canonical_sequence: int | None = None, canonical_digest: str | None = None, template_version: str = "missing-checkin-v1", weekly_authority: WeeklyReminderReservationAuthority | None = None, ) -> ScheduledDeliveryReceipt: """Reserve the one approved static reminder for a missing-check-in window.""" body = _STATIC_OPERATIONAL_REMINDER_TEMPLATES.get(template_version) if body is None: raise CustomerScheduleError("reminder template version is not approved") approval = _require_opaque(operator_approval, "reminder operator approval") task = CustomerScheduleTask(customer_key, "reminder", missing_window) _validate_task(task) template_digest = _digest({"version": template_version, "body": body}) registry_pin = _require_digest(registry_digest, "registry") config_pin = _require_digest(config_digest, "config") if (canonical_sequence is None) != (canonical_digest is None): raise CustomerScheduleError("canonical reminder claim must pin sequence and digest") if canonical_sequence is not None and ( type(canonical_sequence) is not int or canonical_sequence < 0 ): raise CustomerScheduleError("canonical reminder claim sequence is invalid") if canonical_digest is not None: _require_digest(canonical_digest, "canonical reminder claim") destination_digest = _digest(destination) if weekly_authority is not None: if weekly_authority.route_digest != destination_digest: raise CustomerScheduleError("weekly reminder route authority disagrees") if ( weekly_authority.canonical_sequence != canonical_sequence or weekly_authority.canonical_digest != canonical_digest ): raise CustomerScheduleError("weekly reminder canonical authority disagrees") reservation_id = "reminder-" + _digest( { "schedule_key": _schedule_key(task), "template_version": template_version, "destination": destination_digest, "registry": registry_pin, "config": config_pin, "operator_approval": approval, "canonical_sequence": canonical_sequence, "canonical_digest": canonical_digest, } ) if weekly_authority is not None: reservation_id = "weekly-reminder-" + _digest( { "customer_identity": weekly_authority.customer_identity_digest, "kst_day": missing_window.isoformat(), "template_digest": template_digest, } ) with _schedule_lock(profile_root) as (ledger, fence, claims_root): rows = _ensure_ready_locked(ledger, fence, claims_root) matches = sorted( ( row for row in _current_rows(rows).values() if row.get("schedule_key") == _schedule_key(task) ), key=lambda row: int(row["append_sequence"]), ) replacing_known_failure = False if matches: latest = matches[-1] if latest.get("reservation_id") == reservation_id: if ( weekly_authority is not None and weekly_authority_from_receipt(receipt_from_row(latest)) != weekly_authority ): raise CustomerScheduleError("weekly reminder authority pins conflict") return receipt_from_row(latest) if weekly_authority is not None or latest.get("state") != "known_failure": raise CustomerScheduleError("missing-checkin reminder is already terminal or reserved") claim = _claim_path(claims_root, task) try: # The preparing fence makes a crash after the atomic claim # replacement fail closed until recovery inspects the durable # evidence. The old known-failure row remains append-only. _write_fence(fence, "preparing") claim_digest = _replace_tombstone( claim, reservation_id, previous_reservation_id=str(latest["reservation_id"]), ) replacing_known_failure = True except CustomerScheduleError: _set_recovery_fence(fence) raise else: claim = _claim_path(claims_root, task) try: claim_digest = _write_tombstone(claim, reservation_id) except CustomerScheduleError: _set_recovery_fence(fence) raise row = _build_row( sequence=len(rows) + 1, previous_digest=str(rows[-1]["row_digest"]) if rows else None, state="prepared", reservation_id=reservation_id, task=task, body=body, template_digest=template_digest, destination=destination, registry_digest=registry_pin, config_digest=config_pin, legacy_claim_digest=claim_digest, reason=( f"operator-approved-{approval}" if canonical_digest is None else f"operator-approved-{approval}:canonical-{canonical_sequence}-{canonical_digest}" ), weekly_authority=weekly_authority, ) try: _append_row(ledger, row) rows.append(row) _validate_schedule_rows(rows) _check_pairs(rows, claims_root) if replacing_known_failure: _write_fence(fence, "ready") except CustomerScheduleError: _set_recovery_fence(fence) raise return receipt_from_row(row) def mark_customer_task_known_failure( profile_root: Path, reference: object, reason: str ) -> ScheduledDeliveryReceipt: """Record a known provider failure; only reminder reservations may replace it.""" return _append_transition( profile_root, reference, next_state="known_failure", reason=reason ) def abandon_missing_checkin_reminder_for_terminal_morning_response( profile_root: Path, reference: object ) -> ScheduledDeliveryReceipt: """Record a canonical response that suppressed provider delivery.""" receipt = _append_transition( profile_root, reference, next_state="abandoned", reason=_TERMINAL_MORNING_RESPONSE_BEFORE_PROVIDER, ) if receipt.provider_receipt is not None or receipt.message_id is not None: raise CustomerScheduleError("terminal-response abandonment has provider evidence") return receipt def _append_transition( profile_root: Path, reference: object, *, next_state: str, provider_receipt: str | None = None, message_id: str | None = None, reason: str | None = None, expected_weekly_authority: WeeklyReminderReservationAuthority | None = None, ) -> ScheduledDeliveryReceipt: if next_state not in _SCHEDULE_STATES: raise CustomerScheduleError("scheduled-delivery target state is invalid") with _schedule_lock(profile_root) as (ledger, fence, claims_root): rows = _ensure_ready_locked(ledger, fence, claims_root) current, task = _resolve_reference(rows, reference) current_state = str(current["state"]) stored_weekly_authority = weekly_authority_from_receipt(receipt_from_row(current)) if expected_weekly_authority is not None and stored_weekly_authority != expected_weekly_authority: next_state = "abandoned" reason = "weekly_authority_incident" if next_state == "known_failure" and task.kind != "reminder": raise CustomerScheduleError("known failure is limited to reminder reservations") if ( next_state == "abandoned" and reason == _TERMINAL_MORNING_RESPONSE_BEFORE_PROVIDER and ( task.kind != "reminder" or current_state not in {"sending", "abandoned"} ) ): raise CustomerScheduleError( "terminal-response abandonment requires a sending reminder" ) lease_handle: Any | None = None provider_authority = False if current_state == next_state: supplied = { "provider_receipt": provider_receipt, "message_id": message_id, "reason": reason, } if all( value is None or value == current.get(field) for field, value in supplied.items() ): return receipt_from_row(current) raise CustomerScheduleError("conflicting scheduled-delivery replay") allowed = { "prepared": {"sending", "unknown", "abandoned"}, "sending": {"delivered", "unknown", "known_failure", "abandoned"}, "delivered": {"sent_audited"}, }.get(current_state, set()) if next_state not in allowed: if current_state in _SCHEDULE_TERMINAL_STATES: return receipt_from_row(current) raise CustomerScheduleError("scheduled-delivery transition is invalid") if ( next_state == "unknown" and reason == "delivery_unknown_after_restart" and current_state == "sending" and not ( isinstance(reference, ScheduledDeliveryReceipt) and reference.provider_authority ) and _provider_lease_active(profile_root, str(current["reservation_id"])) ): return receipt_from_row(current) reservation_id = str(current["reservation_id"]) if next_state == "sending" and current_state == "prepared": lease_handle = _acquire_provider_lease(profile_root, reservation_id) if lease_handle is None: return receipt_from_row(current) provider_authority = True reason = f"provider-owner-{uuid.uuid4().hex}" if next_state == "delivered": provider_receipt = _require_opaque( provider_receipt, "provider receipt", ) elif next_state == "sent_audited": # The provider receipt is an immutable transport pin. An audit # transition may never replace it with an audit-local identifier. provider_receipt = current.get("provider_receipt") if not isinstance(provider_receipt, str): raise CustomerScheduleError("sent-audited transition has no provider receipt") if next_state == "unknown": reason = reason or "delivery_unknown" row = _build_row( sequence=len(rows) + 1, previous_digest=str(rows[-1]["row_digest"]) if rows else None, state=next_state, reservation_id=str(current["reservation_id"]), task=task, body=current.get("body") if isinstance(current.get("body"), str) else None, template_digest=str(current["template_digest"]), destination=current.get("destination"), registry_digest=str(current["registry_digest"]), config_digest=str(current["config_digest"]), legacy_claim_digest=str(current["legacy_claim_digest"]), provider_receipt=provider_receipt or current.get("provider_receipt"), message_id=message_id or current.get("message_id"), reason=reason, weekly_authority=stored_weekly_authority, ) try: _append_row(ledger, row) if lease_handle is not None: _SCHEDULE_PROVIDER_LEASES[_lease_key(profile_root, reservation_id)] = lease_handle except CustomerScheduleError: if current_state == "sending" and next_state in { "delivered", "unknown", "known_failure", "abandoned", }: _release_provider_lease(profile_root, reservation_id) elif lease_handle is not None: try: fcntl.flock(lease_handle.fileno(), fcntl.LOCK_UN) except OSError: pass try: lease_handle.close() except OSError: pass _set_recovery_fence(fence) raise rows.append(row) try: _validate_schedule_rows(rows) except CustomerScheduleError: if current_state == "sending" and next_state in { "delivered", "unknown", "known_failure", "abandoned", }: _release_provider_lease(profile_root, reservation_id) elif lease_handle is not None: _release_provider_lease(profile_root, reservation_id) _set_recovery_fence(fence) raise if current_state == "sending" and next_state in { "delivered", "unknown", "known_failure", "abandoned", }: _release_provider_lease(profile_root, reservation_id) return receipt_from_row(row, provider_authority=provider_authority) def mark_weekly_reminder_sending( profile_root: Path, reference: ScheduledDeliveryReceipt, authority: WeeklyReminderReservationAuthority, ) -> ScheduledDeliveryReceipt: """Grant provider authority only when every durable weekly pin is current.""" return _append_transition( profile_root, reference, next_state="sending", expected_weekly_authority=authority, ) def mark_customer_task_sending( profile_root: Path, reference: object, ) -> ScheduledDeliveryReceipt: """Record the provider reservation immediately before the provider call. Only the caller that appended the durable prepared-to-sending transition receives ``provider_authority=True``. Replays return the existing receipt without authority. """ return _append_transition(profile_root, reference, next_state="sending") def mark_customer_task_delivered( profile_root: Path, reference: object, provider_receipt: str, message_id: str | None = None, ) -> ScheduledDeliveryReceipt: """Persist a provider receipt exactly once; this function never sends.""" return _append_transition( profile_root, reference, next_state="delivered", provider_receipt=provider_receipt, message_id=message_id, ) def mark_customer_task_unknown( profile_root: Path, reference: object, reason: str = "delivery_unknown", ) -> ScheduledDeliveryReceipt: """Terminalize an uncertain provider outcome; callers must not retry.""" return _append_transition( profile_root, reference, next_state="unknown", reason=reason, ) def mark_customer_task_abandoned( profile_root: Path, reference: object, reason: str = "delivery_abandoned", ) -> ScheduledDeliveryReceipt: """Terminalize a reservation only when the host has abandoned recovery.""" return _append_transition( profile_root, reference, next_state="abandoned", reason=reason, ) def mark_customer_task_sent_audited( profile_root: Path, reference: object, audit_receipt: str | None = None, ) -> ScheduledDeliveryReceipt: """Append the audit terminal without invoking a provider.""" if audit_receipt is not None: _require_opaque(audit_receipt, "audit receipt") return _append_transition( profile_root, reference, next_state="sent_audited", ) def reconcile_customer_task_delivery( profile_root: Path, reference: object, provider_receipt: str | None = None, message_id: str | None = None, audit_receipt: str | None = None, ) -> ScheduledDeliveryReceipt: """Reconcile a persisted/external receipt without a provider retry.""" if audit_receipt is not None: _require_opaque(audit_receipt, "audit receipt") with _schedule_lock(profile_root) as (ledger, fence, claims_root): rows = _ensure_ready_locked(ledger, fence, claims_root) current, _ = _resolve_reference(rows, reference) state = str(current["state"]) if state == "sent_audited": return receipt_from_row(current) if state == "unknown": return receipt_from_row(current) if state == "delivered": existing_provider = current.get("provider_receipt") if provider_receipt is not None and provider_receipt != existing_provider: raise CustomerScheduleError( "delivery receipt conflicts with immutable provider receipt" ) return _append_transition( profile_root, reference, next_state="sent_audited", message_id=message_id, ) if provider_receipt is None and not current.get("provider_receipt"): raise CustomerScheduleError("delivery reconciliation requires a provider receipt") delivered = _append_transition( profile_root, reference, next_state="delivered", provider_receipt=provider_receipt or str(current["provider_receipt"]), message_id=message_id, ) return _append_transition( profile_root, delivered, next_state="sent_audited", message_id=message_id, ) def schedule_delivery_ledger(profile_root: Path) -> tuple[ScheduledDeliveryReceipt, ...]: """Read all current reservation receipts without changing schedule state.""" with _schedule_lock(profile_root) as (ledger, fence, claims_root): rows = _ensure_ready_locked(ledger, fence, claims_root) return tuple(receipt_from_row(row) for row in _current_rows(rows).values()) @contextmanager def schedule_delivery_read_lock(profile_root: Path) -> Iterator[tuple[Path, Path, Path]]: """Lock the existing schedule writer inode without creating or recovering files.""" from checkin_cli.diagnostic_evidence import existing_read_lock ledger, fence, lock_path, claims_root = _schedule_paths(profile_root, create=False) with existing_read_lock(lock_path): state_token = _SCHEDULE_READ_LOCK_STATE.set((ledger, fence, claims_root)) try: yield ledger, fence, claims_root finally: _SCHEDULE_READ_LOCK_STATE.reset(state_token) def _validate_schedule_delivery_read_snapshot_locked( ledger_path: Path, fence_path: Path, claims_root: Path, *, customer_key: str | None = None, ) -> ScheduleDeliveryReadSnapshot: """Validate schedule evidence while the matching shared lock is held.""" held = _SCHEDULE_READ_LOCK_STATE.get() expected = (ledger_path, fence_path, claims_root) if held != expected: raise CustomerScheduleError("schedule read lock is required") if ( customer_key is not None and ( not isinstance(customer_key, str) or _VALID_CUSTOMER_KEY.fullmatch(customer_key) is None ) ): raise CustomerScheduleError("invalid customer key") if ( ledger_path.is_symlink() or fence_path.is_symlink() or claims_root.is_symlink() or not ledger_path.exists() or not fence_path.exists() or not claims_root.exists() or not claims_root.is_dir() ): raise CustomerScheduleError("schedule read state is incomplete") fence_document = _read_fence_document(fence_path) if fence_document is None or fence_document[0].state != "ready": raise CustomerScheduleError("schedule startup fence is not ready") rows = _read_jsonl(ledger_path) if len(rows) > _SCHEDULE_READ_MAX_INVENTORY: raise CustomerScheduleError("schedule ledger inventory is too large") try: _validate_schedule_rows(rows) except CustomerScheduleError: raise except (OSError, TypeError, ValueError) as exc: raise CustomerScheduleError("schedule ledger is corrupt") from exc claims = _claims_by_schedule(claims_root) unpaired = _check_pairs(rows, claims_root, claims=claims) if unpaired: raise CustomerScheduleError("schedule tombstone has no ledger pair") selected_rows = [ row for row in rows if customer_key is None or row.get("customer_key") == customer_key ] selected_claims = { schedule: pair for schedule, pair in claims.items() if customer_key is None or pair[0].customer_key == customer_key } def claim_inventory( values: Mapping[str, tuple[CustomerScheduleTask, Path, bytes, str | None]], ) -> list[dict[str, object]]: return [ { "schedule_key": schedule, "claim_digest": _bytes_digest(pair[2]), "tombstone_reservation": pair[3], } for schedule, pair in sorted(values.items()) ] all_claim_inventory = claim_inventory(claims) selected_claim_inventory = claim_inventory(selected_claims) ledger_digest = _digest(selected_rows) tombstone_inventory_digest = _digest(selected_claim_inventory) all_ledger_digest = _digest(rows) all_tombstone_inventory_digest = _digest(all_claim_inventory) fence_digest = _bytes_digest( (_canonical_json(fence_document[1]) + "\n").encode("utf-8") ) revision_token = _digest( { "schema_version": "schedule_delivery_read_snapshot_v1", "fence_digest": fence_digest, "ledger_digest": all_ledger_digest, "tombstone_inventory_digest": all_tombstone_inventory_digest, } ) return ScheduleDeliveryReadSnapshot( schema_version="schedule_delivery_read_snapshot_v1", fence_epoch=None, row_count=len(selected_rows), tombstone_count=len(selected_claim_inventory), ledger_digest=ledger_digest, tombstone_inventory_digest=tombstone_inventory_digest, fence_digest=fence_digest, revision_token=revision_token, ) def validate_schedule_delivery_read_snapshot( profile_root: Path, customer_key: str | None = None, ) -> ScheduleDeliveryReadSnapshot: """Return a bounded schedule inventory using only the existing read lock.""" try: with schedule_delivery_read_lock(profile_root) as (ledger, fence, claims_root): return _validate_schedule_delivery_read_snapshot_locked( ledger, fence, claims_root, customer_key=customer_key, ) except CustomerScheduleError: raise except (OSError, RuntimeError) as exc: raise CustomerScheduleError("schedule read lock is unavailable") from exc def schedule_delivery_read_snapshot(profile_root: Path) -> tuple[Mapping[str, object], ...]: """Return validated immutable rows under the physically read-only schedule lock.""" with schedule_delivery_read_lock(profile_root) as (ledger, fence, claims): rows = _read_jsonl(ledger) _validate_schedule_delivery_read_snapshot_locked(ledger, fence, claims) return tuple(dict(row) for row in rows)