"""turn the AIGC effect ledger into a transactional outbox Revision ID: 20260731_09 Revises: 20260731_08 Create Date: 2026-07-31 """ from __future__ import annotations from collections.abc import Sequence import sqlalchemy as sa from alembic import op revision: str = "20260731_09" down_revision: str | None = "20260731_08" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None def upgrade() -> None: op.add_column( "pipeline_outbox", sa.Column( "attempt_count", sa.Integer(), nullable=False, server_default="0", ), ) op.add_column( "pipeline_outbox", sa.Column("last_attempt_at", sa.DateTime(), nullable=True), ) op.add_column( "pipeline_outbox", sa.Column("next_retry_at", sa.DateTime(), nullable=True), ) op.add_column( "pipeline_outbox", sa.Column("external_id", sa.String(length=128), nullable=True), ) op.add_column( "pipeline_outbox", sa.Column("external_name", sa.String(length=512), nullable=True), ) op.add_column( "pipeline_outbox", sa.Column("response_json", sa.JSON(), nullable=True), ) outbox = sa.table( "pipeline_outbox", sa.column("status", sa.String(length=32)), ) # Rows produced by the old implementation were written only after the # external call completed, so they represent already-dispatched effects. op.execute( outbox.update() .where(outbox.c.status == "dispatched") .values(status="succeeded") ) def downgrade() -> None: with op.batch_alter_table("pipeline_outbox") as batch_op: batch_op.drop_column("response_json") batch_op.drop_column("external_name") batch_op.drop_column("external_id") batch_op.drop_column("next_retry_at") batch_op.drop_column("last_attempt_at") batch_op.drop_column("attempt_count")