from __future__ import annotations

import hashlib
import json
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, replace
from datetime import datetime
from typing import Any, cast

from sqlalchemy import exists, func, or_, select, text
from sqlalchemy.orm import Session

from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.datasources import (
    DatasourceAccessError,
    DatasourceArtifactReference,
    DatasourceDescriptor,
    DatasourceError,
    DatasourceField,
    DatasourceGovernance,
    DatasourceMaterialization,
    DatasourceMode,
    DatasourceNotFoundError,
    DatasourceOrigin,
    DatasourceOriginReadRequest,
    DatasourcePublicationRequest,
    DatasourcePublicationResult,
    DatasourcePublicationStatus,
    DatasourceReadRequest,
    DatasourceReadResult,
    DatasourceStage,
    DatasourceStageInput,
    DatasourceUnavailableError,
    DatasourceValidationError,
    DatasourceVisibilityPolicyRequest,
    datasource_origins,
    datasource_visibility_policy_provider,
)
from govoplan_core.db.base import utcnow
from govoplan_datasources.backend.db.models import (
    DatasourceGovernanceReferenceRecord,
    DatasourceMaterializationRecord,
    DatasourcePayloadRecord,
    DatasourcePublicationRecord,
    DatasourceRecord,
    DatasourceStageRecord,
)
from govoplan_datasources.backend.governance import (
    RetentionPlan,
    apply_retention_plan,
    build_retention_plan,
    decide_stage,
    ensure_stage_approval_current,
    initialize_stage_approval,
    list_lifecycle_evidence,
    normalize_approval_policy,
    normalize_retention_policy,
    record_lifecycle_evidence,
)
from govoplan_datasources.backend.payloads import (
    DatasourcePayloadBackend,
    PayloadBackendRegistry,
    create_database_rows_payload,
    create_external_payload_reference,
    payload_for_materialization,
    validate_payload_size,
)
from govoplan_datasources.backend.quality import validate_stage
from govoplan_datasources.backend.tabular import (
    MAX_READ_ROWS,
    MAX_STAGE_ROWS,
    encoded_size,
    field_payload,
    fingerprint_rows,
    infer_schema,
    normalize_rows,
)
from govoplan_datasources.backend.visibility import (
    VisibilityPlan,
    apply_visibility,
    normalize_visibility_policy,
    visibility_diagnostics,
    visibility_plan,
    visible_schema,
)


CATALOGUE_READ_SCOPE = "datasources:catalogue:read"
SOURCE_WRITE_SCOPE = "datasources:source:write"
STAGE_WRITE_SCOPE = "datasources:stage:write"
STAGE_APPROVE_SCOPE = "datasources:stage:approve"
ADMIN_SCOPE = "datasources:source:admin"


@dataclass(frozen=True, slots=True)
class _PreparedPublication:
    producer_module: str
    producer_run_ref: str
    idempotency_key: str
    rows: tuple[dict[str, Any], ...] | None
    artifact: DatasourceArtifactReference | None
    schema: tuple[DatasourceField, ...]
    fingerprint: str
    row_count: int
    byte_count: int
    request_hash: str


