feat: define governed rule execution contracts

app/core/data_rules/execution_contracts.py

@@ -0,0 +1,270 @@
+"""Closed contracts for governed rule execution inputs and evidence."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+from typing import Any
+
+from app.core.common.identifiers import ensure_governance_uid
+
+
+SCHEMA_VERSION = "2.0"
+BACKENDS = {"sql_pushdown", "polars_batch", "quality_check"}
+OBJECT_KINDS = {"table", "view", "query", "parquet_artifact"}
+ACCESS_MODES = {"read", "write", "read_write"}
+PLAN_STATUSES = {"compiled", "tested", "published", "revoked"}
+
+_SECRET_KEY_TOKENS = {
+    "apikey",
+    "authorization",
+    "connectionstring",
+    "credential",
+    "credentials",
+    "dsn",
+    "password",
+    "secret",
+    "token",
+}
+_SCHEMA_SNAPSHOT_KEYS = {
+    "schema_ref",
+    "schema_hash",
+    "fields",
+    "source_revision",
+}
+_SCHEMA_FIELD_KEYS = {
+    "name",
+    "type",
+    "nullable",
+    "precision",
+    "scale",
+    "timezone",
+}
+_DATASET_BINDING_KEYS = {
+    "data_source_uid",
+    "object_kind",
+    "object_ref",
+    "schema_snapshot_id",
+    "access_mode",
+    "dialect",
+    "write_mode",
+}
+_EXECUTION_PLAN_KEYS = {
+    "schema_version",
+    "backend",
+    "compiler_version",
+    "rule_version_id",
+    "input_schema_snapshot_id",
+    "output_schema_snapshot_id",
+    "input_binding_id",
+    "output_binding_id",
+    "operations",
+    "status",
+}
+_OPERATION_KEYS = {"kind", "statement", "parameters"}
+
+
+def _normalized_key(value: Any) -> str:
+    return re.sub(r"[^a-z0-9]", "", str(value).lower())
+
+
+def _reject_secret_material(value: Any, path: str = "$") -> None:
+    if isinstance(value, dict):
+        for key, item in value.items():
+            normalized = _normalized_key(key)
+            if any(token in normalized for token in _SECRET_KEY_TOKENS):
+                raise ValueError(
+                    f"secret or credential material is not allowed at {path}.{key}"
+                )
+            _reject_secret_material(item, f"{path}.{key}")
+    elif isinstance(value, list):
+        for index, item in enumerate(value):
+            _reject_secret_material(item, f"{path}[{index}]")
+
+
+def _closed_object(value: Any, allowed: set[str], label: str) -> dict[str, Any]:
+    _reject_secret_material(value)
+    if not isinstance(value, dict):
+        raise ValueError(f"{label} must be an object")
+    unknown = sorted(set(value) - allowed)
+    if unknown:
+        raise ValueError(
+            f"{label} contains unsupported fields: {', '.join(unknown)}"
+        )
+    return copy.deepcopy(value)
+
+
+def _required_fields(value: dict[str, Any], fields: set[str], label: str) -> None:
+    missing = sorted(fields - set(value))
+    if missing:
+        raise ValueError(f"{label} is missing fields: {', '.join(missing)}")
+
+
+def _required_string(value: Any, label: str, maximum: int) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{label} is required")
+    normalized = value.strip()
+    if len(normalized) > maximum:
+        raise ValueError(f"{label} exceeds {maximum} characters")
+    return normalized
+
+
+def _uid(value: Any, label: str) -> str:
+    try:
+        return ensure_governance_uid({"uid": str(value)})
+    except ValueError as exc:
+        raise ValueError(f"{label} must be a valid UUIDv7") from exc
+
+
+def _sha256(value: Any, label: str) -> str:
+    normalized = str(value or "")
+    if not re.fullmatch(r"[0-9a-f]{64}", normalized):
+        raise ValueError(f"{label} must be a sha256 hex digest")
+    return normalized
+
+
+def _canonical_json(value: Any) -> str:
+    try:
+        return json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=False,
+        )
+    except (TypeError, ValueError) as exc:
+        raise ValueError("execution contract must be JSON serializable") from exc
+
+
+def _validate_schema_field(value: Any) -> dict[str, Any]:
+    field = _closed_object(value, _SCHEMA_FIELD_KEYS, "schema field")
+    _required_fields(field, {"name", "type", "nullable"}, "schema field")
+    field["name"] = _required_string(field["name"], "schema field name", 200)
+    field["type"] = _required_string(field["type"], "schema field type", 100)
+    if not isinstance(field["nullable"], bool):
+        raise ValueError("schema field nullable must be a boolean")
+    for key in ("precision", "scale"):
+        if key in field:
+            number = field[key]
+            if isinstance(number, bool) or not isinstance(number, int) or number < 0:
+                raise ValueError(f"schema field {key} must be a non-negative integer")
+    if "timezone" in field:
+        field["timezone"] = _required_string(
+            field["timezone"], "schema field timezone", 100
+        )
+    return field
+
+
+def validate_schema_snapshot(value: Any) -> dict:
+    """Validate immutable, server-resolved schema metadata."""
+
+    snapshot = _closed_object(value, _SCHEMA_SNAPSHOT_KEYS, "schema snapshot")
+    _required_fields(snapshot, _SCHEMA_SNAPSHOT_KEYS, "schema snapshot")
+    snapshot["schema_ref"] = _required_string(
+        snapshot["schema_ref"], "schema_ref", 500
+    )
+    snapshot["schema_hash"] = _sha256(snapshot["schema_hash"], "schema_hash")
+    snapshot["source_revision"] = _required_string(
+        snapshot["source_revision"], "source_revision", 200
+    )
+    raw_fields = snapshot["fields"]
+    if not isinstance(raw_fields, list) or not raw_fields or len(raw_fields) > 1_000:
+        raise ValueError("schema fields must be a non-empty bounded array")
+    fields = [_validate_schema_field(item) for item in raw_fields]
+    names = [item["name"] for item in fields]
+    if len(names) != len(set(names)):
+        raise ValueError("schema field names must be unique")
+    snapshot["fields"] = fields
+    return snapshot
+
+
+def validate_dataset_binding(value: Any) -> dict:
+    """Validate a credential-free, snapshot-pinned physical dataset binding."""
+
+    binding = _closed_object(value, _DATASET_BINDING_KEYS, "dataset binding")
+    _required_fields(binding, _DATASET_BINDING_KEYS, "dataset binding")
+    binding["data_source_uid"] = _uid(
+        binding["data_source_uid"], "data_source_uid"
+    )
+    if binding["object_kind"] not in OBJECT_KINDS:
+        raise ValueError("unsupported dataset object_kind")
+    binding["object_ref"] = _required_string(
+        binding["object_ref"], "object_ref", 1_000
+    )
+    if "://" in binding["object_ref"]:
+        raise ValueError("dataset object_ref cannot contain a connection string")
+    binding["schema_snapshot_id"] = _uid(
+        binding["schema_snapshot_id"], "schema_snapshot_id"
+    )
+    if binding["access_mode"] not in ACCESS_MODES:
+        raise ValueError("unsupported dataset access_mode")
+    binding["dialect"] = _required_string(binding["dialect"], "dialect", 30)
+    binding["write_mode"] = _required_string(
+        binding["write_mode"], "write_mode", 30
+    )
+    return binding
+
+
+def _validate_operation(value: Any) -> dict[str, Any]:
+    operation = _closed_object(value, _OPERATION_KEYS, "execution operation")
+    _required_fields(operation, {"kind"}, "execution operation")
+    operation["kind"] = _required_string(
+        operation["kind"], "execution operation kind", 30
+    )
+    if operation["kind"] == "sql":
+        operation["statement"] = _required_string(
+            operation.get("statement"), "sql execution statement", 20_000
+        )
+    elif "statement" in operation:
+        operation["statement"] = _required_string(
+            operation["statement"], "execution operation statement", 20_000
+        )
+    if "parameters" in operation and not isinstance(operation["parameters"], dict):
+        raise ValueError("execution operation parameters must be an object")
+    return operation
+
+
+def validate_execution_plan_v2(value: Any) -> dict:
+    """Validate an execution plan that contains only immutable references."""
+
+    plan = _closed_object(value, _EXECUTION_PLAN_KEYS, "execution plan")
+    _required_fields(
+        plan,
+        _EXECUTION_PLAN_KEYS - {"status"},
+        "execution plan",
+    )
+    if plan["schema_version"] != SCHEMA_VERSION:
+        raise ValueError(f"execution plan schema_version must be {SCHEMA_VERSION}")
+    if plan["backend"] not in BACKENDS:
+        raise ValueError("unsupported execution backend")
+    plan["compiler_version"] = _required_string(
+        plan["compiler_version"], "compiler_version", 80
+    )
+    for key in (
+        "rule_version_id",
+        "input_schema_snapshot_id",
+        "output_schema_snapshot_id",
+        "input_binding_id",
+        "output_binding_id",
+    ):
+        plan[key] = _uid(plan[key], key)
+    if "status" in plan and plan["status"] not in PLAN_STATUSES:
+        raise ValueError("unsupported execution plan status")
+    raw_operations = plan["operations"]
+    if (
+        not isinstance(raw_operations, list)
+        or not raw_operations
+        or len(raw_operations) > 500
+    ):
+        raise ValueError("execution plan operations must be a non-empty bounded array")
+    plan["operations"] = [_validate_operation(item) for item in raw_operations]
+    _canonical_json(plan)
+    return plan
+
+
+def execution_plan_hash(value: Any) -> str:
+    """Return the SHA-256 digest for the canonical execution-plan JSON."""
+
+    plan = validate_execution_plan_v2(value)
+    return hashlib.sha256(_canonical_json(plan).encode("utf-8")).hexdigest()

migrations/versions/20260723_120_rule_execution_runtime.py

@@ -0,0 +1,84 @@
+"""Add immutable runtime bindings and evidence for governed rule execution."""
+
+from alembic import op
+
+
+revision = "20260723_120"
+down_revision = "20260723_110"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE public.data_schema_snapshots (
+            id UUID PRIMARY KEY,
+            schema_ref VARCHAR(500) NOT NULL,
+            schema_hash CHAR(64) NOT NULL,
+            fields JSONB NOT NULL,
+            source_revision VARCHAR(200) NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (schema_ref, schema_hash)
+        );
+
+        CREATE TABLE public.dataflow_dataset_bindings (
+            id UUID PRIMARY KEY,
+            dataflow_deployment_id UUID NOT NULL REFERENCES public.dataflow_deployments(id),
+            logical_ref VARCHAR(500) NOT NULL,
+            data_source_uid UUID,
+            object_kind VARCHAR(30) NOT NULL CHECK (
+                object_kind IN ('table','view','query','parquet_artifact')
+            ),
+            object_ref VARCHAR(1000) NOT NULL,
+            schema_snapshot_id UUID NOT NULL REFERENCES public.data_schema_snapshots(id),
+            dialect VARCHAR(30) NOT NULL,
+            access_mode VARCHAR(20) NOT NULL CHECK (
+                access_mode IN ('read','write','read_write')
+            ),
+            write_mode VARCHAR(30),
+            binding_hash CHAR(64) NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (dataflow_deployment_id, logical_ref)
+        );
+        CREATE INDEX idx_dataflow_dataset_binding_snapshot
+            ON public.dataflow_dataset_bindings(schema_snapshot_id);
+
+        CREATE TABLE public.rule_compile_evidence (
+            id UUID PRIMARY KEY,
+            rule_execution_plan_id UUID NOT NULL REFERENCES public.rule_execution_plans(id)
+                ON DELETE RESTRICT,
+            compiler_version VARCHAR(80) NOT NULL,
+            compiler_digest CHAR(64) NOT NULL,
+            status VARCHAR(20) NOT NULL CHECK (status IN ('success','failed')),
+            evidence JSONB NOT NULL DEFAULT '{}'::jsonb,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (rule_execution_plan_id, compiler_digest)
+        );
+        CREATE INDEX idx_rule_compile_evidence_plan
+            ON public.rule_compile_evidence(rule_execution_plan_id, created_at DESC);
+
+        CREATE TABLE public.rule_test_evidence (
+            id UUID PRIMARY KEY,
+            rule_execution_plan_id UUID NOT NULL REFERENCES public.rule_execution_plans(id)
+                ON DELETE RESTRICT,
+            test_kind VARCHAR(40) NOT NULL,
+            evidence_hash CHAR(64) NOT NULL,
+            status VARCHAR(20) NOT NULL CHECK (status IN ('success','failed')),
+            evidence JSONB NOT NULL DEFAULT '{}'::jsonb,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (rule_execution_plan_id, test_kind, evidence_hash)
+        );
+        CREATE INDEX idx_rule_test_evidence_plan
+            ON public.rule_test_evidence(rule_execution_plan_id, created_at DESC);
+
+        ALTER TABLE public.rule_generation_runs
+            ADD COLUMN created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
+            ADD COLUMN candidate JSONB;
+        """
+    )
+
+
+def downgrade() -> None:
+    # Immutable runtime evidence and audit linkage are retained on rollback.
+    pass

