diff --git a/backend/tests/alembic_migration_schema_assertions.py b/backend/tests/alembic_migration_schema_assertions.py
new file mode 100644
index 000000000..dc94c5a11
--- /dev/null
+++ b/backend/tests/alembic_migration_schema_assertions.py
@@ -0,0 +1,262 @@
+# #region Test.Alembic.SchemaAssertions [C:4] [TYPE Module] [SEMANTICS test,alembic,migration,schema,source-binding,postgres]
+# @BRIEF DDL and behavior assertions for migration-owned tables, independent of production ORM metadata.
+# @RELATION BINDS_TO -> [Migrations.0051.DeploymentAppliedAt]
+# @RELATION BINDS_TO -> [Migrations.0052.ReferenceDraftInspection]
+# @PRE Callers pass an engine bound to a schema already upgraded to the revision under assertion.
+# @POST Columns, constraints, indexes and insert refusals are checked against hardcoded expected values.
+# @INVARIANT Expectations are literal fixtures and never imported from production ORM metadata.
+
+import re
+import uuid
+
+import pytest
+
+# #region Test.AlembicMigrations.AppliedAtSchema [C:2] [TYPE Function]
+# @POST deployment_records.applied_at/target_digest become nullable after 0051 and keep their declared types.
+def _assert_0051_schema(engine) -> None:
+    from sqlalchemy import inspect
+    import sqlalchemy as sa
+
+    columns = {column["name"]: column for column in inspect(engine).get_columns("deployment_records")}
+    _assert_deployment_applied_columns(columns)
+    assert isinstance(columns["applied_at"]["type"], sa.DateTime)
+# #endregion Test.AlembicMigrations.AppliedAtSchema
+
+
+# #region Test.AlembicMigrations.DeploymentAppliedColumns [C:2] [TYPE Function]
+# @POST The 0051 boundary columns are nullable and target_digest keeps the 64-char width.
+def _assert_deployment_applied_columns(columns: dict) -> None:
+    assert columns["applied_at"]["nullable"] is True
+    assert columns["target_digest"]["nullable"] is True
+    assert columns["target_digest"]["type"].length == 64
+# #endregion Test.AlembicMigrations.DeploymentAppliedColumns
+
+
+# #region Test.AlembicMigrations.InspectionColumnAbsent [C:1] [TYPE Function]
+# @POST The 0052-owned inspection column is absent before 0052 is applied.
+def _assert_no_inspection_snapshot_column(schema_engine) -> None:
+    from sqlalchemy import inspect
+
+    assert "inspection_snapshot" not in {column["name"] for column in inspect(schema_engine).get_columns("dashboard_reference_draft_authorities")}
+# #endregion Test.AlembicMigrations.InspectionColumnAbsent
+
+
+# #region Test.AlembicMigrations.InspectionSchema [C:2] [TYPE Function]
+# @POST The 0052-owned inspection column exists exactly once as nullable JSON without a server default.
+def _assert_0052_schema(engine) -> None:
+    from sqlalchemy import inspect
+    from sqlalchemy.dialects.postgresql import JSONB
+    import sqlalchemy as sa
+
+    columns = [column for column in inspect(engine).get_columns("dashboard_reference_draft_authorities")
+               if column["name"] == "inspection_snapshot"]
+    assert len(columns) == 1
+    assert columns[0]["nullable"] is True
+    assert isinstance(columns[0]["type"], sa.JSON)
+    assert not isinstance(columns[0]["type"], JSONB)
+    assert columns[0].get("default") is None
+# #endregion Test.AlembicMigrations.InspectionSchema
+
+
+# #region Test.AlembicMigrations.SourceBindingSchemaAssert [C:4] [TYPE Function]
+# @BRIEF Assert the source-binding DDL independently of production ORM metadata.
+# @POST Verifies exact columns, FK columns/ondelete, check expressions, index columns/partial predicates.
+def _assert_source_binding_schema(schema_engine, schema: str) -> None:
+    from sqlalchemy import inspect
+
+    table = "git_repository_source_bindings"
+    inspector = inspect(schema_engine)
+    assert table in set(inspector.get_table_names())
+    with schema_engine.connect() as connection:
+        _assert_source_binding_columns(connection, schema, table)
+        _assert_source_binding_constraints(connection, schema, table)
+        _assert_source_binding_indexes(connection, schema, table)
+# #endregion Test.AlembicMigrations.SourceBindingSchemaAssert
+
+
+# #region Test.AlembicMigrations.SourceBindingColumnsAssert [C:2] [TYPE Function]
+# @POST The source-binding table has exactly the expected 16 columns and nullability.
+def _assert_source_binding_columns(connection, schema: str, table: str) -> None:
+    from sqlalchemy import text as sa_text
+
+    column_rows = connection.execute(sa_text(
+        "SELECT column_name, is_nullable FROM information_schema.columns "
+        "WHERE table_schema = :schema AND table_name = :table"
+    ), {"schema": schema, "table": table}).all()
+    expected_columns = {
+        "id": "NO", "repository_id": "NO", "dashboard_uuid": "NO",
+        "source_environment_id": "NO", "source_dashboard_id": "NO",
+        "release_id": "NO", "release_commit_hash": "NO", "status": "NO",
+        "actor_id": "NO", "actor_username": "NO", "created_at": "NO",
+        "retired_at": "YES", "retired_by": "YES", "cas_version": "NO",
+        "request_sha256": "NO", "receipt_sha256": "NO",
+    }
+    assert {row.column_name: row.is_nullable for row in column_rows} == expected_columns
+# #endregion Test.AlembicMigrations.SourceBindingColumnsAssert
+
+
+# #region Test.AlembicMigrations.SourceBindingConstraintsAssert [C:3] [TYPE Function]
+# @POST The two FK/RESTRICT constraints and six checks exist with their exact definitions.
+def _assert_source_binding_constraints(connection, schema: str, table: str) -> None:
+    from sqlalchemy import text as sa_text
+
+    constraint_rows = connection.execute(sa_text(
+        "SELECT conname, contype, pg_get_constraintdef(oid) AS definition "
+        "FROM pg_constraint WHERE conrelid = CAST(:qualified AS regclass)"
+    ), {"qualified": f"{schema}.{table}"}).all()
+    definitions = {row.contype: [] for row in constraint_rows}
+    for row in constraint_rows:
+        definitions[row.contype].append(" ".join(row.definition.replace('"', "").lower().split()))
+    assert set(definitions.get("f", [])) == {
+        "foreign key (repository_id) references git_repositories(id) on delete restrict",
+        "foreign key (release_id) references dashboard_releases(id) on delete restrict",
+    }
+    assert len(definitions.get("c", [])) == 6
+# #endregion Test.AlembicMigrations.SourceBindingConstraintsAssert
+
+
+# #region Test.AlembicMigrations.IndexPredicateNormalize [C:2] [TYPE Function]
+# @POST Normalizes a PG index predicate to the hardcoded expected form; None means no predicate.
+def _normalize_index_predicate(row):
+    if row.predicate is None:
+        return None
+    normalized_predicate = "".join(
+        row.predicate.replace('"', "").lower().replace("::text", "").split())
+    if normalized_predicate.startswith("(") and normalized_predicate.endswith(")"):
+        normalized_predicate = normalized_predicate[1:-1]
+    return re.sub(r"\(([a-z_][a-z0-9_]*)\)", r"\1", normalized_predicate)
+# #endregion Test.AlembicMigrations.IndexPredicateNormalize
+
+
+# #region Test.AlembicMigrations.SourceBindingIndexesAssert [C:3] [TYPE Function]
+# @POST The primary/plain/partial-unique indexes exist with exact uniqueness, predicate and columns.
+def _assert_source_binding_indexes(connection, schema: str, table: str) -> None:
+    from sqlalchemy import text as sa_text
+
+    index_rows = connection.execute(sa_text(
+        "SELECT i.relname AS indexname, x.indisunique AS is_unique, "
+        "pg_get_indexdef(x.indexrelid) AS indexdef, "
+        "pg_get_expr(x.indpred, x.indrelid) AS predicate "
+        "FROM pg_index x JOIN pg_class i ON i.oid = x.indexrelid "
+        "WHERE x.indrelid = CAST(:qualified AS regclass)"
+    ), {"qualified": f"{schema}.{table}"}).all()
+    expected_indexes = {
+        "git_repository_source_bindings_pkey": (True, None, "(id)"),
+        "ix_git_source_binding_repository": (False, None, "(repository_id)"),
+        "uq_git_source_binding_active_repository": (True, "status='active'", "(repository_id)"),
+        "uq_git_source_binding_active_uuid": (True, "status='active'", "(dashboard_uuid)"),
+        "uq_git_source_binding_active_source": (True, "status='active'",
+                                                  "(source_environment_id, source_dashboard_id)"),
+    }
+    actual_indexes = {row.indexname: row for row in index_rows}
+    assert set(actual_indexes) == set(expected_indexes)
+    for name, (unique, predicate, columns) in expected_indexes.items():
+        row = actual_indexes[name]
+        assert bool(row.is_unique) is unique, name
+        normalized_predicate = _normalize_index_predicate(row)
+        assert normalized_predicate == predicate, name
+        assert columns in row.indexdef.replace('"', "").lower(), name
+# #endregion Test.AlembicMigrations.SourceBindingIndexesAssert
+
+
+# #region Test.AlembicMigrations.SourceBindingBehavior [C:4] [TYPE Function]
+# @BRIEF Independent insert-refusal/retired-allowed fixtures for checks and active partial indexes.
+# @POST All six checks reject invalid rows; retired duplicates are allowed; active duplicates are rejected by DB.
+# @INVARIANT Expected values are hardcoded here, not imported from production ORM metadata.
+def _assert_source_binding_behavior(schema_engine) -> None:
+    from datetime import UTC, datetime
+    from sqlalchemy.exc import IntegrityError
+    from sqlalchemy import text as sa_text
+
+    table = "git_repository_source_bindings"
+    env_id = "00000000-0000-4000-8000-000000000001"
+    repo_a = "00000000-0000-4000-8000-000000000011"
+    repo_b = "00000000-0000-4000-8000-000000000012"
+    repo_c = "00000000-0000-4000-8000-000000000013"
+    release_id = "00000000-0000-4000-8000-000000000021"
+    uuid_a = "a2cef831-4ffd-46ec-aaa1-60f71782a91c"
+    uuid_b = "b2cef831-4ffd-46ec-aaa1-60f71782a91c"
+    uuid_c = "c2cef831-4ffd-46ec-aaa1-60f71782a91c"
+    commit = "a" * 40
+    request_hash = "0" * 64
+    receipt_hash = "1" * 64
+    now = datetime.now(UTC)
+
+    with schema_engine.begin() as connection:
+        connection.execute(sa_text(
+            "INSERT INTO deployment_environments (id, name, superset_url, superset_token, is_active) "
+            "VALUES (:id, :name, :url, :token, true)"
+        ), {"id": env_id, "name": "fixture", "url": "http://fixture", "token": ""})
+        for dashboard_id, repo_id in ((810001, repo_a), (810002, repo_b), (810003, repo_c)):
+            connection.execute(sa_text(
+                "INSERT INTO git_repositories "
+                "(id, dashboard_id, local_path, current_branch, sync_status) "
+                "VALUES (:id, :dashboard_id, :path, 'dev', 'CLEAN')"
+            ), {"id": repo_id, "dashboard_id": dashboard_id, "path": f"/tmp/{repo_id}"})
+        deployment_id = connection.execute(sa_text(
+            "INSERT INTO deployment_records "
+            "(repository_id, environment_id, commit_hash, content_hash, deployed_at, status, validation_status) "
+            "VALUES (:repo, :env, :commit, :content, :deployed, 'success', 'validated') RETURNING id"
+        ), {"repo": repo_a, "env": env_id, "commit": commit, "content": "c" * 64,
+            "deployed": now}).scalar_one()
+        connection.execute(sa_text(
+            "INSERT INTO dashboard_releases "
+            "(id, repository_id, deployment_id, name, version, notes, commit_hash, content_hash, status, created_at, created_by) "
+            "VALUES (:id, :repo, :deployment, 'fixture', '1.0.0', 'fixture', :commit, :content, 'ready_to_publish', :created, 'fixture')"
+        ), {"id": release_id, "repo": repo_a, "deployment": deployment_id, "commit": commit,
+            "content": "c" * 64, "created": now})
+
+    # #region Test.AlembicMigrations.SourceBindingInsert [C:1] [TYPE Function]
+    # @BRIEF One valid source-binding row fixture; invalid variants are supplied by the caller.
+    def insert_binding(*, repository_id: str, dashboard_uuid: str, source_environment_id: str,
+                       source_dashboard_id: int = 12, status: str = "active", commit_hash: str = commit,
+                       cas_version: int = 1) -> None:
+        with schema_engine.begin() as connection:
+            connection.execute(sa_text(
+                f"INSERT INTO {table} "
+                "(id, repository_id, dashboard_uuid, source_environment_id, source_dashboard_id, "
+                "release_id, release_commit_hash, status, actor_id, actor_username, created_at, "
+                "cas_version, request_sha256, receipt_sha256) "
+                "VALUES (:id, :repo, :uuid, :source_env, :source_dashboard, :release, :commit, :status, "
+                "'actor', 'operator', :created, :cas, :request, :receipt)"
+            ), {"id": str(uuid.uuid4()), "repo": repository_id, "uuid": dashboard_uuid,
+                "source_env": source_environment_id, "source_dashboard": source_dashboard_id,
+                "release": release_id, "commit": commit_hash, "status": status,
+                "created": now, "cas": cas_version, "request": request_hash, "receipt": receipt_hash})
+    # #endregion Test.AlembicMigrations.SourceBindingInsert
+
+    insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a")
+    insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a",
+                   status="retired")
+    with schema_engine.connect() as connection:
+        assert connection.execute(sa_text(f"SELECT count(*) FROM {table}")).scalar_one() == 2
+
+    for invalid in (
+        {"repository_id": repo_a, "uuid": uuid_b, "source_env": "source-b"},
+        {"repository_id": repo_b, "uuid": uuid_a, "source_env": "source-c"},
+        {"repository_id": repo_c, "uuid": uuid_b, "source_env": "source-a"},
+    ):
+        with pytest.raises(IntegrityError):
+            insert_binding(repository_id=invalid["repository_id"], dashboard_uuid=invalid["uuid"],
+                           source_environment_id=invalid["source_env"])
+
+    invalid_rows = (
+        {"status": "paused"},
+        {"dashboard_uuid": "short"},
+        {"commit_hash": "short"},
+        {"source_environment_id": ""},
+        {"source_dashboard_id": 0},
+        {"cas_version": 0},
+    )
+    for invalid in invalid_rows:
+        values = {
+            "repository_id": repo_a, "dashboard_uuid": uuid_c,
+            "source_environment_id": "source-check", "source_dashboard_id": 12,
+            "status": "retired", "commit_hash": commit, "cas_version": 1,
+        }
+        values.update(invalid)
+        with pytest.raises(IntegrityError):
+            insert_binding(**values)
+# #endregion Test.AlembicMigrations.SourceBindingBehavior
+
+# #endregion Test.Alembic.SchemaAssertions
diff --git a/backend/tests/alembic_migration_support.py b/backend/tests/alembic_migration_support.py
new file mode 100644
index 000000000..46d44a1d7
--- /dev/null
+++ b/backend/tests/alembic_migration_support.py
@@ -0,0 +1,66 @@
+# #region Test.Alembic.MigrationSupport [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,postgres,isolation]
+# @BRIEF Shared PostgreSQL harness for Alembic migration tests: backend gate and per-test schema isolation.
+# @RELATION BINDS_TO -> [Alembic.Env.AlembicEnvModule]
+# @PRE DATABASE_URL is exported by the run-owned test database.
+# @POST PostgreSQL-only test modules import the same backend gate; every isolation caller owns one private schema.
+# @INVARIANT The isolation helper drops only its own schema and never downgrades or drops shared worker tables.
+# @REJECTED Migrating the already-migrated shared worker schema was rejected: both upgrades would be no-ops.
+
+from contextlib import contextmanager
+import os
+import uuid
+
+import pytest
+from sqlalchemy.engine import make_url
+
+# Alembic migrations use PostgreSQL-specific DDL (ALTER COLUMN, partial indexes, JSON).
+_DB_BACKEND = make_url(os.environ.get("DATABASE_URL", "sqlite://")).get_backend_name()
+_REQUIRES_PG = pytest.mark.skipif(
+    _DB_BACKEND != "postgresql",
+    reason="Alembic migrations are PostgreSQL-specific; use a PostgreSQL DATABASE_URL to test",
+)
+
+
+# #region Test.AlembicMigrations.SourceBindingIsolation [C:4] [TYPE Function]
+# @BRIEF Give each migration test its own PostgreSQL schema in the run-owned worker DB.
+# @POST Alembic sees exactly one search_path (no public fallback); cleanup drops only this test's schema.
+# @INVARIANT No downgrade/drop of shared worker tables; no credential files or URL logging.
+@contextmanager
+def _isolated_pg_schema(monkeypatch: pytest.MonkeyPatch):
+    from sqlalchemy import create_engine, text as sa_text
+
+    schema = f"alembic_sb_{uuid.uuid4().hex}"
+    previous_schema = os.environ.get("DB_SCHEMA")
+    engine = None
+    schema_engine = None
+    created = False
+    try:
+        engine = create_engine(os.environ["DATABASE_URL"])
+        with engine.begin() as connection:
+            connection.execute(sa_text(f'CREATE SCHEMA "{schema}"'))
+        created = True
+        schema_engine = create_engine(
+            os.environ["DATABASE_URL"],
+            connect_args={"options": f"-csearch_path={schema}"},
+        )
+        monkeypatch.setenv("DB_SCHEMA", schema)
+        yield schema, schema_engine
+    finally:
+        try:
+            if schema_engine is not None:
+                schema_engine.dispose()
+        finally:
+            try:
+                if created and engine is not None:
+                    with engine.begin() as connection:
+                        connection.execute(sa_text(f'DROP SCHEMA "{schema}" CASCADE'))
+            finally:
+                if engine is not None:
+                    engine.dispose()
+                if previous_schema is None:
+                    monkeypatch.delenv("DB_SCHEMA", raising=False)
+                else:
+                    monkeypatch.setenv("DB_SCHEMA", previous_schema)
+# #endregion Test.AlembicMigrations.SourceBindingIsolation
+
+# #endregion Test.Alembic.MigrationSupport
diff --git a/backend/tests/test_alembic_inplace_upgrade.py b/backend/tests/test_alembic_inplace_upgrade.py
index 05c5b1d7b..cf0c4947f 100644
--- a/backend/tests/test_alembic_inplace_upgrade.py
+++ b/backend/tests/test_alembic_inplace_upgrade.py
@@ -6,7 +6,8 @@
 # @PRE Catalog/publication tables are byte-identical at 0046 and at head: no migration after 0046 touches them
 #      (0047-0049 change draft authorities/receipts/finishes, 0050 adds git_repository_source_bindings,
 #       0051 adds nullable applied_at/target_digest only to deployment_records;
