feat(observatory): seal blocked run preparations

2026-08-30 23:14:05 +03:00
parent 8d5aeb0533
commit 2a5763d3fb
12 changed files with 1935 additions and 25 deletions
@@ -3,6 +3,24 @@
 from k1link.observatory.canonical_result import (
    is_admitted_observatory_recorded_result,
 )
+from k1link.observatory.run_preparations import (
+    MAX_RUN_PREPARATION_RECORDS,
+    MAX_RUN_PREPARATION_STORAGE_BYTES,
+    OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA,
+    OBSERVATORY_RUN_PREPARATION_SCHEMA,
+    RUN_PREPARATION_DATABASE_NAME,
+    RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS,
+    ObservatoryRunPreparation,
+    ObservatoryRunPreparationCapacityError,
+    ObservatoryRunPreparationConflictError,
+    ObservatoryRunPreparationError,
+    ObservatoryRunPreparationIntegrityError,
+    ObservatoryRunPreparationIntent,
+    ObservatoryRunPreparationLedger,
+    ObservatoryRunPreparationNotFoundError,
+    load_observatory_run_preparation_ledger,
+    observatory_run_preparation_request_sha256,
+)
 from k1link.observatory.setups import (
    LABORATORY_SETUP_CATALOG_SCHEMA,
    LABORATORY_SETUP_REGISTRY_SCHEMA,
@@ -13,7 +31,23 @@ from k1link.observatory.setups import (
 __all__ = [
    "LABORATORY_SETUP_CATALOG_SCHEMA",
    "LABORATORY_SETUP_REGISTRY_SCHEMA",
+    "MAX_RUN_PREPARATION_RECORDS",
+    "MAX_RUN_PREPARATION_STORAGE_BYTES",
+    "OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA",
+    "OBSERVATORY_RUN_PREPARATION_SCHEMA",
+    "RUN_PREPARATION_DATABASE_NAME",
+    "RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS",
    "LaboratorySetupRegistry",
    "LaboratorySetupRegistryError",
+    "ObservatoryRunPreparation",
+    "ObservatoryRunPreparationCapacityError",
+    "ObservatoryRunPreparationConflictError",
+    "ObservatoryRunPreparationError",
+    "ObservatoryRunPreparationIntegrityError",
+    "ObservatoryRunPreparationIntent",
+    "ObservatoryRunPreparationLedger",
+    "ObservatoryRunPreparationNotFoundError",
    "is_admitted_observatory_recorded_result",
+    "load_observatory_run_preparation_ledger",
+    "observatory_run_preparation_request_sha256",
 ]
@@ -0,0 +1,712 @@
+"""Durable, observation-only preparation receipts for Observatory runs.
+
+This ledger deliberately does not model a run lifecycle.  Mission Core does
+not currently have a durable dispatcher for the archived Observatory
+RunDefinition, so a preparation can only record the exact admitted input and
+the fail-closed blocker.  A real run ledger must be introduced together with
+an atomic durable dispatch receipt; until then ``run_id`` remains absent.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import threading
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Final, Literal
+from uuid import uuid4
+
+from k1link.artifacts import utc_now_iso
+
+OBSERVATORY_RUN_PREPARATION_SCHEMA: Final = (
+    "missioncore.observatory-run-preparation/v1"
+)
+OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA: Final = (
+    "missioncore.observatory-run-preparation-request/v1"
+)
+RUN_PREPARATION_DATABASE_NAME: Final = "observatory-run-preparations.sqlite3"
+MAX_RUN_PREPARATION_RECORDS: Final = 10_000
+MAX_RUN_PREPARATION_STORAGE_BYTES: Final = 64 * 1024 * 1024
+RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS: Final = 0.1
+_SQLITE_BUSY_TIMEOUT_MILLISECONDS: Final = 100
+
+_PREPARATION_ID = re.compile(r"^observatory-prep-[a-f0-9]{32}$")
+_IDEMPOTENCY_KEY = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$")
+_SESSION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+_IDENTIFIER = re.compile(r"^[a-z][a-z0-9-]{2,95}$")
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_STATE: Literal["blocked"] = "blocked"
+_AUTHORITY: Final = {
+    "commands_enabled": False,
+    "actuation_allowed": False,
+    "navigation_or_safety_accepted": False,
+    "production_accepted": False,
+}
+
+_TABLE_SQL = """
+CREATE TABLE observatory_run_preparations (
+    preparation_id TEXT PRIMARY KEY,
+    idempotency_key TEXT NOT NULL UNIQUE,
+    request_sha256 TEXT NOT NULL,
+    receipt_sha256 TEXT NOT NULL,
+    source_session_id TEXT NOT NULL,
+    source_catalog_sha256 TEXT NOT NULL,
+    setup_id TEXT NOT NULL,
+    definition_id TEXT NOT NULL,
+    definition_version INTEGER NOT NULL CHECK (definition_version > 0),
+    definition_sha256 TEXT NOT NULL,
+    state TEXT NOT NULL CHECK (state = 'blocked'),
+    blocker_code TEXT NOT NULL,
+    blocker_message TEXT NOT NULL,
+    created_at_utc TEXT NOT NULL
+);
+"""
+_SCHEMA_SQL = _TABLE_SQL.replace("CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ", 1)
+
+_EXPECTED_COLUMNS: Final = (
+    ("preparation_id", "TEXT", 0, 1),
+    ("idempotency_key", "TEXT", 1, 0),
+    ("request_sha256", "TEXT", 1, 0),
+    ("receipt_sha256", "TEXT", 1, 0),
+    ("source_session_id", "TEXT", 1, 0),
+    ("source_catalog_sha256", "TEXT", 1, 0),
+    ("setup_id", "TEXT", 1, 0),
+    ("definition_id", "TEXT", 1, 0),
+    ("definition_version", "INTEGER", 1, 0),
+    ("definition_sha256", "TEXT", 1, 0),
+    ("state", "TEXT", 1, 0),
+    ("blocker_code", "TEXT", 1, 0),
+    ("blocker_message", "TEXT", 1, 0),
+    ("created_at_utc", "TEXT", 1, 0),
+)
+
+
+class ObservatoryRunPreparationError(RuntimeError):
+    """Base error for the isolated Observatory preparation ledger."""
+
+
+class ObservatoryRunPreparationConflictError(ObservatoryRunPreparationError):
+    """An idempotency key is already bound to another preparation request."""
+
+
+class ObservatoryRunPreparationNotFoundError(ObservatoryRunPreparationError):
+    """A preparation receipt is not present."""
+
+
+class ObservatoryRunPreparationIntegrityError(ObservatoryRunPreparationError):
+    """The durable ledger violates its fail-closed immutable contract."""
+
+
+class ObservatoryRunPreparationCapacityError(ObservatoryRunPreparationError):
+    """The bounded preparation ledger cannot admit another receipt."""
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryRunPreparationIntent:
+    """Validated, path-free identity captured at the preparation boundary."""
+
+    idempotency_key: str
+    source_session_id: str
+    source_catalog_sha256: str
+    setup_id: str
+    definition_id: str
+    definition_version: int
+    definition_sha256: str
+    blocker_code: str
+    blocker_message: str
+
+    def __post_init__(self) -> None:
+        _validate_idempotency_key(self.idempotency_key)
+        _validate_pattern(self.source_session_id, _SESSION_ID, "source session id")
+        _validate_digest(self.source_catalog_sha256, "source catalog sha256")
+        _validate_pattern(self.setup_id, _IDENTIFIER, "setup id")
+        _validate_pattern(self.definition_id, _IDENTIFIER, "definition id")
+        if (
+            not isinstance(self.definition_version, int)
+            or isinstance(self.definition_version, bool)
+            or self.definition_version < 1
+        ):
+            raise ValueError("definition version must be positive")
+        _validate_digest(self.definition_sha256, "definition sha256")
+        _validate_pattern(self.blocker_code, _IDENTIFIER, "blocker code")
+        _validate_text(self.blocker_message, "blocker message", max_length=1_000)
+
+    @property
+    def request_sha256(self) -> str:
+        """Fingerprint only the exact client payload used for idempotency."""
+
+        return observatory_run_preparation_request_sha256(
+            idempotency_key=self.idempotency_key,
+            source_session_id=self.source_session_id,
+            setup_id=self.setup_id,
+            definition_sha256=self.definition_sha256,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryRunPreparation:
+    preparation_id: str
+    idempotency_key: str
+    request_sha256: str
+    receipt_sha256: str
+    source_session_id: str
+    source_catalog_sha256: str
+    setup_id: str
+    definition_id: str
+    definition_version: int
+    definition_sha256: str
+    blocker_code: str
+    blocker_message: str
+    created_at_utc: str
+    state: Literal["blocked"] = _STATE
+
+    @classmethod
+    def from_intent(
+        cls,
+        intent: ObservatoryRunPreparationIntent,
+        *,
+        preparation_id: str,
+        created_at_utc: str,
+    ) -> ObservatoryRunPreparation:
+        receipt_sha256 = _receipt_sha256(
+            preparation_id=preparation_id,
+            idempotency_key=intent.idempotency_key,
+            request_sha256=intent.request_sha256,
+            source_session_id=intent.source_session_id,
+            source_catalog_sha256=intent.source_catalog_sha256,
+            setup_id=intent.setup_id,
+            definition_id=intent.definition_id,
+            definition_version=intent.definition_version,
+            definition_sha256=intent.definition_sha256,
+            blocker_code=intent.blocker_code,
+            blocker_message=intent.blocker_message,
+            created_at_utc=created_at_utc,
+        )
+        return cls(
+            preparation_id=preparation_id,
+            idempotency_key=intent.idempotency_key,
+            request_sha256=intent.request_sha256,
+            receipt_sha256=receipt_sha256,
+            source_session_id=intent.source_session_id,
+            source_catalog_sha256=intent.source_catalog_sha256,
+            setup_id=intent.setup_id,
+            definition_id=intent.definition_id,
+            definition_version=intent.definition_version,
+            definition_sha256=intent.definition_sha256,
+            blocker_code=intent.blocker_code,
+            blocker_message=intent.blocker_message,
+            created_at_utc=created_at_utc,
+        )
+
+    def __post_init__(self) -> None:
+        _validate_pattern(self.preparation_id, _PREPARATION_ID, "preparation id")
+        intent = ObservatoryRunPreparationIntent(
+            idempotency_key=self.idempotency_key,
+            source_session_id=self.source_session_id,
+            source_catalog_sha256=self.source_catalog_sha256,
+            setup_id=self.setup_id,
+            definition_id=self.definition_id,
+            definition_version=self.definition_version,
+            definition_sha256=self.definition_sha256,
+            blocker_code=self.blocker_code,
+            blocker_message=self.blocker_message,
+        )
+        _validate_digest(self.request_sha256, "request sha256")
+        if self.request_sha256 != intent.request_sha256:
+            raise ObservatoryRunPreparationIntegrityError(
+                "stored preparation request fingerprint changed"
+            )
+        _validate_digest(self.receipt_sha256, "receipt sha256")
+        if self.state != _STATE:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation cannot claim a run lifecycle state"
+            )
+        _validate_timestamp(self.created_at_utc)
+        if self.receipt_sha256 != _receipt_sha256(
+            preparation_id=self.preparation_id,
+            idempotency_key=self.idempotency_key,
+            request_sha256=self.request_sha256,
+            source_session_id=self.source_session_id,
+            source_catalog_sha256=self.source_catalog_sha256,
+            setup_id=self.setup_id,
+            definition_id=self.definition_id,
+            definition_version=self.definition_version,
+            definition_sha256=self.definition_sha256,
+            blocker_code=self.blocker_code,
+            blocker_message=self.blocker_message,
+            created_at_utc=self.created_at_utc,
+        ):
+            raise ObservatoryRunPreparationIntegrityError(
+                "stored preparation receipt fingerprint changed"
+            )
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": OBSERVATORY_RUN_PREPARATION_SCHEMA,
+            "preparation_id": self.preparation_id,
+            "run_id": None,
+            "idempotency_key": self.idempotency_key,
+            "request_sha256": self.request_sha256,
+            "receipt_sha256": self.receipt_sha256,
+            "source": {
+                "session_id": self.source_session_id,
+                "catalog_sha256": self.source_catalog_sha256,
+            },
+            "setup": {
+                "setup_id": self.setup_id,
+                "definition": {
+                    "definition_id": self.definition_id,
+                    "version": self.definition_version,
+                    "definition_sha256": self.definition_sha256,
+                },
+            },
+            "state": self.state,
+            "blocker": {
+                "reason_code": self.blocker_code,
+                "message": self.blocker_message,
+            },
+            "preconditions_passed": True,
+            "submission_allowed": False,
+            "dispatch_receipt": None,
+            "created_at_utc": self.created_at_utc,
+            "authority": dict(_AUTHORITY),
+        }
+
+
+class ObservatoryRunPreparationLedger:
+    """SQLite-backed immutable preparation receipts with durable idempotency."""
+
+    def __init__(
+        self,
+        data_dir: Path,
+        *,
+        clock: Callable[[], str] = utc_now_iso,
+        max_records: int = MAX_RUN_PREPARATION_RECORDS,
+    ) -> None:
+        if (
+            not isinstance(max_records, int)
+            or isinstance(max_records, bool)
+            or not 1 <= max_records <= MAX_RUN_PREPARATION_RECORDS
+        ):
+            raise ValueError("preparation ledger record quota is invalid")
+        self.data_dir = data_dir.expanduser().resolve()
+        self.database_path = self.data_dir / RUN_PREPARATION_DATABASE_NAME
+        self._clock = clock
+        self._max_records = max_records
+        self._lock = threading.RLock()
+        self._initialize()
+
+    def prepare(
+        self,
+        intent: ObservatoryRunPreparationIntent,
+    ) -> tuple[ObservatoryRunPreparation, bool]:
+        """Persist once, returning the original receipt for an exact retry."""
+
+        try:
+            with self._lock, self._connect() as connection:
+                connection.execute("BEGIN IMMEDIATE")
+                existing = connection.execute(
+                    "SELECT * FROM observatory_run_preparations "
+                    "WHERE idempotency_key = ?",
+                    (intent.idempotency_key,),
+                ).fetchone()
+                if existing is not None:
+                    record = _record_from_row(existing)
+                    if record.request_sha256 != intent.request_sha256:
+                        raise ObservatoryRunPreparationConflictError(
+                            "idempotency key is bound to another preparation request"
+                        )
+                    connection.commit()
+                    return record, False
+
+                if self._bounded_record_count(connection) >= self._max_records:
+                    raise ObservatoryRunPreparationCapacityError(
+                        "preparation ledger record quota is exhausted"
+                    )
+                record = ObservatoryRunPreparation.from_intent(
+                    intent,
+                    preparation_id=f"observatory-prep-{uuid4().hex}",
+                    created_at_utc=self._clock(),
+                )
+                connection.execute(
+                    "INSERT INTO observatory_run_preparations "
+                    "(preparation_id, idempotency_key, request_sha256, receipt_sha256, "
+                    "source_session_id, source_catalog_sha256, setup_id, "
+                    "definition_id, definition_version, definition_sha256, "
+                    "state, blocker_code, blocker_message, created_at_utc) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        record.preparation_id,
+                        record.idempotency_key,
+                        record.request_sha256,
+                        record.receipt_sha256,
+                        record.source_session_id,
+                        record.source_catalog_sha256,
+                        record.setup_id,
+                        record.definition_id,
+                        record.definition_version,
+                        record.definition_sha256,
+                        record.state,
+                        record.blocker_code,
+                        record.blocker_message,
+                        record.created_at_utc,
+                    ),
+                )
+                connection.commit()
+                return record, True
+        except ObservatoryRunPreparationError:
+            raise
+        except (OSError, sqlite3.Error) as exc:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger write failed"
+            ) from exc
+
+    def get(self, preparation_id: str) -> ObservatoryRunPreparation:
+        _validate_pattern(preparation_id, _PREPARATION_ID, "preparation id")
+        try:
+            with self._lock, self._connect() as connection:
+                row = connection.execute(
+                    "SELECT * FROM observatory_run_preparations "
+                    "WHERE preparation_id = ?",
+                    (preparation_id,),
+                ).fetchone()
+        except (OSError, sqlite3.Error) as exc:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger read failed"
+            ) from exc
+        if row is None:
+            raise ObservatoryRunPreparationNotFoundError(preparation_id)
+        return _record_from_row(row)
+
+    def get_by_idempotency_key(
+        self,
+        idempotency_key: str,
+    ) -> ObservatoryRunPreparation | None:
+        """Return a sealed receipt without consulting mutable source/catalog state."""
+
+        _validate_idempotency_key(idempotency_key)
+        try:
+            with self._lock, self._connect() as connection:
+                row = connection.execute(
+                    "SELECT * FROM observatory_run_preparations "
+                    "WHERE idempotency_key = ?",
+                    (idempotency_key,),
+                ).fetchone()
+        except (OSError, sqlite3.Error) as exc:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger read failed"
+            ) from exc
+        return None if row is None else _record_from_row(row)
+
+    def _initialize(self) -> None:
+        try:
+            self.data_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+            self._validate_storage_paths()
+            with suppress(OSError):
+                self.data_dir.chmod(0o700)
+            with self._connect() as connection:
+                connection.executescript(_SCHEMA_SQL)
+                self._validate_schema(connection)
+                if self._bounded_record_count(connection) > self._max_records:
+                    raise ObservatoryRunPreparationCapacityError(
+                        "preparation ledger record quota is exceeded"
+                    )
+                connection.commit()
+            self._validate_storage_paths(require_database=True)
+            with suppress(OSError):
+                self.database_path.chmod(0o600)
+            _fsync_directory(self.data_dir)
+        except ObservatoryRunPreparationError:
+            raise
+        except (OSError, sqlite3.Error) as exc:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger initialization failed"
+            ) from exc
+
+    def _validate_schema(self, connection: sqlite3.Connection) -> None:
+        columns = connection.execute(
+            "SELECT name, type, \"notnull\", pk FROM pragma_table_info(?) "
+            "ORDER BY cid LIMIT ?",
+            ("observatory_run_preparations", len(_EXPECTED_COLUMNS) + 1),
+        ).fetchall()
+        actual_columns = tuple(
+            (row["name"], row["type"], row["notnull"], row["pk"])
+            for row in columns
+        )
+        if actual_columns != _EXPECTED_COLUMNS:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger schema is incompatible"
+            )
+        indexes = connection.execute(
+            "SELECT name, \"unique\", partial FROM pragma_index_list(?) "
+            "ORDER BY seq LIMIT 9",
+            ("observatory_run_preparations",),
+        ).fetchall()
+        if len(indexes) > 8:
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger has an unsafe index set"
+            )
+        if not _has_full_binary_unique_index(
+            connection,
+            indexes,
+            "preparation_id",
+        ):
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger lost receipt identity uniqueness"
+            )
+        if not _has_full_binary_unique_index(
+            connection,
+            indexes,
+            "idempotency_key",
+        ):
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger lost idempotency uniqueness"
+            )
+        table = connection.execute(
+            "SELECT sql FROM sqlite_master "
+            "WHERE type = 'table' AND name = 'observatory_run_preparations' "
+            "LIMIT 1"
+        ).fetchone()
+        if (
+            table is None
+            or not isinstance(table["sql"], str)
+            or _normalized_sql(table["sql"]) != _normalized_sql(_TABLE_SQL)
+        ):
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger table contract is incompatible"
+            )
+
+    def _bounded_record_count(self, connection: sqlite3.Connection) -> int:
+        rows = connection.execute(
+            "SELECT 1 FROM observatory_run_preparations LIMIT ?",
+            (self._max_records + 1,),
+        ).fetchall()
+        return len(rows)
+
+    def _validate_storage_paths(self, *, require_database: bool = False) -> None:
+        paths = (
+            self.database_path,
+            Path(f"{self.database_path}-wal"),
+            Path(f"{self.database_path}-shm"),
+        )
+        total_bytes = 0
+        for path in paths:
+            if path.is_symlink() or (path.exists() and not path.is_file()):
+                raise ObservatoryRunPreparationIntegrityError(
+                    "preparation ledger path is unsafe"
+                )
+            if path.exists():
+                total_bytes += path.stat().st_size
+        if require_database and not self.database_path.is_file():
+            raise ObservatoryRunPreparationIntegrityError(
+                "preparation ledger database is unavailable"
+            )
+        if total_bytes > MAX_RUN_PREPARATION_STORAGE_BYTES:
+            raise ObservatoryRunPreparationCapacityError(
+                "preparation ledger storage quota is exceeded"
+            )
+
+    @contextmanager
+    def _connect(self) -> Iterator[sqlite3.Connection]:
+        self._validate_storage_paths()
+        connection = sqlite3.connect(
+            self.database_path,
+            timeout=RUN_PREPARATION_SQLITE_LOCK_TIMEOUT_SECONDS,
+        )
+        connection.row_factory = sqlite3.Row
+        try:
+            connection.execute("PRAGMA foreign_keys = ON")
+            connection.execute("PRAGMA synchronous = FULL")
+            connection.execute(
+                f"PRAGMA busy_timeout = {_SQLITE_BUSY_TIMEOUT_MILLISECONDS}"
+            )
+            connection.execute("PRAGMA journal_mode = WAL")
+            yield connection
+        finally:
+            connection.close()
+
+
+def _record_from_row(row: sqlite3.Row) -> ObservatoryRunPreparation:
+    try:
+        return ObservatoryRunPreparation(
+            preparation_id=row["preparation_id"],
+            idempotency_key=row["idempotency_key"],
+            request_sha256=row["request_sha256"],
+            receipt_sha256=row["receipt_sha256"],
+            source_session_id=row["source_session_id"],
+            source_catalog_sha256=row["source_catalog_sha256"],
+            setup_id=row["setup_id"],
+            definition_id=row["definition_id"],
+            definition_version=row["definition_version"],
+            definition_sha256=row["definition_sha256"],
+            state=row["state"],
+            blocker_code=row["blocker_code"],
+            blocker_message=row["blocker_message"],
+            created_at_utc=row["created_at_utc"],
+        )
+    except (IndexError, KeyError, TypeError, ValueError) as exc:
+        raise ObservatoryRunPreparationIntegrityError(
+            "stored preparation receipt is invalid"
+        ) from exc
+
+
+def load_observatory_run_preparation_ledger(
+    data_dir: Path,
+) -> tuple[ObservatoryRunPreparationLedger | None, str | None]:
+    """Open the optional ledger within a short, fixed startup lock budget."""
+
+    try:
+        return ObservatoryRunPreparationLedger(data_dir), None
+    except (ObservatoryRunPreparationError, OSError) as exc:
+        return None, str(exc)
+
+
+def observatory_run_preparation_request_sha256(
+    *,
+    idempotency_key: str,
+    source_session_id: str,
+    setup_id: str,
+    definition_sha256: str,
+) -> str:
+    """Hash the exact immutable client payload, without mutable resolution state."""
+
+    _validate_idempotency_key(idempotency_key)
+    _validate_pattern(source_session_id, _SESSION_ID, "source session id")
+    _validate_pattern(setup_id, _IDENTIFIER, "setup id")
+    _validate_digest(definition_sha256, "definition sha256")
+    document = {
+        "schema_version": OBSERVATORY_RUN_PREPARATION_REQUEST_SCHEMA,
+        "idempotency_key": idempotency_key,
+        "source_session_id": source_session_id,
+        "setup_id": setup_id,
+        "definition_sha256": definition_sha256,
+    }
+    return hashlib.sha256(_canonical_json(document)).hexdigest()
+
+
+def _receipt_sha256(
+    *,
+    preparation_id: str,
+    idempotency_key: str,
+    request_sha256: str,
+    source_session_id: str,
+    source_catalog_sha256: str,
+    setup_id: str,
+    definition_id: str,
+    definition_version: int,
+    definition_sha256: str,
+    blocker_code: str,
+    blocker_message: str,
+    created_at_utc: str,
+) -> str:
+    document = {
+        "schema_version": OBSERVATORY_RUN_PREPARATION_SCHEMA,
+        "preparation_id": preparation_id,
+        "idempotency_key": idempotency_key,
+        "request_sha256": request_sha256,
+        "source_session_id": source_session_id,
+        "source_catalog_sha256": source_catalog_sha256,
+        "setup_id": setup_id,
+        "definition_id": definition_id,
+        "definition_version": definition_version,
+        "definition_sha256": definition_sha256,
+        "state": _STATE,
+        "blocker_code": blocker_code,
+        "blocker_message": blocker_message,
+        "created_at_utc": created_at_utc,
+    }
+    return hashlib.sha256(_canonical_json(document)).hexdigest()
+
+
+def _has_full_binary_unique_index(
+    connection: sqlite3.Connection,
+    indexes: list[sqlite3.Row],
+    column: str,
+) -> bool:
+    for index in indexes:
+        if not bool(index["unique"]) or bool(index["partial"]):
+            continue
+        fields = connection.execute(
+            "SELECT name, coll, desc, key FROM pragma_index_xinfo(?) "
+            "ORDER BY seqno LIMIT 4",
+            (index["name"],),
+        ).fetchall()
+        if len(fields) > 3:
+            continue
+        key_fields = [field for field in fields if bool(field["key"])]
+        if (
+            len(key_fields) == 1
+            and key_fields[0]["name"] == column
+            and key_fields[0]["coll"] == "BINARY"
+            and not bool(key_fields[0]["desc"])
+        ):
+            return True
+    return False
+
+
+def _normalized_sql(value: str) -> str:
+    return " ".join(value.split()).removesuffix(";")
+
+
+def _validate_idempotency_key(value: object) -> None:
+    _validate_pattern(value, _IDEMPOTENCY_KEY, "idempotency key")
+
+
+def _validate_digest(value: object, label: str) -> None:
+    _validate_pattern(value, _SHA256, label)
+
+
+def _validate_pattern(value: object, pattern: re.Pattern[str], label: str) -> None:
+    if not isinstance(value, str) or pattern.fullmatch(value) is None:
+        raise ValueError(f"{label} is invalid")
+
+
+def _validate_text(value: object, label: str, *, max_length: int) -> None:
+    if (
+        not isinstance(value, str)
+        or not value.strip()
+        or value != value.strip()
+        or len(value) > max_length
+    ):
+        raise ValueError(f"{label} is invalid")
+
+
+def _validate_timestamp(value: object) -> None:
+    _validate_text(value, "created timestamp", max_length=64)
+    assert isinstance(value, str)
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError("created timestamp is invalid") from exc
+    if (
+        parsed.tzinfo is None
+        or parsed.utcoffset() != timedelta(0)
+        or not value.endswith("Z")
+    ):
+        raise ValueError("created timestamp must use UTC")
+
+
+def _canonical_json(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def _fsync_directory(path: Path) -> None:
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
+    flags |= getattr(os, "O_DIRECTORY", 0)
+    descriptor = os.open(path, flags)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)