class SqlDatasourceProvider:
    def __init__(
        self,
        *,
        registry: object | None = None,
        payload_backends: Sequence[DatasourcePayloadBackend] = (),
    ) -> None:
        self._registry = registry
        self._payload_backends = PayloadBackendRegistry(payload_backends)

    def publish_rows(
        self,
        session: object,
        principal: object,
        *,
        request: DatasourcePublicationRequest,
    ) -> DatasourcePublicationResult:
        db, api_principal = _publication_context(session, principal)
        prepared = _prepare_publication(request)
        _lock_publication_identity(
            db,
            tenant_id=api_principal.tenant_id,
            producer_module=prepared.producer_module,
            idempotency_key=prepared.idempotency_key,
        )
        existing = _existing_publication_result(
            db,
            tenant_id=api_principal.tenant_id,
            producer_module=prepared.producer_module,
            idempotency_key=prepared.idempotency_key,
            request_hash=prepared.request_hash,
        )
        if existing is not None:
            return existing

        actor_id = _actor_id(api_principal)
        target, governance, baseline_schema = _publication_validation_context(
            db,
            tenant_id=api_principal.tenant_id,
            request=request,
        )
        validation, publication_status = _validate_publication(
            prepared=prepared,
            quality_policy=governance.quality_policy,
            baseline_schema=baseline_schema,
        )
        if validation["valid"] is not True:
            errors = validation.get("errors", [])
            error_codes = ", ".join(
                str(item.get("code") or "validation.error")
                for item in errors
                if isinstance(item, Mapping)
            )
            suffix = f" ({error_codes})" if error_codes else ""
            raise DatasourceValidationError(
                "Published output failed governed quality or schema validation"
                f"{suffix}."
            )
        reusable_payload = self._publication_payload(
            db,
            tenant_id=api_principal.tenant_id,
            actor_id=actor_id,
            prepared=prepared,
        )
        datasource = _publication_target(
            db,
            tenant_id=api_principal.tenant_id,
            actor_id=actor_id,
            request=request,
            prepared=prepared,
            target=target,
            governance=governance,
            publish_as_current=(
                request.set_current and publication_status != "review_required"
            ),
        )
        materialization = _append_materialization(
            db,
            datasource=datasource,
            rows=prepared.rows or (),
            schema=[field_payload(field) for field in prepared.schema],
            fingerprint=prepared.fingerprint,
            byte_count=prepared.byte_count,
            actor_id=actor_id,
            frozen=request.freeze,
            frozen_label=request.frozen_label,
            source_timestamp=request.source_timestamp,
            provenance={
                **_publication_provenance(request, prepared),
                "publication_validation": validation,
            },
            metadata=dict(request.metadata),
            set_current=(
                request.set_current and publication_status != "review_required"
            ),
            reusable_payload=reusable_payload,
            state=(
                "review_required"
                if publication_status == "review_required"
                else "published"
            ),
        )
        publication = _create_publication_record(
            db,
            tenant_id=api_principal.tenant_id,
            actor_id=actor_id,
            datasource=datasource,
            materialization=materialization,
            request=request,
            prepared=prepared,
            status=publication_status,
        )
        audit_event(
            db,
            tenant_id=api_principal.tenant_id,
            user_id=getattr(api_principal.user, "id", None)
            or api_principal.account_id,
            api_key_id=api_principal.api_key_id,
            action=f"datasource.publication.{publication_status}",
            object_type="datasource_publication",
            object_id=publication.id,
            details={
                "producer_module": prepared.producer_module,
                "producer_run_ref": prepared.producer_run_ref,
                "datasource_ref": _datasource_ref(datasource.id),
                "materialization_ref": _materialization_ref(materialization.id),
                "fingerprint": prepared.fingerprint,
                "row_count": prepared.row_count,
                "payload_backend": (
                    prepared.artifact.backend
                    if prepared.artifact is not None
                    else "database_rows"
                ),
                "policy_hash": validation["policy_hash"],
                "schema_classification": validation["schema_change"][
                    "classification"
                ],
            },
        )
        return DatasourcePublicationResult(
            ref=_publication_ref(publication.id),
            status=cast(DatasourcePublicationStatus, publication.status),
            datasource=_datasource_dto(datasource),
            materialization=_materialization_dto(materialization),
            replayed=False,
        )

    def _publication_payload(
        self,
        session: Session,
        *,
        tenant_id: str,
        actor_id: str | None,
        prepared: _PreparedPublication,
    ) -> DatasourcePayloadRecord | None:
        artifact = prepared.artifact
        if artifact is None:
            return None
        backend = self._payload_backends.require(artifact.backend)
        payload = create_external_payload_reference(
            session,
            tenant_id=tenant_id,
            backend=artifact.backend,
            locator=artifact.locator,
            checksum=artifact.checksum,
            row_count=artifact.row_count,
            byte_count=artifact.byte_count,
            actor_id=actor_id,
            media_type=artifact.media_type,
            checkpoint=artifact.checkpoint,
            metadata={
                **dict(artifact.metadata),
                "publication_fingerprint": artifact.fingerprint,
                "artifact_schema": [
                    field_payload(field) for field in artifact.schema
                ],
                "artifact_validation": dict(artifact.validation),
            },
        )
        try:
            backend.verify(session, payload)
        except Exception:
            session.delete(payload)
            session.flush()
            raise
        return payload

    def list_datasources(
        self,
        session: object,
        principal: object,
        *,
        query: str = "",
        limit: int = 100,
        authority_mode: str | None = None,
        classification: str | None = None,
        publication_state: str | None = None,
        owner_ref: str | None = None,
        responsible_organization_ref: str | None = None,
        affected_ref: str | None = None,
        dependency_ref: str | None = None,
    ) -> Sequence[DatasourceDescriptor]:
        db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE)
        statement = (
            select(DatasourceRecord)
            .where(
                DatasourceRecord.tenant_id == api_principal.tenant_id,
                DatasourceRecord.deleted_at.is_(None),
            )
            .order_by(DatasourceRecord.name.asc(), DatasourceRecord.id.asc())
            .limit(max(1, min(int(limit), 100)))
        )
        cleaned_query = query.strip()
        if cleaned_query:
            pattern = f"%{_escape_like(cleaned_query)}%"
            statement = statement.where(
                or_(
                    DatasourceRecord.name.ilike(pattern, escape="\\"),
                    DatasourceRecord.source_name.ilike(pattern, escape="\\"),
                    DatasourceRecord.description.ilike(pattern, escape="\\"),
                )
            )
        for column, value in (
            (DatasourceRecord.authority_mode, authority_mode),
            (DatasourceRecord.classification, classification),
            (DatasourceRecord.publication_state, publication_state),
            (DatasourceRecord.owner_ref, owner_ref),
            (
                DatasourceRecord.responsible_organization_ref,
                responsible_organization_ref,
            ),
        ):
            cleaned = str(value or "").strip()
            if cleaned:
                statement = statement.where(column == cleaned)
        for relation, value in (
            ("affected", affected_ref),
            ("depends_on", dependency_ref),
        ):
            cleaned = str(value or "").strip()
            if cleaned:
                statement = statement.where(
                    exists().where(
                        DatasourceGovernanceReferenceRecord.datasource_id
                        == DatasourceRecord.id,
                        DatasourceGovernanceReferenceRecord.tenant_id
                        == api_principal.tenant_id,
                        DatasourceGovernanceReferenceRecord.relation == relation,
                        DatasourceGovernanceReferenceRecord.reference == cleaned,
                    )
                )
        result: list[DatasourceDescriptor] = []
        for item in db.scalars(statement):
            if has_scope(api_principal, ADMIN_SCOPE) or self._can_discover(
                db,
                api_principal,
                item=item,
            ):
                result.append(_datasource_dto(item))
        return tuple(result)

    def get_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> DatasourceDescriptor | None:
        db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE)
        item = _datasource_record(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
        )
        if item is None:
            return None
        if not has_scope(api_principal, ADMIN_SCOPE) and not self._can_discover(
            db,
            api_principal,
            item=item,
        ):
            return None
        return _datasource_dto(item)

    def read_datasource(
        self,
        session: object,
        principal: object,
        *,
        request: DatasourceReadRequest,
    ) -> DatasourceReadResult:
        db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=request.datasource_ref,
        )
        limit = max(1, min(int(request.limit), MAX_READ_ROWS))
        offset = max(0, int(request.offset))
        columns = tuple(dict.fromkeys(request.columns))

        read_live = request.consistency == "live" or (
            item.mode == "live"
            and not request.materialization_ref
            and request.consistency == "current"
        )
        try:
            if read_live:
                if not item.provider_ref:
                    raise DatasourceUnavailableError(
                        "This datasource has no live origin."
                    )
                result = self._read_live(
                    db,
                    api_principal,
                    item=item,
                    request=request,
                    limit=limit,
                    offset=offset,
                    columns=columns,
                )
            else:
                result = self._read_materialized(
                    db,
                    api_principal,
                    item=item,
                    request=request,
                    limit=limit,
                    offset=offset,
                    columns=columns,
                )
        except DatasourceAccessError:
            _audit_visibility_read(
                db,
                api_principal,
                item=item,
                outcome="denied",
                consistency=request.consistency,
            )
            raise
        if (
            request.expected_fingerprint
            and request.expected_fingerprint != result.datasource.fingerprint
        ):
            raise DatasourceValidationError(
                "The datasource permitted-view fingerprint changed; refresh the consuming definition."
            )
        visibility_evidence = tuple(
            item
            for item in result.diagnostics
            if item.code.startswith("datasource.visibility_")
        )
        if visibility_evidence:
            _audit_visibility_read(
                db,
                api_principal,
                item=item,
                outcome="filtered",
                consistency=request.consistency,
                diagnostics=visibility_evidence,
            )
        return result

    def _read_materialized(
        self,
        session: Session,
        principal: ApiPrincipal,
        *,
        item: DatasourceRecord,
        request: DatasourceReadRequest,
        limit: int,
        offset: int,
        columns: tuple[str, ...],
    ) -> DatasourceReadResult:
        materialization = _selected_materialization(
            session,
            item=item,
            request=request,
        )
        if materialization is None:
            message = (
                "No frozen state is available for this datasource."
                if request.consistency == "frozen"
                else "This datasource has no materialized state."
            )
            raise DatasourceUnavailableError(message)
        if materialization.disposed_at is not None:
            raise DatasourceUnavailableError(
                "This materialization payload was disposed under its retention policy."
            )
        source_schema = _fields(materialization.schema_)
        snapshot_governance = DatasourceGovernance.from_mapping(
            materialization.governance_snapshot_
        )
        snapshot_policies = (
            (snapshot_governance.visibility_policy,)
            if snapshot_governance.visibility_policy
            and snapshot_governance.visibility_policy
            != _datasource_governance(item).visibility_policy
            else ()
        )
        plan = self._visibility_plan(
            session,
            principal,
            item=item,
            request=request,
            materialized=True,
            schema=source_schema,
            additional_policies=snapshot_policies,
        )
        permitted_schema = visible_schema(source_schema, plan)
        _validate_field_names(permitted_schema, columns)
        payload = payload_for_materialization(session, materialization)
        if plan.applied:
            if payload is None:
                scanned_rows = tuple(materialization.rows[:MAX_STAGE_ROWS])
            else:
                backend = self._payload_backends.require(payload.backend)
                backend.verify(session, payload)
                scanned_rows = tuple(
                    backend.read_rows(
                        session,
                        payload,
                        offset=0,
                        limit=MAX_STAGE_ROWS,
                    )
                )
            permitted_rows, filtered_rows = apply_visibility(
                scanned_rows,
                principal=principal.principal,
                plan=plan,
                columns=columns,
            )
            candidate_rows = permitted_rows[offset : offset + limit]
            rows = _bounded_result_rows(
                candidate_rows,
                max_bytes=request.max_bytes,
            )
            byte_limited = len(rows) < len(candidate_rows)
            total_rows = len(permitted_rows)
            scan_limited = len(scanned_rows) < materialization.row_count
        else:
            if payload is None:
                window = materialization.rows[offset : offset + limit]
            else:
                backend = self._payload_backends.require(payload.backend)
                backend.verify(session, payload)
                window = backend.read_rows(
                    session,
                    payload,
                    offset=offset,
                    limit=limit,
                )
            rows = tuple(_select_columns(row, columns) for row in window)
            scanned_rows = rows
            total_rows = materialization.row_count
            filtered_rows = 0
            scan_limited = False
            byte_limited = False
        view_fingerprint = plan.view_fingerprint(materialization.fingerprint)
        descriptor = _datasource_dto(item)
        result_descriptor = replace(
            descriptor,
            schema=permitted_schema,
            schema_version=str(materialization.schema_version),
            fingerprint=view_fingerprint,
            row_count=total_rows,
            byte_count=encoded_size(rows),
        )
        materialization_dto = replace(
            _materialization_dto(materialization),
            schema=permitted_schema,
            fingerprint=view_fingerprint,
            row_count=total_rows,
            byte_count=encoded_size(rows),
        )
        if plan.applied:
            result_descriptor = _opaque_descriptor(result_descriptor)
            materialization_dto = _opaque_materialization(materialization_dto)
        return DatasourceReadResult(
            datasource=result_descriptor,
            rows=rows,
            total_rows=total_rows,
            truncated=(
                scan_limited or byte_limited or offset + len(rows) < total_rows
            ),
            materialization=materialization_dto,
            returned_bytes=encoded_size(rows),
            effective_row_limit=limit,
            effective_byte_limit=request.max_bytes,
            effective_timeout_ms=request.timeout_ms,
            diagnostics=visibility_diagnostics(
                plan,
                scanned_rows=len(scanned_rows),
                permitted_rows=total_rows,
                filtered_rows=filtered_rows,
                scan_limited=scan_limited,
            ),
        )

    def list_materializations(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> Sequence[DatasourceMaterialization]:
        db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
        )
        statement = (
            select(DatasourceMaterializationRecord)
            .where(
                DatasourceMaterializationRecord.tenant_id == api_principal.tenant_id,
                DatasourceMaterializationRecord.datasource_id == item.id,
            )
            .order_by(DatasourceMaterializationRecord.revision.desc())
        )
        result: list[DatasourceMaterialization] = []
        for row in db.scalars(statement):
            schema = _fields(row.schema_)
            snapshot = DatasourceGovernance.from_mapping(row.governance_snapshot_)
            additional_policies = (
                (snapshot.visibility_policy,)
                if snapshot.visibility_policy
                and snapshot.visibility_policy
                != _datasource_governance(item).visibility_policy
                else ()
            )
            try:
                plan = self._visibility_plan(
                    db,
                    api_principal,
                    item=item,
                    request=DatasourceReadRequest(
                        datasource_ref=datasource_ref,
                        materialization_ref=_materialization_ref(row.id),
                        consistency="frozen",
                    ),
                    materialized=True,
                    schema=schema,
                    additional_policies=additional_policies,
                )
            except DatasourceAccessError:
                continue
            dto = _materialization_dto(row)
            if plan.applied:
                dto = _opaque_materialization(
                    replace(
                        dto,
                        schema=visible_schema(schema, plan),
                        fingerprint=plan.view_fingerprint(row.fingerprint),
                        row_count=None,
                        byte_count=None,
                    )
                )
            result.append(dto)
        return tuple(result)

    def list_stages(
        self,
        session: object,
        principal: object,
        *,
        limit: int = 100,
    ) -> Sequence[DatasourceStage]:
        db, api_principal = _context(
            session,
            principal,
            (CATALOGUE_READ_SCOPE, STAGE_WRITE_SCOPE),
        )
        statement = (
            select(DatasourceStageRecord)
            .where(DatasourceStageRecord.tenant_id == api_principal.tenant_id)
            .order_by(DatasourceStageRecord.created_at.desc())
            .limit(max(1, min(int(limit), 100)))
        )
        return tuple(_stage_dto(row) for row in db.scalars(statement))

    def create_stage(
        self,
        session: object,
        principal: object,
        *,
        stage: DatasourceStageInput,
    ) -> DatasourceStage:
        db, api_principal = _context(session, principal, STAGE_WRITE_SCOPE)
        name = stage.name.strip()
        source_name = stage.source_name.strip()
        if not name:
            raise DatasourceValidationError("Stage name is required.")
        if not source_name:
            raise DatasourceValidationError("Datasource key is required.")
        if stage.shape != "tabular":
            raise DatasourceValidationError(
                "The first staging provider supports tabular data only."
            )
        if stage.mode not in {"static", "cached"}:
            raise DatasourceValidationError(
                "Uploaded stages can be promoted as static or cached data."
            )
        target = None
        if stage.target_datasource_ref:
            target = _required_datasource(
                db,
                tenant_id=api_principal.tenant_id,
                datasource_ref=stage.target_datasource_ref,
            )
            if target.shape != stage.shape or target.mode != stage.mode:
                raise DatasourceValidationError(
                    "A stage can only update a datasource with the same mode and shape."
                )
        governance = (
            stage.governance
            or (_datasource_governance(target) if target is not None else None)
            or _default_governance(
                mode=stage.mode,
                provider_ref=stage.provider_ref,
            )
        )
        rows = normalize_rows(stage.rows)
        schema = infer_schema(rows)
        governance = replace(
            governance,
            visibility_policy=normalize_visibility_policy(
                governance.visibility_policy,
                schema=schema,
            ),
            approval_policy=normalize_approval_policy(governance.approval_policy),
            retention_policy=normalize_retention_policy(governance.retention_policy),
        )
        fingerprint = fingerprint_rows(rows, schema)
        validation = validate_stage(
            rows=rows,
            schema=schema,
            quality_policy=governance.quality_policy,
            baseline_schema=_fields(target.schema_) if target is not None else None,
        )
        item = DatasourceStageRecord(
            tenant_id=api_principal.tenant_id,
            target_datasource_id=target.id if target else None,
            name=name,
            source_name=source_name,
            description=_clean_optional(stage.description),
            kind=stage.kind,
            mode=stage.mode,
            shape=stage.shape,
            state="invalid",
            provider=_clean_optional(stage.provider),
            provider_ref=_clean_optional(stage.provider_ref),
            schema_=[field_payload(field) for field in schema],
            rows=list(rows),
            fingerprint=fingerprint,
            row_count=len(rows),
            byte_count=encoded_size(rows),
            validation_=validation,
            provenance_=dict(stage.provenance),
            metadata_=dict(stage.metadata),
            governance_=governance.to_dict(),
            created_by=_actor_id(api_principal),
        )
        db.add(item)
        db.flush()
        approval = initialize_stage_approval(item, created_at=item.created_at)
        record_lifecycle_evidence(
            db,
            tenant_id=api_principal.tenant_id,
            subject_ref=_stage_ref(item.id),
            event_type="stage.validated",
            actor_ref=_actor_id(api_principal),
            subject_digest=str(approval["subject_digest"]),
            policy_version=str(validation.get("policy_version") or "1"),
            policy_hash=str(validation.get("policy_hash") or ""),
            details={
                "validation_valid": validation.get("valid"),
                "schema_classification": (
                    validation.get("schema_change", {}).get("classification")
                    if isinstance(validation.get("schema_change"), Mapping)
                    else None
                ),
                "approval_required": approval.get("state") != "not_required",
                "approval_policy_hash": approval.get("policy_hash"),
            },
            occurred_at=item.created_at,
        )
        db.flush()
        return _stage_dto(item)

    def decide_stage(
        self,
        session: object,
        principal: object,
        *,
        stage_ref: str,
        decision: str,
        reason: str,
        expected_policy_hash: str,
        expected_subject_digest: str,
    ) -> tuple[DatasourceStage, str | None, bool]:
        db, api_principal = _context(session, principal, STAGE_APPROVE_SCOPE)
        actor_ref = _actor_id(api_principal)
        if actor_ref is None:
            raise DatasourceValidationError(
                "An attributable account is required for datasource approval."
            )
        stage = _required_stage(
            db,
            tenant_id=api_principal.tenant_id,
            stage_ref=stage_ref,
            for_update=True,
        )
        approval, replayed = decide_stage(
            stage,
            actor_ref=actor_ref,
            actor_scopes=tuple(api_principal.scopes),
            decision=decision,
            reason=reason,
            expected_policy_hash=expected_policy_hash,
            expected_subject_digest=expected_subject_digest,
        )
        evidence_hash = None
        if not replayed:
            policy = approval.get("policy")
            policy_mapping = policy if isinstance(policy, Mapping) else {}
            evidence = record_lifecycle_evidence(
                db,
                tenant_id=api_principal.tenant_id,
                subject_ref=_stage_ref(stage.id),
                event_type=(
                    "stage.approved" if decision == "approve" else "stage.rejected"
                ),
                actor_ref=actor_ref,
                subject_digest=str(approval["subject_digest"]),
                policy_version=str(policy_mapping.get("version") or "1"),
                policy_hash=str(approval["policy_hash"]),
                details={
                    "decision": decision,
                    "reason": reason.strip(),
                    "approval_count": approval.get("approval_count"),
                    "required_approvals": approval.get("required_approvals"),
                    "resulting_state": approval.get("state"),
                    "authority_scopes": sorted(api_principal.scopes),
                },
            )
            evidence_hash = evidence.event_hash
        db.flush()
        return _stage_dto(stage), evidence_hash, replayed

    def promote_stage(
        self,
        session: object,
        principal: object,
        *,
        stage_ref: str,
        freeze: bool = False,
        frozen_label: str | None = None,
    ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]:
        db, api_principal = _context(session, principal, STAGE_WRITE_SCOPE)
        stage = _required_stage(
            db,
            tenant_id=api_principal.tenant_id,
            stage_ref=stage_ref,
        )
        if stage.validation_.get("valid") is not True:
            raise DatasourceValidationError(
                "The stage has blocking quality or schema diagnostics and cannot be promoted."
            )
        if stage.state != "ready":
            raise DatasourceValidationError("Only ready stages can be promoted.")
        datasource = (
            db.get(DatasourceRecord, stage.target_datasource_id)
            if stage.target_datasource_id
            else None
        )
        if datasource is not None and (
            datasource.tenant_id != api_principal.tenant_id
            or datasource.deleted_at is not None
        ):
            datasource = None
        current_governance = (
            _datasource_governance(datasource)
            if datasource is not None
            else DatasourceGovernance.from_mapping(stage.governance_)
        )
        ensure_stage_approval_current(
            stage,
            current_policy=current_governance.approval_policy,
        )
        if datasource is None:
            _ensure_source_name_available(
                db,
                tenant_id=api_principal.tenant_id,
                source_name=stage.source_name,
            )
            datasource = DatasourceRecord(
                tenant_id=api_principal.tenant_id,
                source_name=stage.source_name,
                name=stage.name,
                description=stage.description,
                kind=stage.kind,
                mode=stage.mode,
                shape=stage.shape,
                status="active",
                provider=stage.provider or "datasources.stage",
                provider_ref=stage.provider_ref,
                schema_version=1,
                schema_=list(stage.schema_),
                fingerprint=stage.fingerprint,
                row_count=stage.row_count,
                byte_count=stage.byte_count,
                provenance_=dict(stage.provenance_),
                metadata_=dict(stage.metadata_),
                created_by=_actor_id(api_principal),
                updated_by=_actor_id(api_principal),
            )
            _apply_datasource_governance(
                datasource,
                DatasourceGovernance.from_mapping(stage.governance_),
            )
            db.add(datasource)
            db.flush()
            stage.target_datasource_id = datasource.id
        elif datasource.mode != stage.mode or datasource.shape != stage.shape:
            raise DatasourceValidationError(
                "A stage can only update a datasource with the same mode and shape."
            )
        else:
            _apply_datasource_governance(
                datasource,
                DatasourceGovernance.from_mapping(stage.governance_),
            )

        materialization = _append_materialization(
            db,
            datasource=datasource,
            rows=stage.rows,
            schema=stage.schema_,
            fingerprint=stage.fingerprint,
            byte_count=stage.byte_count,
            actor_id=_actor_id(api_principal),
            frozen=freeze,
            frozen_label=frozen_label,
            provenance={
                **dict(stage.provenance_),
                "stage_ref": _stage_ref(stage.id),
                "stage_validation": dict(stage.validation_),
                "stage_approval": dict(stage.approval_),
            },
            metadata=dict(stage.metadata_),
            set_current=True,
        )
        stage.state = "promoted"
        stage.promoted_at = utcnow()
        stage.promoted_materialization_id = materialization.id
        approval_policy = stage.approval_.get("policy")
        policy_mapping = (
            approval_policy if isinstance(approval_policy, Mapping) else {}
        )
        promotion_evidence = record_lifecycle_evidence(
            db,
            tenant_id=api_principal.tenant_id,
            subject_ref=_stage_ref(stage.id),
            event_type="stage.promoted",
            actor_ref=_actor_id(api_principal),
            subject_digest=str(stage.approval_.get("subject_digest") or ""),
            policy_version=str(policy_mapping.get("version") or "1"),
            policy_hash=str(stage.approval_.get("policy_hash") or ""),
            details={
                "datasource_ref": _datasource_ref(datasource.id),
                "materialization_ref": _materialization_ref(materialization.id),
                "revision": materialization.revision,
                "quality_policy_hash": stage.validation_.get("policy_hash"),
                "approval_count": stage.approval_.get("approval_count"),
            },
        )
        materialization.provenance_ = {
            **dict(materialization.provenance_),
            "promotion_evidence_hash": promotion_evidence.event_hash,
        }
        db.flush()
        return _datasource_dto(datasource), _materialization_dto(materialization)

    def prepare_refresh(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> DatasourceStage:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
        )
        if item.mode != "cached" or not item.provider_ref:
            raise DatasourceValidationError(
                "Only cached connector-backed datasources can prepare a refresh."
            )
        rows, origin = self._read_origin_all(
            db,
            api_principal,
            origin_ref=item.provider_ref,
        )
        return self.create_stage(
            db,
            api_principal,
            stage=DatasourceStageInput(
                name=item.name,
                source_name=item.source_name,
                description=item.description,
                kind=cast(Any, item.kind),
                mode="cached",
                shape=cast(Any, item.shape),
                rows=tuple(rows),
                target_datasource_ref=_datasource_ref(item.id),
                provider=item.provider,
                provider_ref=item.provider_ref,
                provenance={
                    "created_via": "datasources.refresh",
                    "origin_ref": origin.ref,
                    "origin_fingerprint": origin.fingerprint,
                    "origin_schema_version": origin.schema_version,
                },
                metadata=dict(item.metadata_),
                governance=_datasource_governance(item),
            ),
        )

    def register_origin(
        self,
        session: object,
        principal: object,
        *,
        origin_ref: str,
        name: str,
        source_name: str,
        mode: DatasourceMode,
        description: str | None = None,
        governance: DatasourceGovernance | None = None,
    ) -> DatasourceDescriptor:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        if mode not in {"live", "cached"}:
            raise DatasourceValidationError(
                "Connector origins can be registered as live or cached datasources."
            )
        origin = self._required_origin(db, api_principal, origin_ref)
        if mode not in origin.supported_modes:
            raise DatasourceValidationError(
                f"The selected origin does not support {mode!r} datasources."
            )
        cleaned_name = name.strip()
        cleaned_source_name = source_name.strip()
        if not cleaned_name or not cleaned_source_name:
            raise DatasourceValidationError(
                "Datasource name and key are required."
            )
        _ensure_source_name_available(
            db,
            tenant_id=api_principal.tenant_id,
            source_name=cleaned_source_name,
        )
        item = DatasourceRecord(
            tenant_id=api_principal.tenant_id,
            source_name=cleaned_source_name,
            name=cleaned_name,
            description=_clean_optional(description) or origin.description,
            kind=origin.kind,
            mode=mode,
            shape=origin.shape,
            status="active",
            provider=origin.provider,
            provider_ref=origin.ref,
            schema_version=_int_schema_version(origin.schema_version),
            schema_=[field_payload(field) for field in origin.schema],
            fingerprint=origin.fingerprint,
            row_count=origin.row_count,
            byte_count=origin.byte_count,
            provenance_={
                "origin_ref": origin.ref,
                "origin_provider": origin.provider,
                "registered_at": utcnow().isoformat(),
            },
            metadata_=_origin_metadata(origin),
            created_by=_actor_id(api_principal),
            updated_by=_actor_id(api_principal),
        )
        _apply_datasource_governance(
            item,
            governance
            or _default_governance(mode=mode, provider_ref=origin.ref),
        )
        db.add(item)
        db.flush()
        if mode == "cached":
            rows, refreshed_origin = self._read_origin_all(
                db,
                api_principal,
                origin_ref=origin_ref,
            )
            self._materialize_origin(
                db,
                item=item,
                origin=refreshed_origin,
                rows=rows,
                actor_id=_actor_id(api_principal),
                set_current=True,
            )
        return _datasource_dto(item)

    def update_datasource_governance(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
        governance: DatasourceGovernance,
    ) -> DatasourceDescriptor:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
            for_update=True,
        )
        _apply_datasource_governance(item, governance)
        item.updated_by = _actor_id(api_principal)
        db.flush()
        return _datasource_dto(item)

    def refresh_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
        )
        if item.mode != "cached" or not item.provider_ref:
            raise DatasourceValidationError(
                "Only cached connector-backed datasources can be refreshed."
            )
        approval_policy = normalize_approval_policy(
            _datasource_governance(item).approval_policy
        )
        if approval_policy["required"]:
            raise DatasourceValidationError(
                "This datasource requires an approved refresh stage before it can become current."
            )
        rows, origin = self._read_origin_all(
            db,
            api_principal,
            origin_ref=item.provider_ref,
        )
        materialization = self._materialize_origin(
            db,
            item=item,
            origin=origin,
            rows=rows,
            actor_id=_actor_id(api_principal),
            set_current=True,
        )
        return _datasource_dto(item), _materialization_dto(materialization)

    def preview_retention(
        self,
        session: object,
        principal: object,
        *,
        as_of: datetime,
    ) -> RetentionPlan:
        db, api_principal = _context(session, principal, ADMIN_SCOPE)
        return build_retention_plan(
            db,
            tenant_id=api_principal.tenant_id,
            as_of=as_of,
        )

    def apply_retention(
        self,
        session: object,
        principal: object,
        *,
        as_of: datetime,
        plan_hash: str,
        target_refs: Sequence[str],
    ) -> tuple[tuple[str, ...], tuple[str, ...]]:
        db, api_principal = _context(session, principal, ADMIN_SCOPE)
        actor_ref = _actor_id(api_principal)
        if actor_ref is None:
            raise DatasourceValidationError(
                "An attributable account is required for retention execution."
            )
        plan = build_retention_plan(
            db,
            tenant_id=api_principal.tenant_id,
            as_of=as_of,
        )
        if plan.plan_hash != plan_hash:
            raise DatasourceValidationError(
                "The retention plan changed; preview it again before applying."
            )
        return apply_retention_plan(
            db,
            tenant_id=api_principal.tenant_id,
            actor_ref=actor_ref,
            plan=plan,
            target_refs=target_refs,
        )

    def list_lifecycle_evidence(
        self,
        session: object,
        principal: object,
        *,
        subject_ref: str | None = None,
        limit: int = 200,
    ):
        db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE)
        return list_lifecycle_evidence(
            db,
            tenant_id=api_principal.tenant_id,
            subject_ref=subject_ref,
            limit=limit,
        )

    def freeze_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
        label: str | None = None,
    ) -> DatasourceMaterialization:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
        )
        if item.mode == "live":
            if not item.provider_ref:
                raise DatasourceUnavailableError(
                    "The live datasource no longer references an origin."
                )
            rows, origin = self._read_origin_all(
                db,
                api_principal,
                origin_ref=item.provider_ref,
            )
            materialization = self._materialize_origin(
                db,
                item=item,
                origin=origin,
                rows=rows,
                actor_id=_actor_id(api_principal),
                frozen=True,
                frozen_label=label,
                set_current=False,
            )
            return _materialization_dto(materialization)

        current = _current_materialization(db, item)
        if current is None:
            raise DatasourceUnavailableError(
                "The datasource has no current state to freeze."
            )
        current_payload = payload_for_materialization(db, current)
        materialization = _append_materialization(
            db,
            datasource=item,
            rows=current.rows if current_payload is None else (),
            schema=current.schema_,
            fingerprint=current.fingerprint,
            byte_count=current.byte_count,
            actor_id=_actor_id(api_principal),
            frozen=True,
            frozen_label=label,
            source_timestamp=current.source_timestamp,
            provenance={
                **dict(current.provenance_),
                "frozen_from": _materialization_ref(current.id),
            },
            metadata=dict(current.metadata_),
            set_current=False,
            reusable_payload=current_payload,
        )
        return _materialization_dto(materialization)

    def retire_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> DatasourceDescriptor:
        db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE)
        item = _required_datasource(
            db,
            tenant_id=api_principal.tenant_id,
            datasource_ref=datasource_ref,
        )
        item.status = "retired"
        item.deleted_at = utcnow()
        item.updated_by = _actor_id(api_principal)
        db.flush()
        return _datasource_dto(item)

    def _read_live(
        self,
        session: Session,
        principal: ApiPrincipal,
        *,
        item: DatasourceRecord,
        request: DatasourceReadRequest,
        limit: int,
        offset: int,
        columns: tuple[str, ...],
    ) -> DatasourceReadResult:
        if not item.provider_ref:
            raise DatasourceUnavailableError(
                "The live datasource no longer references an origin."
            )
        provider = datasource_origins(self._registry)
        if provider is None:
            raise DatasourceUnavailableError(
                "No connector origin provider is currently available."
            )
        plan = self._visibility_plan(
            session,
            principal,
            item=item,
            request=request,
            materialized=False,
            schema=_fields(item.schema_),
        )
        if plan.applied:
            return self._read_live_visible(
                session,
                principal,
                provider=provider,
                item=item,
                request=request,
                plan=plan,
                limit=limit,
                offset=offset,
                columns=columns,
            )
        try:
            result = provider.read_origin(
                session,
                principal,
                request=DatasourceOriginReadRequest(
                    origin_ref=item.provider_ref,
                    limit=limit,
                    offset=offset,
                    columns=columns,
                    expected_fingerprint=request.expected_fingerprint,
                    max_bytes=request.max_bytes,
                    timeout_ms=request.timeout_ms,
                ),
            )
        except DatasourceError:
            raise
        except Exception as exc:
            raise DatasourceUnavailableError(
                f"The datasource origin could not be read: {exc}"
            ) from exc
        descriptor = replace(
            _datasource_dto(item),
            schema=result.origin.schema,
            schema_version=result.origin.schema_version,
            fingerprint=result.origin.fingerprint,
            row_count=result.origin.row_count,
            byte_count=result.origin.byte_count,
            updated_at=result.origin.updated_at,
            metadata=_origin_metadata(result.origin, current=item.metadata_),
        )
        return DatasourceReadResult(
            datasource=descriptor,
            rows=result.rows,
            total_rows=result.total_rows,
            truncated=result.truncated,
            returned_bytes=result.returned_bytes,
            elapsed_ms=result.elapsed_ms,
            effective_row_limit=result.effective_row_limit,
            effective_byte_limit=result.effective_byte_limit,
            effective_timeout_ms=result.effective_timeout_ms,
            diagnostics=result.diagnostics,
        )

    def _read_live_visible(
        self,
        session: Session,
        principal: ApiPrincipal,
        *,
        provider: object,
        item: DatasourceRecord,
        request: DatasourceReadRequest,
        plan: VisibilityPlan,
        limit: int,
        offset: int,
        columns: tuple[str, ...],
    ) -> DatasourceReadResult:
        rows: list[Mapping[str, object]] = []
        first_result = None
        last_result = None
        base_fingerprint: str | None = None
        elapsed_ms = 0
        while len(rows) < MAX_STAGE_ROWS:
            page_limit = min(MAX_READ_ROWS, MAX_STAGE_ROWS - len(rows))
            try:
                page = provider.read_origin(
                    session,
                    principal,
                    request=DatasourceOriginReadRequest(
                        origin_ref=str(item.provider_ref),
                        limit=page_limit,
                        offset=len(rows),
                        expected_fingerprint=base_fingerprint,
                        max_bytes=request.max_bytes,
                        timeout_ms=request.timeout_ms,
                    ),
                )
            except DatasourceError:
                raise
            except Exception as exc:
                raise DatasourceUnavailableError(
                    f"The datasource origin could not be read: {exc}"
                ) from exc
            if first_result is None:
                first_result = page
                base_fingerprint = page.origin.fingerprint
            last_result = page
            elapsed_ms += page.elapsed_ms
            rows.extend(page.rows)
            if not page.truncated or not page.rows:
                break
        if first_result is None or last_result is None:
            raise DatasourceUnavailableError(
                "The datasource origin returned no result."
            )
        source_schema = first_result.origin.schema
        plan = self._visibility_plan(
            session,
            principal,
            item=item,
            request=request,
            materialized=False,
            schema=source_schema,
        )
        permitted_schema = visible_schema(source_schema, plan)
        _validate_field_names(permitted_schema, columns)
        permitted, filtered_rows = apply_visibility(
            rows,
            principal=principal.principal,
            plan=plan,
            columns=columns,
        )
        window = permitted[offset : offset + limit]
        window = _bounded_result_rows(window, max_bytes=request.max_bytes)
        scan_limited = bool(
            last_result.truncated
            and (len(rows) >= MAX_STAGE_ROWS or not last_result.rows)
        )
        total_rows = len(permitted)
        view_fingerprint = plan.view_fingerprint(first_result.origin.fingerprint)
        descriptor = replace(
            _datasource_dto(item),
            schema=permitted_schema,
            schema_version=first_result.origin.schema_version,
            fingerprint=view_fingerprint,
            row_count=total_rows,
            byte_count=encoded_size(window),
            updated_at=first_result.origin.updated_at,
        )
        return DatasourceReadResult(
            datasource=_opaque_descriptor(descriptor),
            rows=window,
            total_rows=total_rows,
            truncated=(
                scan_limited
                or offset + len(window) < total_rows
                or len(window) < min(limit, max(0, total_rows - offset))
            ),
            returned_bytes=encoded_size(window),
            elapsed_ms=elapsed_ms,
            effective_row_limit=limit,
            effective_byte_limit=request.max_bytes,
            effective_timeout_ms=request.timeout_ms,
            diagnostics=(
                *first_result.diagnostics,
                *visibility_diagnostics(
                    plan,
                    scanned_rows=len(rows),
                    permitted_rows=total_rows,
                    filtered_rows=filtered_rows,
                    scan_limited=scan_limited,
                ),
            ),
        )

    def _can_discover(
        self,
        session: Session,
        principal: ApiPrincipal,
        *,
        item: DatasourceRecord,
    ) -> bool:
        try:
            self._visibility_plan(
                session,
                principal,
                item=item,
                request=DatasourceReadRequest(datasource_ref=_datasource_ref(item.id)),
                materialized=False,
                schema=_fields(item.schema_),
                action="discover",
            )
        except (DatasourceAccessError, DatasourceValidationError):
            return False
        return True

    def _visibility_plan(
        self,
        session: Session,
        principal: ApiPrincipal,
        *,
        item: DatasourceRecord,
        request: DatasourceReadRequest,
        materialized: bool,
        schema: Sequence[DatasourceField],
        action: str = "read",
        additional_policies: Sequence[Mapping[str, object]] = (),
    ) -> VisibilityPlan:
        governance = _datasource_governance(item)
        policies: list[Mapping[str, object]] = []
        if governance.visibility_policy:
            policies.append(governance.visibility_policy)
        policies.extend(additional_policies)
        decision_refs: list[str] = []
        policy_provider = datasource_visibility_policy_provider(self._registry)
        if policy_provider is None:
            if governance.access_policy_ref:
                raise DatasourceAccessError(
                    "The configured datasource access policy provider is unavailable."
                )
        else:
            try:
                decision = policy_provider.decide_datasource_visibility(
                    session,
                    request=DatasourceVisibilityPolicyRequest(
                        tenant_id=principal.tenant_id,
                        datasource_ref=_datasource_ref(item.id),
                        principal=principal.principal,
                        action=cast(Any, action),
                        classification=governance.classification,
                        policy_ref=governance.access_policy_ref,
                        consistency=request.consistency,
                        materialization_ref=request.materialization_ref,
                    ),
                )
            except DatasourceError:
                raise
            except Exception as exc:
                raise DatasourceAccessError(
                    "Datasource policy evaluation is unavailable."
                ) from exc
            if not decision.allowed:
                raise DatasourceAccessError(
                    decision.reason or "Datasource policy denied access."
                )
            policies.extend(decision.policies)
            if decision.decision_ref:
                decision_refs.append(decision.decision_ref)
        try:
            return visibility_plan(
                principal=principal.principal,
                policies=policies,
                materialized=materialized,
                schema=schema,
                decision_refs=decision_refs,
            )
        except DatasourceValidationError as exc:
            raise DatasourceAccessError(
                "Datasource visibility policy is invalid and access was denied."
            ) from exc

    def _required_origin(
        self,
        session: Session,
        principal: ApiPrincipal,
        origin_ref: str,
    ) -> DatasourceOrigin:
        provider = datasource_origins(self._registry)
        if provider is None:
            raise DatasourceUnavailableError(
                "No connector origin provider is currently available."
            )
        try:
            origin = provider.get_origin(
                session,
                principal,
                origin_ref=origin_ref,
            )
        except DatasourceError:
            raise
        except Exception as exc:
            raise DatasourceUnavailableError(
                f"The datasource origin could not be inspected: {exc}"
            ) from exc
        if origin is None:
            raise DatasourceNotFoundError("Datasource origin not found.")
        return origin

    def _read_origin_all(
        self,
        session: Session,
        principal: ApiPrincipal,
        *,
        origin_ref: str,
    ) -> tuple[tuple[Mapping[str, object], ...], DatasourceOrigin]:
        provider = datasource_origins(self._registry)
        if provider is None:
            raise DatasourceUnavailableError(
                "No connector origin provider is currently available."
            )
        rows: list[Mapping[str, object]] = []
        origin: DatasourceOrigin | None = None
        offset = 0
        while True:
            try:
                result = provider.read_origin(
                    session,
                    principal,
                    request=DatasourceOriginReadRequest(
                        origin_ref=origin_ref,
                        limit=min(MAX_READ_ROWS, MAX_STAGE_ROWS - offset),
                        offset=offset,
                    ),
                )
            except DatasourceError:
                raise
            except Exception as exc:
                raise DatasourceUnavailableError(
                    f"The datasource origin could not be materialized: {exc}"
                ) from exc
            origin = result.origin
            rows.extend(result.rows)
            offset += len(result.rows)
            if not result.truncated:
                break
            if offset >= MAX_STAGE_ROWS or not result.rows:
                raise DatasourceValidationError(
                    f"Cached and frozen sources are limited to {MAX_STAGE_ROWS:,} rows."
                )
        if origin is None:
            raise DatasourceUnavailableError("The datasource origin returned no metadata.")
        return tuple(rows), origin

    def _materialize_origin(
        self,
        session: Session,
        *,
        item: DatasourceRecord,
        origin: DatasourceOrigin,
        rows: Sequence[Mapping[str, object]],
        actor_id: str | None,
        frozen: bool = False,
        frozen_label: str | None = None,
        set_current: bool,
    ) -> DatasourceMaterializationRecord:
        item.metadata_ = _origin_metadata(origin, current=item.metadata_)
        normalized = normalize_rows(rows)
        schema = infer_schema(normalized) or origin.schema
        fingerprint = fingerprint_rows(normalized, schema)
        return _append_materialization(
            session,
            datasource=item,
            rows=normalized,
            schema=[field_payload(field) for field in schema],
            fingerprint=fingerprint,
            byte_count=encoded_size(normalized),
            actor_id=actor_id,
            frozen=frozen,
            frozen_label=frozen_label,
            source_timestamp=origin.updated_at,
            provenance={
                **dict(item.provenance_),
                "origin_ref": origin.ref,
                "origin_fingerprint": origin.fingerprint,
                "materialized_at": utcnow().isoformat(),
            },
            metadata={
                **dict(item.metadata_),
                "origin_schema_version": origin.schema_version,
            },
            set_current=set_current,
        )


