feat(sp37): add batches.transaction_set_control_number column

Migration 0020 adds the column additively (nullable, no default) and
backfills from raw_result_json.envelope.transaction_set_control_number
for any existing batch rows that already carry it. Required for the
SP37 join-key update so 999 acks can resolve by ST02 (the source 837's
transaction set control number) instead of just ISA13.

backend/src/cyclone/migrations/0020_add_batch_txn_set_control_number.sql

@@ -0,0 +1,27 @@
+-- version: 20
+-- SP37: Batch.transaction_set_control_number = parsed 837's ST02.
+--
+-- Today's 999 ack join (claim_acks.batch_envelope_index, Pass 1) matches
+-- on ``Batch.envelope.control_number == 999's set_control_number``. That
+-- never resolves in production because 999's set_control_number (AK201)
+-- echoes the source 837's ST02 (transaction set control number), not the
+-- ISA13 (interchange control number) that Envelope.control_number stores.
+-- Result: every AK2 set-response against a dzinesco-generated 837 turns
+-- into an orphan.
+--
+-- SP37 fixes this by adding a column populated from the parsed 837's
+-- ST02 on every ``add_record`` write, then updating Pass 1 to match on
+-- it (Task 2). This migration is the additive part: nullable, no
+-- default, backfills from ``raw_result_json.envelope.transaction_set_control_number``
+-- for any pre-existing batch rows that already carry the value.
+--
+-- No new index (column is a primary join key, not a range query; the
+-- existing batches table is small enough for a full scan during the
+-- 999 join — see SP37 §"Migration 0013").
+
+ALTER TABLE batches ADD COLUMN transaction_set_control_number TEXT;
+
+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;

backend/tests/test_db_migrate.py

@@ -126,14 +126,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
     SP28 bumped it to 18 with the claim_acks join table.
     SP32 bumped it to 19 with rendering_provider_npi +
     service_provider_npi on claims and remittances.
