DataOps-platform/app/runner/artifacts.py

"""Digest-bound, bounded Parquet artifacts owned by the DataOps Runner."""

from __future__ import annotations

import hashlib
import json
import os
import re
import tempfile
from collections.abc import Mapping
from contextlib import contextmanager, suppress
from datetime import UTC, datetime, timedelta
from typing import Any

import polars as pl
import pyarrow.parquet as pq
from sqlalchemy import text

from app.core.common.identifiers import (
    ensure_governance_uid,
    new_governance_uid,
)
from app.core.data_rules.execution_contracts import canonical_schema_hash

PARQUET_CONTENT_TYPE = "application/x-parquet"
_DIGEST = re.compile(r"^[0-9a-f]{64}$")


def _parquet_footer_bounds(path: str) -> tuple[int, int]:
    try:
        metadata = pq.ParquetFile(path).metadata
    except Exception as exc:
        raise ValueError("artifact Parquet footer is invalid") from exc
    if metadata is None or metadata.num_row_groups < 1:
        raise ValueError("artifact Parquet footer is incomplete")
    uncompressed = sum(
        metadata.row_group(index).total_byte_size
        for index in range(metadata.num_row_groups)
    )
    if metadata.num_rows < 0 or uncompressed < 0:
        raise ValueError("artifact Parquet footer is invalid")
    return metadata.num_rows, uncompressed


def _now_utc(clock) -> datetime:
    value = clock()
    if not isinstance(value, datetime):
        raise ValueError("artifact clock must return a datetime")
    if value.tzinfo is None:
        value = value.replace(tzinfo=UTC)
    return value.astimezone(UTC)


def _timestamp(value: datetime) -> str:
    return value.astimezone(UTC).isoformat().replace("+00:00", "Z")


def _parse_timestamp(value: Any) -> datetime:
    if not isinstance(value, str) or not value.endswith("Z"):
        raise ValueError("artifact expiry metadata is invalid")
    try:
        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
    except ValueError as exc:
        raise ValueError("artifact expiry metadata is invalid") from exc
    return parsed.astimezone(UTC)


def _uid(value: Any, label: str) -> str:
    try:
        return ensure_governance_uid({"uid": str(value)})
    except ValueError as exc:
        raise ValueError(f"{label} must be a valid UUIDv7") from exc


def _inferred_schema_fields(
    frame: pl.DataFrame | pl.LazyFrame,
) -> list[dict[str, Any]]:
    schema = (
        frame.collect_schema()
        if isinstance(frame, pl.LazyFrame)
        else frame.schema
    )
    fields = []
    for name, dtype in schema.items():
        field: dict[str, Any] = {
            "name": name,
            "nullable": True,
        }
        if dtype == pl.Boolean:
            field["type"] = "boolean"
        elif dtype == pl.Date:
            field["type"] = "date"
        elif dtype == pl.String:
            field["type"] = "string"
        elif dtype.is_integer():
            field["type"] = "integer"
        elif dtype == pl.Float32:
            field["type"] = "float"
        elif dtype == pl.Float64:
            field["type"] = "double"
        elif dtype.is_decimal():
            field.update(
                {
                    "type": "decimal",
                    "precision": dtype.precision,
                    "scale": dtype.scale,
                }
            )
        elif isinstance(dtype, pl.Datetime):
            field["type"] = (
                "timestamptz" if dtype.time_zone else "timestamp"
            )
            if dtype.time_zone:
                field["timezone"] = dtype.time_zone
        else:
            raise ValueError(f"unsupported artifact dtype for {name}")
        fields.append(field)
    return sorted(fields, key=lambda item: item["name"])


def _normalized_schema_fields(value: Any) -> list[dict[str, Any]]:
    canonical_schema_hash(value)
    return sorted(
        [dict(field) for field in value],
        key=lambda item: item["name"],
    )


