from __future__ import annotations

from collections.abc import Mapping, Sequence
from datetime import UTC, datetime, timedelta
import hashlib
import json
from typing import Any
import uuid

from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from govoplan_core.core.records import (
    RecordArchiveProvider,
    RecordArchiveTransferRequest,
    RecordContractError,
    RecordFilingRequest,
    RecordFilingResult,
    RecordSourceLocator,
    RecordSourceProvider,
    RecordSourceReference,
    RecordTransferPackage,
    record_archive_capabilities,
    record_archive_capability,
    record_source_capabilities,
    record_source_capability,
)
from govoplan_core.core.approvals import (
    CAPABILITY_APPROVAL_REQUESTS,
    ApprovalActorSelector,
    ApprovalRequestCreateCommand,
    ApprovalRequestProvider,
    ApprovalStepDefinition,
)
from govoplan_core.core.recovery import (
    RecoveryOperation,
    verify_recovery_evidence_chain,
)
from govoplan_core.core.temporal import current_temporal_data_context
from govoplan_core.db.temporal import apply_temporal_revision_filter
from govoplan_records.backend.db.models import (
    RecordChronologyEntry,
    RecordClassRevision,
    RecordDispositionRevision,
    RecordFilePlanRevision,
    RecordHoldRevision,
    RecordIdentity,
    RecordItem,
    RecordRevision,
    RecordTransferPackageRevision,
    RecordVolumeRevision,
)
from govoplan_records.backend.recovery import current_record_recovery_operation


class RecordStoreError(ValueError):
    pass


class RecordConflictError(RecordStoreError):
    pass


class RecordNotFoundError(RecordStoreError):
    pass


class RecordSourceUnavailableError(RecordStoreError):
    pass