tests/core/data_rules/test_execution_contracts.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_rules.execution_contracts import (
+    execution_plan_hash,
+    validate_dataset_binding,
+    validate_execution_plan_v2,
+    validate_schema_snapshot,
+)
+
+
+def schema_snapshot():
+    return {
+        "schema_ref": "bd:customer:v7",
+        "schema_hash": "a" * 64,
+        "fields": [
+            {"name": "customer_id", "type": "string", "nullable": False},
+            {"name": "mobile", "type": "string", "nullable": True},
+        ],
+        "source_revision": "neo4j:42",
+    }
+
+
+def dataset_binding():
+    return {
+        "data_source_uid": new_governance_uid(),
+        "object_kind": "table",
+        "object_ref": "public.customer",
+        "schema_snapshot_id": new_governance_uid(),
+        "access_mode": "read",
+        "dialect": "postgresql",
+        "write_mode": "append",
+    }
+
+
+def execution_plan():
+    return {
+        "schema_version": "2.0",
+        "backend": "sql_pushdown",
+        "compiler_version": "dataops-sqlglot-1.0",
+        "rule_version_id": new_governance_uid(),
+        "input_schema_snapshot_id": new_governance_uid(),
+        "output_schema_snapshot_id": new_governance_uid(),
+        "input_binding_id": new_governance_uid(),
+        "output_binding_id": new_governance_uid(),
+        "operations": [{"kind": "sql", "statement": "SELECT 1"}],
+    }
+
+
+def test_execution_plan_requires_fixed_schema_dataset_and_backend():
+    plan = validate_execution_plan_v2(execution_plan())
+
+    assert plan["backend"] == "sql_pushdown"
+    assert execution_plan_hash(plan) == execution_plan_hash(
+        dict(reversed(list(plan.items())))
+    )
+
+
+def test_dataset_binding_rejects_credentials_and_unversioned_objects():
+    unsafe = dataset_binding()
+    unsafe.pop("schema_snapshot_id")
+    unsafe["password"] = "unsafe"
+
+    with pytest.raises(ValueError, match="secret|credential"):
+        validate_dataset_binding(unsafe)
+
+
+def test_schema_snapshot_is_closed_and_preserves_typed_fields():
+    snapshot = validate_schema_snapshot(schema_snapshot())
+
+    assert snapshot == schema_snapshot()
+
+    unsafe = copy.deepcopy(snapshot)
+    unsafe["fields"][0]["connection_string"] = "postgresql://unsafe"
+    with pytest.raises(ValueError, match="secret|credential"):
+        validate_schema_snapshot(unsafe)
+
+
+@pytest.mark.parametrize(
+    ("mutation", "message"),
+    [
+        (
+            lambda plan: plan.update({"backend": "generated_python"}),
+            "unsupported execution backend",
+        ),
+        (
+            lambda plan: plan.pop("input_binding_id"),
+            "input_binding_id",
+        ),
+        (
+            lambda plan: plan["operations"][0].update(
+                {"connection_string": "postgresql://unsafe"}
+            ),
+            "secret|credential",
+        ),
+    ],
+)
+def test_execution_plan_rejects_unpinned_or_secret_runtime_material(
+    mutation, message
+):
+    plan = execution_plan()
+    mutation(plan)
+
+    with pytest.raises(ValueError, match=message):
+        validate_execution_plan_v2(plan)