def _schema_contract(value: Any) -> tuple[list[dict[str, Any]], str]:
    fields = _normalized_schema_fields(value)
    return fields, canonical_schema_hash(fields)


def _validate_frame_schema(
    frame: pl.DataFrame | pl.LazyFrame,
    fields: list[dict[str, Any]],
) -> None:
    schema = (
        frame.collect_schema()
        if isinstance(frame, pl.LazyFrame)
        else frame.schema
    )
    expected_names = {field["name"] for field in fields}
    if set(schema.names()) != expected_names:
        raise ValueError("artifact schema fields do not match")
    for field in fields:
        dtype = schema[field["name"]]
        field_type = field["type"]
        matches = (
            (field_type == "boolean" and dtype == pl.Boolean)
            or (field_type == "date" and dtype == pl.Date)
            or (field_type == "string" and dtype == pl.String)
            or (field_type == "integer" and dtype.is_integer())
            or (field_type == "float" and dtype == pl.Float32)
            or (field_type == "double" and dtype == pl.Float64)
            or (
                field_type == "decimal"
                and dtype.is_decimal()
                and dtype.precision == field.get("precision")
                and dtype.scale == field.get("scale")
            )
            or (
                field_type == "timestamp"
                and isinstance(dtype, pl.Datetime)
                and dtype.time_zone is None
            )
            or (
                field_type == "timestamptz"
                and isinstance(dtype, pl.Datetime)
                and dtype.time_zone == field.get("timezone")
            )
        )
        if not matches:
            raise ValueError(
                f"artifact schema type for {field['name']} does not match"
            )
    if isinstance(frame, pl.DataFrame):
        for field in fields:
            if not field["nullable"] and frame[field["name"]].null_count():
                raise ValueError(
                    f"artifact nullable contract for {field['name']} does not match"
                )


def _metadata(value: Any) -> dict[str, str]:
    if not isinstance(value, Mapping):
        raise ValueError("artifact content metadata is missing")
    normalized = {}
    for key, item in value.items():
        name = str(key).lower()
        if name.startswith("x-amz-meta-"):
            name = name[len("x-amz-meta-") :]
        if name in {
            "sha256",
            "row-count",
            "schema-sha256",
            "expires-at",
            "artifact-bytes",
        }:
            normalized[name] = str(item)
    required = {
        "sha256",
        "row-count",
        "schema-sha256",
        "expires-at",
        "artifact-bytes",
    }
    if set(normalized) != required:
        raise ValueError("artifact content metadata is incomplete")
    if sum(len(key) + len(item) for key, item in normalized.items()) > 2_048:
        raise ValueError("artifact content metadata exceeds the safe limit")
    return normalized


