"""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", ]