"""Transactional, server-attested evidence for governed rule execution.""" from __future__ import annotations import hashlib import json import os import re import tempfile 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", } 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 _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 ] class PostgresRuleEvidenceWriter: """Persist one immutable run and at most one expiring violation sample.""" def __init__( self, engine, artifact_store, *, sample_ttl_seconds: int = 3600, ): self.engine = engine self.artifact_store = artifact_store self.sample_ttl_seconds = int(sample_ttl_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") def start( self, *, component_binding_id: str, rule_version_id: str, plan_hash: str, correlation_id: str, dataflow_uid: str, workflow_version: int, node_id: 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") 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, "node_id": node_id, "plan_hash": plan_hash, "rule_version_id": rule, "workflow_version": workflow_version, } ) with self.engine.begin() as connection: deployments = 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 v.dataflow_uid = CAST(:dataflow_uid AS uuid) AND v.version_no = :workflow_version AND d.status IN ('canary','active') ORDER BY CASE d.status WHEN 'active' THEN 0 ELSE 1 END, d.created_at DESC LIMIT 2 """ ), { "component_binding_id": component, "rule_version_id": rule, "plan_hash": plan_hash, "dataflow_uid": dataflow, "workflow_version": workflow_version, }, ).mappings().all() if len(deployments) != 1: raise ValueError( "canonical rule deployment is missing or ambiguous" ) 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 ) 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 ) ON CONFLICT (evidence_key) DO NOTHING RETURNING id::text """ ), { "id": rule_run_id, "deployment_id": deployments[0]["deployment_id"], "component_binding_id": component, "rule_version_id": rule, "plan_hash": plan_hash, "correlation_id": correlation, "evidence_key": evidence_key, }, ).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 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"]), } @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 and ( not isinstance(public_result, dict) or any(str(key).startswith("_") for key in public_result) or "rows" in public_result or len( json.dumps( public_result, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") ) > 32_768 ): raise ValueError("public rule result is not evidence safe") 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 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 ) 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' ) """ ), { "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"], }, ) 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, ), }, ) 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 = 'failed', failure_code = 'sample_handoff_failed', updated_at = CURRENT_TIMESTAMP WHERE rule_run_id = CAST(:rule_run_id AS uuid) AND handoff_status = 'pending' """ ), {"rule_run_id": run_id}, ) with suppress(FileNotFoundError): os.unlink(path) raise def finish(self, rule_run_id: str, result: Any) -> None: run_id = _uid(rule_run_id, "rule run id") normalized = self._validate_finish(result) sample_path = None try: with self.engine.connect() as connection: current = connection.execute( text( """ SELECT status, correlation_id::text 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"}: replay = self.replay(run_id) if replay and replay["status"] == normalized["status"]: 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, 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']}" ) ), }, ) if updated.rowcount != 1: state = connection.execute( text( """ SELECT status, commit_outcome 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"] ): raise RuntimeError( "rule run finalize outcome is unknown" ) except Exception as exc: try: replay = self.replay(run_id) except Exception as recheck_exc: raise RuntimeError( "rule run finalize outcome is unknown" ) from recheck_exc if ( replay is None or replay["status"] != normalized["status"] or replay["commit_outcome"] != normalized["commit_outcome"] ): 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 cleanup_expired(self, *, limit: int = 100) -> 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") removed = 0 with self.engine.begin() as connection: rows = connection.execute( text( """ SELECT id::text, artifact_ref FROM public.rule_violation_samples WHERE 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() for row in rows: try: self.artifact_store.delete(str(row["artifact_ref"])) except Exception: continue connection.execute( text( """ DELETE FROM public.rule_violation_samples WHERE id = CAST(:id AS uuid) """ ), {"id": str(row["id"])}, ) removed += 1 return removed __all__ = ["PostgresRuleEvidenceWriter"]