DataOps-platform/migrations/versions/20260723_120_rule_execution_runtime.py

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