def _origin_metadata(
    origin: DatasourceOrigin,
    *,
    current: Mapping[str, object] | None = None,
) -> dict[str, object]:
    return {
        **dict(current or {}),
        **dict(origin.metadata),
        "source_contract": {
            "source_mode": origin.source_mode,
            "pushdown": {
                "projections": origin.pushdown.projections,
                "pagination": origin.pushdown.pagination,
                "filters": list(origin.pushdown.filters),
                "aggregations": list(origin.pushdown.aggregations),
                "sorting": list(origin.pushdown.sorting),
            },
            "health": {
                "status": origin.health.status,
                "code": origin.health.code,
                "summary": origin.health.summary,
                "checked_at": (
                    origin.health.checked_at.isoformat()
                    if origin.health.checked_at
                    else None
                ),
                "details": dict(origin.health.details),
            },
        },
    }


def _append_materialization(
    session: Session,
    *,
    datasource: DatasourceRecord,
    rows: Sequence[Mapping[str, object]],
    schema: Sequence[Mapping[str, object]],
    fingerprint: str,
    byte_count: int,
    actor_id: str | None,
    frozen: bool = False,
    frozen_label: str | None = None,
    source_timestamp=None,
    provenance: Mapping[str, object] | None = None,
    metadata: Mapping[str, object] | None = None,
    set_current: bool,
    reusable_payload: DatasourcePayloadRecord | None = None,
    state: str = "published",
) -> DatasourceMaterializationRecord:
    datasource = _lock_datasource_for_materialization(session, datasource)
    revision = _allocate_materialization_revision(session, datasource)
    payload = reusable_payload or create_database_rows_payload(
        session,
        tenant_id=datasource.tenant_id,
        rows=rows,
        actor_id=actor_id,
        metadata={
            "datasource_id": datasource.id,
            "fingerprint": fingerprint,
        },
    )
    if payload.tenant_id != datasource.tenant_id or payload.state != "ready":
        raise DatasourceValidationError(
            "Only a ready payload from the same tenant can be materialized."
        )
    validate_payload_size(payload, expected_byte_count=byte_count)
    schema_payload, schema_version = _materialization_schema(
        datasource,
        schema,
    )
    materialization = DatasourceMaterializationRecord(
        tenant_id=datasource.tenant_id,
        datasource_id=datasource.id,
        revision=revision,
        state=state,
        schema_version=max(1, int(schema_version or 1)),
        schema_=schema_payload,
        payload_id=payload.id,
        payload_checksum=payload.checksum,
        rows=[],
        fingerprint=fingerprint,
        row_count=payload.row_count,
        byte_count=payload.byte_count,
        frozen_at=utcnow() if frozen else None,
        frozen_label=_clean_optional(frozen_label),
        source_timestamp=source_timestamp,
        provenance_=dict(provenance or {}),
        metadata_=dict(metadata or {}),
        governance_snapshot_=_datasource_governance(datasource).to_dict(),
        created_by=actor_id,
    )
    session.add(materialization)
    session.flush()
    if set_current:
        _apply_current_materialization(
            datasource,
            materialization=materialization,
            schema=schema_payload,
            schema_version=schema_version,
            actor_id=actor_id,
        )
    session.flush()
    return materialization