class ArtifactStore:
    """Read and write only server-owned, bounded Parquet artifacts."""

    def __init__(
        self,
        client,
        *,
        bucket: str,
        max_artifact_bytes: int,
        max_rows: int,
        memory_limit_bytes: int,
        max_ttl_seconds: int = 86400,
        clock=None,
    ):
        if not re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", bucket):
            raise ValueError("artifact bucket name is invalid")
        self.client = client
        self.bucket = bucket
        self.max_artifact_bytes = int(max_artifact_bytes)
        self.max_rows = int(max_rows)
        self.memory_limit_bytes = int(memory_limit_bytes)
        self.max_ttl_seconds = int(max_ttl_seconds)
        self.clock = clock or (lambda: datetime.now(UTC))
        if (
            self.max_artifact_bytes < 1024
            or self.max_rows < 1
            or self.memory_limit_bytes < self.max_artifact_bytes
            or self.max_ttl_seconds < 1
        ):
            raise ValueError("artifact resource limits are invalid")
        if not self.client.bucket_exists(self.bucket):
            raise ValueError("artifact bucket does not exist")

    def _limits(self, value: Any = None) -> dict[str, int]:
        configured = {
            "max_rows": self.max_rows,
            "max_artifact_bytes": self.max_artifact_bytes,
            "memory_limit_bytes": self.memory_limit_bytes,
        }
        if value is None:
            return configured
        if not isinstance(value, dict) or set(value) != set(configured):
            raise ValueError("artifact limits must have a closed shape")
        result = {}
        for key, ceiling in configured.items():
            item = value[key]
            if (
                isinstance(item, bool)
                or not isinstance(item, int)
                or item < 1
                or item > ceiling
            ):
                raise ValueError(
                    f"artifact {key} exceeds the configured ceiling"
                )
            result[key] = item
        return result

    def _parse_ref(self, ref: Any) -> str:
        prefix = f"minio://{self.bucket}/"
        if not isinstance(ref, str) or not ref.startswith(prefix):
            raise ValueError("artifact reference is not owned by this store")
        key = ref[len(prefix) :]
        match = re.fullmatch(
            r"rules/([0-9a-f-]{36})/([0-9a-f-]{36})\.parquet",
            key,
        )
        if match is None:
            raise ValueError("artifact reference is invalid")
        _uid(match.group(1), "artifact correlation id")
        _uid(match.group(2), "artifact id")
        return key

    def _validated_stat(
        self,
        key: str,
        *,
        expected_digest: str | None = None,
        limits: dict[str, int] | None = None,
    ) -> tuple[Any, dict[str, str]]:
        effective = self._limits(limits)
        stat = self.client.stat_object(self.bucket, key)
        size = int(getattr(stat, "size", -1))
        if size < 1 or size > effective["max_artifact_bytes"]:
            raise ValueError("artifact size exceeds the configured limit")
        if size > effective["memory_limit_bytes"]:
            raise ValueError("artifact download exceeds the memory limit")
        if str(getattr(stat, "content_type", "")).lower() != PARQUET_CONTENT_TYPE:
            raise ValueError("artifact content type is invalid")
        metadata = _metadata(getattr(stat, "metadata", None))
        digest = metadata["sha256"]
        if _DIGEST.fullmatch(digest) is None:
            raise ValueError("artifact digest metadata is invalid")
        if expected_digest is not None and digest != expected_digest:
            raise ValueError("artifact digest does not match")
        try:
            row_count = int(metadata["row-count"])
            metadata_size = int(metadata["artifact-bytes"])
        except (TypeError, ValueError) as exc:
            raise ValueError("artifact count metadata is invalid") from exc
        if row_count < 0 or row_count > effective["max_rows"]:
            raise ValueError("artifact row count exceeds the configured limit")
        if metadata_size != size:
            raise ValueError("artifact size metadata does not match")
        if _DIGEST.fullmatch(metadata["schema-sha256"]) is None:
            raise ValueError("artifact schema metadata is invalid")
        if _parse_timestamp(metadata["expires-at"]) <= _now_utc(self.clock):
            raise ValueError("artifact has expired")
        return stat, metadata

    def write(
        self,
        frame: pl.LazyFrame | pl.DataFrame,
        correlation_id: str,
        ttl_seconds: int,
        *,
        schema_fields: list[dict[str, Any]] | None = None,
        limits: dict[str, int] | None = None,
    ) -> dict[str, Any]:
        effective = self._limits(limits)
        correlation = _uid(correlation_id, "correlation_id")
        if (
            isinstance(ttl_seconds, bool)
            or not isinstance(ttl_seconds, int)
            or ttl_seconds < 1
            or ttl_seconds > self.max_ttl_seconds
        ):
            raise ValueError("artifact TTL is outside the configured limit")
        if isinstance(frame, pl.DataFrame):
            lazy = frame.lazy()
        elif isinstance(frame, pl.LazyFrame):
            lazy = frame
        else:
            raise ValueError("artifact frame must be a Polars frame")
        collected = lazy.head(effective["max_rows"] + 1).collect(
            engine="streaming"
        )
        if collected.height > effective["max_rows"]:
            raise ValueError("artifact row count exceeds the configured limit")
        if collected.estimated_size() > effective["memory_limit_bytes"]:
            raise ValueError("artifact frame exceeds the configured memory limit")
        fields, schema_digest = _schema_contract(
            schema_fields or _inferred_schema_fields(collected)
        )
        _validate_frame_schema(collected, fields)
        expires_at = _timestamp(
            _now_utc(self.clock) + timedelta(seconds=ttl_seconds)
        )
        artifact_id = new_governance_uid()
        key = f"rules/{correlation}/{artifact_id}.parquet"
        path = None
        uploaded = False
        try:
            with tempfile.NamedTemporaryFile(
                prefix="dataops-rule-artifact-",
                suffix=".parquet",
                delete=False,
            ) as handle:
                path = handle.name
            collected.write_parquet(path)
            size = os.path.getsize(path)
            if size < 1 or size > effective["max_artifact_bytes"]:
                raise ValueError("artifact size exceeds the configured limit")
            if size > effective["memory_limit_bytes"]:
                raise ValueError("serialized artifact exceeds the memory limit")
            if (
                size + collected.estimated_size()
                > effective["memory_limit_bytes"]
            ):
                raise ValueError(
                    "artifact serialization exceeds the memory limit"
                )
            digest = hashlib.sha256()
            with open(path, "rb") as handle:
                while chunk := handle.read(1024 * 1024):
                    digest.update(chunk)
            digest_hex = digest.hexdigest()
            with open(path, "rb") as handle:
                self.client.put_object(
                    self.bucket,
                    key,
                    handle,
                    size,
                    content_type=PARQUET_CONTENT_TYPE,
                    metadata={
                        "sha256": digest_hex,
                        "row-count": str(collected.height),
                        "schema-sha256": schema_digest,
                        "expires-at": expires_at,
                        "artifact-bytes": str(size),
                    },
                )
            uploaded = True
            self._validated_stat(
                key,
                expected_digest=digest_hex,
                limits=effective,
            )
            artifact_ref = f"minio://{self.bucket}/{key}"
            with self.stage(
                artifact_ref,
                digest_hex,
                expected_schema_fields=fields,
                limits=effective,
            ):
                pass
        except Exception:
            if uploaded:
                with suppress(Exception):
                    self.client.remove_object(self.bucket, key)
            raise
        finally:
            if path is not None:
                with suppress(FileNotFoundError):
                    os.unlink(path)
        return {
            "artifact_ref": artifact_ref,
            "digest": digest_hex,
            "row_count": collected.height,
            "schema_hash": schema_digest,
            "schema_fields": fields,
            "expires_at": expires_at,
        }

    def write_path(
        self,
        path: str,
        correlation_id: str,
        ttl_seconds: int,
        *,
        schema_fields: list[dict[str, Any]],
        limits: dict[str, int] | None = None,
    ) -> dict[str, Any]:
        """Upload a worker-produced Parquet file without collecting it."""

        effective = self._limits(limits)
        correlation = _uid(correlation_id, "correlation_id")
        if (
            isinstance(ttl_seconds, bool)
            or not isinstance(ttl_seconds, int)
            or ttl_seconds < 1
            or ttl_seconds > self.max_ttl_seconds
        ):
            raise ValueError("artifact TTL is outside the configured limit")
        if not isinstance(path, str) or not os.path.isfile(path):
            raise ValueError("worker artifact path is invalid")
        size = os.path.getsize(path)
        if size < 1 or size > effective["max_artifact_bytes"]:
            raise ValueError("artifact size exceeds the configured limit")
        rows, uncompressed = _parquet_footer_bounds(path)
        if rows > effective["max_rows"]:
            raise ValueError("artifact row count exceeds the configured limit")
        if uncompressed > effective["memory_limit_bytes"]:
            raise ValueError(
                "Parquet footer uncompressed size exceeds the memory limit"
            )
        fields, schema_digest = _schema_contract(schema_fields)
        _validate_frame_schema(pl.scan_parquet(path), fields)
        digest = hashlib.sha256()
        with open(path, "rb") as handle:
            while chunk := handle.read(1024 * 1024):
                digest.update(chunk)
        digest_hex = digest.hexdigest()
        expires_at = _timestamp(
            _now_utc(self.clock) + timedelta(seconds=ttl_seconds)
        )
        artifact_id = new_governance_uid()
        key = f"rules/{correlation}/{artifact_id}.parquet"
        uploaded = False
        try:
            with open(path, "rb") as handle:
                self.client.put_object(
                    self.bucket,
                    key,
                    handle,
                    size,
                    content_type=PARQUET_CONTENT_TYPE,
                    metadata={
                        "sha256": digest_hex,
                        "row-count": str(rows),
                        "schema-sha256": schema_digest,
                        "expires-at": expires_at,
                        "artifact-bytes": str(size),
                    },
                )
            uploaded = True
            artifact_ref = f"minio://{self.bucket}/{key}"
            with self.stage(
                artifact_ref,
                digest_hex,
                expected_schema_fields=fields,
                limits=effective,
            ):
                pass
        except Exception:
            if uploaded:
                with suppress(Exception):
                    self.client.remove_object(self.bucket, key)
            raise
        return {
            "artifact_ref": artifact_ref,
            "digest": digest_hex,
            "row_count": rows,
            "schema_hash": schema_digest,
            "schema_fields": fields,
            "expires_at": expires_at,
        }

    def describe(self, ref: str) -> dict[str, Any]:
        """Return validated object metadata without exposing MinIO credentials."""

        key = self._parse_ref(ref)
        _stat, metadata = self._validated_stat(key)
        return {
            "artifact_ref": ref,
            "digest": metadata["sha256"],
            "row_count": int(metadata["row-count"]),
            "schema_hash": metadata["schema-sha256"],
            "expires_at": metadata["expires-at"],
        }

    def read(
        self,
        ref: str,
        expected_digest: str,
        *,
        expected_schema_fields: list[dict[str, Any]] | None = None,
        limits: dict[str, int] | None = None,
    ) -> pl.LazyFrame:
        with self.stage(
            ref,
            expected_digest,
            expected_schema_fields=expected_schema_fields,
            limits=limits,
        ) as path:
            effective = self._limits(limits)
            try:
                frame = pl.read_parquet(
                    path,
                    n_rows=effective["max_rows"] + 1,
                    memory_map=False,
                )
            except Exception as exc:
                raise ValueError("artifact is not valid Parquet") from exc
        if frame.estimated_size() > effective["memory_limit_bytes"]:
            raise ValueError("artifact frame exceeds the configured memory limit")
        _validate_frame_schema(
            frame, _normalized_schema_fields(expected_schema_fields)
        )
        return frame.lazy()

    @contextmanager
    def stage(
        self,
        ref: str,
        expected_digest: str,
        *,
        expected_schema_fields: list[dict[str, Any]] | None = None,
        limits: dict[str, int] | None = None,
    ):
        effective = self._limits(limits)
        if _DIGEST.fullmatch(str(expected_digest or "")) is None:
            raise ValueError("expected artifact digest is invalid")
        key = self._parse_ref(ref)
        _stat, metadata = self._validated_stat(
            key,
            expected_digest=expected_digest,
            limits=effective,
        )
        if expected_schema_fields is None:
            raise ValueError("expected artifact schema fields are required")
        fields = _normalized_schema_fields(expected_schema_fields)
        if canonical_schema_hash(fields) != metadata["schema-sha256"]:
            raise ValueError("artifact schema contract is not expected")
        response = self.client.get_object(self.bucket, key)
        digest = hashlib.sha256()
        path = None
        size = 0
        try:
            try:
                with tempfile.NamedTemporaryFile(
                    prefix="dataops-rule-stage-",
                    suffix=".parquet",
                    delete=False,
                ) as handle:
                    path = handle.name
                    while chunk := response.read(1024 * 1024):
                        size += len(chunk)
                        if size > effective["max_artifact_bytes"]:
                            raise ValueError(
                                "artifact size exceeds the configured limit"
                            )
                        digest.update(chunk)
                        handle.write(chunk)
            finally:
                response.close()
                release = getattr(response, "release_conn", None)
                if callable(release):
                    release()
            if digest.hexdigest() != expected_digest:
                raise ValueError("artifact digest does not match content")
            footer_rows, uncompressed = _parquet_footer_bounds(path)
            if footer_rows != int(metadata["row-count"]):
                raise ValueError("artifact row count does not match metadata")
            if footer_rows > effective["max_rows"]:
                raise ValueError(
                    "artifact row count exceeds the configured limit"
                )
            if uncompressed > effective["memory_limit_bytes"]:
                raise ValueError(
                    "Parquet footer uncompressed size exceeds the memory limit"
                )
            lazy = pl.scan_parquet(path)
            _validate_frame_schema(lazy, fields)
            yield path
        finally:
            if path is not None:
                with suppress(FileNotFoundError):
                    os.unlink(path)

    def cleanup_expired(self, correlation_id: str) -> int:
        correlation = _uid(correlation_id, "correlation_id")
        prefix = f"rules/{correlation}/"
        removed = 0
        for item in self.client.list_objects(
            self.bucket,
            prefix=prefix,
            recursive=True,
        ):
            key = str(getattr(item, "object_name", ""))
            if not key.startswith(prefix):
                continue
            try:
                self._parse_ref(f"minio://{self.bucket}/{key}")
                stat = self.client.stat_object(self.bucket, key)
                metadata = _metadata(getattr(stat, "metadata", None))
                expired = _parse_timestamp(
                    metadata["expires-at"]
                ) <= _now_utc(self.clock)
            except ValueError:
                continue
            if expired:
                self.client.remove_object(self.bucket, key)
                removed += 1
        return removed

    def delete(self, ref: str) -> None:
        """Delete one exact store-owned artifact after validating its key."""

        key = self._parse_ref(ref)
        self.client.remove_object(self.bucket, key)


