DataOps-platform/app/runner/rule_evidence.py

"""Transactional, server-attested evidence for governed rule execution."""

from __future__ import annotations

import hashlib
import json
import os
import re
import tempfile
import uuid
from contextlib import suppress
from typing import Any

import polars as pl
from sqlalchemy import text

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

_DIGEST = re.compile(r"^[0-9a-f]{64}$")
_FINAL_STATUSES = {"success", "failed", "unknown", "cancelled"}
_COMMIT_OUTCOMES = {
    "not_applicable",
    "not_committed",
    "committed",
    "unknown",
}
_FINISH_KEYS = {
    "status",
    "rows_in",
    "rows_out",
    "rows_rejected",
    "rows_quarantined",
    "commit_outcome",
    "timings",
    "public_result",
    "violation_sample",
    "sample_count",
    "redaction_policy",
}
_PUBLIC_RESULT_KEYS = {
    "affected_rows",
    "artifact_ref",
    "commit_outcome",
    "component_binding_id",
    "digest",
    "execution_plan_hash",
    "expires_at",
    "output_artifact",
    "row_count",
    "rows_aggregated",
    "rows_deduplicated",
    "rows_filtered",
    "rows_in",
    "rows_join_dropped",
    "rows_out",
    "rows_quarantined",
    "rows_rejected",
    "rule_version_id",
    "schema_hash",
    "violation_count",
    "violations",
}
_ATTESTATION_RESULT_KEYS = {
    "component_binding_id",
    "execution_plan_hash",
    "rule_version_id",
}
_BACKEND_PUBLIC_RESULT_KEYS = {
    "sql_pushdown": _ATTESTATION_RESULT_KEYS
    | {
        "commit_outcome",
        "output_artifact",
        "rows_in",
        "rows_out",
        "rows_quarantined",
        "rows_rejected",
    },
    "polars_batch": _ATTESTATION_RESULT_KEYS
    | {
        "artifact_ref",
        "commit_outcome",
        "digest",
        "expires_at",
        "output_artifact",
        "row_count",
        "rows_aggregated",
        "rows_deduplicated",
        "rows_filtered",
        "rows_in",
        "rows_join_dropped",
        "rows_out",
        "rows_quarantined",
        "rows_rejected",
        "schema_hash",
        "violation_count",
        "violations",
    },
    "quality_check": _ATTESTATION_RESULT_KEYS
    | {
        "commit_outcome",
        "rows_in",
        "rows_out",
        "rows_quarantined",
        "rows_rejected",
        "violation_count",
        "violations",
    },
}


def _uid(value: Any, label: str) -> str:
    try:
        return ensure_governance_uid({"uid": str(value)})
    except ValueError as exc:
        raise ValueError(f"{label} is invalid") from exc


def _uuid(value: Any, label: str) -> str:
    try:
        return str(uuid.UUID(str(value)))
    except (TypeError, ValueError, AttributeError) as exc:
        raise ValueError(f"{label} is invalid") from exc


def _canonical_digest(value: Any) -> str:
    encoded = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _bounded_count(value: Any, label: str) -> int:
    if isinstance(value, bool):
        raise ValueError(f"{label} is invalid")
    try:
        normalized = int(value or 0)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{label} is invalid") from exc
    if normalized < 0 or normalized > 10_000_000_000:
        raise ValueError(f"{label} is outside the evidence limit")
    return normalized


def _sample_fields(sample: list[dict[str, Any]]) -> list[dict[str, Any]]:
    names = sorted({str(key) for row in sample for key in row})
    if not names:
        raise ValueError("violation sample has no fields")
    return [
        {"name": name, "type": "string", "nullable": True}
        for name in names
    ]


def validate_public_rule_result(
    value: Any,
    *,
    backend: str | None = None,
) -> dict[str, Any]:
    allowed_keys = (
        _BACKEND_PUBLIC_RESULT_KEYS.get(backend)
        if backend is not None
        else _PUBLIC_RESULT_KEYS
    )
    if (
        not isinstance(value, dict)
        or allowed_keys is None
        or set(value) - allowed_keys
        or any(str(key).startswith("_") for key in value)
        or len(
            json.dumps(
                value,
                sort_keys=True,
                separators=(",", ":"),
                ensure_ascii=False,
            ).encode("utf-8")
        )
        > 32_768
    ):
        raise ValueError("public rule result is not evidence safe")
    for key, item in value.items():
        if key == "violations":
            if not isinstance(item, list) or len(item) > 100:
                raise ValueError(
                    "public rule result is not evidence safe"
                )
            for summary in item:
                if (
                    not isinstance(summary, dict)
                    or set(summary) != {"step_id", "count"}
                    or not isinstance(summary["step_id"], str)
                    or isinstance(summary["count"], bool)
                    or not isinstance(summary["count"], int)
                    or summary["count"] < 0
                ):
                    raise ValueError(
                        "public rule result is not evidence safe"
                    )
        elif isinstance(item, (dict, list, tuple, set)):
            raise ValueError("public rule result is not evidence safe")
    return dict(value)