def _lock_datasource_for_materialization(
    session: Session,
    datasource: DatasourceRecord,
) -> DatasourceRecord:
    locked = session.scalar(
        select(DatasourceRecord)
        .where(
            DatasourceRecord.id == datasource.id,
            DatasourceRecord.tenant_id == datasource.tenant_id,
            DatasourceRecord.deleted_at.is_(None),
        )
        .with_for_update()
        .execution_options(populate_existing=True)
    )
    if locked is None:
        raise DatasourceUnavailableError(
            "The datasource is no longer available for materialization."
        )
    return locked


def _allocate_materialization_revision(
    session: Session,
    datasource: DatasourceRecord,
) -> int:
    """Allocate under the datasource row lock held by the caller."""

    return int(
        session.scalar(
            select(func.max(DatasourceMaterializationRecord.revision)).where(
                DatasourceMaterializationRecord.datasource_id == datasource.id
            )
        )
        or 0
    ) + 1


def _materialization_schema(
    datasource: DatasourceRecord,
    schema: Sequence[Mapping[str, object]],
) -> tuple[list[dict[str, object]], int]:
    schema_payload = [dict(field) for field in schema]
    schema_changed = datasource.schema_ != schema_payload
    schema_version = (
        datasource.schema_version + 1
        if schema_changed
        else datasource.schema_version
    )
    return schema_payload, int(schema_version or 1)