class SqlRecordRegistry:
    def __init__(self, registry: object | None = None) -> None:
        self.registry = registry

    def catalog(
        self, session: Session, principal: object
    ) -> dict[str, list[dict[str, Any]]]:
        tenant_id = _tenant(principal)
        file_plan_query = session.query(RecordFilePlanRevision).filter(
            RecordFilePlanRevision.tenant_id == tenant_id
        )
        class_query = session.query(RecordClassRevision).filter(
            RecordClassRevision.tenant_id == tenant_id
        )
        file_plan = (
            apply_temporal_revision_filter(file_plan_query, RecordFilePlanRevision)
            .order_by(RecordFilePlanRevision.code, RecordFilePlanRevision.label)
            .all()
        )
        classes = (
            apply_temporal_revision_filter(class_query, RecordClassRevision)
            .order_by(RecordClassRevision.label)
            .all()
        )
        return {
            "file_plan": [_file_plan_dict(item) for item in file_plan],
            "classes": [_class_dict(item) for item in classes],
        }

    def write_file_plan_node(
        self,
        session: Session,
        principal: object,
        *,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(payload)
        replay = (
            session.query(RecordFilePlanRevision)
            .filter(
                RecordFilePlanRevision.tenant_id == tenant_id,
                RecordFilePlanRevision.idempotency_key
                == str(payload["idempotency_key"]),
            )
            .one_or_none()
        )
        if replay is not None:
            _verify_replay(replay.request_sha256, request_hash)
            return _file_plan_dict(replay)

        node_id = _text(payload, "node_id")
        current = _current_file_plan(session, tenant_id, node_id, lock=True)
        expected = payload.get("expected_revision")
        _validate_expected(
            current.revision if current else None, expected, label="File-plan node"
        )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at if current else None, recorded_at)
        parent_node_id = _optional_text(payload.get("parent_node_id"))
        _validate_file_plan_parent(
            session,
            tenant_id=tenant_id,
            node_id=node_id,
            parent_node_id=parent_node_id,
        )
        duplicate = (
            session.query(RecordFilePlanRevision)
            .filter(
                RecordFilePlanRevision.tenant_id == tenant_id,
                RecordFilePlanRevision.code == _text(payload, "code"),
                RecordFilePlanRevision.node_id != node_id,
                RecordFilePlanRevision.superseded_at.is_(None),
            )
            .first()
        )
        if duplicate is not None:
            raise RecordConflictError(
                "A current file-plan node already uses this code."
            )
        if current is not None:
            current.superseded_at = recorded_at
        row = RecordFilePlanRevision(
            tenant_id=tenant_id,
            node_id=node_id,
            revision=(current.revision + 1) if current else 1,
            previous_revision_id=current.id if current else None,
            parent_node_id=parent_node_id,
            code=_text(payload, "code"),
            label=_text(payload, "label"),
            description=_optional_text(payload.get("description")),
            active=bool(payload.get("active", True)),
            valid_from=_optional_timestamp(payload.get("valid_from"), "valid_from"),
            valid_to=_optional_timestamp(payload.get("valid_to"), "valid_to"),
            recorded_at=recorded_at,
            institutional_context=_mapping(payload.get("institutional_context")),
            changed_by=_actor(principal),
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        _validate_interval(row.valid_from, row.valid_to)
        session.add(row)
        session.flush()
        return _file_plan_dict(row)

    def write_record_class(
        self,
        session: Session,
        principal: object,
        *,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(payload)
        replay = (
            session.query(RecordClassRevision)
            .filter(
                RecordClassRevision.tenant_id == tenant_id,
                RecordClassRevision.idempotency_key == str(payload["idempotency_key"]),
            )
            .one_or_none()
        )
        if replay is not None:
            _verify_replay(replay.request_sha256, request_hash)
            return _class_dict(replay)
        class_id = _text(payload, "class_id")
        node_id = _text(payload, "file_plan_node_id")
        if _current_file_plan(session, tenant_id, node_id) is None:
            raise RecordStoreError(
                "The record class requires an existing file-plan node."
            )
        current = _current_class(session, tenant_id, class_id, lock=True)
        _validate_expected(
            current.revision if current else None,
            payload.get("expected_revision"),
            label="Record class",
        )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at if current else None, recorded_at)
        duplicate = (
            session.query(RecordClassRevision)
            .filter(
                RecordClassRevision.tenant_id == tenant_id,
                RecordClassRevision.key == _text(payload, "key"),
                RecordClassRevision.class_id != class_id,
                RecordClassRevision.superseded_at.is_(None),
            )
            .first()
        )
        if duplicate is not None:
            raise RecordConflictError("A current record class already uses this key.")
        if current is not None:
            current.superseded_at = recorded_at
        row = RecordClassRevision(
            tenant_id=tenant_id,
            class_id=class_id,
            revision=(current.revision + 1) if current else 1,
            previous_revision_id=current.id if current else None,
            file_plan_node_id=node_id,
            key=_text(payload, "key"),
            label=_text(payload, "label"),
            description=_optional_text(payload.get("description")),
            metadata_requirements=_text_list(payload.get("metadata_requirements")),
            allowed_source_types=_text_list(payload.get("allowed_source_types")),
            retention_period_days=_optional_int(payload.get("retention_period_days")),
            closure_trigger=_optional_text(payload.get("closure_trigger")),
            access_mode=_text(payload, "access_mode", default="tenant"),
            active=bool(payload.get("active", True)),
            valid_from=_optional_timestamp(payload.get("valid_from"), "valid_from"),
            valid_to=_optional_timestamp(payload.get("valid_to"), "valid_to"),
            recorded_at=recorded_at,
            institutional_context=_mapping(payload.get("institutional_context")),
            changed_by=_actor(principal),
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        _validate_interval(row.valid_from, row.valid_to)
        session.add(row)
        session.flush()
        return _class_dict(row)

    def create_record(
        self,
        session: Session,
        principal: object,
        *,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(payload)
        replay = _replay_event(
            session,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return self.get_record(
                session,
                principal,
                record_id=replay.record_id,
                revision=replay.record_revision,
            )["record"]
        class_id = _text(payload, "class_id")
        node_id = _text(payload, "file_plan_node_id")
        record_class = _current_class(session, tenant_id, class_id)
        if record_class is None or not record_class.active:
            raise RecordStoreError("The selected record class is not available.")
        if record_class.file_plan_node_id != node_id:
            raise RecordStoreError(
                "The record class does not belong to the selected file-plan node."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        record_id = _optional_text(payload.get("record_id")) or str(uuid.uuid4())
        if _text(payload, "access_mode", default="tenant") != "tenant":
            raise RecordStoreError(
                "Restricted records require the object-grant slice and cannot be created yet."
            )
        duplicate_number = (
            session.query(RecordIdentity.id)
            .filter(
                RecordIdentity.tenant_id == tenant_id,
                RecordIdentity.record_number == _text(payload, "record_number"),
            )
            .first()
        )
        if duplicate_number is not None:
            raise RecordConflictError("A record already uses this record number.")
        identity = RecordIdentity(
            tenant_id=tenant_id,
            record_id=record_id,
            record_number=_text(payload, "record_number"),
            created_by=_actor(principal),
        )
        session.add(identity)
        session.flush()
        snapshot = _record_snapshot(payload)
        row = RecordRevision(
            tenant_id=tenant_id,
            record_id=record_id,
            identity_id=identity.id,
            revision=1,
            class_id=class_id,
            file_plan_node_id=node_id,
            title=_text(payload, "title"),
            description=_optional_text(payload.get("description")),
            state=_text(payload, "state", default="open"),
            source_authority_mode=_text(
                payload, "source_authority_mode", default="native_authoritative"
            ),
            access_mode=_text(payload, "access_mode", default="tenant"),
            purpose=_text(payload, "purpose"),
            classification=_optional_text(payload.get("classification")),
            responsible_unit_id=_optional_text(payload.get("responsible_unit_id")),
            responsible_function_id=_optional_text(
                payload.get("responsible_function_id")
            ),
            external_reference=_mapping(payload.get("external_reference")),
            institutional_context=_mapping(payload.get("institutional_context")),
            search_text=_record_search_text(payload, identity.record_number),
            valid_from=_optional_timestamp(payload.get("valid_from"), "valid_from")
            or recorded_at,
            valid_to=_optional_timestamp(payload.get("valid_to"), "valid_to"),
            recorded_at=recorded_at,
            changed_by=_actor(principal),
            snapshot=snapshot,
        )
        _validate_interval(row.valid_from, row.valid_to)
        session.add(row)
        _append_event(
            session,
            principal,
            row=row,
            event_type="record.created",
            summary=f"Record {identity.record_number} created",
            occurred_at=recorded_at,
            purpose=row.purpose,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={"change_reason": _text(payload, "change_reason")},
        )
        session.flush()
        return _record_dict(row, identity)

    def update_record(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_event(
            session,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return self.get_record(
                session,
                principal,
                record_id=record_id,
                revision=replay.record_revision,
            )["record"]
        current = _current_record(session, tenant_id, record_id, lock=True)
        if current is None:
            raise RecordNotFoundError("Record not found.")
        expected = int(payload.get("expected_revision") or 0)
        if current.revision != expected:
            raise RecordConflictError(
                f"Record revision conflict: expected {expected}, current revision is {current.revision}."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at, recorded_at)
        identity = session.get(RecordIdentity, current.identity_id)
        if identity is None:
            raise RecordStoreError("Record identity is missing.")
        if (
            current.state not in {"planned", "open"}
            and "state" in payload
            and str(payload["state"]) != current.state
        ):
            raise RecordStoreError(
                "Lifecycle state can only change through a governed lifecycle action."
            )
        values = _revision_values(current, payload)
        if values["access_mode"] != "tenant":
            raise RecordStoreError(
                "Restricted records require the object-grant slice and cannot be enabled yet."
            )
        record_class = _current_class(session, tenant_id, values["class_id"])
        if (
            record_class is None
            or record_class.file_plan_node_id != values["file_plan_node_id"]
        ):
            raise RecordStoreError(
                "The selected record class and file-plan node do not match."
            )
        current.superseded_at = recorded_at
        row = RecordRevision(
            tenant_id=tenant_id,
            record_id=record_id,
            identity_id=current.identity_id,
            revision=current.revision + 1,
            previous_revision_id=current.id,
            recorded_at=recorded_at,
            changed_by=_actor(principal),
            search_text=" ".join(
                value
                for value in (
                    identity.record_number,
                    str(values["title"]),
                    str(values["description"] or ""),
                    str(values["classification"] or ""),
                )
                if value
            ).lower(),
            snapshot={**dict(current.snapshot), **_json_mapping(payload)},
            **values,
        )
        _validate_interval(row.valid_from, row.valid_to)
        session.add(row)
        _append_event(
            session,
            principal,
            row=row,
            event_type="record.revised",
            summary=f"Record {identity.record_number} revised",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "change_reason": _text(payload, "change_reason"),
                "previous_revision": current.revision,
            },
        )
        session.flush()
        return _record_dict(row, identity)

    def list_records(
        self,
        session: Session,
        principal: object,
        *,
        query: str | None = None,
        state: str | None = None,
        class_id: str | None = None,
        file_plan_node_id: str | None = None,
        offset: int = 0,
        limit: int = 100,
    ) -> tuple[list[dict[str, Any]], int]:
        tenant_id = _tenant(principal)
        statement = (
            session.query(RecordRevision, RecordIdentity)
            .join(RecordIdentity, RecordIdentity.id == RecordRevision.identity_id)
            .filter(RecordRevision.tenant_id == tenant_id)
        )
        statement = apply_temporal_revision_filter(statement, RecordRevision)
        if query and query.strip():
            pattern = f"%{query.strip().lower()}%"
            statement = statement.filter(RecordRevision.search_text.ilike(pattern))
        if state:
            statement = statement.filter(RecordRevision.state == state)
        if class_id:
            statement = statement.filter(RecordRevision.class_id == class_id)
        if file_plan_node_id:
            statement = statement.filter(
                RecordRevision.file_plan_node_id == file_plan_node_id
            )
        total = statement.count()
        rows = (
            statement.order_by(
                RecordIdentity.record_number, RecordRevision.recorded_at.desc()
            )
            .offset(offset)
            .limit(limit)
            .all()
        )
        return [_record_dict(row, identity) for row, identity in rows], total

    def get_record(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        revision: int | None = None,
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        query = session.query(RecordRevision).filter(
            RecordRevision.tenant_id == tenant_id,
            RecordRevision.record_id == record_id,
        )
        if revision is None:
            query = apply_temporal_revision_filter(query, RecordRevision)
        else:
            query = query.filter(RecordRevision.revision == revision)
        row = query.order_by(RecordRevision.recorded_at.desc()).first()
        if row is None:
            raise RecordNotFoundError(
                "Record not found in the selected temporal context."
            )
        identity = session.get(RecordIdentity, row.identity_id)
        if identity is None:
            raise RecordStoreError("Record identity is missing.")
        volumes = _record_volumes(session, tenant_id=tenant_id, record_id=record_id)
        items = _record_items(session, tenant_id=tenant_id, record_id=record_id)
        chronology = _record_chronology(
            session, tenant_id=tenant_id, record_id=record_id
        )
        holds = _record_holds(session, tenant_id=tenant_id, record_id=record_id)
        dispositions = _record_dispositions(
            session, tenant_id=tenant_id, record_id=record_id
        )
        transfer_packages = _record_transfer_packages(
            session, tenant_id=tenant_id, record_id=record_id
        )
        return {
            "record": _record_dict(row, identity),
            "volumes": [_volume_dict(item) for item in volumes],
            "items": [_item_dict(item) for item in items],
            "chronology": [_chronology_dict(item) for item in chronology],
            "holds": [_hold_dict(item) for item in holds],
            "dispositions": [_disposition_dict(item) for item in dispositions],
            "transfer_packages": [
                _transfer_package_dict(item) for item in transfer_packages
            ],
            "access_explanation": {
                "decision": "allowed",
                "reason": "Current tenant and Records permission were evaluated for this read.",
                "purpose": row.purpose,
                "current_authorization": True,
                "access_mode": row.access_mode,
                "limitations": (
                    ["Object-level restricted-record grants are not yet implemented."]
                    if row.access_mode == "restricted"
                    else []
                ),
            },
        }

    def close_record(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_event(
            session,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return self.get_record(
                session,
                principal,
                record_id=record_id,
                revision=replay.record_revision,
            )["record"]
        current = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        _validate_expected(
            current.revision,
            payload.get("expected_revision"),
            label="Record",
        )
        if current.state != "open":
            raise RecordStoreError("Only open records can be closed.")
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at, recorded_at)
        record_class = _current_class(session, tenant_id, current.class_id)
        if record_class is None:
            raise RecordStoreError("The current record class is unavailable.")
        restart_retention = bool(payload.get("restart_retention", False))
        if current.retention_started_at is not None and not restart_retention:
            retention_started_at = current.retention_started_at
            retention_due_at = current.retention_due_at
            retention_rule = dict(current.retention_rule)
        else:
            retention_started_at, retention_due_at, retention_rule = (
                _calculate_retention(
                    record_class,
                    closed_at=recorded_at,
                    explicit_trigger_at=_optional_timestamp(
                        payload.get("retention_trigger_at"),
                        "retention_trigger_at",
                    ),
                )
            )
        state = "retention_running" if retention_due_at is not None else "closed"
        row, identity = _revise_record_state(
            session,
            principal,
            current=current,
            recorded_at=recorded_at,
            values={
                "state": state,
                "closed_at": recorded_at,
                "retention_started_at": retention_started_at,
                "retention_due_at": retention_due_at,
                "retention_rule": retention_rule,
                "appraisal_state": None,
                "appraisal": {},
            },
            snapshot_patch={
                "lifecycle": {
                    "event": "closed",
                    "reason": _text(payload, "reason"),
                    "retention_rule": retention_rule,
                }
            },
        )
        _append_event(
            session,
            principal,
            row=row,
            event_type="record.closed",
            summary=f"Record {identity.record_number} closed",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "reason": _text(payload, "reason"),
                "retention_started_at": _datetime_text(retention_started_at),
                "retention_due_at": _datetime_text(retention_due_at),
                "restart_retention": restart_retention,
            },
        )
        session.flush()
        return _record_dict(row, identity)

    def reopen_record(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_event(
            session,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return self.get_record(
                session,
                principal,
                record_id=record_id,
                revision=replay.record_revision,
            )["record"]
        current = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        _validate_expected(
            current.revision,
            payload.get("expected_revision"),
            label="Record",
        )
        if current.state not in {"closed", "retention_running", "appraised"}:
            raise RecordStoreError(
                "Only closed, retention-running, or appraised records can be reopened."
            )
        if _current_disposition_for_record(session, tenant_id, record_id) is not None:
            raise RecordStoreError(
                "A record with a disposition proposal cannot be reopened."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at, recorded_at)
        row, identity = _revise_record_state(
            session,
            principal,
            current=current,
            recorded_at=recorded_at,
            values={"state": "open", "appraisal_state": None, "appraisal": {}},
            snapshot_patch={
                "lifecycle": {
                    "event": "reopened",
                    "reason": _text(payload, "reason"),
                }
            },
        )
        _append_event(
            session,
            principal,
            row=row,
            event_type="record.reopened",
            summary=f"Record {identity.record_number} reopened",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "reason": _text(payload, "reason"),
                "retention_schedule_preserved": True,
            },
        )
        session.flush()
        return _record_dict(row, identity)

    def appraise_record(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_event(
            session,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return self.get_record(
                session,
                principal,
                record_id=record_id,
                revision=replay.record_revision,
            )["record"]
        current = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        _validate_expected(
            current.revision,
            payload.get("expected_revision"),
            label="Record",
        )
        if current.state not in {"closed", "retention_running"}:
            raise RecordStoreError("Only closed records can be appraised.")
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at, recorded_at)
        override = bool(payload.get("override_retention_not_due", False))
        if (
            current.retention_due_at is not None
            and _aware(recorded_at) < _aware(current.retention_due_at)
            and not override
        ):
            raise RecordStoreError(
                "The retention period has not elapsed; an administrator override is required."
            )
        outcome = _text(payload, "outcome")
        if outcome not in {"retain", "transfer", "destroy", "reclassify"}:
            raise RecordStoreError("Unsupported appraisal outcome.")
        appraisal = {
            "outcome": outcome,
            "reason": _text(payload, "reason"),
            "policy_refs": _text_list(payload.get("policy_refs")),
            "retention_not_due_override": override,
            "appraised_at": _datetime_text(recorded_at),
            "appraised_by": _actor(principal),
        }
        row, identity = _revise_record_state(
            session,
            principal,
            current=current,
            recorded_at=recorded_at,
            values={
                "state": "appraised",
                "appraisal_state": outcome,
                "appraisal": appraisal,
            },
            snapshot_patch={"appraisal": appraisal},
        )
        _append_event(
            session,
            principal,
            row=row,
            event_type="record.appraised",
            summary=f"Record {identity.record_number} appraised for {outcome}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload=appraisal,
        )
        session.flush()
        return _record_dict(row, identity)

    def apply_hold(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_by_key(
            session,
            RecordHoldRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return _hold_dict(replay)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        _validate_expected(
            record.revision,
            payload.get("expected_record_revision"),
            label="Record",
        )
        hold_id = _optional_text(payload.get("hold_id")) or str(uuid.uuid4())
        if _current_hold(session, tenant_id, hold_id) is not None:
            raise RecordConflictError("A current hold already uses this identifier.")
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        effective_from = (
            _optional_timestamp(payload.get("effective_from"), "effective_from")
            or recorded_at
        )
        effective_to = _optional_timestamp(payload.get("effective_to"), "effective_to")
        _validate_interval(effective_from, effective_to)
        row = RecordHoldRevision(
            tenant_id=tenant_id,
            hold_id=hold_id,
            record_id=record_id,
            revision=1,
            status="active",
            reason=_text(payload, "reason"),
            authority=_text(payload, "authority"),
            scope=_mapping(payload.get("scope")),
            effective_from=effective_from,
            effective_to=effective_to,
            policy_refs=_text_list(payload.get("policy_refs")),
            institutional_context={
                **dict(record.institutional_context),
                **_mapping(payload.get("institutional_context")),
            },
            recorded_at=recorded_at,
            changed_by=_actor(principal),
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.hold_applied",
            summary=f"Hold applied: {row.authority}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=row.idempotency_key,
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "hold_id": hold_id,
                "reason": row.reason,
                "authority": row.authority,
                "effective_from": _datetime_text(effective_from),
                "effective_to": _datetime_text(effective_to),
            },
        )
        session.flush()
        return _hold_dict(row)

    def release_hold(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        hold_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(
            {"record_id": record_id, "hold_id": hold_id, **payload}
        )
        replay = _replay_by_key(
            session,
            RecordHoldRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return _hold_dict(replay)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        current = _current_hold(session, tenant_id, hold_id, lock=True)
        if current is None or current.record_id != record_id:
            raise RecordNotFoundError("Record hold not found.")
        _validate_expected(
            current.revision,
            payload.get("expected_hold_revision"),
            label="Record hold",
        )
        if current.status != "active":
            raise RecordStoreError("Only an active hold can be released.")
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at, recorded_at)
        current.superseded_at = recorded_at
        row = RecordHoldRevision(
            tenant_id=tenant_id,
            hold_id=hold_id,
            record_id=record_id,
            revision=current.revision + 1,
            previous_revision_id=current.id,
            status="released",
            reason=current.reason,
            authority=current.authority,
            scope={
                **dict(current.scope),
                "release_reason": _text(payload, "reason"),
            },
            effective_from=current.effective_from,
            effective_to=current.effective_to or recorded_at,
            released_at=recorded_at,
            policy_refs=list(current.policy_refs),
            institutional_context=dict(current.institutional_context),
            recorded_at=recorded_at,
            changed_by=_actor(principal),
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.hold_released",
            summary=f"Hold released: {row.authority}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=row.idempotency_key,
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "hold_id": hold_id,
                "reason": _text(payload, "reason"),
                "released_revision": row.revision,
            },
        )
        session.flush()
        return _hold_dict(row)

    def propose_disposition(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_by_key(
            session,
            RecordDispositionRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return _disposition_dict(replay)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        _validate_expected(
            record.revision,
            payload.get("expected_record_revision"),
            label="Record",
        )
        if record.state != "appraised":
            raise RecordStoreError(
                "A disposition proposal requires an appraised record."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _require_no_active_hold(session, tenant_id, record_id, at=recorded_at)
        if _current_disposition_for_record(session, tenant_id, record_id) is not None:
            raise RecordConflictError(
                "The record already has a current disposition proposal."
            )
        action = _text(payload, "action")
        if action not in {"retain", "transfer", "destroy", "reclassify"}:
            raise RecordStoreError("Unsupported disposition action.")
        if record.appraisal_state != action:
            raise RecordStoreError(
                "The disposition action must match the current appraisal outcome."
            )
        disposition_id = _optional_text(payload.get("disposition_id")) or str(
            uuid.uuid4()
        )
        evidence = _record_evidence_manifest(
            session, tenant_id=tenant_id, record=record
        )
        subject_sha256 = _request_hash(evidence)
        policy_refs = _text_list(payload.get("policy_refs"))
        approval_request_id: str | None = None
        status = "review_unavailable"
        approval_provider = self._tenant_capability(
            CAPABILITY_APPROVAL_REQUESTS, session, tenant_id
        )
        if isinstance(approval_provider, ApprovalRequestProvider):
            approval = approval_provider.create_request(
                session,
                principal,
                command=ApprovalRequestCreateCommand(
                    title=f"Disposition of record {evidence['record_number']}",
                    description=_text(payload, "reason"),
                    subject_module="records",
                    subject_type="record_disposition",
                    subject_id=disposition_id,
                    subject_version=str(record.revision),
                    subject_digest=subject_sha256,
                    steps=(
                        ApprovalStepDefinition(
                            key="records-disposition-review",
                            label="Independent disposition review",
                            selectors=(
                                ApprovalActorSelector(
                                    kind="any_account",
                                    value="*",
                                    label="Authorized Records reviewer",
                                ),
                            ),
                            required_approvals=1,
                            forbidden_evidence_roles=("proposer",),
                        ),
                    ),
                    separation_of_duties=True,
                    policy_refs=tuple(policy_refs),
                    evidence_actors={"proposer": ((_actor(principal) or "unknown"),)},
                    metadata={
                        "record_id": record_id,
                        "record_revision": record.revision,
                        "action": action,
                    },
                ),
                idempotency_key=f"records-disposition:{_text(payload, 'idempotency_key')}",
            )
            approval_request_id = approval.id
            status = "review_pending"
        consequence_preview = _disposition_consequence_preview(
            action=action,
            record=record,
            evidence=evidence,
            approvals_available=approval_request_id is not None,
        )
        row = RecordDispositionRevision(
            tenant_id=tenant_id,
            disposition_id=disposition_id,
            record_id=record_id,
            revision=1,
            action=action,
            status=status,
            reason=_text(payload, "reason"),
            subject_revision=record.revision,
            subject_sha256=subject_sha256,
            consequence_preview=consequence_preview,
            policy_refs=policy_refs,
            approval_request_id=approval_request_id,
            proposed_by=_actor(principal),
            institutional_context={
                **dict(record.institutional_context),
                **_mapping(payload.get("institutional_context")),
            },
            recorded_at=recorded_at,
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.disposition_proposed",
            summary=f"Disposition proposed: {action}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=row.idempotency_key,
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "disposition_id": disposition_id,
                "action": action,
                "subject_sha256": subject_sha256,
                "approval_request_id": approval_request_id,
                "consequence_preview": consequence_preview,
            },
        )
        session.flush()
        return _disposition_dict(row)

    def finalize_disposition(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        disposition_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(
            {"record_id": record_id, "disposition_id": disposition_id, **payload}
        )
        replay = _replay_by_key(
            session,
            RecordDispositionRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            record = _required_current_record(
                session, tenant_id=tenant_id, record_id=record_id
            )
            identity = session.get(RecordIdentity, record.identity_id)
            if identity is None:
                raise RecordStoreError("Record identity is missing.")
            return {
                "disposition": _disposition_dict(replay),
                "record": _record_dict(record, identity),
            }
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        current = _current_disposition(session, tenant_id, disposition_id, lock=True)
        if current is None or current.record_id != record_id:
            raise RecordNotFoundError("Record disposition not found.")
        _validate_expected(
            current.revision,
            payload.get("expected_disposition_revision"),
            label="Record disposition",
        )
        if current.status != "review_pending":
            raise RecordStoreError(
                "The disposition is not awaiting an available approval."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _require_no_active_hold(session, tenant_id, record_id, at=recorded_at)
        evidence = _record_evidence_manifest(
            session, tenant_id=tenant_id, record=record
        )
        if _request_hash(evidence) != current.subject_sha256:
            raise RecordConflictError(
                "The record evidence changed after the disposition was proposed."
            )
        approval_provider = self._tenant_capability(
            CAPABILITY_APPROVAL_REQUESTS, session, tenant_id
        )
        if not isinstance(approval_provider, ApprovalRequestProvider):
            raise RecordSourceUnavailableError(
                "The Approvals module is required to finalize a disposition."
            )
        if not current.approval_request_id:
            raise RecordStoreError("The disposition has no approval request.")
        approval = approval_provider.check_approved(
            session,
            principal,
            request_id=current.approval_request_id,
            subject_module="records",
            subject_type="record_disposition",
            subject_id=disposition_id,
            subject_version=str(current.subject_revision),
            subject_digest=current.subject_sha256,
        )
        if not approval.approved:
            raise RecordStoreError("The disposition has not been approved.")
        current.superseded_at = recorded_at
        disposition = RecordDispositionRevision(
            tenant_id=tenant_id,
            disposition_id=disposition_id,
            record_id=record_id,
            revision=current.revision + 1,
            previous_revision_id=current.id,
            action=current.action,
            status="approved",
            reason=current.reason,
            subject_revision=current.subject_revision,
            subject_sha256=current.subject_sha256,
            consequence_preview=dict(current.consequence_preview),
            policy_refs=list(current.policy_refs),
            approval_request_id=current.approval_request_id,
            proposed_by=current.proposed_by,
            reviewed_by=_actor(principal),
            reviewed_at=recorded_at,
            institutional_context=dict(current.institutional_context),
            recorded_at=recorded_at,
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(disposition)
        state = {
            "retain": "closed",
            "transfer": "transfer_pending",
            "destroy": "destruction_pending",
            "reclassify": "open",
        }[current.action]
        revised, identity = _revise_record_state(
            session,
            principal,
            current=record,
            recorded_at=recorded_at,
            values={"state": state},
            snapshot_patch={
                "disposition": {
                    "id": disposition_id,
                    "action": current.action,
                    "status": "approved",
                    "subject_sha256": current.subject_sha256,
                }
            },
        )
        _append_event(
            session,
            principal,
            row=revised,
            event_type="record.disposition_approved",
            summary=f"Disposition approved: {current.action}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=disposition.idempotency_key,
            request_hash=request_hash,
            institutional_context=disposition.institutional_context,
            payload={
                "disposition_id": disposition_id,
                "approval_request_id": current.approval_request_id,
                "subject_sha256": current.subject_sha256,
                "record_state": state,
            },
        )
        session.flush()
        return {
            "disposition": _disposition_dict(disposition),
            "record": _record_dict(revised, identity),
        }

    def withdraw_disposition(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        disposition_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(
            {"record_id": record_id, "disposition_id": disposition_id, **payload}
        )
        replay = _replay_by_key(
            session,
            RecordDispositionRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return _disposition_dict(replay)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        current = _current_disposition(session, tenant_id, disposition_id, lock=True)
        if current is None or current.record_id != record_id:
            raise RecordNotFoundError("Record disposition not found.")
        _validate_expected(
            current.revision,
            payload.get("expected_disposition_revision"),
            label="Record disposition",
        )
        if current.status not in {"review_pending", "review_unavailable"}:
            raise RecordStoreError(
                "Only an unapproved disposition proposal can be withdrawn."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _validate_later_revision(current.recorded_at, recorded_at)
        current.superseded_at = recorded_at
        row = RecordDispositionRevision(
            tenant_id=tenant_id,
            disposition_id=disposition_id,
            record_id=record_id,
            revision=current.revision + 1,
            previous_revision_id=current.id,
            action=current.action,
            status="withdrawn",
            reason=current.reason,
            subject_revision=current.subject_revision,
            subject_sha256=current.subject_sha256,
            consequence_preview={
                **dict(current.consequence_preview),
                "withdrawal_reason": _text(payload, "reason"),
            },
            policy_refs=list(current.policy_refs),
            approval_request_id=current.approval_request_id,
            proposed_by=current.proposed_by,
            reviewed_by=_actor(principal),
            reviewed_at=recorded_at,
            institutional_context=dict(current.institutional_context),
            recorded_at=recorded_at,
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.disposition_withdrawn",
            summary=f"Disposition withdrawn: {current.action}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=row.idempotency_key,
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "disposition_id": disposition_id,
                "reason": _text(payload, "reason"),
                "superseded_revision": current.revision,
            },
        )
        session.flush()
        return _disposition_dict(row)

    def archive_providers(
        self, session: Session, principal: object
    ) -> list[dict[str, Any]]:
        tenant_id = _tenant(principal)
        providers: list[dict[str, Any]] = []
        for capability_name in record_archive_capabilities(self.registry):
            provider = self._tenant_capability(capability_name, session, tenant_id)
            if not isinstance(provider, RecordArchiveProvider):
                continue
            state = provider.state()
            providers.append(
                {
                    "id": state.provider_id,
                    "label": state.label,
                    "profiles": list(state.profiles),
                    "authority_modes": list(state.authority_modes),
                    "healthy": state.healthy,
                    "checked_at": _datetime_text(state.checked_at),
                    "last_success_at": _datetime_text(state.last_success_at),
                    "freshness_seconds": state.freshness_seconds,
                    "limitations": list(state.limitations),
                    "simulated": state.simulated,
                }
            )
        return providers

    def recovery_status(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id
        )
        identity = session.get(RecordIdentity, record.identity_id)
        if identity is None:
            raise RecordStoreError("Record identity is missing.")
        source_checks: list[dict[str, Any]] = []
        for item in (
            session.query(RecordItem)
            .filter(
                RecordItem.tenant_id == tenant_id,
                RecordItem.record_id == record_id,
            )
            .order_by(RecordItem.sequence)
            .all()
        ):
            check = {
                "item_id": item.id,
                "source_module": item.source_module,
                "resource_type": item.resource_type,
                "resource_id": item.resource_id,
                "source_revision": item.source_revision,
                "stored_content_sha256": item.content_sha256,
                "status": "unavailable",
                "error": None,
            }
            provider = self._tenant_capability(
                record_source_capability(item.source_module), session, tenant_id
            )
            if not isinstance(provider, RecordSourceProvider):
                check["error"] = "The source provider is not enabled."
                source_checks.append(check)
                continue
            try:
                reference = provider.resolve(
                    session,
                    principal,
                    locator=RecordSourceLocator(
                        tenant_id=tenant_id,
                        source_module=item.source_module,
                        resource_type=item.resource_type,
                        resource_id=item.resource_id,
                        source_revision=item.source_revision,
                        metadata=dict(item.source_metadata),
                    ),
                    purpose=item.purpose,
                )
            except Exception as exc:
                check["error"] = f"Source revalidation failed ({type(exc).__name__})."
                source_checks.append(check)
                continue
            resolved_digest = (reference.content_sha256 or "").removeprefix(
                "sha256:"
            ) or None
            check["resolved_content_sha256"] = resolved_digest
            if item.content_sha256 and resolved_digest != item.content_sha256:
                check["status"] = "digest_mismatch"
                check["error"] = (
                    "The source digest no longer matches the filed evidence."
                )
            else:
                check["status"] = "verified"
            source_checks.append(check)

        package_checks = []
        for package in _record_transfer_packages(
            session, tenant_id=tenant_id, record_id=record_id
        ):
            calculated = _request_hash(dict(package.manifest))
            package_checks.append(
                {
                    "package_id": package.package_id,
                    "revision": package.revision,
                    "status": package.status,
                    "simulated": package.simulated,
                    "stored_manifest_sha256": package.manifest_sha256,
                    "calculated_manifest_sha256": calculated,
                    "manifest_verified": calculated == package.manifest_sha256,
                    "receipt_sha256": package.receipt_sha256,
                    "recovery_operation_id": package.recovery_operation_id,
                }
            )

        operations = (
            session.query(RecoveryOperation)
            .filter(
                RecoveryOperation.module_id == "records",
                RecoveryOperation.resource_type == "record",
                RecoveryOperation.resource_id.in_((record_id, identity.record_number)),
            )
            .order_by(RecoveryOperation.created_at)
            .all()
        )
        operation_checks = [
            {
                "operation_id": operation.id,
                "operation_type": operation.operation_type,
                "status": operation.status,
                "mode": operation.mode,
                "checkpoint_count": operation.checkpoint_count,
                "evidence_head_sha256": operation.evidence_head_sha256,
                "evidence_chain_verified": verify_recovery_evidence_chain(
                    session, operation.id
                ),
                "completed_at": _datetime_text(operation.completed_at),
            }
            for operation in operations
        ]
        evidence = _record_evidence_manifest(
            session, tenant_id=tenant_id, record=record
        )
        failures = [
            check for check in source_checks if check["status"] not in {"verified"}
        ]
        failures.extend(
            check for check in package_checks if not check["manifest_verified"]
        )
        failures.extend(
            check for check in operation_checks if not check["evidence_chain_verified"]
        )
        return {
            "record_id": record_id,
            "record_revision": record.revision,
            "evidence_sha256": _request_hash(evidence),
            "healthy": not failures,
            "source_checks": source_checks,
            "package_checks": package_checks,
            "recovery_operations": operation_checks,
            "failure_count": len(failures),
            "limitations": [
                "Source checks use current authorization and never bypass the owning module.",
                "A verified simulation receipt does not prove archival custody.",
            ],
        }

    def prepare_transfer_package(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_by_key(
            session,
            RecordTransferPackageRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return _transfer_package_dict(replay)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        _validate_expected(
            record.revision,
            payload.get("expected_record_revision"),
            label="Record",
        )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _require_no_active_hold(session, tenant_id, record_id, at=recorded_at)
        disposition_id = _text(payload, "disposition_id")
        disposition = _current_disposition(session, tenant_id, disposition_id)
        if (
            disposition is None
            or disposition.record_id != record_id
            or disposition.action != "transfer"
            or disposition.status != "approved"
        ):
            raise RecordStoreError(
                "An approved transfer disposition is required before packaging."
            )
        provider_id = _text(payload, "provider_id")
        provider = self._tenant_capability(
            record_archive_capability(provider_id), session, tenant_id
        )
        if not isinstance(provider, RecordArchiveProvider):
            raise RecordSourceUnavailableError(
                "The selected record archive provider is unavailable."
            )
        provider_state = provider.state()
        profile = _text(payload, "profile")
        if not provider_state.healthy or profile not in provider_state.profiles:
            raise RecordSourceUnavailableError(
                "The selected archive profile is unavailable or unhealthy."
            )
        package_id = _optional_text(payload.get("package_id")) or str(uuid.uuid4())
        manifest = {
            "format": "govoplan-record-transfer-manifest",
            "version": 1,
            "package_id": package_id,
            "provider_id": provider_id,
            "profile": profile,
            "prepared_at": _datetime_text(recorded_at),
            "record": _record_evidence_manifest(
                session, tenant_id=tenant_id, record=record
            ),
            "disposition": _disposition_dict(disposition),
            "institutional_context": dict(record.institutional_context),
        }
        manifest_sha256 = _request_hash(manifest)
        row = RecordTransferPackageRevision(
            tenant_id=tenant_id,
            package_id=package_id,
            record_id=record_id,
            disposition_id=disposition_id,
            revision=1,
            record_revision=record.revision,
            provider_id=provider_id,
            profile=profile,
            status="prepared",
            authority_mode=provider_state.authority_modes[0],
            manifest=manifest,
            manifest_sha256=manifest_sha256,
            receipt={},
            recovery_operation_id=current_record_recovery_operation(),
            simulated=provider_state.simulated,
            institutional_context=dict(record.institutional_context),
            recorded_at=recorded_at,
            changed_by=_actor(principal),
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.transfer_package_prepared",
            summary=f"Transfer package prepared for {provider_state.label}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=row.idempotency_key,
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "package_id": package_id,
                "provider_id": provider_id,
                "profile": profile,
                "manifest_sha256": manifest_sha256,
                "simulated": provider_state.simulated,
            },
        )
        session.flush()
        return _transfer_package_dict(row)

    def dispatch_transfer_package(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        package_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash(
            {"record_id": record_id, "package_id": package_id, **payload}
        )
        replay = _replay_by_key(
            session,
            RecordTransferPackageRevision,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            return _transfer_package_dict(replay)
        record = _required_current_record(
            session, tenant_id=tenant_id, record_id=record_id, lock=True
        )
        current = _current_transfer_package(session, tenant_id, package_id, lock=True)
        if current is None or current.record_id != record_id:
            raise RecordNotFoundError("Record transfer package not found.")
        _validate_expected(
            current.revision,
            payload.get("expected_package_revision"),
            label="Record transfer package",
        )
        if current.status != "prepared":
            raise RecordStoreError(
                "Only a prepared package can be dispatched; unknown outcomes require reconciliation."
            )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        _require_no_active_hold(session, tenant_id, record_id, at=recorded_at)
        provider = self._tenant_capability(
            record_archive_capability(current.provider_id), session, tenant_id
        )
        if not isinstance(provider, RecordArchiveProvider):
            raise RecordSourceUnavailableError(
                "The record archive provider is unavailable."
            )
        if not provider.state().simulated:
            raise RecordSourceUnavailableError(
                "Real archive dispatch requires a configured target-specific recovery profile."
            )
        try:
            receipt = provider.dispatch(
                session,
                principal,
                request=RecordArchiveTransferRequest(
                    package=RecordTransferPackage(
                        tenant_id=tenant_id,
                        package_id=package_id,
                        record_id=record_id,
                        record_revision=current.record_revision,
                        profile=current.profile,
                        manifest_sha256=current.manifest_sha256,
                        manifest=dict(current.manifest),
                    ),
                    purpose=_text(payload, "purpose"),
                    idempotency_key=_text(payload, "idempotency_key"),
                    institutional_context=dict(current.institutional_context),
                ),
            )
        except RecordContractError as exc:
            raise RecordStoreError(str(exc)) from exc
        if (
            receipt.package_id != package_id
            or receipt.provider_id != current.provider_id
        ):
            raise RecordStoreError(
                "The archive provider returned a receipt for another package."
            )
        if not receipt.simulated:
            raise RecordStoreError(
                "The simulation provider returned a receipt that could be mistaken for a real custody transfer."
            )
        current.superseded_at = recorded_at
        status = {
            "accepted": ("simulated_accepted" if receipt.simulated else "accepted"),
            "rejected": "rejected",
            "outcome_unknown": "outcome_unknown",
        }[receipt.outcome]
        receipt_data = {
            "provider_id": receipt.provider_id,
            "package_id": receipt.package_id,
            "outcome": receipt.outcome,
            "observed_at": _datetime_text(receipt.observed_at),
            "external_reference": receipt.external_reference,
            "retry_safe": receipt.retry_safe,
            "simulated": receipt.simulated,
            "metadata": dict(receipt.metadata),
        }
        row = RecordTransferPackageRevision(
            tenant_id=tenant_id,
            package_id=package_id,
            record_id=record_id,
            disposition_id=current.disposition_id,
            revision=current.revision + 1,
            previous_revision_id=current.id,
            record_revision=current.record_revision,
            provider_id=current.provider_id,
            profile=current.profile,
            status=status,
            authority_mode=current.authority_mode,
            manifest=dict(current.manifest),
            manifest_sha256=current.manifest_sha256,
            receipt=receipt_data,
            receipt_sha256=receipt.receipt_sha256.removeprefix("sha256:"),
            external_reference=receipt.external_reference,
            recovery_operation_id=current_record_recovery_operation()
            or current.recovery_operation_id,
            simulated=receipt.simulated,
            institutional_context=dict(current.institutional_context),
            recorded_at=recorded_at,
            changed_by=_actor(principal),
            idempotency_key=_text(payload, "idempotency_key"),
            request_sha256=request_hash,
        )
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.transfer_package_dispatched",
            summary=f"Transfer package result: {status}",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=row.idempotency_key,
            request_hash=request_hash,
            institutional_context=row.institutional_context,
            payload={
                "package_id": package_id,
                "provider_id": current.provider_id,
                "status": status,
                "receipt_sha256": row.receipt_sha256,
                "simulated": row.simulated,
                "external_reference": row.external_reference,
            },
        )
        session.flush()
        return _transfer_package_dict(row)

    def create_volume(
        self,
        session: Session,
        principal: object,
        *,
        record_id: str,
        payload: Mapping[str, object],
    ) -> dict[str, Any]:
        tenant_id = _tenant(principal)
        request_hash = _request_hash({"record_id": record_id, **payload})
        replay = _replay_event(
            session,
            tenant_id=tenant_id,
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
        )
        if replay is not None:
            volume_id = str(replay.payload.get("volume_id") or "")
            row = _current_volume(session, tenant_id, volume_id)
            if row is None:
                raise RecordStoreError("Replayed record volume is missing.")
            return _volume_dict(row)
        record = _current_record(session, tenant_id, record_id, lock=True)
        if record is None:
            raise RecordNotFoundError("Record not found.")
        sequence = (
            int(
                session.query(func.max(RecordVolumeRevision.sequence))
                .filter(
                    RecordVolumeRevision.tenant_id == tenant_id,
                    RecordVolumeRevision.record_id == record_id,
                    RecordVolumeRevision.superseded_at.is_(None),
                )
                .scalar()
                or 0
            )
            + 1
        )
        recorded_at = _timestamp(payload.get("recorded_at"), "recorded_at")
        row = RecordVolumeRevision(
            tenant_id=tenant_id,
            volume_id=_optional_text(payload.get("volume_id")) or str(uuid.uuid4()),
            record_id=record_id,
            revision=1,
            sequence=sequence,
            label=_text(payload, "label"),
            state="open",
            valid_from=_optional_timestamp(payload.get("valid_from"), "valid_from")
            or recorded_at,
            valid_to=_optional_timestamp(payload.get("valid_to"), "valid_to"),
            recorded_at=recorded_at,
            changed_by=_actor(principal),
        )
        _validate_interval(row.valid_from, row.valid_to)
        session.add(row)
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.volume_created",
            summary=f"Volume {row.label} created",
            occurred_at=recorded_at,
            purpose=_text(payload, "purpose"),
            idempotency_key=_text(payload, "idempotency_key"),
            request_hash=request_hash,
            institutional_context=record.institutional_context,
            payload={"volume_id": row.volume_id, "sequence": sequence},
        )
        session.flush()
        return _volume_dict(row)

    def file(
        self,
        session: object,
        principal: object,
        *,
        request: RecordFilingRequest,
    ) -> RecordFilingResult:
        if not isinstance(session, Session):
            raise RecordStoreError("Record filing requires a database session.")
        tenant_id = _tenant(principal)
        if request.tenant_id != tenant_id:
            raise RecordStoreError("Record filing cannot cross tenants.")
        request_hash = _request_hash(_filing_request_mapping(request))
        replay = (
            session.query(RecordItem)
            .filter(
                RecordItem.tenant_id == tenant_id,
                RecordItem.idempotency_key == request.idempotency_key,
            )
            .one_or_none()
        )
        if replay is not None:
            _verify_replay(replay.request_sha256, request_hash)
            return _filing_result(replay, replayed=True)
        record = _current_record(session, tenant_id, request.record_id, lock=True)
        if record is None:
            raise RecordNotFoundError("Record not found.")
        if record.state != "open":
            raise RecordStoreError("Only open records accept new items.")
        record_class = _current_class(session, tenant_id, record.class_id)
        if record_class is None:
            raise RecordStoreError("The current record class is unavailable.")
        if request.volume_id:
            volume = _current_volume(session, tenant_id, request.volume_id)
            if volume is None or volume.record_id != request.record_id:
                raise RecordStoreError(
                    "The selected record volume does not belong to this record."
                )
        reference = self._resolve_source(session, principal, request=request)
        source_key = (
            f"{reference.locator.source_module}:{reference.locator.resource_type}"
        )
        if record_class.allowed_source_types and (
            source_key not in record_class.allowed_source_types
            and reference.locator.resource_type not in record_class.allowed_source_types
        ):
            raise RecordStoreError("The record class does not permit this source type.")
        sequence = (
            int(
                session.query(func.max(RecordItem.sequence))
                .filter(
                    RecordItem.tenant_id == tenant_id,
                    RecordItem.record_id == request.record_id,
                )
                .scalar()
                or 0
            )
            + 1
        )
        filed_at = datetime.now(UTC)
        row = RecordItem(
            tenant_id=tenant_id,
            record_id=request.record_id,
            volume_id=request.volume_id,
            sequence=sequence,
            source_module=reference.locator.source_module,
            resource_type=reference.locator.resource_type,
            resource_id=reference.locator.resource_id,
            source_revision=reference.locator.source_revision,
            label=reference.label,
            relationship=request.relationship,
            filing_reason=request.filing_reason,
            purpose=request.purpose,
            authority_mode=reference.authority_mode,
            content_sha256=(reference.content_sha256 or "").removeprefix("sha256:")
            or None,
            content_type=reference.content_type,
            size_bytes=reference.size_bytes,
            source_valid_from=reference.valid_from,
            source_valid_to=reference.valid_to,
            source_recorded_at=reference.recorded_at,
            launch_url=reference.launch_url,
            filed_at=filed_at,
            filed_by=_actor(principal),
            actor_assignment_id=_actor_assignment(principal),
            actor_delegation_id=_actor_delegation(principal),
            institutional_context=dict(request.institutional_context),
            source_metadata=dict(reference.metadata),
            filing_metadata=dict(request.metadata),
            idempotency_key=request.idempotency_key,
            request_sha256=request_hash,
        )
        session.add(row)
        session.flush()
        _append_event(
            session,
            principal,
            row=record,
            event_type="record.item_filed",
            summary=f"{reference.label} filed",
            occurred_at=filed_at,
            purpose=request.purpose,
            idempotency_key=request.idempotency_key,
            request_hash=request_hash,
            institutional_context=dict(request.institutional_context),
            payload={
                "item_id": row.id,
                "sequence": sequence,
                "source_module": row.source_module,
                "resource_type": row.resource_type,
                "resource_id": row.resource_id,
                "source_revision": row.source_revision,
            },
        )
        session.flush()
        return _filing_result(row, replayed=False)

    def source_providers(
        self,
        session: Session,
        principal: object,
    ) -> list[dict[str, object]]:
        tenant_id = _tenant(principal)
        providers: list[dict[str, object]] = []
        for capability_name in record_source_capabilities(self.registry):
            provider = self._tenant_capability(capability_name, session, tenant_id)
            if not isinstance(provider, RecordSourceProvider):
                continue
            providers.append(
                {
                    "id": provider.provider_id,
                    "source_module": capability_name.removeprefix("records.source."),
                    "resource_types": list(provider.resource_types()),
                }
            )
        return providers

    def _resolve_source(
        self,
        session: Session,
        principal: object,
        *,
        request: RecordFilingRequest,
    ) -> RecordSourceReference:
        capability_name = record_source_capability(request.source.source_module)
        provider = self._tenant_capability(capability_name, session, request.tenant_id)
        if not isinstance(provider, RecordSourceProvider):
            raise RecordSourceUnavailableError(
                f"No enabled record source provider is available for {request.source.source_module}."
            )
        if request.source.resource_type not in provider.resource_types():
            raise RecordStoreError(
                "The record source provider does not support this resource type."
            )
        try:
            reference = provider.resolve(
                session,
                principal,
                locator=request.source,
                purpose=request.purpose,
            )
        except RecordContractError as exc:
            raise RecordStoreError(str(exc)) from exc
        if reference.locator != request.source:
            raise RecordStoreError(
                "The source provider returned a different source revision."
            )
        return reference

    def _tenant_capability(
        self, name: str, session: Session, tenant_id: str
    ) -> object | None:
        if self.registry is None:
            return None
        try:
            if hasattr(self.registry, "tenant_capability"):
                return self.registry.tenant_capability(
                    name, session, tenant_id=tenant_id
                )
            if hasattr(self.registry, "capability"):
                return self.registry.capability(name)
        except Exception as exc:
            raise RecordSourceUnavailableError(
                f"Record capability {name} is unavailable: {exc}"
            ) from exc
        return None


def _current_file_plan(
    session: Session, tenant_id: str, node_id: str, *, lock: bool = False
) -> RecordFilePlanRevision | None:
    query = session.query(RecordFilePlanRevision).filter(
        RecordFilePlanRevision.tenant_id == tenant_id,
        RecordFilePlanRevision.node_id == node_id,
        RecordFilePlanRevision.superseded_at.is_(None),
    )
    return query.with_for_update().one_or_none() if lock else query.one_or_none()


def _validate_file_plan_parent(
    session: Session,
    *,
    tenant_id: str,
    node_id: str,
    parent_node_id: str | None,
) -> None:
    current_id = parent_node_id
    visited: set[str] = set()
    while current_id is not None:
        if current_id == node_id or current_id in visited:
            raise RecordStoreError("The file-plan parent would create a cycle.")
        visited.add(current_id)
        parent = _current_file_plan(session, tenant_id, current_id)
        if parent is None:
            raise RecordStoreError("The parent file-plan node does not exist.")
        current_id = parent.parent_node_id


def _current_class(
    session: Session, tenant_id: str, class_id: str, *, lock: bool = False
) -> RecordClassRevision | None:
    query = session.query(RecordClassRevision).filter(
        RecordClassRevision.tenant_id == tenant_id,
        RecordClassRevision.class_id == class_id,
        RecordClassRevision.superseded_at.is_(None),
    )
    return query.with_for_update().one_or_none() if lock else query.one_or_none()


def _current_record(
    session: Session, tenant_id: str, record_id: str, *, lock: bool = False
) -> RecordRevision | None:
    query = session.query(RecordRevision).filter(
        RecordRevision.tenant_id == tenant_id,
        RecordRevision.record_id == record_id,
        RecordRevision.superseded_at.is_(None),
    )
    return query.with_for_update().one_or_none() if lock else query.one_or_none()


def _required_current_record(
    session: Session,
    *,
    tenant_id: str,
    record_id: str,
    lock: bool = False,
) -> RecordRevision:
    row = _current_record(session, tenant_id, record_id, lock=lock)
    if row is None:
        raise RecordNotFoundError("Record not found.")
    return row


def _current_hold(
    session: Session,
    tenant_id: str,
    hold_id: str,
    *,
    lock: bool = False,
) -> RecordHoldRevision | None:
    query = session.query(RecordHoldRevision).filter(
        RecordHoldRevision.tenant_id == tenant_id,
        RecordHoldRevision.hold_id == hold_id,
        RecordHoldRevision.superseded_at.is_(None),
    )
    return query.with_for_update().one_or_none() if lock else query.one_or_none()


def _current_disposition(
    session: Session,
    tenant_id: str,
    disposition_id: str,
    *,
    lock: bool = False,
) -> RecordDispositionRevision | None:
    query = session.query(RecordDispositionRevision).filter(
        RecordDispositionRevision.tenant_id == tenant_id,
        RecordDispositionRevision.disposition_id == disposition_id,
        RecordDispositionRevision.superseded_at.is_(None),
    )
    return query.with_for_update().one_or_none() if lock else query.one_or_none()


def _current_disposition_for_record(
    session: Session, tenant_id: str, record_id: str
) -> RecordDispositionRevision | None:
    return (
        session.query(RecordDispositionRevision)
        .filter(
            RecordDispositionRevision.tenant_id == tenant_id,
            RecordDispositionRevision.record_id == record_id,
            RecordDispositionRevision.superseded_at.is_(None),
            RecordDispositionRevision.status != "withdrawn",
        )
        .one_or_none()
    )


def _current_transfer_package(
    session: Session,
    tenant_id: str,
    package_id: str,
    *,
    lock: bool = False,
) -> RecordTransferPackageRevision | None:
    query = session.query(RecordTransferPackageRevision).filter(
        RecordTransferPackageRevision.tenant_id == tenant_id,
        RecordTransferPackageRevision.package_id == package_id,
        RecordTransferPackageRevision.superseded_at.is_(None),
    )
    return query.with_for_update().one_or_none() if lock else query.one_or_none()


def _current_volume(
    session: Session, tenant_id: str, volume_id: str
) -> RecordVolumeRevision | None:
    return (
        session.query(RecordVolumeRevision)
        .filter(
            RecordVolumeRevision.tenant_id == tenant_id,
            RecordVolumeRevision.volume_id == volume_id,
            RecordVolumeRevision.superseded_at.is_(None),
        )
        .one_or_none()
    )


def _record_volumes(
    session: Session, *, tenant_id: str, record_id: str
) -> Sequence[RecordVolumeRevision]:
    query = session.query(RecordVolumeRevision).filter(
        RecordVolumeRevision.tenant_id == tenant_id,
        RecordVolumeRevision.record_id == record_id,
    )
    return (
        apply_temporal_revision_filter(query, RecordVolumeRevision)
        .order_by(RecordVolumeRevision.sequence)
        .all()
    )


def _record_items(
    session: Session, *, tenant_id: str, record_id: str
) -> Sequence[RecordItem]:
    context = current_temporal_data_context()
    query = session.query(RecordItem).filter(
        RecordItem.tenant_id == tenant_id,
        RecordItem.record_id == record_id,
    )
    if context.recorded_at is not None:
        query = query.filter(RecordItem.filed_at <= context.recorded_at)
    instant = context.validity_instant
    if instant is not None:
        query = query.filter(
            or_(
                RecordItem.source_valid_from.is_(None),
                RecordItem.source_valid_from <= instant,
            ),
            or_(
                RecordItem.source_valid_to.is_(None),
                RecordItem.source_valid_to > instant,
            ),
        )
    return query.order_by(RecordItem.sequence).all()


def _record_chronology(
    session: Session, *, tenant_id: str, record_id: str
) -> Sequence[RecordChronologyEntry]:
    context = current_temporal_data_context()
    query = session.query(RecordChronologyEntry).filter(
        RecordChronologyEntry.tenant_id == tenant_id,
        RecordChronologyEntry.record_id == record_id,
    )
    if context.recorded_at is not None:
        query = query.filter(RecordChronologyEntry.occurred_at <= context.recorded_at)
    return query.order_by(RecordChronologyEntry.occurred_at.desc()).all()


def _record_holds(
    session: Session, *, tenant_id: str, record_id: str
) -> Sequence[RecordHoldRevision]:
    query = session.query(RecordHoldRevision).filter(
        RecordHoldRevision.tenant_id == tenant_id,
        RecordHoldRevision.record_id == record_id,
    )
    return (
        apply_temporal_revision_filter(
            query,
            RecordHoldRevision,
            valid_from=None,
            valid_to=None,
        )
        .order_by(RecordHoldRevision.recorded_at.desc())
        .all()
    )


def _record_dispositions(
    session: Session, *, tenant_id: str, record_id: str
) -> Sequence[RecordDispositionRevision]:
    query = session.query(RecordDispositionRevision).filter(
        RecordDispositionRevision.tenant_id == tenant_id,
        RecordDispositionRevision.record_id == record_id,
    )
    return (
        apply_temporal_revision_filter(
            query,
            RecordDispositionRevision,
            valid_from=None,
            valid_to=None,
        )
        .order_by(RecordDispositionRevision.recorded_at.desc())
        .all()
    )


def _record_transfer_packages(
    session: Session, *, tenant_id: str, record_id: str
) -> Sequence[RecordTransferPackageRevision]:
    query = session.query(RecordTransferPackageRevision).filter(
        RecordTransferPackageRevision.tenant_id == tenant_id,
        RecordTransferPackageRevision.record_id == record_id,
    )
    return (
        apply_temporal_revision_filter(
            query,
            RecordTransferPackageRevision,
            valid_from=None,
            valid_to=None,
        )
        .order_by(RecordTransferPackageRevision.recorded_at.desc())
        .all()
    )


def _active_holds(
    session: Session,
    *,
    tenant_id: str,
    record_id: str,
    at: datetime,
) -> Sequence[RecordHoldRevision]:
    return (
        session.query(RecordHoldRevision)
        .filter(
            RecordHoldRevision.tenant_id == tenant_id,
            RecordHoldRevision.record_id == record_id,
            RecordHoldRevision.superseded_at.is_(None),
            RecordHoldRevision.status == "active",
            RecordHoldRevision.effective_from <= at,
            or_(
                RecordHoldRevision.effective_to.is_(None),
                RecordHoldRevision.effective_to > at,
            ),
        )
        .order_by(RecordHoldRevision.recorded_at)
        .all()
    )


def _require_no_active_hold(
    session: Session, tenant_id: str, record_id: str, *, at: datetime
) -> None:
    holds = _active_holds(session, tenant_id=tenant_id, record_id=record_id, at=at)
    if holds:
        raise RecordConflictError(
            f"Disposition is blocked by {len(holds)} active record hold(s)."
        )


def _append_event(
    session: Session,
    principal: object,
    *,
    row: RecordRevision,
    event_type: str,
    summary: str,
    occurred_at: datetime,
    purpose: str,
    idempotency_key: str,
    request_hash: str,
    institutional_context: Mapping[str, object],
    payload: Mapping[str, object],
) -> RecordChronologyEntry:
    event = RecordChronologyEntry(
        tenant_id=row.tenant_id,
        record_id=row.record_id,
        event_id=str(uuid.uuid4()),
        event_type=event_type,
        record_revision=row.revision,
        summary=summary[:500],
        occurred_at=occurred_at,
        actor_id=_actor(principal),
        actor_assignment_id=_actor_assignment(principal),
        actor_delegation_id=_actor_delegation(principal),
        purpose=purpose,
        idempotency_key=idempotency_key,
        request_sha256=request_hash,
        institutional_context=dict(institutional_context),
        payload=dict(payload),
    )
    session.add(event)
    return event


def _replay_event(
    session: Session,
    *,
    tenant_id: str,
    idempotency_key: str,
    request_hash: str,
) -> RecordChronologyEntry | None:
    event = (
        session.query(RecordChronologyEntry)
        .filter(
            RecordChronologyEntry.tenant_id == tenant_id,
            RecordChronologyEntry.idempotency_key == idempotency_key,
        )
        .one_or_none()
    )
    if event is not None:
        _verify_replay(event.request_sha256, request_hash)
    return event


def _replay_by_key(
    session: Session,
    model: type,
    *,
    tenant_id: str,
    idempotency_key: str,
    request_hash: str,
):
    row = (
        session.query(model)
        .filter(
            model.tenant_id == tenant_id,
            model.idempotency_key == idempotency_key,
        )
        .one_or_none()
    )
    if row is not None:
        _verify_replay(row.request_sha256, request_hash)
    return row


def _verify_replay(actual_hash: str, requested_hash: str) -> None:
    if actual_hash != requested_hash:
        raise RecordConflictError(
            "The idempotency key was already used with a different request."
        )


def _validate_expected(current: int | None, expected: object, *, label: str) -> None:
    if current is None:
        if expected is not None:
            raise RecordConflictError(
                f"{label} revision conflict: no current revision exists."
            )
        return
    if expected is None or int(expected) != current:
        raise RecordConflictError(
            f"{label} revision conflict: expected {expected}, current revision is {current}."
        )


def _validate_later_revision(previous: datetime | None, current: datetime) -> None:
    if previous is not None and _aware(current) <= _aware(previous):
        raise RecordConflictError(
            "A new revision must be recorded after the current revision."
        )


def _validate_interval(valid_from: datetime | None, valid_to: datetime | None) -> None:
    if valid_from and valid_to and _aware(valid_to) <= _aware(valid_from):
        raise RecordStoreError("valid_to must be after valid_from.")


def _revise_record_state(
    session: Session,
    principal: object,
    *,
    current: RecordRevision,
    recorded_at: datetime,
    values: Mapping[str, object],
    snapshot_patch: Mapping[str, object],
) -> tuple[RecordRevision, RecordIdentity]:
    identity = session.get(RecordIdentity, current.identity_id)
    if identity is None:
        raise RecordStoreError("Record identity is missing.")
    revision_values = _revision_values(current, {})
    revision_values.update(values)
    current.superseded_at = recorded_at
    row = RecordRevision(
        tenant_id=current.tenant_id,
        record_id=current.record_id,
        identity_id=current.identity_id,
        revision=current.revision + 1,
        previous_revision_id=current.id,
        recorded_at=recorded_at,
        changed_by=_actor(principal),
        search_text=current.search_text,
        snapshot={**dict(current.snapshot), **dict(snapshot_patch)},
        **revision_values,
    )
    session.add(row)
    return row, identity


def _calculate_retention(
    record_class: RecordClassRevision,
    *,
    closed_at: datetime,
    explicit_trigger_at: datetime | None,
) -> tuple[datetime | None, datetime | None, dict[str, Any]]:
    period_days = record_class.retention_period_days
    trigger_mode = (record_class.closure_trigger or "record_closed").strip().lower()
    if period_days is None:
        return (
            None,
            None,
            {
                "class_id": record_class.class_id,
                "class_revision": record_class.revision,
                "period_days": None,
                "trigger": trigger_mode,
            },
        )
    if trigger_mode in {"record_closed", "closed", "closure"}:
        started_at = closed_at
    elif trigger_mode in {"calendar_year_end", "year_end"}:
        started_at = closed_at.replace(
            year=closed_at.year + 1,
            month=1,
            day=1,
            hour=0,
            minute=0,
            second=0,
            microsecond=0,
        )
    elif trigger_mode in {"calendar_month_end", "month_end"}:
        if closed_at.month == 12:
            started_at = closed_at.replace(
                year=closed_at.year + 1,
                month=1,
                day=1,
                hour=0,
                minute=0,
                second=0,
                microsecond=0,
            )
        else:
            started_at = closed_at.replace(
                month=closed_at.month + 1,
                day=1,
                hour=0,
                minute=0,
                second=0,
                microsecond=0,
            )
    elif trigger_mode == "explicit":
        if explicit_trigger_at is None:
            raise RecordStoreError(
                "This record class requires an explicit retention trigger date."
            )
        started_at = explicit_trigger_at
    else:
        raise RecordStoreError(
            f"Unsupported record-class closure trigger: {trigger_mode}."
        )
    due_at = started_at + timedelta(days=period_days)
    return (
        started_at,
        due_at,
        {
            "class_id": record_class.class_id,
            "class_revision": record_class.revision,
            "period_days": period_days,
            "trigger": trigger_mode,
            "started_at": _datetime_text(started_at),
            "due_at": _datetime_text(due_at),
        },
    )


def _record_evidence_manifest(
    session: Session, *, tenant_id: str, record: RecordRevision
) -> dict[str, Any]:
    identity = session.get(RecordIdentity, record.identity_id)
    if identity is None:
        raise RecordStoreError("Record identity is missing.")
    record_class = (
        session.query(RecordClassRevision)
        .filter(
            RecordClassRevision.tenant_id == tenant_id,
            RecordClassRevision.class_id == record.class_id,
            RecordClassRevision.revision
            == int(record.retention_rule.get("class_revision") or 0),
        )
        .one_or_none()
    )
    if record_class is None:
        record_class = _current_class(session, tenant_id, record.class_id)
    items = (
        session.query(RecordItem)
        .filter(
            RecordItem.tenant_id == tenant_id,
            RecordItem.record_id == record.record_id,
        )
        .order_by(RecordItem.sequence)
        .all()
    )
    active_holds = _active_holds(
        session,
        tenant_id=tenant_id,
        record_id=record.record_id,
        at=datetime.now(UTC),
    )
    return {
        "record_id": record.record_id,
        "record_number": identity.record_number,
        "record_revision": record.revision,
        "recorded_at": _datetime_text(record.recorded_at),
        "state": record.state,
        "class": (
            {
                "class_id": record_class.class_id,
                "revision": record_class.revision,
                "key": record_class.key,
                "retention_period_days": record_class.retention_period_days,
                "closure_trigger": record_class.closure_trigger,
            }
            if record_class is not None
            else {"class_id": record.class_id, "revision": None}
        ),
        "retention": {
            "closed_at": _datetime_text(record.closed_at),
            "started_at": _datetime_text(record.retention_started_at),
            "due_at": _datetime_text(record.retention_due_at),
            "rule": dict(record.retention_rule),
        },
        "appraisal": dict(record.appraisal),
        "items": [
            {
                "item_id": item.id,
                "sequence": item.sequence,
                "source_module": item.source_module,
                "resource_type": item.resource_type,
                "resource_id": item.resource_id,
                "source_revision": item.source_revision,
                "authority_mode": item.authority_mode,
                "content_sha256": item.content_sha256,
                "filed_at": _datetime_text(item.filed_at),
            }
            for item in items
        ],
        "active_holds": [
            {
                "hold_id": hold.hold_id,
                "revision": hold.revision,
                "authority": hold.authority,
                "effective_from": _datetime_text(hold.effective_from),
                "effective_to": _datetime_text(hold.effective_to),
            }
            for hold in active_holds
        ],
        "institutional_context": dict(record.institutional_context),
    }


def _disposition_consequence_preview(
    *,
    action: str,
    record: RecordRevision,
    evidence: Mapping[str, object],
    approvals_available: bool,
) -> dict[str, Any]:
    descriptions = {
        "retain": "Retain the record under its current classification without an external effect.",
        "transfer": "Prepare an archive-neutral package; dispatch remains a separate governed action.",
        "destroy": "Mark destruction as pending only; no content or source object is deleted by approval.",
        "reclassify": "Return the record to an open state so a governed classification revision can follow.",
    }
    return {
        "action": action,
        "description": descriptions[action],
        "record_revision": record.revision,
        "item_count": len(list(evidence.get("items") or [])),
        "destructive_effect": False,
        "external_effect": False,
        "requires_independent_approval": True,
        "approval_capability_available": approvals_available,
        "limitations": [
            "Approval changes lifecycle state only; destructive and external effects require separate evidence-bound operations."
        ],
    }


def _revision_values(
    current: RecordRevision, payload: Mapping[str, object]
) -> dict[str, Any]:
    def selected(name: str) -> object:
        return payload[name] if name in payload else getattr(current, name)

    return {
        "class_id": str(selected("class_id")),
        "file_plan_node_id": str(selected("file_plan_node_id")),
        "title": str(selected("title")),
        "description": _optional_text(selected("description")),
        "state": str(selected("state")),
        "source_authority_mode": current.source_authority_mode,
        "access_mode": str(selected("access_mode")),
        "purpose": current.purpose,
        "classification": _optional_text(selected("classification")),
        "responsible_unit_id": _optional_text(selected("responsible_unit_id")),
        "responsible_function_id": _optional_text(selected("responsible_function_id")),
        "external_reference": dict(current.external_reference),
        "institutional_context": _mapping(selected("institutional_context")),
        "valid_from": (
            _optional_timestamp(payload.get("valid_from"), "valid_from")
            if "valid_from" in payload
            else current.valid_from
        ),
        "valid_to": (
            _optional_timestamp(payload.get("valid_to"), "valid_to")
            if "valid_to" in payload
            else current.valid_to
        ),
        "closed_at": current.closed_at,
        "retention_started_at": current.retention_started_at,
        "retention_due_at": current.retention_due_at,
        "retention_rule": dict(current.retention_rule),
        "appraisal_state": current.appraisal_state,
        "appraisal": dict(current.appraisal),
    }


def _record_snapshot(payload: Mapping[str, object]) -> dict[str, Any]:
    return _json_mapping(payload)


def _record_search_text(payload: Mapping[str, object], record_number: str) -> str:
    return " ".join(
        value
        for value in (
            record_number,
            str(payload.get("title") or ""),
            str(payload.get("description") or ""),
            str(payload.get("classification") or ""),
        )
        if value
    ).lower()


def _file_plan_dict(row: RecordFilePlanRevision) -> dict[str, Any]:
    return {
        "node_id": row.node_id,
        "revision": row.revision,
        "parent_node_id": row.parent_node_id,
        "code": row.code,
        "label": row.label,
        "description": row.description,
        "active": row.active,
        "valid_from": _datetime_text(row.valid_from),
        "valid_to": _datetime_text(row.valid_to),
        "recorded_at": _datetime_text(row.recorded_at),
        "institutional_context": dict(row.institutional_context),
    }


def _class_dict(row: RecordClassRevision) -> dict[str, Any]:
    return {
        "class_id": row.class_id,
        "revision": row.revision,
        "file_plan_node_id": row.file_plan_node_id,
        "key": row.key,
        "label": row.label,
        "description": row.description,
        "metadata_requirements": list(row.metadata_requirements),
        "allowed_source_types": list(row.allowed_source_types),
        "retention_period_days": row.retention_period_days,
        "closure_trigger": row.closure_trigger,
        "access_mode": row.access_mode,
        "active": row.active,
        "valid_from": _datetime_text(row.valid_from),
        "valid_to": _datetime_text(row.valid_to),
        "recorded_at": _datetime_text(row.recorded_at),
        "institutional_context": dict(row.institutional_context),
    }


def _record_dict(row: RecordRevision, identity: RecordIdentity) -> dict[str, Any]:
    return {
        "reference": {
            "kind": "record",
            "owner_module": "records",
            "object_id": row.record_id,
            "tenant_id": row.tenant_id,
            "version": str(row.revision),
        },
        "record_id": row.record_id,
        "record_number": identity.record_number,
        "revision": row.revision,
        "class_id": row.class_id,
        "file_plan_node_id": row.file_plan_node_id,
        "title": row.title,
        "description": row.description,
        "state": row.state,
        "source_authority_mode": row.source_authority_mode,
        "access_mode": row.access_mode,
        "purpose": row.purpose,
        "classification": row.classification,
        "responsible_unit_id": row.responsible_unit_id,
        "responsible_function_id": row.responsible_function_id,
        "external_reference": dict(row.external_reference),
        "institutional_context": dict(row.institutional_context),
        "valid_from": _datetime_text(row.valid_from),
        "valid_to": _datetime_text(row.valid_to),
        "recorded_at": _datetime_text(row.recorded_at),
        "closed_at": _datetime_text(row.closed_at),
        "retention_started_at": _datetime_text(row.retention_started_at),
        "retention_due_at": _datetime_text(row.retention_due_at),
        "retention_rule": dict(row.retention_rule),
        "appraisal_state": row.appraisal_state,
        "appraisal": dict(row.appraisal),
    }


def _volume_dict(row: RecordVolumeRevision) -> dict[str, Any]:
    return {
        "volume_id": row.volume_id,
        "record_id": row.record_id,
        "revision": row.revision,
        "sequence": row.sequence,
        "label": row.label,
        "state": row.state,
        "valid_from": _datetime_text(row.valid_from),
        "valid_to": _datetime_text(row.valid_to),
        "recorded_at": _datetime_text(row.recorded_at),
    }


def _item_dict(row: RecordItem) -> dict[str, Any]:
    return {
        "item_id": row.id,
        "record_id": row.record_id,
        "volume_id": row.volume_id,
        "sequence": row.sequence,
        "source": {
            "source_module": row.source_module,
            "resource_type": row.resource_type,
            "resource_id": row.resource_id,
            "source_revision": row.source_revision,
        },
        "label": row.label,
        "relationship": row.relationship,
        "filing_reason": row.filing_reason,
        "purpose": row.purpose,
        "authority_mode": row.authority_mode,
        "content_sha256": row.content_sha256,
        "content_type": row.content_type,
        "size_bytes": row.size_bytes,
        "source_valid_from": _datetime_text(row.source_valid_from),
        "source_valid_to": _datetime_text(row.source_valid_to),
        "source_recorded_at": _datetime_text(row.source_recorded_at),
        "launch_url": row.launch_url,
        "filed_at": _datetime_text(row.filed_at),
        "filed_by": row.filed_by,
        "institutional_context": dict(row.institutional_context),
        "source_metadata": dict(row.source_metadata),
        "filing_metadata": dict(row.filing_metadata),
    }


def _chronology_dict(row: RecordChronologyEntry) -> dict[str, Any]:
    return {
        "event_id": row.event_id,
        "event_type": row.event_type,
        "record_revision": row.record_revision,
        "summary": row.summary,
        "occurred_at": _datetime_text(row.occurred_at),
        "actor_id": row.actor_id,
        "purpose": row.purpose,
        "institutional_context": dict(row.institutional_context),
        "payload": dict(row.payload),
    }


def _hold_dict(row: RecordHoldRevision) -> dict[str, Any]:
    return {
        "hold_id": row.hold_id,
        "record_id": row.record_id,
        "revision": row.revision,
        "status": row.status,
        "reason": row.reason,
        "authority": row.authority,
        "scope": dict(row.scope),
        "effective_from": _datetime_text(row.effective_from),
        "effective_to": _datetime_text(row.effective_to),
        "released_at": _datetime_text(row.released_at),
        "policy_refs": list(row.policy_refs),
        "institutional_context": dict(row.institutional_context),
        "recorded_at": _datetime_text(row.recorded_at),
        "changed_by": row.changed_by,
    }


def _disposition_dict(row: RecordDispositionRevision) -> dict[str, Any]:
    return {
        "disposition_id": row.disposition_id,
        "record_id": row.record_id,
        "revision": row.revision,
        "action": row.action,
        "status": row.status,
        "reason": row.reason,
        "subject_revision": row.subject_revision,
        "subject_sha256": row.subject_sha256,
        "consequence_preview": dict(row.consequence_preview),
        "policy_refs": list(row.policy_refs),
        "approval_request_id": row.approval_request_id,
        "proposed_by": row.proposed_by,
        "reviewed_by": row.reviewed_by,
        "reviewed_at": _datetime_text(row.reviewed_at),
        "institutional_context": dict(row.institutional_context),
        "recorded_at": _datetime_text(row.recorded_at),
    }


def _transfer_package_dict(row: RecordTransferPackageRevision) -> dict[str, Any]:
    return {
        "package_id": row.package_id,
        "record_id": row.record_id,
        "disposition_id": row.disposition_id,
        "revision": row.revision,
        "record_revision": row.record_revision,
        "provider_id": row.provider_id,
        "profile": row.profile,
        "status": row.status,
        "authority_mode": row.authority_mode,
        "manifest": dict(row.manifest),
        "manifest_sha256": row.manifest_sha256,
        "receipt": dict(row.receipt),
        "receipt_sha256": row.receipt_sha256,
        "external_reference": row.external_reference,
        "recovery_operation_id": row.recovery_operation_id,
        "simulated": row.simulated,
        "institutional_context": dict(row.institutional_context),
        "recorded_at": _datetime_text(row.recorded_at),
        "changed_by": row.changed_by,
    }


def _filing_request_mapping(request: RecordFilingRequest) -> dict[str, object]:
    return {
        "tenant_id": request.tenant_id,
        "record_id": request.record_id,
        "source": {
            "tenant_id": request.source.tenant_id,
            "source_module": request.source.source_module,
            "resource_type": request.source.resource_type,
            "resource_id": request.source.resource_id,
            "source_revision": request.source.source_revision,
            "metadata": dict(request.source.metadata),
        },
        "purpose": request.purpose,
        "filing_reason": request.filing_reason,
        "idempotency_key": request.idempotency_key,
        "volume_id": request.volume_id,
        "relationship": request.relationship,
        "institutional_context": dict(request.institutional_context),
        "metadata": dict(request.metadata),
    }


def _filing_result(row: RecordItem, *, replayed: bool) -> RecordFilingResult:
    locator = RecordSourceLocator(
        tenant_id=row.tenant_id,
        source_module=row.source_module,
        resource_type=row.resource_type,
        resource_id=row.resource_id,
        source_revision=row.source_revision,
    )
    reference = RecordSourceReference(
        locator=locator,
        label=row.label,
        authority_mode=row.authority_mode,  # type: ignore[arg-type]
        content_sha256=row.content_sha256,
        content_type=row.content_type,
        size_bytes=row.size_bytes,
        valid_from=row.source_valid_from,
        valid_to=row.source_valid_to,
        recorded_at=row.source_recorded_at,
        launch_url=row.launch_url,
        metadata=dict(row.source_metadata),
    )
    return RecordFilingResult(
        record_id=row.record_id,
        item_id=row.id,
        sequence=row.sequence,
        source=reference,
        filed_at=_aware(row.filed_at),
        replayed=replayed,
    )


def _tenant(principal: object) -> str:
    value = _principal_value(principal, "tenant_id")
    if not value:
        raise RecordStoreError("Records operations require a tenant-bound principal.")
    return value


def _actor(principal: object) -> str | None:
    for name in ("account_id", "identity_id", "membership_id"):
        value = _principal_value(principal, name)
        if value:
            return value
    return None


def _actor_assignment(principal: object) -> str | None:
    return _principal_value(principal, "acting_assignment_id") or _principal_value(
        principal, "assignment_id"
    )


def _actor_delegation(principal: object) -> str | None:
    return _principal_value(principal, "acting_delegation_id") or _principal_value(
        principal, "delegation_id"
    )


def _principal_value(principal: object, name: str) -> str | None:
    value = str(getattr(principal, name, "") or "").strip()
    return value or None


def _text(
    payload: Mapping[str, object], name: str, *, default: str | None = None
) -> str:
    value = str(payload.get(name, default) or "").strip()
    if not value:
        raise RecordStoreError(f"{name} is required.")
    return value


def _optional_text(value: object) -> str | None:
    text = str(value or "").strip()
    return text or None


def _text_list(value: object) -> list[str]:
    if value is None:
        return []
    if not isinstance(value, (list, tuple)):
        raise RecordStoreError("Expected a list of text values.")
    return list(dict.fromkeys(str(item).strip() for item in value if str(item).strip()))


def _mapping(value: object) -> dict[str, Any]:
    if value is None:
        return {}
    if not isinstance(value, Mapping):
        raise RecordStoreError("Expected an object value.")
    return _json_mapping(value)


def _json_mapping(value: Mapping[str, object]) -> dict[str, Any]:
    return json.loads(json.dumps(dict(value), default=_json_default))


def _json_default(value: object) -> str:
    if isinstance(value, datetime):
        return _datetime_text(value) or ""
    return str(value)


def _request_hash(payload: Mapping[str, object]) -> str:
    encoded = json.dumps(
        dict(payload), sort_keys=True, separators=(",", ":"), default=_json_default
    )
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def _timestamp(value: object, name: str) -> datetime:
    parsed = _optional_timestamp(value, name)
    if parsed is None:
        raise RecordStoreError(f"{name} is required.")
    return parsed


def _optional_timestamp(value: object, name: str) -> datetime | None:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        parsed = value
    else:
        try:
            parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        except ValueError as exc:
            raise RecordStoreError(f"{name} must be an ISO 8601 timestamp.") from exc
    if parsed.tzinfo is None:
        raise RecordStoreError(f"{name} must include a timezone.")
    return parsed.astimezone(UTC)


def _optional_int(value: object) -> int | None:
    return None if value is None or value == "" else int(value)


def _aware(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


def _datetime_text(value: datetime | None) -> str | None:
    return (
        _aware(value).isoformat().replace("+00:00", "Z") if value is not None else None
    )


__all__ = [
    "RecordConflictError",
    "RecordNotFoundError",
    "RecordSourceUnavailableError",
    "RecordStoreError",
    "SqlRecordRegistry",
]