-#       0052 adds nullable inspection_snapshot only to draft authorities).
+#       0052 adds nullable inspection_snapshot only to draft authorities;
+#       0053-0055 add finish intent/settings and run-notification tables, not catalog/publication tables).
 # @POST The same rows, payload bytes and digests exist after `alembic upgrade head`, and the current read
 #       path returns them by exact revision, including a receipt-bound publication.
 # @INVARIANT The upgrade never rewrites an approved catalog row, an approval receipt or a publication operation.
@@ -28,7 +29,7 @@ from sqlalchemy import text
 from tests.test_alembic_migrations import _REQUIRES_PG, _isolated_pg_schema, _run_alembic_upgrade
 
 PREVIOUS_RELEASE_HEAD = "0046_mapper_llm_plans"
-CURRENT_HEAD = "0052_draft_inspection"
+CURRENT_HEAD = "0055_scenario_run_notifications"
 EXPECTED_FORWARD_REVISIONS = (
     "0047_reference_draft_projection",
     "0048_reference_draft_finish",
@@ -36,6 +37,9 @@ EXPECTED_FORWARD_REVISIONS = (
     "0050_git_source_bind",
     "0051_deployment_applied_at",
     "0052_draft_inspection",
+    "0053_reference_finish_intent",
+    "0054_reference_finish_settings",
+    "0055_scenario_run_notifications",
 )
 _PRE_BUILD_IDENTITY_AUTHORITY_KEYS = (
     "superset_build",
@@ -47,8 +51,9 @@ _PRE_BUILD_IDENTITY_AUTHORITY_KEYS = (
 
 
 # #region Test.Alembic.InPlaceUpgrade.ExactForwardChain [C:2] [TYPE Function]
-# @BRIEF The integrated previous-release path includes every exact forward revision through 0052.
-# @RATIONALE Integration adds the 0052 draft inspection column while preserving every earlier revision and approved-reference assertion.
+# @BRIEF The integrated previous-release path includes every exact forward revision through CURRENT_HEAD.
+# @RATIONALE Integration adds the 0052 inspection column and the 0053-0055 finish/settings/notification revisions
+#            while preserving every earlier revision and approved-reference assertion.
 # @REJECTED Deriving the expected head from the chain under test would silently accept a removed or foreign revision.
 def test_previous_release_forward_chain_is_exact() -> None:
     from alembic.script import ScriptDirectory
@@ -65,6 +70,11 @@ def test_previous_release_forward_chain_is_exact() -> None:
 # @PRE The schema is already at PREVIOUS_RELEASE_HEAD inside an isolated PostgreSQL schema.
 # @POST One published catalog revision with an entry + authority, one draft approval receipt and one
 #       publication operation exist, shaped like the previous release wrote them.
+# @PRE The previous release wrote the authority envelope before the build-identity members existed,
+#      so the fixture persists that exact shape instead of the current builder output.
+# @REJECTED Updating the appended envelope afterwards: the PG append-only trigger refuses it by design.
+# @RATIONALE Draft authorities and approval receipts changed after 0046 (0047-0049), so the
+#            previous-release rows are written with the 0046 column set, not the current ORM.
 def _seed_approved_reference(engine) -> dict:
     from sqlalchemy.orm import Session
 
@@ -89,9 +99,6 @@ def _seed_approved_reference(engine) -> dict:
         seed = seed_catalog(db, repository=CATALOG_REPOSITORY,
             dashboard_key=f"dashboard-{wrapper.entry.dashboard_id}", actor_id="dr-audit",
             reason="previous release catalog seed", idempotency_key=f"dr-audit:seed:{uuid4()}")
-        # @PRE The previous release wrote the authority envelope before the build-identity members existed,
-        #      so the fixture persists that exact shape instead of the current builder output.
-        # @REJECTED Updating the appended envelope afterwards: the PG append-only trigger refuses it by design.
         built = build_published_entry_authority(wrapper=wrapper,
             catalog_id=seed.revision.catalog_id, catalog_revision_id=invocation_id,
             release_id=RELEASE, repository_id=CATALOG_REPOSITORY)
@@ -130,8 +137,6 @@ def _seed_approved_reference(engine) -> dict:
                 catalog_revision_id=selector.catalog_revision_id).scalar()),
             authority=dict(db.query(BaselineCatalogEntryAuthority.authority).filter_by(
                 catalog_revision_id=selector.catalog_revision_id).scalar()))