def _apply_current_materialization(
    datasource: DatasourceRecord,
    *,
    materialization: DatasourceMaterializationRecord,
    schema: list[dict[str, object]],
    schema_version: int,
    actor_id: str | None,
) -> None:
    datasource.current_materialization_id = materialization.id
    datasource.schema_ = schema
    datasource.schema_version = schema_version
    datasource.fingerprint = materialization.fingerprint
    datasource.row_count = materialization.row_count
    datasource.byte_count = materialization.byte_count
    datasource.updated_by = actor_id


def _selected_materialization(
    session: Session,
    *,
    item: DatasourceRecord,
    request: DatasourceReadRequest,
) -> DatasourceMaterializationRecord | None:
    if request.materialization_ref:
        return _materialization_record(
            session,
            tenant_id=item.tenant_id,
            datasource_id=item.id,
            materialization_ref=request.materialization_ref,
        )
    if request.consistency == "frozen":
        return session.scalar(
            select(DatasourceMaterializationRecord)
            .where(
                DatasourceMaterializationRecord.datasource_id == item.id,
                DatasourceMaterializationRecord.tenant_id == item.tenant_id,
                DatasourceMaterializationRecord.frozen_at.is_not(None),
            )
            .order_by(DatasourceMaterializationRecord.revision.desc())
            .limit(1)
        )
    return _current_materialization(session, item)


