test: guard migration identity by full filename stem

Author: eliperez-dev

docs/PIPELINE-MIGRATION.md

@@ -18,3 +18,25 @@ surfaces disabled, and treats disappearance as `not-observed`, never closure.
 Failures preserve the previous validated release and emit a restricted
 failure report. Acquisition terms approval and publication approval remain
 separate human gates.
+
+## Migration runner contract
+
+`pipeline/scripts/maintenance/apply-migrations.py` discovers every `*.sql` file
+in the migration directory and sorts the paths lexically. The migration identity
+stored in `uec.schema_migrations.version` is the complete filename stem, not the
+numeric prefix. Consequently, `020_release_manifests.sql` and
+`020_suppression_aware_v2_history.sql` have distinct identities and are both
+applied in that lexical order. Migration files must not be renamed, edited,
+deleted, squashed, or resequenced after they have been applied; a new change
+gets a new filename.
+
+The runner records the SHA-256 digest of each migration and refuses to execute
+an already-recorded identity when its file content has changed. Database setup
+and each migration run in separate transactions. A failed migration rolls back
+its SQL and ledger insert, while earlier successful migrations remain recorded
+so a subsequent run can retry and resume at the failed identity.
+
+The database connection is retried up to ten times after an operational
+connection error, waiting one second between attempts. Connection retries do
+not alter migration ordering or checksum behavior. These rules are tested by
+`pipeline/tests/test_apply_migrations.py` without connecting to a database.

pipeline/tests/test_apply_migrations.py

@@ -63,6 +63,20 @@ def execute(self, query, params=None):
 
 
 class MigrationRunnerTests(unittest.TestCase):
+    def test_existing_020_migrations_use_distinct_full_stem_identities(self):
+        migration_dir = Path(__file__).parents[1] / "migrations"
+
+        files = MODULE.migration_files(migration_dir)
+        names = [path.name for path in files]
+        stems = [path.stem for path in files]
+
+        first = "020_release_manifests.sql"
+        second = "020_suppression_aware_v2_history.sql"
+        self.assertIn(first, names)
+        self.assertIn(second, names)
+        self.assertNotEqual(Path(first).stem, Path(second).stem)
+        self.assertLess(stems.index(Path(first).stem), stems.index(Path(second).stem))
+
     def test_failed_migration_is_not_recorded_and_retry_resumes(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)