-    # @RATIONALE Draft authorities and approval receipts changed after 0046 (0047-0049), so the
-    #            previous-release rows are written with the 0046 column set, not the current ORM.
     now = datetime.now(UTC)
     with engine.begin() as connection:
         connection.execute(text(
@@ -183,8 +188,11 @@ def _whole_rows(connection, seeded: dict) -> dict:
     }
     return {name: tuple(connection.execute(text(sql), params).scalars().all())
         for name, (sql, params) in queries.items()}
+# #endregion Test.Alembic.InPlaceUpgrade.RowSnapshot
 
 
+# #region Test.Alembic.InPlaceUpgrade.DraftAuthorityRow [C:1] [TYPE Function]
+# @POST The legacy draft-authority row is compared column-by-column as exact text.
 def _draft_authority_row(connection, draft_id: str) -> dict:
     return dict(connection.execute(text(
         "SELECT draft_id, owner_id, owner_username, environment_id, dashboard_id, "
@@ -193,12 +201,21 @@ def _draft_authority_row(connection, draft_id: str) -> dict:
         "state, cas_version, invalidation_code, created_at, updated_at "
         "FROM dashboard_reference_draft_authorities WHERE draft_id = :draft_id"),
         {"draft_id": draft_id}).mappings().one())
-# #endregion Test.Alembic.InPlaceUpgrade.RowSnapshot
+# #endregion Test.Alembic.InPlaceUpgrade.DraftAuthorityRow
+
+
+# #region Test.Alembic.InPlaceUpgrade.SeededReferencePresent [C:2] [TYPE Function]
+# @POST The seeded catalog, authority and receipt row groups are all non-empty before the upgrade.
+def _assert_reference_rows_seeded(before: dict) -> None:
+    assert before["catalog_revisions"] and before["catalog_memberships"]
+    assert before["entry_authorities"] and before["publication_operations"]
+    assert before["approval_receipts"]
+# #endregion Test.Alembic.InPlaceUpgrade.SeededReferencePresent
 
 
 # #region Test.Alembic.InPlaceUpgrade.PreviousRelease [C:4] [TYPE Function]
 # @BRIEF A database at the previous release head upgrades to head and keeps every approved reference row.
-# @INVARIANT The alembic_version row moves 0046_mapper_llm_plans -> 0052_draft_inspection exactly once.
+# @INVARIANT The alembic_version row moves 0046_mapper_llm_plans -> 0055_scenario_run_notifications exactly once.
 @_REQUIRES_PG
 def test_previous_release_upgrades_in_place_preserving_approved_reference(
         monkeypatch: pytest.MonkeyPatch) -> None:
@@ -211,9 +228,7 @@ def test_previous_release_upgrades_in_place_preserving_approved_reference(
         with schema_engine.connect() as connection:
             before = _whole_rows(connection, seeded)
             before_authority = _draft_authority_row(connection, seeded["draft_id"])
-        assert before["catalog_revisions"] and before["catalog_memberships"]
-        assert before["entry_authorities"] and before["publication_operations"]
-        assert before["approval_receipts"]
+        _assert_reference_rows_seeded(before)
 
         _run_alembic_upgrade()
 
@@ -237,7 +252,7 @@ def test_previous_release_upgrades_in_place_preserving_approved_reference(
 # #endregion Test.Alembic.InPlaceUpgrade.PreviousRelease
 
 
-# #region Test.Alembic.InPlaceUpgrade.CurrentRead [C:4] [TYPE Function]
+# #region Test.Alembic.InPlaceUpgrade.CurrentRead [C:3] [TYPE Function]
 # @POST Exact revision read keeps the seeded digest, receipt binding and the honest legacy authority shape.
 def _assert_current_code_reads_exact_reference(engine, seeded: dict) -> None:
     from sqlalchemy.orm import Session
@@ -250,28 +265,56 @@ def _assert_current_code_reads_exact_reference(engine, seeded: dict) -> None:
 
     with Session(engine) as db:
         head = read_catalog_head(db, seeded["catalog_id"])
-        assert head.cas_version == 2 and head.status == "active"
+        _assert_catalog_head_current(head)
         snapshot = load_receipted_published_catalog(db, ExactCatalogRevision(kind="revision",
             catalog_id=seeded["catalog_id"],
             catalog_revision_id=seeded["catalog_revision_id"]))
-        assert snapshot.integrity_sha256 == seeded["integrity_sha256"]
-        assert snapshot.revision.catalog_digest == seeded["catalog_digest"]
-        assert snapshot.revision.catalog_revision_id == seeded["catalog_revision_id"]
-        assert snapshot.publication.operation_id == seeded["operation_id"]
-        assert snapshot.publication.commit_sha == seeded["commit_sha"]
-        assert snapshot.publication.render_sha256 == seeded["render_sha256"]
-        authority = snapshot.entry_authorities[0].authority
-        # @INVARIANT A pre-build-identity envelope stays readable and is not rewritten into a different shape.
-        assert authority.model_dump(mode="json", exclude_unset=True) == seeded["authority"]
+        _assert_published_snapshot_exact(snapshot, seeded)
+        _assert_legacy_authority_shape(snapshot, seeded)
         receipt = db.get(ReferenceDraftApprovalReceipt, seeded["receipt_id"])
-        assert receipt is not None
-        assert (receipt.draft_id, receipt.candidate_id, receipt.decision) == (
-            seeded["draft_id"], seeded["candidate_id"], "approved")
-        assert receipt.reason == seeded["receipt_reason"]
-        assert receipt.actor_id == "dr-audit-owner" and receipt.cas_version == 2
+        _assert_receipt_unchanged(receipt, seeded)
 # #endregion Test.Alembic.InPlaceUpgrade.CurrentRead
 
 
+# #region Test.Alembic.InPlaceUpgrade.CatalogHeadCurrent [C:1] [TYPE Function]
+# @POST The readable head is the second active revision from the seeded fixture.
+def _assert_catalog_head_current(head) -> None:
+    assert head.cas_version == 2 and head.status == "active"
+# #endregion Test.Alembic.InPlaceUpgrade.CatalogHeadCurrent
+
+
+# #region Test.Alembic.InPlaceUpgrade.PublishedSnapshotExact [C:2] [TYPE Function]
+# @POST The receipted snapshot preserves integrity, digest, revision, publication identity and render hash.
+def _assert_published_snapshot_exact(snapshot, seeded: dict) -> None:
+    assert snapshot.integrity_sha256 == seeded["integrity_sha256"]
+    assert snapshot.revision.catalog_digest == seeded["catalog_digest"]
+    assert snapshot.revision.catalog_revision_id == seeded["catalog_revision_id"]
+    assert snapshot.publication.operation_id == seeded["operation_id"]
+    assert snapshot.publication.commit_sha == seeded["commit_sha"]
+    assert snapshot.publication.render_sha256 == seeded["render_sha256"]
+# #endregion Test.Alembic.InPlaceUpgrade.PublishedSnapshotExact
+
+
+# #region Test.Alembic.InPlaceUpgrade.LegacyAuthorityShape [C:2] [TYPE Function]
+# @BRIEF The pre-build-identity envelope is compared exactly against the seeded authority JSON.
+# @INVARIANT A pre-build-identity envelope stays readable and is not rewritten into a different shape.
+def _assert_legacy_authority_shape(snapshot, seeded: dict) -> None:
+    authority = snapshot.entry_authorities[0].authority
+    assert authority.model_dump(mode="json", exclude_unset=True) == seeded["authority"]
+# #endregion Test.Alembic.InPlaceUpgrade.LegacyAuthorityShape
+
+
+# #region Test.Alembic.InPlaceUpgrade.ReceiptUnchanged [C:2] [TYPE Function]
+# @POST The approval receipt keeps its decision, reason and actor/cas version.
+def _assert_receipt_unchanged(receipt, seeded: dict) -> None:
+    assert receipt is not None
+    assert (receipt.draft_id, receipt.candidate_id, receipt.decision) == (
+        seeded["draft_id"], seeded["candidate_id"], "approved")
+    assert receipt.reason == seeded["receipt_reason"]
+    assert receipt.actor_id == "dr-audit-owner" and receipt.cas_version == 2
+# #endregion Test.Alembic.InPlaceUpgrade.ReceiptUnchanged
+
+
 # #region Test.Alembic.InPlaceUpgrade.LegacyRefusal [C:3] [TYPE Function]
 # @BRIEF An rc2-shaped schema (non-empty, no alembic_version) is refused without losing or stamping data.
 # @RATIONALE The 2026-10-20 finding names this exact refusal; PG proves transactional DDL leaves no partial state.
diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py
index 796b29eec..015efc2b8 100644
--- a/backend/tests/test_alembic_migrations.py
+++ b/backend/tests/test_alembic_migrations.py
@@ -1,71 +1,50 @@
-# #region Test.Alembic.Migrations [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,schema,postgres]
-# @BRIEF Verifies the single baseline Alembic upgrade.
-# @RELATION BINDS_TO -> [Alembic.AddDeploymentRecords]
-# @INVARIANT  Running `alembic upgrade head` twice is idempotent (no-op on second run) on PostgreSQL.
-# @PRE  Alembic is installed, migration files exist in alembic/versions/.
-# @POST  Schema matches expected state after upgrade.
+# #region Test.Alembic.Migrations [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,schema,postgres,current-head]
+# @BRIEF Verify the fresh Alembic chain, the pinned current head, and the fixed 0049→0050 source-binding boundary.
+# @RELATION BINDS_TO -> [Migrations.0051.DeploymentAppliedAt]
+# @RELATION BINDS_TO -> [Migrations.0052.ReferenceDraftInspection]
+# @INVARIANT Running `alembic upgrade head` twice is idempotent (no-op on second run) on PostgreSQL.
+# @PRE Alembic is installed, migration files exist in alembic/versions/.
+# @POST Schema matches expected state after upgrade; named current-chain tests assert CURRENT_ALEMBIC_HEAD.
 # @TEST_INVARIANT: alembic_fresh_upgrade -> VERIFIED_BY: [Test.AlembicMigrations.TestFreshUpgradeCreatesTables]
+# @TEST_INVARIANT: alembic_current_head_pin -> VERIFIED_BY: [Test.AlembicMigrations.CurrentHeadContract]
 # @REJECTED Full migration chain idempotency on SQLite — initial migration uses
 #           op.create_index() which conflicts with SQLite auto-created indexes
 #           for UNIQUE constraints. Migrations are designed for PostgreSQL.
+# @REJECTED Moving `_run_alembic_upgrade` into a helper module was rejected: test_alembic_url_interpolation
+#           execs that top-level function from this module's AST to prove exact URL interpolation.
 
-from contextlib import contextmanager
-import os
 from pathlib import Path
-import re
-import uuid
-import pytest
+import os
 import sys
 
-from sqlalchemy.engine import make_url
+import pytest
+
+from tests.alembic_migration_schema_assertions import (
+    _assert_0051_schema,
+    _assert_0052_schema,
+    _assert_no_inspection_snapshot_column,
+    _assert_source_binding_behavior,
+    _assert_source_binding_schema,
+)
+from tests.alembic_migration_support import (  # noqa: F401  re-exported for sibling test modules
+    _DB_BACKEND,
+    _REQUIRES_PG,
+    _isolated_pg_schema,
+)
+
+# One hardcoded pin for the current chain head; bump it when a new migration becomes
+# head (root integrates 0056_reference_finish_completion after this preparation).
+CURRENT_ALEMBIC_HEAD = "0055_scenario_run_notifications"
 
 # Ensure backend/src is importable for model metadata
 sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
 
-# Alembic migrations use PostgreSQL-specific features (CREATE INDEX may conflict
-# with SQLite auto-indexes for UNIQUE constraints). Skip full-chain tests on SQLite.
-_DB_BACKEND = make_url(os.environ.get("DATABASE_URL", "sqlite://")).get_backend_name()
-_REQUIRES_PG = pytest.mark.skipif(
-    _DB_BACKEND != "postgresql",
-    reason="Alembic migrations are PostgreSQL-specific; use a PostgreSQL DATABASE_URL to test",
-)
-
 
 # #region Test.AlembicMigrations.TestFreshUpgradeCreatesTables [C:3] [TYPE Function]
 # @BRIEF Run alembic upgrade head on a fresh PostgreSQL DB and verify key tables exist.
-# @POST All expected tables exist after upgrade.
+# @POST All expected tables exist after upgrade and alembic_version names CURRENT_ALEMBIC_HEAD.
 # @TEST_INVARIANT: alembic_fresh_upgrade -> VERIFIED_BY: [Test.AlembicMigrations.TestFreshUpgradeCreatesTables]
 @_REQUIRES_PG
-# #region Test.AlembicMigrations.AppliedAtSchema
-def _assert_0051_schema(engine) -> None:
-    from sqlalchemy import inspect
-    import sqlalchemy as sa
-
-    columns = {column["name"]: column for column in inspect(engine).get_columns("deployment_records")}
-    assert columns["applied_at"]["nullable"] is True
-    assert columns["target_digest"]["nullable"] is True
-    assert isinstance(columns["applied_at"]["type"], sa.DateTime)
-    assert columns["target_digest"]["type"].length == 64
-# #endregion Test.AlembicMigrations.AppliedAtSchema
-
-
-# #region Test.AlembicMigrations.InspectionSchema [C:2] [TYPE Function]
-# @POST The 0052-owned inspection column exists exactly once as nullable JSON without a server default.
-def _assert_0052_schema(engine) -> None:
-    from sqlalchemy import inspect
-    from sqlalchemy.dialects.postgresql import JSONB
-    import sqlalchemy as sa
-
-    columns = [column for column in inspect(engine).get_columns("dashboard_reference_draft_authorities")
-               if column["name"] == "inspection_snapshot"]
-    assert len(columns) == 1
-    assert columns[0]["nullable"] is True
-    assert isinstance(columns[0]["type"], sa.JSON)
-    assert not isinstance(columns[0]["type"], JSONB)
-    assert columns[0].get("default") is None
-# #endregion Test.AlembicMigrations.InspectionSchema
-
-
 def test_fresh_upgrade_creates_tables() -> None:
     """Fresh Alembic upgrade creates all expected tables (PostgreSQL only)."""
     from sqlalchemy import create_engine, inspect, text as sa_text
@@ -82,7 +61,7 @@ def test_fresh_upgrade_creates_tables() -> None:
         result = conn.execute(sa_text("SELECT version_num FROM alembic_version"))
         version = result.scalar_one()
         assert version is not None and len(version) > 0
-        assert version == "0052_draft_inspection"
+        assert version == CURRENT_ALEMBIC_HEAD
     _assert_0051_schema(engine)
     _assert_0052_schema(engine)
     roles_cols = {c["name"] for c in inspector.get_columns("roles")}
@@ -94,219 +73,6 @@ def test_fresh_upgrade_creates_tables() -> None:
 # #endregion Test.AlembicMigrations.TestFreshUpgradeCreatesTables
 
 
-
-# #region Test.AlembicMigrations.SourceBindingIsolation [C:4] [TYPE Function]
-# @BRIEF Give each migration test its own PostgreSQL schema in the run-owned worker DB.
-# @POST Alembic sees exactly one search_path (no public fallback); cleanup drops only this test's schema.
-# @INVARIANT No downgrade/drop of shared worker tables; no credential files or URL logging.
-@contextmanager
-def _isolated_pg_schema(monkeypatch: pytest.MonkeyPatch):
-    from sqlalchemy import create_engine, text as sa_text
-
-    schema = f"alembic_sb_{uuid.uuid4().hex}"
-    previous_schema = os.environ.get("DB_SCHEMA")
-    engine = None
-    schema_engine = None
-    created = False
-    try:
-        engine = create_engine(os.environ["DATABASE_URL"])
-        with engine.begin() as connection:
-            connection.execute(sa_text(f'CREATE SCHEMA "{schema}"'))
-        created = True
-        schema_engine = create_engine(
-            os.environ["DATABASE_URL"],
-            connect_args={"options": f"-csearch_path={schema}"},
-        )
-        monkeypatch.setenv("DB_SCHEMA", schema)
-        yield schema, schema_engine
-    finally:
-        try:
-            if schema_engine is not None:
-                schema_engine.dispose()
-        finally:
-            try:
-                if created and engine is not None:
-                    with engine.begin() as connection:
-                        connection.execute(sa_text(f'DROP SCHEMA "{schema}" CASCADE'))
-            finally:
-                if engine is not None:
-                    engine.dispose()
-                if previous_schema is None:
-                    monkeypatch.delenv("DB_SCHEMA", raising=False)
-                else:
-                    monkeypatch.setenv("DB_SCHEMA", previous_schema)
-# #endregion Test.AlembicMigrations.SourceBindingIsolation
-
-
-# #region Test.AlembicMigrations.SourceBindingSchemaAssert [C:4] [TYPE Function]
-# @BRIEF Assert the source-binding DDL independently of production ORM metadata.
-# @POST Verifies exact columns, FK columns/ondelete, check expressions, index columns/partial predicates.
-def _assert_source_binding_schema(schema_engine, schema: str) -> None:
-    from sqlalchemy import inspect, text as sa_text
-
-    table = "git_repository_source_bindings"
-    inspector = inspect(schema_engine)
-    assert table in set(inspector.get_table_names())
-    with schema_engine.connect() as connection:
-        column_rows = connection.execute(sa_text(
-            "SELECT column_name, is_nullable FROM information_schema.columns "
-            "WHERE table_schema = :schema AND table_name = :table"
-        ), {"schema": schema, "table": table}).all()
-        expected_columns = {
-            "id": "NO", "repository_id": "NO", "dashboard_uuid": "NO",
-            "source_environment_id": "NO", "source_dashboard_id": "NO",
-            "release_id": "NO", "release_commit_hash": "NO", "status": "NO",
-            "actor_id": "NO", "actor_username": "NO", "created_at": "NO",
-            "retired_at": "YES", "retired_by": "YES", "cas_version": "NO",
-            "request_sha256": "NO", "receipt_sha256": "NO",
-        }
-        assert {row.column_name: row.is_nullable for row in column_rows} == expected_columns
-
-        constraint_rows = connection.execute(sa_text(
-            "SELECT conname, contype, pg_get_constraintdef(oid) AS definition "
-            "FROM pg_constraint WHERE conrelid = CAST(:qualified AS regclass)"
-        ), {"qualified": f"{schema}.{table}"}).all()
-        definitions = {row.contype: [] for row in constraint_rows}
-        for row in constraint_rows:
-            definitions[row.contype].append(" ".join(row.definition.replace('"', "").lower().split()))
-        assert set(definitions.get("f", [])) == {
-            "foreign key (repository_id) references git_repositories(id) on delete restrict",
-            "foreign key (release_id) references dashboard_releases(id) on delete restrict",
-        }
-        assert len(definitions.get("c", [])) == 6
-
-        index_rows = connection.execute(sa_text(
-            "SELECT i.relname AS indexname, x.indisunique AS is_unique, "
-            "pg_get_indexdef(x.indexrelid) AS indexdef, "
-            "pg_get_expr(x.indpred, x.indrelid) AS predicate "
-            "FROM pg_index x JOIN pg_class i ON i.oid = x.indexrelid "
-            "WHERE x.indrelid = CAST(:qualified AS regclass)"
-        ), {"qualified": f"{schema}.{table}"}).all()
-        expected_indexes = {
-            "git_repository_source_bindings_pkey": (True, None, "(id)"),
-            "ix_git_source_binding_repository": (False, None, "(repository_id)"),
-            "uq_git_source_binding_active_repository": (True, "status='active'", "(repository_id)"),
-            "uq_git_source_binding_active_uuid": (True, "status='active'", "(dashboard_uuid)"),
-            "uq_git_source_binding_active_source": (True, "status='active'",
-                                                      "(source_environment_id, source_dashboard_id)"),
-        }
-        actual_indexes = {row.indexname: row for row in index_rows}
-        assert set(actual_indexes) == set(expected_indexes)
-        for name, (unique, predicate, columns) in expected_indexes.items():
-            row = actual_indexes[name]
-            assert bool(row.is_unique) is unique, name
-            if row.predicate is None:
-                normalized_predicate = None
-            else:
-                normalized_predicate = "".join(
-                    row.predicate.replace('"', "").lower().replace("::text", "").split())
-                if normalized_predicate.startswith("(") and normalized_predicate.endswith(")"):
-                    normalized_predicate = normalized_predicate[1:-1]
-                normalized_predicate = re.sub(
-                    r"\(([a-z_][a-z0-9_]*)\)", r"\1", normalized_predicate)
-            assert normalized_predicate == predicate, name
-            assert columns in row.indexdef.replace('"', "").lower(), name
-# #endregion Test.AlembicMigrations.SourceBindingSchemaAssert
-
-
-# #region Test.AlembicMigrations.SourceBindingBehavior [C:4] [TYPE Function]
-# @BRIEF Independent insert-refusal/retired-allowed fixtures for checks and active partial indexes.
-# @POST All six checks reject invalid rows; retired duplicates are allowed; active duplicates are rejected by DB.
-# @INVARIANT Expected values are hardcoded here, not imported from production ORM metadata.
-def _assert_source_binding_behavior(schema_engine) -> None:
-    from datetime import UTC, datetime
-    from sqlalchemy.exc import IntegrityError
-    from sqlalchemy import text as sa_text
-
-    table = "git_repository_source_bindings"
-    env_id = "00000000-0000-4000-8000-000000000001"
-    repo_a = "00000000-0000-4000-8000-000000000011"
-    repo_b = "00000000-0000-4000-8000-000000000012"
-    repo_c = "00000000-0000-4000-8000-000000000013"
-    release_id = "00000000-0000-4000-8000-000000000021"
-    uuid_a = "a2cef831-4ffd-46ec-aaa1-60f71782a91c"
-    uuid_b = "b2cef831-4ffd-46ec-aaa1-60f71782a91c"
-    uuid_c = "c2cef831-4ffd-46ec-aaa1-60f71782a91c"
-    commit = "a" * 40
-    request_hash = "0" * 64
-    receipt_hash = "1" * 64
-    now = datetime.now(UTC)
-
-    with schema_engine.begin() as connection:
-        connection.execute(sa_text(
-            "INSERT INTO deployment_environments (id, name, superset_url, superset_token, is_active) "
-            "VALUES (:id, :name, :url, :token, true)"
-        ), {"id": env_id, "name": "fixture", "url": "http://fixture", "token": ""})
-        for dashboard_id, repo_id in ((810001, repo_a), (810002, repo_b), (810003, repo_c)):
-            connection.execute(sa_text(
-                "INSERT INTO git_repositories "
-                "(id, dashboard_id, local_path, current_branch, sync_status) "
-                "VALUES (:id, :dashboard_id, :path, 'dev', 'CLEAN')"
-            ), {"id": repo_id, "dashboard_id": dashboard_id, "path": f"/tmp/{repo_id}"})
-        deployment_id = connection.execute(sa_text(
-            "INSERT INTO deployment_records "
-            "(repository_id, environment_id, commit_hash, content_hash, deployed_at, status, validation_status) "
-            "VALUES (:repo, :env, :commit, :content, :deployed, 'success', 'validated') RETURNING id"
-        ), {"repo": repo_a, "env": env_id, "commit": commit, "content": "c" * 64,
-            "deployed": now}).scalar_one()
-        connection.execute(sa_text(
-            "INSERT INTO dashboard_releases "
-            "(id, repository_id, deployment_id, name, version, notes, commit_hash, content_hash, status, created_at, created_by) "
-            "VALUES (:id, :repo, :deployment, 'fixture', '1.0.0', 'fixture', :commit, :content, 'ready_to_publish', :created, 'fixture')"
-        ), {"id": release_id, "repo": repo_a, "deployment": deployment_id, "commit": commit,
-            "content": "c" * 64, "created": now})
-
-    def insert_binding(*, repository_id: str, dashboard_uuid: str, source_environment_id: str,
-                       source_dashboard_id: int = 12, status: str = "active", commit_hash: str = commit,
-                       cas_version: int = 1) -> None:
-        with schema_engine.begin() as connection:
-            connection.execute(sa_text(
-                f"INSERT INTO {table} "
-                "(id, repository_id, dashboard_uuid, source_environment_id, source_dashboard_id, "
-                "release_id, release_commit_hash, status, actor_id, actor_username, created_at, "
-                "cas_version, request_sha256, receipt_sha256) "
-                "VALUES (:id, :repo, :uuid, :source_env, :source_dashboard, :release, :commit, :status, "
-                "'actor', 'operator', :created, :cas, :request, :receipt)"
-            ), {"id": str(uuid.uuid4()), "repo": repository_id, "uuid": dashboard_uuid,
-                "source_env": source_environment_id, "source_dashboard": source_dashboard_id,
-                "release": release_id, "commit": commit_hash, "status": status,
-                "created": now, "cas": cas_version, "request": request_hash, "receipt": receipt_hash})
-
-    insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a")
-    insert_binding(repository_id=repo_a, dashboard_uuid=uuid_a, source_environment_id="source-a",
-                   status="retired")
-    with schema_engine.connect() as connection:
-        assert connection.execute(sa_text(f"SELECT count(*) FROM {table}")).scalar_one() == 2
-
-    for invalid in (
-        {"repository_id": repo_a, "uuid": uuid_b, "source_env": "source-b"},
-        {"repository_id": repo_b, "uuid": uuid_a, "source_env": "source-c"},
-        {"repository_id": repo_c, "uuid": uuid_b, "source_env": "source-a"},
-    ):
-        with pytest.raises(IntegrityError):
-            insert_binding(repository_id=invalid["repository_id"], dashboard_uuid=invalid["uuid"],
-                           source_environment_id=invalid["source_env"])
-
-    invalid_rows = (
-        {"status": "paused"},
-        {"dashboard_uuid": "short"},
-        {"commit_hash": "short"},
-        {"source_environment_id": ""},
-        {"source_dashboard_id": 0},
-        {"cas_version": 0},
-    )
-    for invalid in invalid_rows:
-        values = {
-            "repository_id": repo_a, "dashboard_uuid": uuid_c,
-            "source_environment_id": "source-check", "source_dashboard_id": 12,
-            "status": "retired", "commit_hash": commit, "cas_version": 1,
-        }
-        values.update(invalid)
-        with pytest.raises(IntegrityError):
-            insert_binding(**values)
-# #endregion Test.AlembicMigrations.SourceBindingBehavior
-
-
 # #region Test.AlembicMigrations.SourceBindingTable [C:4] [TYPE Function]
 # @BRIEF Fresh PostgreSQL chain must let revision 0050 own the source-binding DDL, not 0001 create_all.
 # @TEST_INVARIANT: source_binding_owned_by_0050 -> VERIFIED_BY: [Test.AlembicMigrations.SourceBindingTable]
@@ -328,12 +94,12 @@ def test_fresh_upgrade_owns_source_binding_table(monkeypatch: pytest.MonkeyPatch
         _assert_0052_schema(schema_engine)
         with schema_engine.connect() as connection:
             version = connection.execute(sa_text("SELECT version_num FROM alembic_version")).scalar_one()
-            assert version == "0052_draft_inspection"
+            assert version == CURRENT_ALEMBIC_HEAD
 # #endregion Test.AlembicMigrations.SourceBindingTable
 
 
 # #region Test.AlembicMigrations.SourceBindingUpgradePath [C:4] [TYPE Function]
-# @BRIEF Prove the exact 0049→0050 source-binding boundary, then retain full upgrade-to-0052 coverage.
+# @BRIEF Prove the exact 0049→0050 source-binding boundary, then replay the full chain to CURRENT_ALEMBIC_HEAD.
 @_REQUIRES_PG
 def test_source_binding_upgrade_from_0049_to_0050(monkeypatch: pytest.MonkeyPatch) -> None:
     from sqlalchemy import inspect, text as sa_text
@@ -352,7 +118,7 @@ def test_source_binding_upgrade_from_0049_to_0050(monkeypatch: pytest.MonkeyPatc
         _assert_0052_schema(schema_engine)
         with schema_engine.connect() as connection:
             version = connection.execute(sa_text("SELECT version_num FROM alembic_version")).scalar_one()
-            assert version == "0052_draft_inspection"
+            assert version == CURRENT_ALEMBIC_HEAD
 # #endregion Test.AlembicMigrations.SourceBindingUpgradePath
 
 
@@ -367,17 +133,19 @@ def test_deployment_applied_at_upgrade_from_0050_to_0051(monkeypatch: pytest.Mon
         before = {c["name"] for c in inspect(schema_engine).get_columns("deployment_records")}
         assert "applied_at" not in before and "target_digest" not in before
         _run_alembic_upgrade("0051_deployment_applied_at")
-        columns = {c["name"]: c for c in inspect(schema_engine).get_columns("deployment_records")}
-        assert columns["applied_at"]["nullable"] is True
-        assert columns["target_digest"]["nullable"] is True
-        assert columns["target_digest"]["type"].length == 64
+        _assert_0051_schema(schema_engine)
         with schema_engine.connect() as connection:
             version = connection.execute(sa_text("SELECT version_num FROM alembic_version")).scalar_one()
             assert version == "0051_deployment_applied_at"
-        assert "inspection_snapshot" not in {column["name"] for column in inspect(schema_engine).get_columns("dashboard_reference_draft_authorities")}
+        _assert_no_inspection_snapshot_column(schema_engine)
 # #endregion Test.AlembicMigrations.DeploymentAppliedAt
 
 
+# #region Test.AlembicMigrations.BaselineIncompleteSchema [C:2] [TYPE Function]
+# @BRIEF A legacy table cannot be silently stamped as the reset baseline.
+# @POST The baseline upgrade raises its typed refusal before any stamp or DDL runs.
+# @RELATION VERIFIES -> [Migrations.Baseline.Upgrade]
+# @TEST_INVARIANT: incomplete_existing_schema_refused -> VERIFIED_BY: [Test.AlembicMigrations.BaselineIncompleteSchema]
 def test_baseline_refuses_incomplete_existing_schema(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     """A legacy table cannot be silently stamped as the reset baseline."""
     from sqlalchemy import create_engine, text
@@ -391,6 +159,7 @@ def test_baseline_refuses_incomplete_existing_schema(tmp_path: Path, monkeypatch
     monkeypatch.setenv("DATABASE_URL", database_url)
     with pytest.raises(RuntimeError, match="Refusing to stamp an incomplete existing schema"):
         _run_alembic_upgrade()
+# #endregion Test.AlembicMigrations.BaselineIncompleteSchema
 
 
 # #region Test.AlembicMigrations.RevisionIdLength [C:2] [TYPE Function]
@@ -413,6 +182,24 @@ def test_revision_ids_fit_varchar32() -> None:
 # #endregion Test.AlembicMigrations.RevisionIdLength
 
 
+# #region Test.AlembicMigrations.CurrentHeadContract [C:2] [TYPE Function]
+# @BRIEF Offline structural guard: the real migration scripts resolve to exactly CURRENT_ALEMBIC_HEAD.
+# @POST ScriptDirectory reports one head and it equals the hardcoded chain pin.
+# @TEST_INVARIANT: alembic_current_head_pin -> VERIFIED_BY: [Test.AlembicMigrations.CurrentHeadContract]
+# @REJECTED Reading the head from a database was rejected: this guard must run without PostgreSQL.
+def test_current_head_revision_is_pinned_offline() -> None:
+    """Static guard: the declared chain pin matches the single real Alembic head."""
+    from alembic.config import Config as AlembicConfig
+    from alembic.script import ScriptDirectory
+
+    backend_dir = Path(__file__).parent.parent
+    alembic_cfg = AlembicConfig(str(backend_dir / "alembic.ini"))
+    alembic_cfg.set_main_option("script_location", str(backend_dir / "alembic"))
+    heads = tuple(ScriptDirectory.from_config(alembic_cfg).get_heads())
+    assert heads == (CURRENT_ALEMBIC_HEAD,)
+# #endregion Test.AlembicMigrations.CurrentHeadContract
+
+
 # #region Test.AlembicMigrations.RunAlembicUpgrade [C:1] [TYPE Function]
 def _run_alembic_upgrade(revision: str = "head") -> None:
     """Run a named Alembic upgrade programmatically against DATABASE_URL."""
diff --git a/backend/tests/test_reference_draft_inspection_migration.py b/backend/tests/test_reference_draft_inspection_migration.py
index f9713064b..329b14078 100644
--- a/backend/tests/test_reference_draft_inspection_migration.py
+++ b/backend/tests/test_reference_draft_inspection_migration.py
@@ -34,44 +34,107 @@ def install_shape_double(monkeypatch, baseline_owner, *, column=None, drop_mutat
     columns = [{"name": "draft_id", "type": sa.String(36), "nullable": False, "default": None}]
     if column is not None:
         columns.append(column)
+    # #region Test.Migrations.ReferenceDraftInspection.ShapeInspectColumns [C:1] [TYPE Function]
+    # @BRIEF Returns a deep copy of the simulated authority column list.
     def inspect_columns(table):
         assert table == _TABLE
         return deepcopy(columns)
+    # #endregion Test.Migrations.ReferenceDraftInspection.ShapeInspectColumns
+    # #region Test.Migrations.ReferenceDraftInspection.ShapeDropColumn [C:1] [TYPE Function]
+    # @BRIEF Simulates an owned column drop and optionally mutates the simulated column list.
     def drop_column(table, name):
         assert (table, name) == (_TABLE, _COLUMN)
         if drop_mutates:
             columns[:] = [item for item in columns if item["name"] != name]
+    # #endregion Test.Migrations.ReferenceDraftInspection.ShapeDropColumn
     inspector = SimpleNamespace(get_columns=inspect_columns)
-    monkeypatch.setattr(baseline_owner.sa, "inspect", lambda _bind: inspector)
+    monkeypatch.setattr(
+        baseline_owner.sa,
+        "inspect",
+        # #region Test.Migrations.ReferenceDraftInspection.ShapeInspectorCallback [C:1] [TYPE Function]
+        # @BRIEF Binds the injected inspector to any inspected bind.
+        lambda _bind: inspector
+        # #endregion Test.Migrations.ReferenceDraftInspection.ShapeInspectorCallback
+    )
     dropped = Mock(side_effect=drop_column)
     monkeypatch.setattr(baseline_owner.op, "drop_column", dropped)
     return columns, dropped
 # #endregion Test.Migrations.ReferenceDraftInspection.ShapeDouble
 
 
-# #region Test.Migrations.ReferenceDraftInspection.Ownership [C:3] [TYPE Function]
-def test_current_metadata_field_is_deferred_before_historical_0041_validation(baseline_owner):
+# #region Test.Migrations.ReferenceDraftInspection.MetadataDefersInspection [C:1] [TYPE Function]
+# @POST Current metadata still declares the 0052-owned inspection column.
+def _assert_metadata_defers_inspection_column(baseline_owner) -> None:
     assert _COLUMN in baseline_owner.Base.metadata.tables[_TABLE].c
+# #endregion Test.Migrations.ReferenceDraftInspection.MetadataDefersInspection
+
+
+# #region Test.Migrations.ReferenceDraftInspection.ExpectedColumns0041 [C:2] [TYPE Function]
+# @POST The historical 0041 expected-column set excludes the future 0052 inspection column.
+def _assert_0041_expected_columns_exclude_inspection() -> None:
     path = Path(__file__).parents[1] / "alembic/versions/0041_reference_draft_authority.py"
     tree = ast.parse(path.read_text())
     expected = next(node.value for node in tree.body if isinstance(node, ast.Assign)
         and any(isinstance(target, ast.Name) and target.id == "_EXPECTED_COLUMNS" for target in node.targets))
     assert _COLUMN not in [key.value for key in expected.keys]
+# #endregion Test.Migrations.ReferenceDraftInspection.ExpectedColumns0041
+
+
+# #region Test.Migrations.ReferenceDraftInspection.BaselineRestoreCalls [C:2] [TYPE Function]
+# @POST Collects every helper call made by the baseline upgrade body.
+def _baseline_restore_calls(baseline_owner) -> list:
     source = ast.parse(Path(baseline_owner.__file__).read_text())
     upgrade = next(node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade")
-    calls = [node.func.id for statement in upgrade.body for node in ast.walk(statement)
+    return [node.func.id for statement in upgrade.body for node in ast.walk(statement)
         if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)]
+# #endregion Test.Migrations.ReferenceDraftInspection.BaselineRestoreCalls
+
+
+# #region Test.Migrations.ReferenceDraftInspection.BaselineRestoreOrder [C:1] [TYPE Function]
+# @POST The baseline upgrade calls the seven historical-shape restorers in the fixed order.
+def _assert_baseline_restore_order(baseline_owner) -> None:
+    calls = _baseline_restore_calls(baseline_owner)
     required = ["_create_frozen_catalog_tables", "_create_metadata_without_visual_capture_tables",
         "_restore_pre_0047_draft_shape", "_restore_pre_0048_finish_shape", "_restore_pre_0049_release_shape",
         "_restore_pre_0051_receipt_shape", "_restore_pre_0052_inspection_shape"]
     assert [call for call in calls if call in required] == required
-    migration = ast.parse((path.parent / "0052_reference_draft_inspection.py").read_text())
-    parent = next(node.value.value for node in migration.body if isinstance(node, ast.Assign)
+# #endregion Test.Migrations.ReferenceDraftInspection.BaselineRestoreOrder
+
+
+# #region Test.Migrations.ReferenceDraftInspection.Migration0052Tree [C:1] [TYPE Function]
+# @POST Parses the 0052 migration source exactly once for the ownership checks.
+def _0052_migration_tree():
+    path = Path(__file__).parents[1] / "alembic/versions/0052_reference_draft_inspection.py"
+    return ast.parse(path.read_text())
+# #endregion Test.Migrations.ReferenceDraftInspection.Migration0052Tree
+
+
+# #region Test.Migrations.ReferenceDraftInspection.DownRevision0052 [C:1] [TYPE Function]
+# @POST 0052 is chained directly onto the 0051 deployment boundary.
+def _assert_0052_down_revision(tree) -> None:
+    parent = next(node.value.value for node in tree.body if isinstance(node, ast.Assign)
         and any(isinstance(target, ast.Name) and target.id == "down_revision" for target in node.targets))
     assert parent == "0051_deployment_applied_at"
-    add = next(node for node in migration.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade")
+# #endregion Test.Migrations.ReferenceDraftInspection.DownRevision0052
+
+
+# #region Test.Migrations.ReferenceDraftInspection.InspectionColumn0052 [C:1] [TYPE Function]
+# @POST 0052 owns exactly one add_column of the inspection snapshot.
+def _assert_0052_single_add_column(tree) -> None:
+    add = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "upgrade")
     assert len(add.body) == 1
     assert isinstance(add.body[0], ast.Expr) and add.body[0].value.func.attr == "add_column"
+# #endregion Test.Migrations.ReferenceDraftInspection.InspectionColumn0052
+
+
+# #region Test.Migrations.ReferenceDraftInspection.Ownership [C:3] [TYPE Function]
+def test_current_metadata_field_is_deferred_before_historical_0041_validation(baseline_owner):
+    _assert_metadata_defers_inspection_column(baseline_owner)
+    _assert_0041_expected_columns_exclude_inspection()
+    _assert_baseline_restore_order(baseline_owner)
+    migration = _0052_migration_tree()
+    _assert_0052_down_revision(migration)
+    _assert_0052_single_add_column(migration)
 # #endregion Test.Migrations.ReferenceDraftInspection.Ownership
 
 
@@ -109,43 +172,63 @@ def test_baseline_rechecks_absence_after_drop(monkeypatch, baseline_owner):
 # #endregion Test.Migrations.ReferenceDraftInspection.DropVerification
 
 
+# #region Test.Migrations.ReferenceDraftInspection.LegacyRowsBefore0052 [C:2] [TYPE Function]
+# @POST Captures the pre-0052 column absence and the exact legacy row/authority snapshots.
+def _assert_legacy_rows_before_0052(engine, seeded: dict):
+    from tests.test_alembic_inplace_upgrade import _whole_rows
+
+    assert _COLUMN not in {column["name"] for column in sa.inspect(engine).get_columns(_TABLE)}
+    with engine.connect() as connection:
+        assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0051_deployment_applied_at"
+        before = _whole_rows(connection, seeded)
+        before_authority = connection.execute(sa.text(
+            "SELECT to_jsonb(t)::text FROM dashboard_reference_draft_authorities t WHERE draft_id=:draft_id"),
+            {"draft_id": seeded["draft_id"]}).scalar_one()
+    assert all(before.values())
+    return before, before_authority
+# #endregion Test.Migrations.ReferenceDraftInspection.LegacyRowsBefore0052
+
+
+# #region Test.Migrations.ReferenceDraftInspection.LegacyRows0052 [C:2] [TYPE Function]
+# @POST Exact 0052 replay adds SQL NULL only; every legacy byte group stays equal and the version stays 0052.
+def _assert_0052_replay_preserves(engine, seeded: dict, before: dict, before_authority: str) -> None:
+    from tests.test_alembic_inplace_upgrade import _whole_rows
+    from tests.test_alembic_migrations import _assert_0052_schema
+
+    _assert_0052_schema(engine)
+    with engine.connect() as connection:
+        assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0052_draft_inspection"
+        assert _whole_rows(connection, seeded) == before
+        assert connection.execute(sa.text(
+            "SELECT (to_jsonb(t)-'inspection_snapshot')::text FROM dashboard_reference_draft_authorities t "
+            "WHERE draft_id=:draft_id"), {"draft_id": seeded["draft_id"]}).scalar_one() == before_authority
+        added = connection.execute(sa.text(
+            "SELECT inspection_snapshot, inspection_snapshot IS NULL AS inspection_is_sql_null "
+            "FROM dashboard_reference_draft_authorities WHERE draft_id=:draft_id"),
+            {"draft_id": seeded["draft_id"]}).mappings().one()
+        assert dict(added) == {"inspection_snapshot": None, "inspection_is_sql_null": True}
+# #endregion Test.Migrations.ReferenceDraftInspection.LegacyRows0052
+
+
 # #region Test.Migrations.ReferenceDraftInspection.LegacyUpgrade [C:4] [TYPE Function]
 # @PRE Root supplies isolated migrated PostgreSQL; a valid 0046 fixture is advanced explicitly to 0051 before the measured step.
-# @POST Exact 0051→0052 adds SQL NULL only; every old authority/catalog/membership/publication/receipt byte group stays equal.
+# @POST Exact 0051→0052 adds SQL NULL only; replaying exact 0052 is a no-op and the version remains 0052.
 @_REQUIRES_PG
 def test_0051_to_0052_preserves_all_legacy_rows_and_is_idempotent(monkeypatch):
     from tests.test_alembic_inplace_upgrade import (
-        PREVIOUS_RELEASE_HEAD, _seed_approved_reference, _whole_rows, _assert_current_code_reads_exact_reference)
-    from tests.test_alembic_migrations import _isolated_pg_schema, _run_alembic_upgrade, _assert_0052_schema
+        PREVIOUS_RELEASE_HEAD, _assert_current_code_reads_exact_reference, _seed_approved_reference)
+    from tests.test_alembic_migrations import _isolated_pg_schema, _run_alembic_upgrade
 
     with _isolated_pg_schema(monkeypatch) as (_schema, engine):
         _run_alembic_upgrade(PREVIOUS_RELEASE_HEAD)
         seeded = _seed_approved_reference(engine)
         _run_alembic_upgrade("0051_deployment_applied_at")
-        assert _COLUMN not in {column["name"] for column in sa.inspect(engine).get_columns(_TABLE)}
-        with engine.connect() as connection:
-            assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0051_deployment_applied_at"
-            before = _whole_rows(connection, seeded)
-            before_authority = connection.execute(sa.text(
-                "SELECT to_jsonb(t)::text FROM dashboard_reference_draft_authorities t WHERE draft_id=:draft_id"),
-                {"draft_id": seeded["draft_id"]}).scalar_one()
-        assert all(before.values())
+        before, before_authority = _assert_legacy_rows_before_0052(engine, seeded)
         _run_alembic_upgrade("0052_draft_inspection")
         for pass_index in (1, 2):
-            _assert_0052_schema(engine)
-            with engine.connect() as connection:
-                assert connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one() == "0052_draft_inspection"
-                assert _whole_rows(connection, seeded) == before
-                assert connection.execute(sa.text(
-                    "SELECT (to_jsonb(t)-'inspection_snapshot')::text FROM dashboard_reference_draft_authorities t "
-                    "WHERE draft_id=:draft_id"), {"draft_id": seeded["draft_id"]}).scalar_one() == before_authority
-                added = connection.execute(sa.text(
-                    "SELECT inspection_snapshot, inspection_snapshot IS NULL AS inspection_is_sql_null "
-                    "FROM dashboard_reference_draft_authorities WHERE draft_id=:draft_id"),
-                    {"draft_id": seeded["draft_id"]}).mappings().one()
-                assert dict(added) == {"inspection_snapshot": None, "inspection_is_sql_null": True}
+            _assert_0052_replay_preserves(engine, seeded, before, before_authority)
             if pass_index == 1:
-                _run_alembic_upgrade()
+                _run_alembic_upgrade("0052_draft_inspection")
         _assert_current_code_reads_exact_reference(engine, seeded)
 # #endregion Test.Migrations.ReferenceDraftInspection.LegacyUpgrade
 # #endregion Test.Migrations.ReferenceDraftInspection