def _current_materialization(
    session: Session,
    item: DatasourceRecord,
) -> DatasourceMaterializationRecord | None:
    if not item.current_materialization_id:
        return None
    return session.scalar(
        select(DatasourceMaterializationRecord).where(
            DatasourceMaterializationRecord.id == item.current_materialization_id,
            DatasourceMaterializationRecord.datasource_id == item.id,
            DatasourceMaterializationRecord.tenant_id == item.tenant_id,
        )
    )


def _datasource_record(
    session: Session,
    *,
    tenant_id: str,
    datasource_ref: str,
    for_update: bool = False,
) -> DatasourceRecord | None:
    datasource_id = _strip_ref(datasource_ref, "datasource:")
    if datasource_id is None:
        return None
    statement = select(DatasourceRecord).where(
            DatasourceRecord.id == datasource_id,
            DatasourceRecord.tenant_id == tenant_id,
            DatasourceRecord.deleted_at.is_(None),
        )
    if for_update:
        statement = statement.with_for_update()
    return session.scalar(statement)


def _required_datasource(
    session: Session,
    *,
    tenant_id: str,
    datasource_ref: str,
    for_update: bool = False,
) -> DatasourceRecord:
    item = _datasource_record(
        session,
        tenant_id=tenant_id,
        datasource_ref=datasource_ref,
        for_update=for_update,
    )
    if item is None:
        raise DatasourceNotFoundError("Datasource not found.")
    return item


def _materialization_record(
    session: Session,
    *,
    tenant_id: str,
    datasource_id: str,
    materialization_ref: str,
) -> DatasourceMaterializationRecord | None:
    materialization_id = _strip_ref(materialization_ref, "materialization:")
    if materialization_id is None:
        return None
    return session.scalar(
        select(DatasourceMaterializationRecord).where(
            DatasourceMaterializationRecord.id == materialization_id,
            DatasourceMaterializationRecord.tenant_id == tenant_id,
            DatasourceMaterializationRecord.datasource_id == datasource_id,
        )
    )


def _required_stage(
    session: Session,
    *,
    tenant_id: str,
    stage_ref: str,
    for_update: bool = False,
) -> DatasourceStageRecord:
    stage_id = _strip_ref(stage_ref, "stage:")
    if stage_id is None:
        raise DatasourceNotFoundError("Datasource stage not found.")
    statement = select(DatasourceStageRecord).where(
            DatasourceStageRecord.id == stage_id,
            DatasourceStageRecord.tenant_id == tenant_id,
        )
    if for_update:
        statement = statement.with_for_update()
    item = session.scalar(statement)
    if item is None:
        raise DatasourceNotFoundError("Datasource stage not found.")
    return item


def _ensure_source_name_available(
    session: Session,
    *,
    tenant_id: str,
    source_name: str,
) -> None:
    existing = session.scalar(
        select(DatasourceRecord.id).where(
            DatasourceRecord.tenant_id == tenant_id,
            DatasourceRecord.source_name == source_name,
        )
    )
    if existing is not None:
        raise DatasourceValidationError(
            f"A datasource with key {source_name!r} already exists."
        )


def _default_governance(
    *,
    mode: str,
    provider_ref: str | None,
) -> DatasourceGovernance:
    if mode == "live":
        authority_mode = "external_authoritative"
    elif mode == "cached" and provider_ref:
        authority_mode = "external_mirror"
    else:
        authority_mode = "native_authoritative"
    return DatasourceGovernance(
        authoritative_source_ref=provider_ref,
        authority_mode=cast(Any, authority_mode),
        purposes=("governed_data_processing",),
        publication_state="internal",
    )


def _datasource_governance(item: DatasourceRecord) -> DatasourceGovernance:
    return DatasourceGovernance.from_mapping(
        {
            "owner_ref": item.owner_ref,
            "steward_ref": item.steward_ref,
            "responsible_organization_ref": item.responsible_organization_ref,
            "responsible_function_ref": item.responsible_function_ref,
            "authoritative_source_ref": item.authoritative_source_ref,
            "authority_mode": item.authority_mode,
            "legal_basis_refs": item.legal_basis_refs,
            "purposes": item.purposes,
            "semantic_definition": item.semantic_definition,
            "schema_owner_ref": item.schema_owner_ref,
            "official_keys": item.official_keys,
            "classification": item.classification,
            "privacy_profile_ref": item.privacy_profile_ref,
            "retention_policy_ref": item.retention_policy_ref,
            "access_policy_ref": item.access_policy_ref,
            "visibility_policy": item.visibility_policy,
            "hold_refs": item.hold_refs,
            "publication_state": item.publication_state,
            "transfer_agreement_ref": item.transfer_agreement_ref,
            "freshness_policy": item.freshness_policy,
            "quality_policy": item.quality_policy,
            "approval_policy": item.approval_policy,
            "retention_policy": item.retention_policy,
            "known_limits": item.known_limits,
            "correction_procedure_ref": item.correction_procedure_ref,
            "affected_refs": item.affected_refs,
            "dependency_refs": item.dependency_refs,
        }
    )


def _apply_datasource_governance(
    item: DatasourceRecord,
    governance: DatasourceGovernance,
) -> None:
    item.owner_ref = governance.owner_ref
    item.steward_ref = governance.steward_ref
    item.responsible_organization_ref = governance.responsible_organization_ref
    item.responsible_function_ref = governance.responsible_function_ref
    item.authoritative_source_ref = governance.authoritative_source_ref
    item.authority_mode = governance.authority_mode
    item.legal_basis_refs = list(governance.legal_basis_refs)
    item.purposes = list(governance.purposes)
    item.semantic_definition = governance.semantic_definition
    item.schema_owner_ref = governance.schema_owner_ref
    item.official_keys = list(governance.official_keys)
    item.classification = governance.classification
    item.privacy_profile_ref = governance.privacy_profile_ref
    item.retention_policy_ref = governance.retention_policy_ref
    item.access_policy_ref = governance.access_policy_ref
    item.visibility_policy = dict(
        normalize_visibility_policy(
            governance.visibility_policy,
            schema=_fields(item.schema_),
        )
    )
    item.hold_refs = list(governance.hold_refs)
    item.publication_state = governance.publication_state
    item.transfer_agreement_ref = governance.transfer_agreement_ref
    item.freshness_policy = dict(governance.freshness_policy)
    item.quality_policy = dict(governance.quality_policy)
    item.approval_policy = normalize_approval_policy(governance.approval_policy)
    item.retention_policy = normalize_retention_policy(governance.retention_policy)
    item.known_limits = list(governance.known_limits)
    item.correction_procedure_ref = governance.correction_procedure_ref
    item.affected_refs = list(governance.affected_refs)
    item.dependency_refs = list(governance.dependency_refs)
    item.governance_references = [
        DatasourceGovernanceReferenceRecord(
            tenant_id=item.tenant_id,
            relation=relation,
            reference=reference,
        )
        for relation, references in (
            ("affected", governance.affected_refs),
            ("depends_on", governance.dependency_refs),
        )
        for reference in references
    ]


def _datasource_dto(item: DatasourceRecord) -> DatasourceDescriptor:
    capabilities = ["read", "preview", "freeze"]
    if item.mode == "cached" and item.provider_ref:
        capabilities.append("refresh")
    return DatasourceDescriptor(
        ref=_datasource_ref(item.id),
        source_name=item.source_name,
        name=item.name,
        description=item.description,
        kind=cast(Any, item.kind),
        mode=cast(Any, item.mode),
        shape=cast(Any, item.shape),
        status=item.status,
        provider=item.provider,
        provider_ref=item.provider_ref,
        schema=_fields(item.schema_),
        schema_version=str(item.schema_version),
        fingerprint=item.fingerprint,
        current_materialization_ref=(
            _materialization_ref(item.current_materialization_id)
            if item.current_materialization_id
            else None
        ),
        row_count=item.row_count,
        byte_count=item.byte_count,
        updated_at=item.updated_at,
        capabilities=tuple(capabilities),
        provenance=dict(item.provenance_),
        metadata=dict(item.metadata_),
        governance=_datasource_governance(item),
    )


def _materialization_dto(
    item: DatasourceMaterializationRecord,
) -> DatasourceMaterialization:
    return DatasourceMaterialization(
        ref=_materialization_ref(item.id),
        datasource_ref=_datasource_ref(item.datasource_id),
        revision=item.revision,
        state=item.state,
        fingerprint=item.fingerprint,
        schema=_fields(item.schema_),
        row_count=item.row_count,
        byte_count=item.byte_count,
        frozen_at=item.frozen_at,
        frozen_label=item.frozen_label,
        source_timestamp=item.source_timestamp,
        created_at=item.created_at,
        disposed_at=item.disposed_at,
        disposition=dict(item.disposition_),
        provenance=dict(item.provenance_),
        metadata=dict(item.metadata_),
        governance=DatasourceGovernance.from_mapping(item.governance_snapshot_),
    )


