DataOps-platform/migrations/versions/20260719_80_mcp_gateway_control_plane.py

"""Add persistent MCP Gateway scope, canary evidence, and idempotency state."""

from alembic import op

revision = "20260719_80"
down_revision = "20260719_70"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        ALTER TABLE public.dataflow_workflow_versions
            ADD COLUMN business_domain VARCHAR(200) NOT NULL DEFAULT 'legacy',
            ADD COLUMN created_by_subject VARCHAR(200) NOT NULL DEFAULT 'legacy',
            ADD COLUMN write_authorized BOOLEAN NOT NULL DEFAULT FALSE;

        CREATE INDEX idx_workflow_versions_gateway_scope
            ON public.dataflow_workflow_versions(
                business_domain, environment, status, version_no DESC
            );

        ALTER TABLE public.workflow_plan_audits
            ADD COLUMN actor_subject VARCHAR(200) NOT NULL DEFAULT 'legacy';

        ALTER TABLE public.workflow_plan_audits
            DROP CONSTRAINT IF EXISTS workflow_plan_audits_decision_check;
        ALTER TABLE public.workflow_plan_audits
            DROP CONSTRAINT IF EXISTS ck_workflow_plan_audits_decision;
        ALTER TABLE public.workflow_plan_audits
            ADD CONSTRAINT ck_workflow_plan_audits_decision
                CHECK (decision IN (
                    'allowed', 'rejected', 'failed', 'recorded',
                    'idempotent_replay'
                ));

        CREATE TABLE public.workflow_canary_evidence (
            id UUID PRIMARY KEY,
            workflow_version_id UUID NOT NULL
                REFERENCES public.dataflow_workflow_versions(id)
                ON DELETE CASCADE,
            engine_execution_id VARCHAR(255) NOT NULL,
            baseline_execution_id VARCHAR(255),
            status VARCHAR(20) NOT NULL
                CHECK (status IN ('started', 'passed', 'failed')),
            sample_runs INTEGER NOT NULL DEFAULT 0 CHECK (sample_runs >= 0),
            verified_by VARCHAR(200),
            verification_summary JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            verified_at TIMESTAMPTZ,
            UNIQUE (engine_execution_id)
        );
        CREATE INDEX idx_workflow_canary_version_created
            ON public.workflow_canary_evidence(
                workflow_version_id, created_at DESC
            );

        CREATE TABLE public.workflow_gateway_operations (
            id UUID PRIMARY KEY,
            workflow_version_id UUID
                REFERENCES public.dataflow_workflow_versions(id)
                ON DELETE SET NULL,
            action VARCHAR(80) NOT NULL,
            idempotency_key VARCHAR(500) NOT NULL,
            status VARCHAR(20) NOT NULL
                CHECK (status IN ('claimed', 'succeeded', 'failed')),
            result JSONB NOT NULL DEFAULT '{}'::jsonb,
            safe_error VARCHAR(1000),
            actor_subject VARCHAR(200) NOT NULL,
            correlation_id UUID NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (action, idempotency_key)
        );
        CREATE INDEX idx_workflow_gateway_operations_version
            ON public.workflow_gateway_operations(
                workflow_version_id, action, created_at DESC
            );

        ALTER TABLE public.workflow_runs
            ADD COLUMN parent_run_id UUID
                REFERENCES public.workflow_runs(id) ON DELETE SET NULL,
            ADD COLUMN attempt INTEGER NOT NULL DEFAULT 1 CHECK (attempt > 0),
            ADD COLUMN run_metadata JSONB NOT NULL DEFAULT '{}'::jsonb;
        CREATE INDEX idx_workflow_runs_parent
            ON public.workflow_runs(parent_run_id, attempt);
        """
    )


def downgrade() -> None:
    op.execute(
        """
        DROP TABLE IF EXISTS public.workflow_gateway_operations;
        DROP TABLE IF EXISTS public.workflow_canary_evidence;

        DROP INDEX IF EXISTS public.idx_workflow_runs_parent;
        ALTER TABLE public.workflow_runs
            DROP COLUMN IF EXISTS run_metadata,
            DROP COLUMN IF EXISTS attempt,
            DROP COLUMN IF EXISTS parent_run_id;

        UPDATE public.workflow_plan_audits
        SET decision = 'recorded'
        WHERE decision = 'idempotent_replay';

        ALTER TABLE public.workflow_plan_audits
            DROP CONSTRAINT IF EXISTS ck_workflow_plan_audits_decision;
        ALTER TABLE public.workflow_plan_audits
            ADD CONSTRAINT workflow_plan_audits_decision_check
                CHECK (decision IN (
                    'allowed', 'rejected', 'failed', 'recorded'
                ));
        ALTER TABLE public.workflow_plan_audits
            DROP COLUMN IF EXISTS actor_subject;

        DROP INDEX IF EXISTS public.idx_workflow_versions_gateway_scope;
        ALTER TABLE public.dataflow_workflow_versions
            DROP COLUMN IF EXISTS write_authorized,
            DROP COLUMN IF EXISTS created_by_subject,
            DROP COLUMN IF EXISTS business_domain;
        """
    )