"""Add the durable single-use Runner task execution ledger.""" from alembic import op revision = "20260719_70" down_revision = "20260718_60" branch_labels = None depends_on = None def upgrade() -> None: op.execute( """ CREATE TABLE public.runner_task_executions ( token_jti UUID PRIMARY KEY, task_uid UUID NOT NULL, dataflow_uid UUID NOT NULL, workflow_version INTEGER NOT NULL CHECK (workflow_version > 0), correlation_id UUID NOT NULL, node_id VARCHAR(100) NOT NULL, node_type VARCHAR(100) NOT NULL, data_source_uid UUID, idempotency_key VARCHAR(500), status VARCHAR(20) NOT NULL CHECK (status IN ('running', 'success', 'failed', 'unknown')), commit_outcome VARCHAR(30) NOT NULL CHECK (commit_outcome IN ( 'not_applicable', 'not_committed', 'committed', 'unknown' )), safe_detail VARCHAR(500), expires_at TIMESTAMPTZ NOT NULL, started_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, finished_at TIMESTAMPTZ, UNIQUE (task_uid) ); CREATE INDEX idx_runner_task_executions_correlation ON public.runner_task_executions(correlation_id, started_at DESC); CREATE INDEX idx_runner_task_executions_datasource ON public.runner_task_executions(data_source_uid, started_at DESC) WHERE data_source_uid IS NOT NULL; CREATE INDEX idx_runner_task_executions_expiry ON public.runner_task_executions(expires_at); """ ) def downgrade() -> None: op.execute( "DROP TABLE IF EXISTS public.runner_task_executions;" )