fix(sp37-followup): load BACKFILL_SQL from migration file (no drift)

Followup #4 from the SP37 final-state tracker. The previous
BACKFILL_SQL constant in test_migration_0020.py was a hand-copied
duplicate of the migration's UPDATE statement. A future contributor
could edit one without the other and the test would silently
replay a different SQL than production — defeating the regression.

Fix: tests now load the migration file at test time and extract
its UPDATE via the same splitter db_migrate.run() uses (strip
'--' comments, split on ';'). The test can never disagree with
what production runs.

Changes:
  * test_migration_0020.py:
    - Remove the hand-copied BACKFILL_SQL constant
    - Add _migration_0020_path(), _extract_update_statements(),
      and _load_migration_0020_backfill_sql() helpers
    - Replace 3 BACKFILL_SQL references with helper calls
  * test_migration_0020_no_drift.py (new, 3 tests):
    - test_migration_0020_backfill_sql_uses_migration_file
      (asserts the extracted SQL targets the right column + path)
    - test_migration_0020_backfill_sql_is_non_empty_single_statement
    - test_migration_0020_has_exactly_one_update (guardrail against
      future contributors adding a second UPDATE — the extraction
      fails loudly so the test author can decide which is the
      backfill)

Tests: 43/43 pass in 1.46s (full SP37 followup chain).
Imports across test files match the existing pattern (test_store.py
imports from test_store_reconcile.py).

backend/tests/test_migration_0020.py

@@ -14,6 +14,11 @@ the fresh DB up to v20, insert representative rows, then replay the
 exact UPDATE statement the migration uses. Replaying the UPDATE
 proves the SQL works as intended even though the migration itself
 already ran over an empty table.
