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

"""Add trusted generation receipts and compile/test publication gates."""

from alembic import op

revision = "20260723_200"
down_revision = "20260723_190"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        DO $$
        BEGIN
            IF EXISTS (
                SELECT 1 FROM public.data_rule_versions
            ) OR EXISTS (
                SELECT 1 FROM public.rule_execution_plans
            ) THEN
                RAISE EXCEPTION
                    'pre-Task7 rule versions or execution plans lack trusted '
                    'generation, compile, test, and publication provenance; '
                    'export and remove every legacy row, run migration 200, '
                    'then rebuild each rule through the Task7 publication '
                    'chain';
            END IF;
        END $$;

        ALTER TABLE public.data_rule_versions
            DROP CONSTRAINT data_rule_versions_status_check;
        ALTER TABLE public.data_rule_versions
            ADD CONSTRAINT data_rule_versions_status_check
            CHECK (status IN ('draft','validated','published','revoked'));

        ALTER TABLE public.rule_execution_plans
            DROP CONSTRAINT rule_execution_plans_status_check;
        ALTER TABLE public.rule_execution_plans
            ADD CONSTRAINT rule_execution_plans_status_check
            CHECK (status IN ('compiled','tested','published','revoked'));

        ALTER TABLE public.rule_generation_runs
            ADD COLUMN receipt_hash CHAR(64),
            ADD COLUMN receipt_consumed_at TIMESTAMPTZ,
            ADD COLUMN validation_context JSONB,
            ADD COLUMN model_hash CHAR(64),
            ADD COLUMN prompt_hash CHAR(64);
        CREATE UNIQUE INDEX uq_rule_generation_linked_version
            ON public.rule_generation_runs(rule_version_id)
            WHERE rule_version_id IS NOT NULL;
        CREATE UNIQUE INDEX uq_rule_generation_receipt_hash
            ON public.rule_generation_runs(receipt_hash)
            WHERE receipt_hash IS NOT NULL;

        CREATE TABLE public.rule_generation_attempts (
            id UUID PRIMARY KEY,
            generation_run_id UUID NOT NULL
                REFERENCES public.rule_generation_runs(id) ON DELETE RESTRICT,
            attempt_no INTEGER NOT NULL CHECK (attempt_no BETWEEN 0 AND 2),
            candidate_hash CHAR(64) NOT NULL,
            candidate JSONB,
            error_code VARCHAR(80),
            model_hash CHAR(64) NOT NULL,
            prompt_hash CHAR(64) NOT NULL,
            context_hash CHAR(64) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('valid','invalid')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (generation_run_id, attempt_no)
        );

        CREATE TABLE public.rule_validation_profiles (
            id UUID PRIMARY KEY,
            rule_version_id UUID NOT NULL UNIQUE
                REFERENCES public.data_rule_versions(id) ON DELETE RESTRICT,
            input_schema_snapshot_id UUID NOT NULL
                REFERENCES public.data_schema_snapshots(id) ON DELETE RESTRICT,
            input_schema_hash CHAR(64) NOT NULL,
            input_fields JSONB NOT NULL,
            output_schema_snapshot_id UUID NOT NULL
                REFERENCES public.data_schema_snapshots(id) ON DELETE RESTRICT,
            output_schema_hash CHAR(64) NOT NULL,
            output_fields JSONB NOT NULL,
            input_sample_artifact_ref VARCHAR(500) NOT NULL,
            input_sample_artifact_digest CHAR(64) NOT NULL,
            golden_output_artifact_ref VARCHAR(500),
            golden_output_artifact_digest CHAR(64),
            context_hash CHAR(64) NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (
                (golden_output_artifact_ref IS NULL
                 AND golden_output_artifact_digest IS NULL)
                OR
                (golden_output_artifact_ref IS NOT NULL
                 AND golden_output_artifact_digest IS NOT NULL)
            )
        );

        CREATE TABLE public.rule_logical_plans (
            id UUID PRIMARY KEY,
            rule_version_id UUID NOT NULL
                REFERENCES public.data_rule_versions(id) ON DELETE RESTRICT,
            validation_profile_id UUID NOT NULL
                REFERENCES public.rule_validation_profiles(id)
                ON DELETE RESTRICT,
            compiler_version VARCHAR(80) NOT NULL,
            backend VARCHAR(30) NOT NULL CHECK (
                backend IN ('sql_pushdown','polars_batch')
            ),
            plan JSONB NOT NULL,
            plan_hash CHAR(64) NOT NULL,
            schema_hashes JSONB NOT NULL,
            capabilities JSONB NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('compiled','tested','published','revoked')
            ),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (rule_version_id, plan_hash)
        );

        CREATE TABLE public.rule_logical_compile_evidence (
            id UUID PRIMARY KEY,
            logical_plan_id UUID NOT NULL
                REFERENCES public.rule_logical_plans(id) ON DELETE RESTRICT,
            compiler_version VARCHAR(80) NOT NULL,
            compiler_digest CHAR(64) NOT NULL,
            plan_hash CHAR(64) NOT NULL,
            schema_hashes JSONB NOT NULL,
            capabilities JSONB NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('success','failed')),
            created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (logical_plan_id, compiler_digest)
        );

        CREATE TABLE public.rule_logical_test_evidence (
            id UUID PRIMARY KEY,
            logical_plan_id UUID NOT NULL
                REFERENCES public.rule_logical_plans(id) ON DELETE RESTRICT,
            test_kind VARCHAR(40) NOT NULL,
            evidence_hash CHAR(64) NOT NULL,
            run_id UUID NOT NULL,
            plan_hash CHAR(64) NOT NULL,
            schema_hashes JSONB NOT NULL,
            evidence JSONB NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('success','failed')),
            created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (logical_plan_id, test_kind, evidence_hash)
        );

        ALTER TABLE public.rule_compile_evidence
            ADD COLUMN plan_hash CHAR(64),
            ADD COLUMN schema_hashes JSONB,
            ADD COLUMN binding_hashes JSONB,
            ADD COLUMN capabilities JSONB,
            ADD COLUMN legacy_untrusted BOOLEAN NOT NULL DEFAULT TRUE,
            ADD COLUMN created_by UUID REFERENCES public.users(id)
                ON DELETE SET NULL;
        UPDATE public.rule_compile_evidence e
        SET plan_hash = p.plan_hash,
            schema_hashes = p.schema_hashes,
            binding_hashes = jsonb_build_object(
                'input', COALESCE(p.plan->>'input_binding_hash', ''),
                'output', COALESCE(p.plan->>'output_binding_hash', '')
            ),
            capabilities = COALESCE(
                p.plan->'capabilities',
                jsonb_build_object('backend', p.backend)
            )
        FROM public.rule_execution_plans p
        WHERE p.id = e.rule_execution_plan_id;
        ALTER TABLE public.rule_compile_evidence
            ALTER COLUMN plan_hash SET NOT NULL,
            ALTER COLUMN schema_hashes SET NOT NULL,
            ALTER COLUMN binding_hashes SET NOT NULL,
            ALTER COLUMN capabilities SET NOT NULL;

        ALTER TABLE public.rule_test_evidence
            ADD COLUMN plan_hash CHAR(64),
            ADD COLUMN schema_hashes JSONB,
            ADD COLUMN binding_hashes JSONB,
            ADD COLUMN run_id UUID,
            ADD COLUMN legacy_untrusted BOOLEAN NOT NULL DEFAULT TRUE,
            ADD COLUMN created_by UUID REFERENCES public.users(id)
                ON DELETE SET NULL;
        UPDATE public.rule_test_evidence e
        SET plan_hash = p.plan_hash,
            schema_hashes = p.schema_hashes,
            binding_hashes = jsonb_build_object(
                'input', COALESCE(p.plan->>'input_binding_hash', ''),
                'output', COALESCE(p.plan->>'output_binding_hash', '')
            ),
            run_id = e.id
        FROM public.rule_execution_plans p
        WHERE p.id = e.rule_execution_plan_id;
        ALTER TABLE public.rule_test_evidence
            ALTER COLUMN plan_hash SET NOT NULL,
            ALTER COLUMN schema_hashes SET NOT NULL,
            ALTER COLUMN binding_hashes SET NOT NULL,
            ALTER COLUMN run_id SET NOT NULL;

        CREATE TABLE public.rule_validation_attempts (
            id UUID PRIMARY KEY,
            rule_version_id UUID NOT NULL
                REFERENCES public.data_rule_versions(id) ON DELETE RESTRICT,
            actor_uid UUID REFERENCES public.users(id) ON DELETE SET NULL,
            error_code VARCHAR(80) NOT NULL,
            error_hash CHAR(64) NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE public.rule_publication_audits (
            id UUID PRIMARY KEY,
            rule_version_id UUID NOT NULL
                REFERENCES public.data_rule_versions(id) ON DELETE RESTRICT,
            rule_execution_plan_id UUID
                REFERENCES public.rule_execution_plans(id) ON DELETE RESTRICT,
            actor_uid UUID REFERENCES public.users(id) ON DELETE SET NULL,
            action VARCHAR(30) NOT NULL CHECK (
                action IN (
                    'draft_created','compile_succeeded','test_succeeded',
                    'published','revoked'
                )
            ),
            from_status VARCHAR(20),
            to_status VARCHAR(20) NOT NULL,
            evidence_hash CHAR(64),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_rule_publication_audit_version
            ON public.rule_publication_audits(
                rule_version_id, created_at DESC
            );
        CREATE INDEX idx_published_rule_catalog
            ON public.data_rule_versions(status, published_at DESC)
            WHERE status = 'published';
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "trusted rule publication evidence is forward-only and cannot be "
        "downgraded without invalidating signed receipts and audit history"
    )