+    SP37 bumped it to 20 with batch transaction_set_control_number.
     """
     engine = _fresh_engine(tmp_path)
     db_migrate.run(engine)
     v_after_first = _user_version(engine)
-    assert v_after_first == 19, f"expected head=19, got {v_after_first}"
+    assert v_after_first == 20, f"expected head=20, got {v_after_first}"
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 19, "second run should not bump version"
+    assert _user_version(engine) == 20, "second run should not bump version"
 
 
 def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -159,7 +160,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
     engine = _fresh_engine(tmp_path)
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 19, f"expected head=19, got {_user_version(engine)}"
+    assert _user_version(engine) == 20, f"expected head=20, got {_user_version(engine)}"
 
     # Two claims in one batch with the same patient_control_number
     # must be insertable. If 0015's table recreation re-introduced a

backend/tests/test_migration_0020.py

@@ -0,0 +1,173 @@
+"""Tests for migration 0020_add_batch_txn_set_control_number.sql.
+
+SP37 adds a nullable ``batches.transaction_set_control_number`` column
+populated from the parsed 837's ST02 (transaction set control number)
+on every write. The 999 ack join (Pass 1) needs to resolve by ST02,
+not ISA13, so this column is the join key. This migration is purely
+additive: nullable, no default, backfills from
+``raw_result_json.envelope.transaction_set_control_number`` where the
+source JSON already carries it.
+
+For the backfill-shape tests we point ``db_migrate.MIGRATIONS_DIR``
+at the real migrations directory, apply all migrations once to bring
+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.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+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 _fresh_engine(path: Path) -> sa.Engine:
+    return sa.create_engine(f"sqlite:///{path}", future=True)
+
+
+def _table_info(engine: sa.Engine, table: str) -> list[tuple]:
+    """Return PRAGMA table_info rows for ``table`` as plain tuples."""
+    with engine.connect() as conn:
+        return list(conn.exec_driver_sql(f"PRAGMA table_info({table});").tuples())
+
+
+def _real_migrations_dir() -> Path:
+    return Path(__file__).parent.parent / "src" / "cyclone" / "migrations"
+
+
+@pytest.fixture
+def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Yield an engine at v20 against which every real migration has run.
+
+    Points ``db_migrate.MIGRATIONS_DIR`` at the real migrations
+    directory so the test exercises the actual 0020 file, then runs
+    the migration runner on a per-test fresh DB.
+    """
+    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", _real_migrations_dir())
+    engine = _fresh_engine(tmp_path / "mig0020.db")
+    db_migrate.run(engine)
+
+    # Confirm head is 20 (every migration applied).
+    with engine.connect() as conn:
+        v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
+    assert v == 20, f"expected migration head=20, got {v}"
+
+    yield engine
+    engine.dispose()
+
+
+def test_migration_0020_creates_column(migrated_engine) -> None:
+    """Migration 0020 adds ``transaction_set_control_number`` to ``batches``."""
+    cols = _table_info(migrated_engine, "batches")
+    col_names = {row[1] for row in cols}
+    assert "transaction_set_control_number" in col_names, (
+        "batches.transaction_set_control_number missing — migration 0020 did not run. "
+        f"Existing columns: {sorted(col_names)}"
+    )
+
+
+def test_migration_0020_column_is_nullable(migrated_engine) -> None:
+    """The new column is nullable (no DEFAULT, no NOT NULL) so existing
+    batches that don't yet carry the ST02 stay valid."""
+    cols = {row[1]: row for row in _table_info(migrated_engine, "batches")}
+    row = cols["transaction_set_control_number"]
+    # PRAGMA table_info tuples: (cid, name, type, notnull, dflt_value, pk)
+    assert row[3] == 0, f"notnull flag must be 0 (nullable), got {row[3]}"
+    assert row[4] is None, f"dflt_value must be NULL, got {row[4]!r}"
+    assert "TEXT" in (row[2] or "").upper(), f"expected TEXT column, got {row[2]!r}"
+
+
+def test_migration_0020_backfills_when_key_present(
+    migrated_engine: sa.Engine,
+) -> None:
+    """Replay the migration's backfill UPDATE against a row whose
+    ``raw_result_json`` carries the key — must populate the new column.
+
+    Replay (rather than waiting for ``db_migrate.run()`` to do it) is
+    the only way to test the SQL against a row that didn't exist when
+    init_db() ran; the migration's UPDATE naturally runs only over
+    pre-existing rows.
+    """
+    st02_value = "ST0001"
+    raw_json = {"envelope": {"transaction_set_control_number": st02_value}}
+
+    with migrated_engine.begin() as conn:
+        conn.exec_driver_sql(
+            "INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
+            "VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)",
+            (json.dumps(raw_json),),
+        )
+        conn.exec_driver_sql(BACKFILL_SQL)
+
+    with migrated_engine.connect() as conn:
+        row = conn.exec_driver_sql(
+            "SELECT transaction_set_control_number FROM batches WHERE id='B-ST02-1'"
+        ).first()
+    assert row is not None
+    assert row[0] == st02_value, (
+        f"backfill failed: expected {st02_value!r}, got {row[0]!r}"
+    )
+
+
+def test_migration_0020_backfill_conditional_on_key_present(
+    migrated_engine: sa.Engine,
+) -> None:
+    """Replay the UPDATE against a row whose ``raw_result_json`` does NOT
+    carry the key — must stay NULL. Proves the UPDATE is conditional
+    (the ``json_extract(...) IS NOT NULL`` guard) rather than
+    unconditionally overwriting with NULL."""
+    raw_json = {"envelope": {"control_number": "ISA0001"}}  # no txn-set key
+    with migrated_engine.begin() as conn:
+        conn.exec_driver_sql(
+            "INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
+            "VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)",
+            (json.dumps(raw_json),),
+        )
+        conn.exec_driver_sql(BACKFILL_SQL)
+
+    with migrated_engine.connect() as conn:
+        row = conn.exec_driver_sql(
+            "SELECT transaction_set_control_number FROM batches WHERE id='B-NOST-1'"
+        ).first()
+    assert row is not None
+    assert row[0] is None, (
+        f"backfill must not overwrite when key is absent; got {row[0]!r}"
+    )
+
+
+def test_migration_0020_backfill_handles_null_raw_result_json(
+    migrated_engine: sa.Engine,
+) -> None:
+    """A Batch row with ``raw_result_json IS NULL`` (the prior SP's
+    unparsed state) must not crash the backfill and must stay NULL."""
+    with migrated_engine.begin() as conn:
+        conn.exec_driver_sql(
+            "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)
+
+    with migrated_engine.connect() as conn:
+        row = conn.exec_driver_sql(
+            "SELECT transaction_set_control_number FROM batches WHERE id='B-NULL-1'"
+        ).first()
+    assert row is not None
+    assert row[0] is None