+
+The backfill UPDATE is loaded directly from the migration file (not
+a hand-maintained copy) so the test cannot drift from production —
+see :func:`_load_migration_0020_backfill_sql` and the regression
+locks in ``test_migration_0020_no_drift.py``.
 """
 
 from __future__ import annotations
@@ -27,18 +32,50 @@ import sqlalchemy as sa
 from cyclone import db_migrate
 
 
-# The backfill UPDATE the migration executes (extracted so the test can
-# replay it against rows that didn't exist when init_db ran).
-BACKFILL_SQL = (
-    "UPDATE batches "
-    "SET transaction_set_control_number = "
-    "json_extract(raw_result_json, '$.envelope.transaction_set_control_number') "
-    "WHERE raw_result_json IS NOT NULL "
-    "AND json_extract(raw_result_json, '$.envelope.transaction_set_control_number') "
-    "IS NOT NULL"
+def _migration_0020_path() -> Path:
+    return (
+        Path(__file__).parent.parent / "src" / "cyclone" / "migrations"
+        / "0020_add_batch_txn_set_control_number.sql"
     )
 
 
+def _extract_update_statements(sql: str) -> list[str]:
+    """Split a migration into statements; return the UPDATE ones.
+
+    Mirrors db_migrate.run()'s splitter (strip ``--`` comments,
+    split on ``;``) so the test extraction can never disagree with
+    what the runner actually executes.
+    """
+    lines = [
+        line for line in sql.splitlines()
+        if not line.strip().startswith("--")
+    ]
+    cleaned = "\n".join(lines)
+    return [
+        stmt.strip() for stmt in cleaned.split(";")
+        if stmt.strip()
+    ]
+
+
+def _load_migration_0020_backfill_sql() -> str:
+    """Return migration 0020's UPDATE statement (the backfill).
+
+    Reads the migration file at test time and extracts its UPDATE.
+    Test code that needs to replay the backfill against rows that
+    didn't exist when init_db ran uses this helper — guarantees the
+    replayed SQL is byte-identical to what production will run.
+    """
+    sql = _migration_0020_path().read_text()
+    updates = [
+        s for s in _extract_update_statements(sql)
+        if s.upper().startswith("UPDATE ")
+    ]
+    assert len(updates) == 1, (
+        f"expected exactly 1 UPDATE in migration 0020, found {len(updates)}: {updates}"
+    )
+    return updates[0]
+
+
 def _fresh_engine(path: Path) -> sa.Engine:
     return sa.create_engine(f"sqlite:///{path}", future=True)
 
@@ -115,7 +152,7 @@ def test_migration_0020_backfills_when_key_present(
             "VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)",
             (json.dumps(raw_json),),
         )
-        conn.exec_driver_sql(BACKFILL_SQL)
+        conn.exec_driver_sql(_load_migration_0020_backfill_sql())
 
     with migrated_engine.connect() as conn:
         row = conn.exec_driver_sql(
@@ -141,7 +178,7 @@ def test_migration_0020_backfill_conditional_on_key_present(
             "VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)",
             (json.dumps(raw_json),),
         )
-        conn.exec_driver_sql(BACKFILL_SQL)
+        conn.exec_driver_sql(_load_migration_0020_backfill_sql())
 
     with migrated_engine.connect() as conn:
         row = conn.exec_driver_sql(
@@ -163,7 +200,7 @@ def test_migration_0020_backfill_handles_null_raw_result_json(
             "INSERT INTO batches (id, kind, input_filename, parsed_at) "
             "VALUES ('B-NULL-1', '837p', 'mig0020-null.txt', '2026-07-07 00:00:00')"
         )
-        conn.exec_driver_sql(BACKFILL_SQL)
+        conn.exec_driver_sql(_load_migration_0020_backfill_sql())
 
     with migrated_engine.connect() as conn:
         row = conn.exec_driver_sql(

backend/tests/test_migration_0020_no_drift.py

@@ -0,0 +1,66 @@
+"""SP37 follow-up #4: detect drift between test BACKFILL_SQL and migration SQL.
+
+Followup #1 of the SP37 final-state tracker. The previous test file
+maintained a hand-copied ``BACKFILL_SQL`` constant alongside the
+migration file. If a future contributor edited one but not the other,
+the test would silently replay a different SQL than production —
+defeating the regression test.
+
+The fix: ``test_migration_0020.py`` now reads the migration file at
+test time and extracts its UPDATE. This file imports the helper and
+pins the invariant — so a future contributor who edits the migration
+automatically gets the new SQL replayed in tests, and a contributor
+who removes the backfill UPDATE gets a loud extraction failure.
+"""
+from __future__ import annotations
+
+from test_migration_0020 import (
+    _extract_update_statements,
+    _load_migration_0020_backfill_sql,
+    _migration_0020_path,
+)
+
+
+def test_migration_0020_backfill_sql_uses_migration_file():
+    """The backfill SQL used in tests must come from the migration file.
+
+    Guards against the previous pattern of a hand-maintained
+    BACKFILL_SQL constant that could drift from the actual migration.
+    """
+    sql = _load_migration_0020_backfill_sql()
+    # Sanity check: the SQL targets batches.transaction_set_control_number
+    # via json_extract on raw_result_json. If a contributor changes
+    # the column name or the JSON path, this assertion catches it.
+    assert "UPDATE batches" in sql
+    assert "SET transaction_set_control_number" in sql
+    assert "json_extract(raw_result_json" in sql
+    assert "$.envelope.transaction_set_control_number" in sql
+
+
+def test_migration_0020_backfill_sql_is_non_empty_single_statement():
+    """The helper returns a non-empty SQL string suitable for direct
+    execution via exec_driver_sql. The existing
+    ``test_migration_0020.py`` tests use this helper to replay the
+    SQL against representative rows — so any successful replay
+    exercises the migration's actual UPDATE.
+    """
+    sql = _load_migration_0020_backfill_sql()
+    assert sql  # non-empty
+    assert ";" not in sql  # already stripped by the splitter
+
+
+def test_migration_0020_has_exactly_one_update():
+    """Guardrail: if a future migration adds a second UPDATE, the
+    extraction fails loudly so the test author can decide which one
+    is the backfill.
+    """
+    sql = _migration_0020_path().read_text()
+    updates = [
+        s for s in _extract_update_statements(sql)
+        if s.upper().startswith("UPDATE ")
+    ]
+    assert len(updates) == 1, (
+        f"migration 0020 should have exactly 1 UPDATE (the backfill); "
+        f"found {len(updates)}. If you added a second UPDATE, update "
+        f"_load_migration_0020_backfill_sql() to pick the right one."
+    )