class PostgresArtifactResolver:
    """Resolve a canonical artifact binding without accepting caller paths."""

    def __init__(self, engine, artifact_store: ArtifactStore):
        self.engine = engine
        self.artifact_store = artifact_store

    def resolve(
        self,
        *,
        binding_id: str,
        correlation_id: str,
        kind: str,
    ) -> dict[str, Any]:
        binding = _uid(binding_id, "artifact binding id")
        correlation = _uid(correlation_id, "artifact correlation id")
        if kind not in {"input", "lookup", "output"}:
            raise ValueError("artifact kind is invalid")
        statement = text(
            """
            SELECT
                a.artifact_ref,
                a.artifact_digest,
                a.row_count,
                a.schema_hash,
                a.schema_fields,
                a.expires_at,
                b.binding_hash
            FROM public.rule_run_artifacts a
            JOIN public.dataflow_dataset_bindings b
              ON b.id = a.binding_id
            WHERE a.binding_id = CAST(:binding_id AS uuid)
              AND a.correlation_id = CAST(:correlation_id AS uuid)
              AND a.artifact_kind = :artifact_kind
              AND a.expires_at > CURRENT_TIMESTAMP
              AND b.object_kind = 'parquet_artifact'
              AND b.access_mode IN ('read', 'read_write')
            ORDER BY a.created_at DESC
            LIMIT 1
            """
        )
        with self.engine.connect() as connection:
            row = connection.execute(
                statement,
                {
                    "binding_id": binding,
                    "correlation_id": correlation,
                    "artifact_kind": kind,
                },
            ).mappings().one_or_none()
        if row is None:
            raise ValueError("canonical artifact binding was not found")
        artifact_ref = str(row["artifact_ref"])
        key = self.artifact_store._parse_ref(artifact_ref)
        if not key.startswith(f"rules/{correlation}/"):
            raise ValueError(
                "catalog artifact does not match the execution correlation"
            )
        described = self.artifact_store.describe(artifact_ref)
        row_fields = row["schema_fields"]
        if isinstance(row_fields, str):
            row_fields = json.loads(row_fields)
        if (
            described["digest"] != str(row["artifact_digest"])
            or described["row_count"] != int(row["row_count"])
            or described["schema_hash"] != str(row["schema_hash"])
        ):
            raise ValueError("catalog artifact metadata does not match storage")
        return {
            **described,
            "schema_fields": _normalized_schema_fields(row_fields),
            "binding_hash": str(row["binding_hash"]),
        }

    def attest_binding(
        self,
        *,
        binding_id: str,
        binding_hash: str,
        access_mode: str,
    ) -> dict[str, str]:
        binding = _uid(binding_id, "artifact binding id")
        if _DIGEST.fullmatch(str(binding_hash or "")) is None:
            raise ValueError("artifact binding hash is invalid")
        allowed = {
            "read": {"read", "read_write"},
            "write": {"write", "read_write"},
        }.get(access_mode)
        if allowed is None:
            raise ValueError("artifact access mode is invalid")
        with self.engine.connect() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT binding_hash, access_mode, object_kind
                    FROM public.dataflow_dataset_bindings
                    WHERE id = CAST(:binding_id AS uuid)
                    """
                ),
                {"binding_id": binding},
            ).mappings().one_or_none()
        if (
            row is None
            or row["object_kind"] != "parquet_artifact"
            or row["access_mode"] not in allowed
            or str(row["binding_hash"]) != binding_hash
        ):
            raise ValueError("canonical artifact binding no longer matches")
        return {"binding_hash": str(row["binding_hash"])}

    def register(
        self,
        *,
        binding_id: str,
        correlation_id: str,
        artifact: dict[str, Any],
        kind: str,
        binding_hash: str,
    ) -> dict[str, Any]:
        binding = _uid(binding_id, "artifact binding id")
        correlation = _uid(correlation_id, "artifact correlation id")
        if kind not in {"input", "lookup", "output"}:
            raise ValueError("artifact kind is invalid")
        self.attest_binding(
            binding_id=binding,
            binding_hash=binding_hash,
            access_mode="write" if kind == "output" else "read",
        )
        if not isinstance(artifact, dict):
            raise ValueError("artifact metadata is invalid")
        artifact_ref = artifact.get("artifact_ref")
        key = self.artifact_store._parse_ref(artifact_ref)
        if not key.startswith(f"rules/{correlation}/"):
            raise ValueError(
                "artifact does not match the execution correlation"
            )
        described = self.artifact_store.describe(artifact_ref)
        for key in (
            "artifact_ref",
            "digest",
            "row_count",
            "schema_hash",
            "expires_at",
        ):
            if described[key] != artifact.get(key):
                raise ValueError("artifact metadata does not match storage")
        fields = _normalized_schema_fields(artifact.get("schema_fields"))
        if canonical_schema_hash(fields) != described["schema_hash"]:
            raise ValueError("artifact schema contract does not match storage")
        inserted = None
        with self.engine.begin() as connection:
            inserted = connection.execute(
                text(
                    """
                    INSERT INTO public.rule_run_artifacts (
                        id, correlation_id, binding_id, artifact_ref,
                        artifact_digest, row_count, schema_hash, schema_fields,
                        artifact_kind, expires_at
                    ) VALUES (
                        CAST(:id AS uuid), CAST(:correlation_id AS uuid),
                        CAST(:binding_id AS uuid), :artifact_ref,
                        :artifact_digest, :row_count, :schema_hash,
                        CAST(:schema_fields AS jsonb), :artifact_kind,
                        CAST(:expires_at AS timestamptz)
                    )
                    ON CONFLICT (
                        correlation_id, binding_id, artifact_kind
                    ) DO NOTHING
                    RETURNING artifact_ref, artifact_digest, row_count,
                              schema_hash, schema_fields, expires_at
                    """
                ),
                {
                    "id": new_governance_uid(),
                    "correlation_id": correlation,
                    "binding_id": binding,
                    "artifact_ref": described["artifact_ref"],
                    "artifact_digest": described["digest"],
                    "row_count": described["row_count"],
                    "schema_hash": described["schema_hash"],
                    "schema_fields": json.dumps(
                        fields,
                        sort_keys=True,
                        separators=(",", ":"),
                    ),
                    "artifact_kind": kind,
                    "expires_at": described["expires_at"],
                },
            ).mappings().one_or_none()
            if inserted is None:
                inserted = connection.execute(
                    text(
                        """
                        SELECT artifact_ref, artifact_digest, row_count,
                               schema_hash, schema_fields, expires_at
                        FROM public.rule_run_artifacts
                        WHERE correlation_id = CAST(:correlation_id AS uuid)
                          AND binding_id = CAST(:binding_id AS uuid)
                          AND artifact_kind = :artifact_kind
                        FOR UPDATE
                        """
                    ),
                    {
                        "correlation_id": correlation,
                        "binding_id": binding,
                        "artifact_kind": kind,
                    },
                ).mappings().one_or_none()
        if inserted is None:
            self.artifact_store.delete(artifact_ref)
            raise ValueError("immutable artifact catalog handoff was not found")
        existing = dict(inserted)
        existing_fields = existing["schema_fields"]
        if isinstance(existing_fields, str):
            existing_fields = json.loads(existing_fields)
        registered = {
            "artifact_ref": str(existing["artifact_ref"]),
            "digest": str(existing["artifact_digest"]),
            "row_count": int(existing["row_count"]),
            "schema_hash": str(existing["schema_hash"]),
            "schema_fields": _normalized_schema_fields(existing_fields),
            "expires_at": (
                _timestamp(existing["expires_at"])
                if isinstance(existing["expires_at"], datetime)
                else str(existing["expires_at"])
            ),
        }
        if registered["digest"] != described["digest"]:
            self.artifact_store.delete(artifact_ref)
            raise ValueError(
                "immutable artifact catalog digest conflicts with retry"
            )
        if registered["artifact_ref"] != artifact_ref:
            self.artifact_store.delete(artifact_ref)
            stored = self.artifact_store.describe(
                registered["artifact_ref"]
            )
            for key in (
                "digest",
                "row_count",
                "schema_hash",
                "expires_at",
            ):
                if stored[key] != registered[key]:
                    raise ValueError(
                        "immutable artifact catalog does not match storage"
                    )
        return registered

    def cleanup_expired(self, *, limit: int = 100) -> int:
        if (
            isinstance(limit, bool)
            or not isinstance(limit, int)
            or limit < 1
            or limit > 1_000
        ):
            raise ValueError("artifact cleanup limit is invalid")
        removed = 0
        with self.engine.begin() as connection:
            rows = connection.execute(
                text(
                    """
                    SELECT id::text AS id, artifact_ref
                    FROM public.rule_run_artifacts
                    WHERE expires_at <= CURRENT_TIMESTAMP
                    ORDER BY expires_at, id
                    LIMIT :limit
                    FOR UPDATE SKIP LOCKED
                    """
                ),
                {"limit": limit},
            ).mappings().all()
            for row in rows:
                self.artifact_store.delete(str(row["artifact_ref"]))
                connection.execute(
                    text(
                        """
                        DELETE FROM public.rule_run_artifacts
                        WHERE id = CAST(:id AS uuid)
                        """
                    ),
                    {"id": str(row["id"])},
                )
                removed += 1
        return removed