def _stage_dto(item: DatasourceStageRecord) -> DatasourceStage:
    return DatasourceStage(
        ref=_stage_ref(item.id),
        name=item.name,
        source_name=item.source_name,
        kind=cast(Any, item.kind),
        mode=cast(Any, item.mode),
        shape=cast(Any, item.shape),
        state=item.state,
        target_datasource_ref=(
            _datasource_ref(item.target_datasource_id)
            if item.target_datasource_id
            else None
        ),
        fingerprint=item.fingerprint,
        schema=_fields(item.schema_),
        row_count=item.row_count,
        byte_count=item.byte_count,
        validation=dict(item.validation_),
        approval=dict(item.approval_),
        created_at=item.created_at,
        promoted_at=item.promoted_at,
        promoted_materialization_ref=(
            _materialization_ref(item.promoted_materialization_id)
            if item.promoted_materialization_id
            else None
        ),
        provenance=dict(item.provenance_),
        metadata=dict(item.metadata_),
        governance=DatasourceGovernance.from_mapping(item.governance_),
    )


def _fields(payload: Sequence[Mapping[str, object]]) -> tuple[DatasourceField, ...]:
    return tuple(
        DatasourceField(
            name=str(item.get("name", "")),
            data_type=str(item.get("data_type", "unknown")),
            nullable=bool(item.get("nullable", True)),
            classification=str(item.get("classification") or "internal"),
        )
        for item in payload
    )


def _context(
    session: object,
    principal: object,
    required_scope: str | Sequence[str],
) -> tuple[Session, ApiPrincipal]:
    if not isinstance(session, Session):
        raise TypeError("Datasource providers require a SQLAlchemy session.")
    if not isinstance(principal, ApiPrincipal):
        raise DatasourceAccessError("A tenant API principal is required.")
    required_scopes = (
        (required_scope,) if isinstance(required_scope, str) else tuple(required_scope)
    )
    if not (
        any(has_scope(principal, scope) for scope in required_scopes)
        or has_scope(principal, ADMIN_SCOPE)
    ):
        raise DatasourceAccessError(
            f"Missing one of the required scopes: {', '.join(required_scopes)}"
        )
    return session, principal


def _publication_context(
    session: object,
    principal: object,
) -> tuple[Session, ApiPrincipal]:
    return _context(session, principal, SOURCE_WRITE_SCOPE)


def _validate_columns(
    schema: Sequence[Mapping[str, object]],
    columns: tuple[str, ...],
) -> None:
    if not columns:
        return
    known = {str(field.get("name", "")) for field in schema}
    unknown = [column for column in columns if column not in known]
    if unknown:
        raise DatasourceValidationError(
            f"Unknown datasource columns: {', '.join(unknown)}"
        )


def _validate_field_names(
    schema: Sequence[DatasourceField],
    columns: tuple[str, ...],
) -> None:
    _validate_columns(
        tuple({"name": field.name} for field in schema),
        columns,
    )


def _bounded_result_rows(
    rows: Sequence[Mapping[str, object]],
    *,
    max_bytes: int,
) -> tuple[Mapping[str, object], ...]:
    result: list[Mapping[str, object]] = []
    byte_limit = max(1, int(max_bytes))
    for row in rows:
        candidate = (*result, row)
        if encoded_size(candidate) > byte_limit:
            break
        result.append(row)
    return tuple(result)


def _opaque_descriptor(item: DatasourceDescriptor) -> DatasourceDescriptor:
    return replace(
        item,
        provider=None,
        provider_ref=None,
        provenance={"visibility": "policy_filtered"},
        metadata={},
        governance=DatasourceGovernance(
            classification=item.governance.classification,
            publication_state=item.governance.publication_state,
        ),
    )


def _opaque_materialization(
    item: DatasourceMaterialization,
) -> DatasourceMaterialization:
    return replace(
        item,
        provenance={"visibility": "policy_filtered"},
        metadata={},
        governance=DatasourceGovernance(
            classification=item.governance.classification,
            publication_state=item.governance.publication_state,
        ),
    )


def _audit_visibility_read(
    session: Session,
    principal: ApiPrincipal,
    *,
    item: DatasourceRecord,
    outcome: str,
    consistency: str,
    diagnostics: Sequence[object] = (),
) -> None:
    visibility_summary: dict[str, object] = {}
    for diagnostic in diagnostics:
        if getattr(diagnostic, "code", None) != "datasource.visibility_applied":
            continue
        details = getattr(diagnostic, "details", {})
        if isinstance(details, Mapping):
            visibility_summary = {
                key: details[key]
                for key in (
                    "policy_count",
                    "field_rule_count",
                    "row_filter_count",
                    "scanned_rows",
                    "permitted_rows",
                    "filtered_rows",
                )
                if key in details
            }
    audit_event(
        session,
        tenant_id=principal.tenant_id,
        user_id=getattr(principal.user, "id", None) or principal.account_id,
        api_key_id=principal.api_key_id,
        action=f"datasource.read.{outcome}",
        object_type="datasource",
        object_id=item.id,
        details={
            "outcome": outcome,
            "consistency": consistency,
            "policy_ref_configured": bool(item.access_policy_ref),
            "local_policy_configured": bool(item.visibility_policy),
            "diagnostic_codes": [
                str(getattr(diagnostic, "code", ""))
                for diagnostic in diagnostics
                if getattr(diagnostic, "code", None)
            ],
            "visibility_summary": visibility_summary,
        },
    )


def _select_columns(
    row: Mapping[str, object],
    columns: tuple[str, ...],
) -> Mapping[str, object]:
    if not columns:
        return dict(row)
    return {column: row.get(column) for column in columns}


def _strip_ref(value: str, prefix: str) -> str | None:
    cleaned = str(value or "").strip()
    if not cleaned:
        return None
    if cleaned.startswith(prefix):
        return cleaned[len(prefix) :]
    return cleaned if ":" not in cleaned else None


def _datasource_ref(item_id: str) -> str:
    return f"datasource:{item_id}"


def _materialization_ref(item_id: str) -> str:
    return f"materialization:{item_id}"


def _stage_ref(item_id: str) -> str:
    return f"stage:{item_id}"


def _publication_ref(item_id: str) -> str:
    return f"publication:{item_id}"


def _actor_id(principal: ApiPrincipal) -> str | None:
    return principal.account_id or principal.membership_id or principal.identity_id


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


def _int_schema_version(value: str) -> int:
    try:
        return max(1, int(value))
    except (TypeError, ValueError):
        return 1


def _escape_like(value: str) -> str:
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _valid_source_name(value: str) -> bool:
    return bool(re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,119}", value))


def _prepare_publication(
    request: DatasourcePublicationRequest,
) -> _PreparedPublication:
    producer_module = request.producer_module.strip()
    producer_run_ref = request.producer_run_ref.strip()
    idempotency_key = request.idempotency_key.strip()
    _validate_publication_identity(
        producer_module=producer_module,
        producer_run_ref=producer_run_ref,
        idempotency_key=idempotency_key,
    )
    if (request.rows is None) == (request.artifact is None):
        raise DatasourceValidationError(
            "A publication requires exactly one inline row payload or durable "
            "artifact reference."
        )
    artifact = request.artifact
    if artifact is None:
        normalized: tuple[dict[str, Any], ...] | None = normalize_rows(
            request.rows or ()
        )
        schema = infer_schema(normalized)
        fingerprint = fingerprint_rows(normalized, schema)
        row_count = len(normalized)
        byte_count = encoded_size(normalized)
    else:
        normalized = None
        schema = _validated_artifact_schema(artifact)
        fingerprint = _validated_sha256(
            artifact.fingerprint,
            label="Artifact publication fingerprints",
        )
        _validated_sha256(
            artifact.checksum,
            label="Artifact publication checksums",
        )
        if artifact.row_count < 0 or artifact.byte_count < 0:
            raise DatasourceValidationError(
                "Artifact publication sizes cannot be negative."
            )
        row_count = artifact.row_count
        byte_count = artifact.byte_count
    return _PreparedPublication(
        producer_module=producer_module,
        producer_run_ref=producer_run_ref,
        idempotency_key=idempotency_key,
        rows=normalized,
        artifact=artifact,
        schema=schema,
        fingerprint=fingerprint,
        row_count=row_count,
        byte_count=byte_count,
        request_hash=_publication_request_hash(
            request,
            normalized=normalized,
            fingerprint=fingerprint,
        ),
    )


def _validated_artifact_schema(
    artifact: DatasourceArtifactReference,
) -> tuple[DatasourceField, ...]:
    schema = tuple(artifact.schema)
    names = [field.name.strip() for field in schema]
    if not schema or any(not name for name in names):
        raise DatasourceValidationError(
            "Artifact publications require a non-empty schema."
        )
    if len(names) != len(set(names)):
        raise DatasourceValidationError(
            "Artifact publication schema field names must be unique."
        )
    return schema


def _validated_sha256(value: str, *, label: str) -> str:
    cleaned = value.strip().casefold()
    try:
        valid = len(cleaned) == 64 and int(cleaned, 16) >= 0
    except ValueError:
        valid = False
    if not valid:
        raise DatasourceValidationError(f"{label} must be SHA-256 values.")
    return cleaned