tests/test_data_rule_schema.py

@@ -10,6 +10,12 @@ MIGRATION = (
     / "versions"
     / "20260723_110_ai_data_rules.py"
 )
+RUNTIME_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_120_rule_execution_runtime.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -64,3 +70,33 @@ def test_ai_data_rule_migration_is_forward_preserving():
 
     assert "DROP TABLE" not in downgrade.upper()
     assert "pass" in downgrade
+
+
+def test_rule_execution_runtime_migration_adds_pinned_runtime_evidence():
+    source = RUNTIME_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_120"' in source
+    assert 'down_revision = "20260723_110"' in source
+    for table in (
+        "data_schema_snapshots",
+        "dataflow_dataset_bindings",
+        "rule_compile_evidence",
+        "rule_test_evidence",
+    ):
+        assert f"CREATE TABLE public.{table}" in source
+    for expected in (
+        "rule_execution_plan_id UUID NOT NULL",
+        "created_by UUID REFERENCES public.users(id)",
+        "candidate JSONB",
+        "UNIQUE (schema_ref, schema_hash)",
+        "UNIQUE (dataflow_deployment_id, logical_ref)",
+    ):
+        assert expected in source
+
+
+def test_rule_execution_runtime_migration_is_forward_preserving():
+    source = RUNTIME_MIGRATION.read_text(encoding="utf-8")
+    downgrade = source.split("def downgrade()", 1)[1]
+
+    assert "DROP TABLE" not in downgrade.upper()
+    assert "pass" in downgrade