class PostgresRuleEvidenceWriter:
    """Persist one immutable run and at most one expiring violation sample."""

    def __init__(
        self,
        engine,
        artifact_store,
        *,
        sample_ttl_seconds: int = 3600,
        lease_seconds: int = 300,
        cleanup_claim_seconds: int = 300,
    ):
        self.engine = engine
        self.artifact_store = artifact_store
        self.sample_ttl_seconds = int(sample_ttl_seconds)
        self.lease_seconds = int(lease_seconds)
        self.cleanup_claim_seconds = int(cleanup_claim_seconds)
        if (
            self.sample_ttl_seconds < 1
            or self.sample_ttl_seconds
            > self.artifact_store.max_ttl_seconds
        ):
            raise ValueError("violation sample TTL is invalid")
        if self.lease_seconds < 30 or self.lease_seconds > 900:
            raise ValueError("rule execution lease is invalid")
        if (
            self.cleanup_claim_seconds < 30
            or self.cleanup_claim_seconds > 3600
        ):
            raise ValueError("rule cleanup lease is invalid")
        self.heartbeat_interval_seconds = max(
            5.0,
            min(60.0, self.lease_seconds / 3),
        )

    def start(
        self,
        *,
        component_binding_id: str,
        rule_version_id: str,
        plan_hash: str,
        correlation_id: str,
        dataflow_uid: str,
        deployment_id: str,
        environment: str,
        workflow_version: int,
        node_id: str,
        lease_owner: str,
    ) -> str:
        component = _uid(component_binding_id, "component binding id")
        rule = _uid(rule_version_id, "rule version id")
        correlation = _uid(correlation_id, "correlation id")
        dataflow = _uid(dataflow_uid, "dataflow id")
        deployment = _uid(deployment_id, "deployment id")
        owner = _uuid(lease_owner, "lease owner")
        if environment not in {"development", "test", "production"}:
            raise ValueError("deployment environment is invalid")
        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
            raise ValueError("plan hash is invalid")
        if (
            isinstance(workflow_version, bool)
            or not isinstance(workflow_version, int)
            or workflow_version < 1
        ):
            raise ValueError("workflow version is invalid")
        if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,99}", str(node_id)):
            raise ValueError("node id is invalid")
        evidence_key = _canonical_digest(
            {
                "component_binding_id": component,
                "correlation_id": correlation,
                "dataflow_uid": dataflow,
                "deployment_id": deployment,
                "environment": environment,
                "node_id": node_id,
                "plan_hash": plan_hash,
                "rule_version_id": rule,
                "workflow_version": workflow_version,
            }
        )
        with self.engine.begin() as connection:
            canonical = connection.execute(
                text(
                    """
                    SELECT d.id::text AS deployment_id
                    FROM public.dataflow_component_bindings b
                    JOIN public.dataflow_versions v
                      ON v.id = b.dataflow_version_id
                    JOIN public.rule_execution_plans p
                      ON p.component_binding_id = b.id
                    JOIN public.dataflow_deployments d
                      ON d.dataflow_version_id = v.id
                    WHERE b.id = CAST(:component_binding_id AS uuid)
                      AND b.rule_version_id =
                          CAST(:rule_version_id AS uuid)
                      AND p.plan_hash = :plan_hash
                      AND p.status = 'published'
                      AND b.component_id = :node_id
                      AND v.dataflow_uid = CAST(:dataflow_uid AS uuid)
                      AND v.version_no = :workflow_version
                      AND d.id = CAST(:deployment_id AS uuid)
                      AND d.environment = :environment
                      AND d.status IN ('canary','active')
                    """
                ),
                {
                    "component_binding_id": component,
                    "rule_version_id": rule,
                    "plan_hash": plan_hash,
                    "dataflow_uid": dataflow,
                    "deployment_id": deployment,
                    "environment": environment,
                    "workflow_version": workflow_version,
                    "node_id": node_id,
                },
            ).mappings().one_or_none()
            if canonical is None:
                raise ValueError("canonical rule deployment does not match")
            existing = connection.execute(
                text(
                    """
                    SELECT id::text, status, lease_owner::text,
                           lease_expires_at
                    FROM public.rule_runs
                    WHERE evidence_key = :evidence_key
                    FOR UPDATE
                    """
                ),
                {"evidence_key": evidence_key},
            ).mappings().one_or_none()
            if existing is not None:
                if (
                    existing["status"] == "running"
                    and existing["lease_expires_at"] is not None
                ):
                    expired = connection.execute(
                        text(
                            """
                            SELECT :lease_expires_at <= CURRENT_TIMESTAMP
                            """
                        ),
                        {
                            "lease_expires_at": existing[
                                "lease_expires_at"
                            ]
                        },
                    ).scalar_one()
                    if expired:
                        connection.execute(
                            text(
                                """
                                UPDATE public.rule_runs
                                SET status = 'unknown',
                                    commit_outcome = 'unknown',
                                    failure_code = 'execution_lease_expired',
                                    finished_at = CURRENT_TIMESTAMP,
                                    updated_at = CURRENT_TIMESTAMP
                                WHERE id = CAST(:id AS uuid)
                                  AND status = 'running'
                                """
                            ),
                            {"id": existing["id"]},
                        )
                return str(existing["id"])
            rule_run_id = new_governance_uid()
            selected = connection.execute(
                text(
                    """
                    INSERT INTO public.rule_runs (
                        id, deployment_id, component_binding_id,
                        rule_version_id, plan_hash, status,
                        correlation_id, evidence_key, started_at
                        , attempt_no, lease_owner, lease_expires_at,
                        heartbeat_at
                    ) VALUES (
                        CAST(:id AS uuid), CAST(:deployment_id AS uuid),
                        CAST(:component_binding_id AS uuid),
                        CAST(:rule_version_id AS uuid), :plan_hash,
                        'running', CAST(:correlation_id AS uuid),
                        :evidence_key, CURRENT_TIMESTAMP, 1,
                        CAST(:lease_owner AS uuid),
                        CURRENT_TIMESTAMP
                            + make_interval(secs => :lease_seconds),
                        CURRENT_TIMESTAMP
                    )
                    ON CONFLICT (evidence_key) DO NOTHING
                    RETURNING id::text
                    """
                ),
                {
                    "id": rule_run_id,
                    "deployment_id": canonical["deployment_id"],
                    "component_binding_id": component,
                    "rule_version_id": rule,
                    "plan_hash": plan_hash,
                    "correlation_id": correlation,
                    "evidence_key": evidence_key,
                    "lease_owner": owner,
                    "lease_seconds": self.lease_seconds,
                },
            ).scalar_one_or_none()
            if selected is None:
                selected = connection.execute(
                    text(
                        """
                        SELECT id::text
                        FROM public.rule_runs
                        WHERE evidence_key = :evidence_key
                        """
                    ),
                    {"evidence_key": evidence_key},
                ).scalar_one()
        return str(selected)

    def heartbeat(self, rule_run_id: str, lease_owner: str) -> None:
        run_id = _uid(rule_run_id, "rule run id")
        owner = _uuid(lease_owner, "lease owner")
        with self.engine.begin() as connection:
            updated = connection.execute(
                text(
                    """
                    UPDATE public.rule_runs
                    SET heartbeat_at = CURRENT_TIMESTAMP,
                        lease_expires_at = CURRENT_TIMESTAMP
                            + make_interval(secs => :lease_seconds),
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = CAST(:id AS uuid)
                      AND status = 'running'
                      AND lease_owner = CAST(:lease_owner AS uuid)
                    """
                ),
                {
                    "id": run_id,
                    "lease_owner": owner,
                    "lease_seconds": self.lease_seconds,
                },
            )
            if updated.rowcount != 1:
                raise ValueError("rule execution lease is not owned")

    def stage_sql_output(
        self,
        rule_run_id: str,
        *,
        output_binding_id: str,
        ttl_seconds: int | None = None,
    ) -> str:
        run_id = _uid(rule_run_id, "rule run id")
        binding_id = _uid(output_binding_id, "output binding id")
        if ttl_seconds is None:
            ttl_seconds = min(3600, self.sample_ttl_seconds)
        if (
            isinstance(ttl_seconds, bool)
            or not isinstance(ttl_seconds, int)
            or ttl_seconds < 1
            or ttl_seconds > self.sample_ttl_seconds
        ):
            raise ValueError("SQL staging TTL is invalid")
        receipt_id = new_governance_uid()
        with self.engine.begin() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT r.deployment_id::text, r.correlation_id::text,
                           b.binding_hash, b.object_ref, b.object_kind,
                           b.access_mode
                    FROM public.rule_runs r
                    JOIN public.dataflow_dataset_bindings b
                      ON b.id = CAST(:binding_id AS uuid)
                     AND b.dataflow_deployment_id = r.deployment_id
                    WHERE r.id = CAST(:run_id AS uuid)
                      AND r.status = 'running'
                    FOR SHARE OF r, b
                    """
                ),
                {"run_id": run_id, "binding_id": binding_id},
            ).mappings().one_or_none()
            if (
                row is None
                or row["object_kind"] not in {"table", "view"}
                or row["access_mode"] not in {"write", "read_write"}
            ):
                raise ValueError("SQL staging output binding is invalid")
            relation_digest = _canonical_digest(
                {
                    "binding_hash": str(row["binding_hash"]),
                    "object_kind": str(row["object_kind"]),
                    "object_ref": str(row["object_ref"]),
                }
            )
            selected = connection.execute(
                text(
                    """
                    INSERT INTO public.rule_sql_staging_receipts (
                        id, producer_rule_run_id, deployment_id,
                        correlation_id, output_binding_id,
                        output_binding_hash, relation_ref,
                        relation_digest, commit_outcome, status,
                        expires_at
                    ) VALUES (
                        CAST(:id AS uuid), CAST(:run_id AS uuid),
                        CAST(:deployment_id AS uuid),
                        CAST(:correlation_id AS uuid),
                        CAST(:binding_id AS uuid), :binding_hash,
                        :relation_ref, :relation_digest, 'committed',
                        'pending', CURRENT_TIMESTAMP
                            + make_interval(secs => :ttl_seconds)
                    )
                    ON CONFLICT (
                        producer_rule_run_id, output_binding_id
                    ) DO NOTHING
                    RETURNING id::text
                    """
                ),
                {
                    "id": receipt_id,
                    "run_id": run_id,
                    "deployment_id": row["deployment_id"],
                    "correlation_id": row["correlation_id"],
                    "binding_id": binding_id,
                    "binding_hash": str(row["binding_hash"]),
                    "relation_ref": str(row["object_ref"]),
                    "relation_digest": relation_digest,
                    "ttl_seconds": ttl_seconds,
                },
            ).scalar_one_or_none()
            if selected is None:
                selected = connection.execute(
                    text(
                        """
                        SELECT id::text
                        FROM public.rule_sql_staging_receipts
                        WHERE producer_rule_run_id = CAST(:run_id AS uuid)
                          AND output_binding_id =
                              CAST(:binding_id AS uuid)
                        """
                    ),
                    {"run_id": run_id, "binding_id": binding_id},
                ).scalar_one()
        return f"dataops-staging://{selected}"

    def resolve_sql_staging(
        self,
        receipt_ref: str,
        *,
        deployment_id: str,
        correlation_id: str,
        input_binding_id: str,
    ) -> dict[str, str]:
        match = re.fullmatch(
            r"dataops-staging://([0-9a-f-]{36})",
            str(receipt_ref or ""),
        )
        if match is None:
            raise ValueError("SQL staging receipt is invalid")
        receipt_id = _uid(match.group(1), "SQL staging receipt id")
        deployment = _uid(deployment_id, "deployment id")
        correlation = _uid(correlation_id, "correlation id")
        binding_id = _uid(input_binding_id, "input binding id")
        with self.engine.connect() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT s.relation_ref, s.relation_digest,
                           s.output_binding_hash, b.binding_hash,
                           b.object_ref, b.object_kind, b.access_mode
                    FROM public.rule_sql_staging_receipts s
                    JOIN public.rule_runs r
                      ON r.id = s.producer_rule_run_id
                    JOIN public.dataflow_dataset_bindings b
                      ON b.id = s.output_binding_id
                    WHERE s.id = CAST(:id AS uuid)
                      AND s.deployment_id =
                          CAST(:deployment_id AS uuid)
                      AND s.correlation_id =
                          CAST(:correlation_id AS uuid)
                      AND s.output_binding_id =
                          CAST(:input_binding_id AS uuid)
                      AND s.status = 'ready'
                      AND s.expires_at > CURRENT_TIMESTAMP
                      AND s.commit_outcome = 'committed'
                      AND r.status = 'success'
                      AND r.commit_outcome = 'committed'
                      AND b.binding_hash = s.output_binding_hash
                      AND b.access_mode IN ('read','read_write')
                    """
                ),
                {
                    "id": receipt_id,
                    "deployment_id": deployment,
                    "correlation_id": correlation,
                    "input_binding_id": binding_id,
                },
            ).mappings().one_or_none()
        if row is None:
            raise ValueError("SQL staging receipt is not executable")
        expected_digest = _canonical_digest(
            {
                "binding_hash": str(row["binding_hash"]),
                "object_kind": str(row["object_kind"]),
                "object_ref": str(row["object_ref"]),
            }
        )
        if (
            expected_digest != str(row["relation_digest"])
            or str(row["relation_ref"]) != str(row["object_ref"])
        ):
            raise ValueError("SQL staging receipt attestation does not match")
        return {
            "relation_ref": str(row["relation_ref"]),
            "relation_digest": str(row["relation_digest"]),
        }

    def replay(self, rule_run_id: str) -> dict[str, Any] | None:
        run_id = _uid(rule_run_id, "rule run id")
        with self.engine.connect() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT status, commit_outcome, public_result
                    FROM public.rule_runs
                    WHERE id = CAST(:id AS uuid)
                    """
                ),
                {"id": run_id},
            ).mappings().one_or_none()
        if row is None:
            raise ValueError("rule run was not found")
        if row["status"] in {"queued", "running"}:
            return None
        result = row["public_result"]
        if isinstance(result, str):
            result = json.loads(result)
        return {
            **(dict(result) if isinstance(result, dict) else {}),
            "status": str(row["status"]),
            "commit_outcome": str(row["commit_outcome"]),
        }

    def replay_by_lease_owner(
        self,
        *,
        lease_owner: str,
        deployment_id: str,
        correlation_id: str,
        component_binding_id: str,
        rule_version_id: str,
        plan_hash: str,
    ) -> dict[str, Any] | None:
        owner = _uuid(lease_owner, "lease owner")
        deployment = _uid(deployment_id, "deployment id")
        correlation = _uid(correlation_id, "correlation id")
        component = _uid(
            component_binding_id,
            "component binding id",
        )
        rule = _uid(rule_version_id, "rule version id")
        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
            raise ValueError("plan hash is invalid")
        with self.engine.connect() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT id::text
                    FROM public.rule_runs
                    WHERE lease_owner = CAST(:lease_owner AS uuid)
                      AND deployment_id =
                          CAST(:deployment_id AS uuid)
                      AND correlation_id =
                          CAST(:correlation_id AS uuid)
                      AND component_binding_id =
                          CAST(:component_binding_id AS uuid)
                      AND rule_version_id =
                          CAST(:rule_version_id AS uuid)
                      AND plan_hash = :plan_hash
                    """
                ),
                {
                    "lease_owner": owner,
                    "deployment_id": deployment,
                    "correlation_id": correlation,
                    "component_binding_id": component,
                    "rule_version_id": rule,
                    "plan_hash": plan_hash,
                },
            ).scalar_one_or_none()
        if row is None:
            return None
        return self.replay(str(row))

    def reconcile_expired_lease(
        self,
        *,
        lease_owner: str,
        deployment_id: str,
        correlation_id: str,
        component_binding_id: str,
        rule_version_id: str,
        plan_hash: str,
    ) -> dict[str, Any]:
        owner = _uuid(lease_owner, "lease owner")
        deployment = _uid(deployment_id, "deployment id")
        correlation = _uid(correlation_id, "correlation id")
        component = _uid(
            component_binding_id,
            "component binding id",
        )
        rule = _uid(rule_version_id, "rule version id")
        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
            raise ValueError("plan hash is invalid")
        parameters = {
            "lease_owner": owner,
            "deployment_id": deployment,
            "correlation_id": correlation,
            "component_binding_id": component,
            "rule_version_id": rule,
            "plan_hash": plan_hash,
        }
        with self.engine.begin() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT id::text, status, lease_expires_at,
                           evidence_digest
                    FROM public.rule_runs
                    WHERE lease_owner = CAST(:lease_owner AS uuid)
                      AND deployment_id =
                          CAST(:deployment_id AS uuid)
                      AND correlation_id =
                          CAST(:correlation_id AS uuid)
                      AND component_binding_id =
                          CAST(:component_binding_id AS uuid)
                      AND rule_version_id =
                          CAST(:rule_version_id AS uuid)
                      AND plan_hash = :plan_hash
                    FOR UPDATE
                    """
                ),
                parameters,
            ).mappings().one_or_none()
            if row is None:
                return {"state": "missing"}
            if (
                row["status"] == "running"
                and row["lease_expires_at"] is not None
            ):
                expired = connection.execute(
                    text(
                        "SELECT :lease_expires_at <= CURRENT_TIMESTAMP"
                    ),
                    {"lease_expires_at": row["lease_expires_at"]},
                ).scalar_one()
                if expired:
                    connection.execute(
                        text(
                            """
                            UPDATE public.rule_runs
                            SET status = 'unknown',
                                commit_outcome = 'unknown',
                                failure_code =
                                    'execution_lease_expired',
                                lease_expires_at = NULL,
                                finished_at = CURRENT_TIMESTAMP,
                                updated_at = CURRENT_TIMESTAMP
                            WHERE id = CAST(:id AS uuid)
                              AND status = 'running'
                            """
                        ),
                        {"id": row["id"]},
                    )
                    connection.execute(
                        text(
                            """
                            UPDATE public.rule_sql_staging_receipts
                            SET status = 'failed',
                                commit_outcome = 'unknown',
                                updated_at = CURRENT_TIMESTAMP
                            WHERE producer_rule_run_id =
                                CAST(:id AS uuid)
                              AND status = 'pending'
                            """
                        ),
                        {"id": row["id"]},
                    )
                    row = {**row, "status": "unknown"}
            if row["status"] == "running":
                return {"state": "running"}
            run_id = str(row["id"])
        replay = self.replay(run_id)
        result = {
            key: value
            for key, value in replay.items()
            if key != "status"
        }
        return {
            "state": "terminal",
            "status": str(replay["status"]),
            "commit_outcome": str(replay["commit_outcome"]),
            "result": result,
            "result_digest": _canonical_digest(result),
            "evidence_digest": str(row["evidence_digest"] or ""),
        }

    @staticmethod
    def _validate_finish(result: Any) -> dict[str, Any]:
        if not isinstance(result, dict) or set(result) - _FINISH_KEYS:
            raise ValueError("rule evidence result has unsupported fields")
        status = result.get("status")
        commit_outcome = result.get("commit_outcome", "not_applicable")
        if status not in _FINAL_STATUSES:
            raise ValueError("rule evidence status is invalid")
        if commit_outcome not in _COMMIT_OUTCOMES:
            raise ValueError("rule evidence commit outcome is invalid")
        timings = result.get("timings", {})
        if (
            not isinstance(timings, dict)
            or set(timings) != {"duration_ms"}
            or isinstance(timings.get("duration_ms"), bool)
            or not isinstance(timings.get("duration_ms"), int)
            or timings["duration_ms"] < 0
            or timings["duration_ms"] > 86_400_000
        ):
            raise ValueError("rule evidence timings are invalid")
        public_result = result.get("public_result")
        if public_result is not None:
            validate_public_rule_result(public_result)
        sample = result.get("violation_sample")
        if sample is not None:
            if (
                status != "success"
                or not isinstance(sample, list)
                or not 1 <= len(sample) <= 100
                or result.get("sample_count") != len(sample)
                or result.get("redaction_policy")
                != "rule-violation-default-v1"
            ):
                raise ValueError("violation sample is invalid")
            for row in sample:
                if not isinstance(row, dict):
                    raise ValueError("violation sample row is invalid")
                for value in row.values():
                    if value not in {None, "[REDACTED]"}:
                        raise ValueError(
                            "violation sample contains unredacted values"
                        )
        return {
            **result,
            "rows_in": _bounded_count(result.get("rows_in"), "rows_in"),
            "rows_out": _bounded_count(result.get("rows_out"), "rows_out"),
            "rows_rejected": _bounded_count(
                result.get("rows_rejected"), "rows_rejected"
            ),
            "rows_quarantined": _bounded_count(
                result.get("rows_quarantined"), "rows_quarantined"
            ),
            "commit_outcome": commit_outcome,
        }

    def _prepare_sample(
        self,
        run_id: str,
        correlation_id: str,
        sample: list[dict[str, Any]],
        redaction_policy: str,
    ) -> tuple[str, dict[str, Any], str]:
        fields = _sample_fields(sample)
        frame = pl.DataFrame(
            {
                field["name"]: [
                    row.get(field["name"]) for row in sample
                ]
                for field in fields
            },
            schema={field["name"]: pl.String for field in fields},
        )
        with tempfile.NamedTemporaryFile(
            prefix="dataops-rule-violation-",
            suffix=".parquet",
            delete=False,
        ) as handle:
            path = handle.name
        sample_id = None
        prepared = None
        uploaded = False
        try:
            frame.write_parquet(path)
            prepared = self.artifact_store.prepare_path(
                path,
                correlation_id,
                self.sample_ttl_seconds,
                schema_fields=fields,
                limits={
                    "max_rows": min(100, self.artifact_store.max_rows),
                    "max_artifact_bytes": min(
                        4 * 1024 * 1024,
                        self.artifact_store.max_artifact_bytes,
                    ),
                    "memory_limit_bytes": min(
                        16 * 1024 * 1024,
                        self.artifact_store.memory_limit_bytes,
                    ),
                },
            )
            sample_id = new_governance_uid()
            with self.engine.begin() as connection:
                row = connection.execute(
                    text(
                        """
                        SELECT id::text, artifact_ref, artifact_digest,
                               schema_hash, sample_count,
                               redaction_policy, expires_at, handoff_status
                        FROM public.rule_violation_samples
                        WHERE rule_run_id = CAST(:rule_run_id AS uuid)
                        FOR UPDATE
                        """
                    ),
                    {"rule_run_id": run_id},
                ).mappings().one_or_none()
                if row is None:
                    connection.execute(
                        text(
                            """
                            INSERT INTO public.rule_violation_samples (
                                id, rule_run_id, artifact_ref,
                                artifact_digest, schema_hash, sample_count,
                                redaction_policy, expires_at,
                                handoff_status, schema_fields
                            ) VALUES (
                                CAST(:id AS uuid),
                                CAST(:rule_run_id AS uuid), :artifact_ref,
                                :artifact_digest, :schema_hash,
                                :sample_count, :redaction_policy,
                                CAST(:expires_at AS timestamptz), 'pending',
                                CAST(:schema_fields AS jsonb)
                            )
                            """
                        ),
                        {
                            "id": sample_id,
                            "rule_run_id": run_id,
                            "artifact_ref": prepared["artifact_ref"],
                            "artifact_digest": prepared["digest"],
                            "schema_hash": prepared["schema_hash"],
                            "sample_count": len(sample),
                            "redaction_policy": redaction_policy,
                            "expires_at": prepared["expires_at"],
                            "schema_fields": json.dumps(fields),
                        },
                    )
                else:
                    if (
                        str(row["artifact_digest"]) != prepared["digest"]
                        or int(row["sample_count"]) != len(sample)
                        or str(row["redaction_policy"])
                        != redaction_policy
                    ):
                        raise ValueError(
                            "violation sample evidence is immutable"
                        )
                    sample_id = str(row["id"])
                    prepared.update(
                        {
                            "artifact_ref": str(row["artifact_ref"]),
                            "digest": str(row["artifact_digest"]),
                            "schema_hash": str(row["schema_hash"]),
                            "expires_at": str(row["expires_at"]),
                        }
                    )
                    if row["handoff_status"] == "ready":
                        stored = self.artifact_store.describe(
                            prepared["artifact_ref"]
                        )
                        if (
                            stored["digest"] != prepared["digest"]
                            or stored["schema_hash"]
                            != prepared["schema_hash"]
                            or stored["row_count"] != len(sample)
                        ):
                            raise ValueError(
                                "ready violation sample does not match storage"
                            )
                        return sample_id, prepared, path
                    if row["handoff_status"] == "failed":
                        raise ValueError(
                            "violation sample handoff already failed"
                        )
            self.artifact_store.upload_path(
                path,
                prepared,
                limits={
                    "max_rows": min(100, self.artifact_store.max_rows),
                    "max_artifact_bytes": min(
                        4 * 1024 * 1024,
                        self.artifact_store.max_artifact_bytes,
                    ),
                    "memory_limit_bytes": min(
                        16 * 1024 * 1024,
                        self.artifact_store.memory_limit_bytes,
                    ),
                },
            )
            uploaded = True
            with self.engine.begin() as connection:
                updated = connection.execute(
                    text(
                        """
                        UPDATE public.rule_violation_samples
                        SET handoff_status = 'ready',
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = CAST(:id AS uuid)
                          AND handoff_status = 'pending'
                          AND artifact_digest = :artifact_digest
                        """
                    ),
                    {
                        "id": sample_id,
                        "artifact_digest": prepared["digest"],
                    },
                )
                if updated.rowcount != 1:
                    state = connection.execute(
                        text(
                            """
                            SELECT handoff_status
                            FROM public.rule_violation_samples
                            WHERE id = CAST(:id AS uuid)
                            """
                        ),
                        {"id": sample_id},
                    ).scalar_one_or_none()
                    if state != "ready":
                        raise RuntimeError(
                            "violation sample finalize outcome is unknown"
                        )
            return sample_id, prepared, path
        except Exception:
            if sample_id is not None and prepared is not None:
                try:
                    with self.engine.connect() as connection:
                        committed = connection.execute(
                            text(
                                """
                                SELECT handoff_status, artifact_digest
                                FROM public.rule_violation_samples
                                WHERE id = CAST(:id AS uuid)
                                """
                            ),
                            {"id": sample_id},
                        ).mappings().one_or_none()
                    if (
                        committed is not None
                        and committed["handoff_status"] == "ready"
                        and str(committed["artifact_digest"])
                        == prepared["digest"]
                    ):
                        return sample_id, prepared, path
                except Exception:
                    pass
            with self.engine.begin() as connection:
                connection.execute(
                    text(
                        """
                        UPDATE public.rule_violation_samples
                        SET handoff_status = :handoff_status,
                            failure_code = :failure_code,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE rule_run_id = CAST(:rule_run_id AS uuid)
                          AND handoff_status = 'pending'
                        """
                    ),
                    {
                        "rule_run_id": run_id,
                        "handoff_status": (
                            "unknown" if uploaded else "failed"
                        ),
                        "failure_code": (
                            "sample_finalize_unknown"
                            if uploaded
                            else "sample_handoff_failed"
                        ),
                    },
                )
            raise
        finally:
            with suppress(FileNotFoundError):
                os.unlink(path)

    def finish(self, rule_run_id: str, result: Any) -> None:
        run_id = _uid(rule_run_id, "rule run id")
        normalized = self._validate_finish(result)
        evidence_digest = _canonical_digest(normalized)
        sample_path = None
        try:
            with self.engine.connect() as connection:
                current = connection.execute(
                    text(
                        """
                        SELECT status, correlation_id::text,
                               evidence_digest
                        FROM public.rule_runs
                        WHERE id = CAST(:id AS uuid)
                        """
                    ),
                    {"id": run_id},
                ).mappings().one_or_none()
            if current is None:
                raise ValueError("rule run was not found")
            if current["status"] not in {"queued", "running"}:
                if (
                    current["status"] == normalized["status"]
                    and str(current["evidence_digest"] or "")
                    == evidence_digest
                ):
                    return
                raise ValueError("rule run evidence is immutable")
            if normalized.get("violation_sample"):
                _sample_id, _prepared, sample_path = self._prepare_sample(
                    run_id,
                    str(current["correlation_id"]),
                    normalized["violation_sample"],
                    normalized["redaction_policy"],
                )
            try:
                with self.engine.begin() as connection:
                    updated = connection.execute(
                        text(
                            """
                            UPDATE public.rule_runs
                            SET rows_in = :rows_in,
                                rows_out = :rows_out,
                                rows_rejected = :rows_rejected,
                                rows_quarantined = :rows_quarantined,
                                status = :status,
                                timings = CAST(:timings AS jsonb),
                                commit_outcome = :commit_outcome,
                                public_result = CAST(:public_result AS jsonb),
                                failure_code = :failure_code,
                                evidence_digest = :evidence_digest,
                                lease_expires_at = NULL,
                                finished_at = CURRENT_TIMESTAMP,
                                updated_at = CURRENT_TIMESTAMP
                            WHERE id = CAST(:id AS uuid)
                              AND status IN ('queued','running')
                            """
                        ),
                        {
                            "id": run_id,
                            "rows_in": normalized["rows_in"],
                            "rows_out": normalized["rows_out"],
                            "rows_rejected": normalized[
                                "rows_rejected"
                            ],
                            "rows_quarantined": normalized[
                                "rows_quarantined"
                            ],
                            "status": normalized["status"],
                            "timings": json.dumps(
                                normalized["timings"]
                            ),
                            "commit_outcome": normalized[
                                "commit_outcome"
                            ],
                            "public_result": (
                                json.dumps(
                                    normalized.get("public_result")
                                )
                                if normalized.get("public_result")
                                is not None
                                else None
                            ),
                            "failure_code": (
                                None
                                if normalized["status"] == "success"
                                else (
                                    f"execution_{normalized['status']}"
                                )
                            ),
                            "evidence_digest": evidence_digest,
                        },
                    )
                    if updated.rowcount != 1:
                        state = connection.execute(
                            text(
                                """
                                SELECT status, commit_outcome,
                                       evidence_digest
                                FROM public.rule_runs
                                WHERE id = CAST(:id AS uuid)
                                """
                            ),
                            {"id": run_id},
                        ).mappings().one_or_none()
                        if (
                            state is None
                            or state["status"] != normalized["status"]
                            or state["commit_outcome"]
                            != normalized["commit_outcome"]
                            or str(state["evidence_digest"] or "")
                            != evidence_digest
                        ):
                            raise RuntimeError(
                                "rule run finalize outcome is unknown"
                            )
                    receipt_status = (
                        "ready"
                        if normalized["status"] == "success"
                        and normalized["commit_outcome"] == "committed"
                        else "failed"
                    )
                    connection.execute(
                        text(
                            """
                            UPDATE public.rule_sql_staging_receipts
                            SET status = :status,
                                ready_at = CASE
                                    WHEN :status = 'ready'
                                    THEN CURRENT_TIMESTAMP
                                    ELSE ready_at
                                END,
                                commit_outcome = CASE
                                    WHEN :status = 'ready'
                                    THEN 'committed'
                                    ELSE 'unknown'
                                END,
                                updated_at = CURRENT_TIMESTAMP
                            WHERE producer_rule_run_id =
                                CAST(:run_id AS uuid)
                              AND status = 'pending'
                            """
                        ),
                        {
                            "run_id": run_id,
                            "status": receipt_status,
                        },
                    )
            except Exception as exc:
                try:
                    with self.engine.connect() as connection:
                        terminal = connection.execute(
                            text(
                                """
                                SELECT status, commit_outcome,
                                       evidence_digest
                                FROM public.rule_runs
                                WHERE id = CAST(:id AS uuid)
                                """
                            ),
                            {"id": run_id},
                        ).mappings().one_or_none()
                except Exception as recheck_exc:
                    raise RuntimeError(
                        "rule run finalize outcome is unknown"
                    ) from recheck_exc
                if (
                    terminal is None
                    or terminal["status"] != normalized["status"]
                    or terminal["commit_outcome"]
                    != normalized["commit_outcome"]
                    or str(terminal["evidence_digest"] or "")
                    != evidence_digest
                ):
                    raise RuntimeError(
                        "rule run finalize outcome is unknown"
                    ) from exc
        finally:
            if sample_path is not None:
                with suppress(FileNotFoundError):
                    os.unlink(sample_path)

    def reconcile_samples(self, *, limit: int = 100) -> dict[str, int]:
        if isinstance(limit, bool) or not isinstance(limit, int):
            raise ValueError("cleanup limit is invalid")
        if limit < 1 or limit > 1000:
            raise ValueError("cleanup limit is invalid")
        claim = new_governance_uid()
        with self.engine.begin() as connection:
            candidates = connection.execute(
                text(
                    """
                    SELECT id::text, artifact_ref, artifact_digest,
                           schema_hash, schema_fields, sample_count,
                           handoff_status,
                           expires_at <= CURRENT_TIMESTAMP AS expired
                    FROM public.rule_violation_samples
                    WHERE (
                        cleanup_claim IS NULL
                        OR cleanup_claim_expires_at <= CURRENT_TIMESTAMP
                    )
                      AND (
                          handoff_status IN ('pending','unknown')
                          OR (
                              expires_at <= CURRENT_TIMESTAMP
                              AND handoff_status IN (
                                  'legacy','ready','failed'
                              )
                          )
                      )
                    ORDER BY expires_at, id
                    FOR UPDATE SKIP LOCKED
                    LIMIT :limit
                    """
                ),
                {"limit": limit},
            ).mappings().all()
            rows = []
            for row in candidates:
                updated = connection.execute(
                    text(
                        """
                        UPDATE public.rule_violation_samples
                        SET cleanup_claim = CAST(:claim AS uuid),
                            cleanup_claim_expires_at =
                                CURRENT_TIMESTAMP
                                + make_interval(
                                    secs => :claim_seconds
                                ),
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = CAST(:id AS uuid)
                          AND (
                              cleanup_claim IS NULL
                              OR cleanup_claim_expires_at
                                  <= CURRENT_TIMESTAMP
                          )
                        """
                    ),
                    {
                        "id": str(row["id"]),
                        "claim": claim,
                        "claim_seconds": self.cleanup_claim_seconds,
                    },
                )
                if int(updated.rowcount or 0) == 1:
                    rows.append(row)
        metrics = {
            "claimed": len(rows),
            "ready": 0,
            "failed": 0,
            "expired_deleted": 0,
        }
        for row in rows:
            row_id = str(row["id"])
            if bool(row["expired"]):
                try:
                    self.artifact_store.delete(str(row["artifact_ref"]))
                except Exception:
                    with self.engine.begin() as connection:
                        connection.execute(
                            text(
                                """
                                UPDATE public.rule_violation_samples
                                SET cleanup_claim = NULL,
                                    cleanup_claim_expires_at = NULL,
                                    updated_at = CURRENT_TIMESTAMP
                                WHERE id = CAST(:id AS uuid)
                                  AND cleanup_claim =
                                      CAST(:claim AS uuid)
                                """
                            ),
                            {"id": row_id, "claim": claim},
                        )
                    continue
                with self.engine.begin() as connection:
                    deleted = connection.execute(
                        text(
                            """
                            DELETE FROM public.rule_violation_samples
                            WHERE id = CAST(:id AS uuid)
                              AND cleanup_claim = CAST(:claim AS uuid)
                            """
                        ),
                        {"id": row_id, "claim": claim},
                    )
                metrics["expired_deleted"] += int(
                    deleted.rowcount or 0
                )
                continue
            next_status = row["handoff_status"]
            failure_code = None
            try:
                fields = row["schema_fields"]
                if isinstance(fields, str):
                    fields = json.loads(fields)
                if not isinstance(fields, list):
                    raise ValueError(
                        "violation sample schema is unavailable"
                    )
                described = self.artifact_store.describe_optional(
                    str(row["artifact_ref"])
                )
                if described is None:
                    raise ValueError(
                        "violation sample object is missing"
                    )
                if (
                    described["digest"]
                    != str(row["artifact_digest"])
                    or described["schema_hash"] != row["schema_hash"]
                    or described["row_count"]
                    != int(row["sample_count"])
                ):
                    raise ValueError(
                        "violation sample attestation does not match"
                    )
                with self.artifact_store.stage(
                    str(row["artifact_ref"]),
                    str(row["artifact_digest"]),
                    expected_schema_fields=fields,
                    limits={
                        "max_rows": min(
                            100,
                            self.artifact_store.max_rows,
                        ),
                        "max_artifact_bytes": min(
                            4 * 1024 * 1024,
                            self.artifact_store.max_artifact_bytes,
                        ),
                        "memory_limit_bytes": min(
                            16 * 1024 * 1024,
                            self.artifact_store.memory_limit_bytes,
                        ),
                    },
                ):
                    pass
                next_status = "ready"
                metrics["ready"] += 1
            except ValueError:
                next_status = "failed"
                failure_code = "sample_reconcile_invalid"
                metrics["failed"] += 1
            except Exception:
                next_status = row["handoff_status"]
            with self.engine.begin() as connection:
                connection.execute(
                    text(
                        """
                        UPDATE public.rule_violation_samples
                        SET handoff_status = :handoff_status,
                            failure_code = :failure_code,
                            cleanup_claim = NULL,
                            cleanup_claim_expires_at = NULL,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = CAST(:id AS uuid)
                          AND cleanup_claim = CAST(:claim AS uuid)
                        """
                    ),
                    {
                        "id": row_id,
                        "claim": claim,
                        "handoff_status": next_status,
                        "failure_code": failure_code,
                    },
                )
        return metrics

    def cleanup_sql_staging(self, *, limit: int = 100) -> int:
        if (
            isinstance(limit, bool)
            or not isinstance(limit, int)
            or limit < 1
            or limit > 1000
        ):
            raise ValueError("cleanup limit is invalid")
        claim = new_governance_uid()
        with self.engine.begin() as connection:
            candidates = connection.execute(
                text(
                    """
                    SELECT id::text
                    FROM public.rule_sql_staging_receipts
                    WHERE expires_at <= CURRENT_TIMESTAMP
                      AND status IN ('pending','ready','failed')
                      AND (
                          cleanup_claim IS NULL
                          OR cleanup_claim_expires_at
                              <= CURRENT_TIMESTAMP
                      )
                    ORDER BY expires_at, id
                    FOR UPDATE SKIP LOCKED
                    LIMIT :limit
                    """
                ),
                {"limit": limit},
            ).mappings().all()
            rows = []
            for row in candidates:
                updated = connection.execute(
                    text(
                        """
                        UPDATE public.rule_sql_staging_receipts
                        SET cleanup_claim = CAST(:claim AS uuid),
                            cleanup_claim_expires_at =
                                CURRENT_TIMESTAMP
                                + make_interval(
                                    secs => :claim_seconds
                                ),
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = CAST(:id AS uuid)
                          AND (
                              cleanup_claim IS NULL
                              OR cleanup_claim_expires_at
                                  <= CURRENT_TIMESTAMP
                          )
                        """
                    ),
                    {
                        "id": str(row["id"]),
                        "claim": claim,
                        "claim_seconds": self.cleanup_claim_seconds,
                    },
                )
                if int(updated.rowcount or 0) == 1:
                    rows.append(row)
        finalized = 0
        for row in rows:
            with self.engine.begin() as connection:
                updated = connection.execute(
                    text(
                        """
                        UPDATE public.rule_sql_staging_receipts
                        SET status = 'expired',
                            cleanup_claim = NULL,
                            cleanup_claim_expires_at = NULL,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = CAST(:id AS uuid)
                          AND cleanup_claim = CAST(:claim AS uuid)
                          AND expires_at <= CURRENT_TIMESTAMP
                        """
                    ),
                    {"id": str(row["id"]), "claim": claim},
                )
            finalized += int(updated.rowcount or 0)
        return finalized

    def cleanup_expired(self, *, limit: int = 100) -> int:
        samples = self.reconcile_samples(limit=limit)
        receipts = self.cleanup_sql_staging(limit=limit)
        return samples["expired_deleted"] + receipts


__all__ = [
    "PostgresRuleEvidenceWriter",
    "validate_public_rule_result",
]