def _validate_publication(
    *,
    prepared: _PreparedPublication,
    quality_policy: Mapping[str, object],
    baseline_schema: Sequence[DatasourceField] | None,
) -> tuple[dict[str, object], DatasourcePublicationStatus]:
    if prepared.artifact is None:
        validation = validate_stage(
            rows=prepared.rows or (),
            schema=prepared.schema,
            quality_policy=quality_policy,
            baseline_schema=baseline_schema,
        )
        status: DatasourcePublicationStatus = (
            "published_with_warnings"
            if validation.get("warnings")
            else "published"
        )
        return validation, status

    raw_rules = quality_policy.get("rules", [])
    rules = (
        [item for item in raw_rules if isinstance(item, Mapping)]
        if isinstance(raw_rules, Sequence)
        and not isinstance(raw_rules, (str, bytes))
        else []
    )
    metadata_rule_types = {"required_fields", "field"}
    metadata_policy = {
        **dict(quality_policy),
        "rules": [
            dict(item)
            for item in rules
            if str(item.get("type") or "") in metadata_rule_types
        ],
    }
    validation = validate_stage(
        rows=(),
        schema=prepared.schema,
        quality_policy=metadata_policy,
        baseline_schema=baseline_schema,
    )
    if validation["valid"] is not True:
        return validation, "review_required"

    evidence = dict(prepared.artifact.validation)
    content_rule_ids = {
        str(item.get("id") or f"rule-{index + 1}")
        for index, item in enumerate(rules)
        if str(item.get("type") or "") not in metadata_rule_types
    }
    policy_probe = validate_stage(
        rows=(),
        schema=prepared.schema,
        quality_policy=quality_policy,
        baseline_schema=baseline_schema,
    )
    evidence_rule_ids = {
        str(item)
        for item in evidence.get("rules_evaluated", [])
        if str(item).strip()
    } if isinstance(evidence.get("rules_evaluated"), Sequence) and not isinstance(
        evidence.get("rules_evaluated"), (str, bytes)
    ) else set()
    evidence_verified = (
        not content_rule_ids
        or (
            evidence.get("valid") is True
            and str(evidence.get("policy_hash") or "")
            == str(policy_probe["policy_hash"])
            and str(evidence.get("payload_checksum") or "").casefold()
            == prepared.artifact.checksum.casefold()
            and content_rule_ids.issubset(evidence_rule_ids)
        )
    )
    requested_status = str(evidence.get("status") or "verified")
    if evidence.get("valid") is False or requested_status == "failed":
        errors = evidence.get("errors")
        validation["valid"] = False
        validation["errors"] = (
            list(errors)
            if isinstance(errors, Sequence)
            and not isinstance(errors, (str, bytes))
            else [
                {
                    "severity": "error",
                    "code": "quality.artifact_validation",
                    "message": "Artifact validation evidence reports failure.",
                }
            ]
        )
        return validation, "review_required"

    evidence_warnings = evidence.get("warnings")
    if isinstance(evidence_warnings, Sequence) and not isinstance(
        evidence_warnings, (str, bytes)
    ):
        validation["warnings"] = [
            *list(validation.get("warnings", [])),
            *list(evidence_warnings),
        ]
    validation["artifact_evidence"] = {
        "verified": evidence_verified,
        "rules_required": sorted(content_rule_ids),
        "rules_evaluated": sorted(evidence_rule_ids),
        "payload_checksum": prepared.artifact.checksum.casefold(),
    }
    if not evidence_verified or requested_status == "review_required":
        return validation, "review_required"
    if validation.get("warnings") or requested_status == "warning":
        return validation, "published_with_warnings"
    return validation, "published"


def _validate_publication_identity(
    *,
    producer_module: str,
    producer_run_ref: str,
    idempotency_key: str,
) -> None:
    values = (
        (producer_module, 100, "A producer module"),
        (producer_run_ref, 500, "A producer run reference"),
        (idempotency_key, 255, "An idempotency key"),
    )
    for value, maximum, label in values:
        if not value or len(value) > maximum:
            raise DatasourceValidationError(
                f"{label} of at most {maximum} characters is required."
            )


def _publication_lock_key(
    *,
    tenant_id: str,
    producer_module: str,
    idempotency_key: str,
) -> int:
    payload = (
        f"govoplan.datasources.publication\0{tenant_id}\0"
        f"{producer_module}\0{idempotency_key}"
    ).encode("utf-8")
    return int.from_bytes(
        hashlib.sha256(payload).digest()[:8],
        "big",
        signed=True,
    )


def _lock_publication_identity(
    session: Session,
    *,
    tenant_id: str,
    producer_module: str,
    idempotency_key: str,
) -> None:
    if session.get_bind().dialect.name != "postgresql":
        return
    session.execute(
        text("SELECT pg_advisory_xact_lock(:key)"),
        {
            "key": _publication_lock_key(
                tenant_id=tenant_id,
                producer_module=producer_module,
                idempotency_key=idempotency_key,
            )
        },
    )


def _existing_publication_result(
    session: Session,
    *,
    tenant_id: str,
    producer_module: str,
    idempotency_key: str,
    request_hash: str,
) -> DatasourcePublicationResult | None:
    publication = session.scalar(
        select(DatasourcePublicationRecord).where(
            DatasourcePublicationRecord.tenant_id == tenant_id,
            DatasourcePublicationRecord.producer_module == producer_module,
            DatasourcePublicationRecord.idempotency_key == idempotency_key,
        )
    )
    if publication is None:
        return None
    if publication.request_hash != request_hash:
        raise DatasourceValidationError(
            "The publication idempotency key was already used with different output."
        )
    datasource = session.get(DatasourceRecord, publication.datasource_id)
    materialization = session.get(
        DatasourceMaterializationRecord,
        publication.materialization_id,
    )
    if (
        datasource is None
        or materialization is None
        or datasource.tenant_id != tenant_id
        or materialization.tenant_id != tenant_id
    ):
        raise DatasourceUnavailableError(
            "The prior publication result is no longer available."
        )
    return DatasourcePublicationResult(
        ref=_publication_ref(publication.id),
        status=cast(DatasourcePublicationStatus, publication.status),
        datasource=_datasource_dto(datasource),
        materialization=_materialization_dto(materialization),
        replayed=True,
    )


def _publication_target(
    session: Session,
    *,
    tenant_id: str,
    actor_id: str | None,
    request: DatasourcePublicationRequest,
    prepared: _PreparedPublication,
    target: DatasourceRecord | None,
    governance: DatasourceGovernance,
    publish_as_current: bool,
) -> DatasourceRecord:
    if target is not None:
        datasource = target
        if request.governance is not None:
            _apply_datasource_governance(datasource, governance)
            datasource.updated_by = actor_id
        return datasource
    name = str(request.name or "").strip()
    source_name = str(request.source_name or "").strip()
    if not name:
        raise DatasourceValidationError(
            "A datasource name is required for a new publication target."
        )
    if not _valid_source_name(source_name):
        raise DatasourceValidationError(
            "Datasource keys must start with a letter or underscore and contain "
            "only letters, numbers, and underscores."
        )
    _ensure_source_name_available(
        session,
        tenant_id=tenant_id,
        source_name=source_name,
    )
    datasource = DatasourceRecord(
        tenant_id=tenant_id,
        source_name=source_name,
        name=name,
        description=_clean_optional(request.description),
        kind="custom",
        mode="static",
        shape="tabular",
        status="active",
        provider=prepared.producer_module,
        provider_ref=prepared.producer_run_ref,
        schema_version=1,
        schema_=[field_payload(field) for field in prepared.schema],
        fingerprint=prepared.fingerprint if publish_as_current else "",
        row_count=prepared.row_count if publish_as_current else None,
        byte_count=prepared.byte_count if publish_as_current else None,
        provenance_={
            **dict(request.provenance),
            "producer_module": prepared.producer_module,
            "producer_run_ref": prepared.producer_run_ref,
        },
        metadata_=dict(request.metadata),
        created_by=actor_id,
        updated_by=actor_id,
    )
    _apply_datasource_governance(
        datasource,
        governance,
    )
    session.add(datasource)
    session.flush()
    return datasource


def _publication_validation_context(
    session: Session,
    *,
    tenant_id: str,
    request: DatasourcePublicationRequest,
) -> tuple[
    DatasourceRecord | None,
    DatasourceGovernance,
    tuple[DatasourceField, ...] | None,
]:
    if not request.target_datasource_ref:
        return (
            None,
            request.governance
            or _default_governance(mode="static", provider_ref=None),
            None,
        )
    target = _required_datasource(
        session,
        tenant_id=tenant_id,
        datasource_ref=request.target_datasource_ref,
        for_update=True,
    )
    if target.mode == "live" or target.shape != "tabular":
        raise DatasourceValidationError(
            "Produced rows require a static or cached tabular datasource."
        )
    return (
        target,
        request.governance or _datasource_governance(target),
        _fields(target.schema_) if target.current_materialization_id else None,
    )


def _publication_provenance(
    request: DatasourcePublicationRequest,
    prepared: _PreparedPublication,
) -> dict[str, object]:
    return {
        **dict(request.provenance),
        "producer_module": prepared.producer_module,
        "producer_run_ref": prepared.producer_run_ref,
        "idempotency_key": prepared.idempotency_key,
        "published_at": utcnow().isoformat(),
    }


def _create_publication_record(
    session: Session,
    *,
    tenant_id: str,
    actor_id: str | None,
    datasource: DatasourceRecord,
    materialization: DatasourceMaterializationRecord,
    request: DatasourcePublicationRequest,
    prepared: _PreparedPublication,
    status: DatasourcePublicationStatus,
) -> DatasourcePublicationRecord:
    publication = DatasourcePublicationRecord(
        tenant_id=tenant_id,
        producer_module=prepared.producer_module,
        producer_run_ref=prepared.producer_run_ref,
        idempotency_key=prepared.idempotency_key,
        request_hash=prepared.request_hash,
        datasource_id=datasource.id,
        materialization_id=materialization.id,
        status=status,
        details_={
            "fingerprint": prepared.fingerprint,
            "row_count": prepared.row_count,
            "set_current": (
                request.set_current and status != "review_required"
            ),
            "frozen": request.freeze,
            "payload_backend": (
                prepared.artifact.backend
                if prepared.artifact is not None
                else "database_rows"
            ),
            "validation": materialization.provenance_.get(
                "publication_validation",
                {},
            ),
        },
        created_by=actor_id,
    )
    session.add(publication)
    session.flush()
    return publication


def _publication_request_hash(
    request: DatasourcePublicationRequest,
    *,
    normalized: Sequence[Mapping[str, object]] | None,
    fingerprint: str,
) -> str:
    payload = {
        "producer_module": request.producer_module.strip(),
        "producer_run_ref": request.producer_run_ref.strip(),
        "target_datasource_ref": request.target_datasource_ref,
        "name": request.name,
        "source_name": request.source_name,
        "description": request.description,
        "rows": [dict(row) for row in (normalized or ())],
        "artifact": (
            {
                "backend": request.artifact.backend,
                "locator": request.artifact.locator,
                "checksum": request.artifact.checksum.casefold(),
                "row_count": request.artifact.row_count,
                "byte_count": request.artifact.byte_count,
                "schema": [
                    field_payload(field) for field in request.artifact.schema
                ],
                "fingerprint": request.artifact.fingerprint.casefold(),
                "media_type": request.artifact.media_type,
                "checkpoint": dict(request.artifact.checkpoint),
                "metadata": dict(request.artifact.metadata),
                "validation": dict(request.artifact.validation),
            }
            if request.artifact is not None
            else None
        ),
        "fingerprint": fingerprint,
        "freeze": request.freeze,
        "frozen_label": request.frozen_label,
        "set_current": request.set_current,
        "source_timestamp": request.source_timestamp,
        "provenance": dict(request.provenance),
        "metadata": dict(request.metadata),
    }
    encoded = json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


__all__ = [
    "ADMIN_SCOPE",
    "CATALOGUE_READ_SCOPE",
    "SOURCE_WRITE_SCOPE",
    "STAGE_APPROVE_SCOPE",
    "STAGE_WRITE_SCOPE",
    "SqlDatasourceProvider",
]