Repair missing BPMN runtime columns

src/govoplan_workflow_engine/backend/migrations/versions/e4a1f8c2d7b6_v0114_repair_bpmn_runtime_columns.py

@@ -0,0 +1,43 @@
+"""v0.1.14 repair BPMN runtime columns
+
+Revision ID: e4a1f8c2d7b6
+Revises: b2e4f6a8c0d1
+Create Date: 2026-08-03 00:00:00.000000
+
+The runtime columns briefly existed only in a modified, already-released
+``e9a4c6b8d2f1`` migration. Installations that applied the earlier form of
+that revision therefore need a new forward migration.
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "e4a1f8c2d7b6"
+down_revision = "b2e4f6a8c0d1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    columns = {
+        column["name"]
+        for column in inspector.get_columns("workflow_definition_revisions")
+    }
+    with op.batch_alter_table("workflow_definition_revisions") as batch_op:
+        if "bpmn_runtime_kind" not in columns:
+            batch_op.add_column(
+                sa.Column("bpmn_runtime_kind", sa.String(length=20), nullable=True)
+            )
+        if "bpmn_executable" not in columns:
+            batch_op.add_column(
+                sa.Column("bpmn_executable", sa.Boolean(), nullable=True)
+            )
+
+
+def downgrade() -> None:
+    # These columns belong to the schema represented by the parent revision.
+    # The repair revision only makes that already-declared state reliable.
+    pass

tests/test_migrations.py

@@ -6,7 +6,7 @@ from pathlib import Path
 import shutil
 
 from alembic.runtime.migration import MigrationContext
-from sqlalchemy import create_engine, inspect
+from sqlalchemy import create_engine, inspect, text
 
 from govoplan_core.db.migrations import migrate_database
 from dataclasses import replace
@@ -29,7 +29,7 @@ class WorkflowMigrationTests(unittest.TestCase):
             try:
                 with engine.connect() as connection:
                     self.assertIn(
-                        "b2e4f6a8c0d1",
+                        "e4a1f8c2d7b6",
                         set(MigrationContext.configure(connection).get_current_heads()),
                     )
                     self.assertEqual(
@@ -101,7 +101,9 @@ class WorkflowMigrationTests(unittest.TestCase):
                 engine_manifest.migration_spec.script_location or ""
             )
             for path in current_revisions.glob("*.py"):
-                if path.name.startswith(("0b4e7c9a2d6f_", "b2e4f6a8c0d1_")):
+                if path.name.startswith(
+                    ("0b4e7c9a2d6f_", "b2e4f6a8c0d1_", "e4a1f8c2d7b6_")
+                ):
                     continue
                 shutil.copy2(path, legacy_revisions / path.name)
             legacy_manifest = replace(
@@ -144,7 +146,7 @@ class WorkflowMigrationTests(unittest.TestCase):
                 manifest_factories=(get_manifest,),
             )
 
-            self.assertIn("b2e4f6a8c0d1", result.current_revision or "")
+            self.assertIn("e4a1f8c2d7b6", result.current_revision or "")
             engine = create_engine(url)
             try:
                 upgraded_tables = set(inspect(engine).get_table_names())
@@ -166,6 +168,64 @@ class WorkflowMigrationTests(unittest.TestCase):
             finally:
                 engine.dispose()
 
+    def test_repair_revision_adds_columns_missing_from_applied_parent(self) -> None:
+        with tempfile.TemporaryDirectory(
+            prefix="govoplan-workflow-engine-repair-"
+        ) as directory:
+            url = f"sqlite:///{Path(directory) / 'workflow.db'}"
+            migrate_database(
+                database_url=url,
+                enabled_modules=("workflow_engine",),
+                manifest_factories=(get_manifest,),
+            )
+            engine = create_engine(url)
+            try:
+                with engine.begin() as connection:
+                    connection.execute(
+                        text(
+                            "ALTER TABLE workflow_definition_revisions "
+                            "DROP COLUMN bpmn_runtime_kind"
+                        )
+                    )
+                    connection.execute(
+                        text(
+                            "ALTER TABLE workflow_definition_revisions "
+                            "DROP COLUMN bpmn_executable"
+                        )
+                    )
+                    connection.execute(
+                        text(
+                            "UPDATE alembic_version SET version_num = "
+                            "'b2e4f6a8c0d1' WHERE version_num = 'e4a1f8c2d7b6'"
+                        )
+                    )
+            finally:
+                engine.dispose()
+
+            migrate_database(
+                database_url=url,
+                enabled_modules=("workflow_engine",),
+                manifest_factories=(get_manifest,),
+            )
+
+            engine = create_engine(url)
+            try:
+                columns = {
+                    item["name"]
+                    for item in inspect(engine).get_columns(
+                        "workflow_definition_revisions"
+                    )
+                }
+                self.assertIn("bpmn_runtime_kind", columns)
+                self.assertIn("bpmn_executable", columns)
+                with engine.connect() as connection:
+                    self.assertIn(
+                        "e4a1f8c2d7b6",
+                        set(MigrationContext.configure(connection).get_current_heads()),
+                    )
+            finally:
+                engine.dispose()
+
 
 if __name__ == "__main__":
     unittest.main()