feat(sp39): Resubmission model + migration 0021 + idempotency tests

backend/src/cyclone/db.py

@@ -952,3 +952,33 @@ class Session(Base):
     user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
     expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+
+class Resubmission(Base):
+    """SP39: audit row recording that a claim was pushed to SFTP as
+    part of a corrected-file resubmission. One row per claim per push.
+
+    Status is derived at read-time by joining against claim_acks (via
+    the existing SP28/31 auto-link) + remittances (via CLP->claim).
+    The corrected-v2 regen preserves the original claim_id from
+    raw_json, so inbound 999 acks auto-link back to the original claim
+    row and the join works without any new matching logic.
+    """
+
+    __tablename__ = "resubmissions"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    claim_id: Mapped[str] = mapped_column(String, nullable=False, index=True)
+    batch_id: Mapped[str] = mapped_column(String, nullable=False, index=True)
+    resubmitted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    source_corrected_path: Mapped[str] = mapped_column(String, nullable=False)
+    interchange_control_number: Mapped[str] = mapped_column(String, nullable=False)
+    group_control_number: Mapped[str] = mapped_column(String, nullable=False)
+
+    __table_args__ = (
+        Index(
+            "ux_resubmissions_claim_icn",
+            "claim_id", "interchange_control_number",
+            unique=True,
+        ),
+    )

backend/src/cyclone/migrations/0021_resubmissions.sql

@@ -0,0 +1,30 @@
+-- version: 21
+-- SP39: resubmissions audit table for tracking corrected-file SFTP pushes.
+--
+-- One row per claim per push. Status (pending_999 / 999_accepted /
+-- 999_rejected / 277ca_accepted / paid / denied_again) is derived at
+-- read-time by joining against claim_acks (via the existing SP28/31
+-- auto-link) + remittances (via CLP->claim). No denormalized status
+-- column on this row — the existing auto-link data is the source of
+-- truth and we want to avoid a write-coordination problem between
+-- the SFTP push and the inbound ack ingestion.
+--
+-- Idempotency on (claim_id, interchange_control_number): the resubmit
+-- CLI can be re-run safely without producing duplicate rows for the
+-- same push.
+
+CREATE TABLE resubmissions (
+    id                          INTEGER PRIMARY KEY AUTOINCREMENT,
+    claim_id                    TEXT NOT NULL,
+    batch_id                    TEXT NOT NULL,
+    resubmitted_at              DATETIME NOT NULL,
+    source_corrected_path       TEXT NOT NULL,
+    interchange_control_number  TEXT NOT NULL,
+    group_control_number        TEXT NOT NULL
+);
+
+CREATE INDEX ix_resubmissions_claim_id ON resubmissions(claim_id);
+CREATE INDEX ix_resubmissions_batch_id ON resubmissions(batch_id);
+
+CREATE UNIQUE INDEX ux_resubmissions_claim_icn
+    ON resubmissions(claim_id, interchange_control_number);

backend/tests/test_db_migrate.py

@@ -127,14 +127,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
     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.
+    SP39 bumped it to 21 with the resubmissions audit table.
     """
     engine = _fresh_engine(tmp_path)
     db_migrate.run(engine)
     v_after_first = _user_version(engine)
-    assert v_after_first == 20, f"expected head=20, got {v_after_first}"
+    assert v_after_first == 21, f"expected head=21, got {v_after_first}"
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 20, "second run should not bump version"
+    assert _user_version(engine) == 21, "second run should not bump version"
 
 
 def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -160,7 +161,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) == 20, f"expected head=20, got {_user_version(engine)}"
+    assert _user_version(engine) == 21, f"expected head=21, 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
@@ -187,3 +188,50 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
         assert [r[0] for r in rows] == ["CLM-1", "CLM-2"]
         assert [float(r[1]) for r in rows] == [100.0, 200.0]
 
+
+
+# ---------------------------------------------------------------------------
+# SP39: migration 0021 creates the resubmissions table
+# ---------------------------------------------------------------------------
+
+
+def test_migration_0021_creates_resubmissions_table(tmp_path: Path) -> None:
+    """SP39: 0021_resubmissions.sql adds the resubmissions audit table
+    with the documented columns + unique constraint on
+    (claim_id, interchange_control_number)."""
+    # Point the runner at the REAL migrations dir so we exercise 0021.
+    real_dir = Path(db_migrate.__file__).parent / "migrations"
+    import cyclone.db_migrate as real_migrate_mod
+    real_dir_str = str(real_dir)
+
+    engine = _fresh_engine(tmp_path)
+    # monkeypatch the module-level MIGRATIONS_DIR
+    import importlib
+    monkey_save = real_migrate_mod.MIGRATIONS_DIR
+    real_migrate_mod.MIGRATIONS_DIR = Path(real_dir_str)
+    try:
+        db_migrate.run(engine)
+    finally:
+        real_migrate_mod.MIGRATIONS_DIR = monkey_save
+
+    # Verify the table exists with the expected columns + unique index.
+    with engine.connect() as conn:
+        version = conn.exec_driver_sql("PRAGMA user_version").scalar()
+        assert version >= 21
+        cols = conn.exec_driver_sql(
+            "PRAGMA table_info(resubmissions)"
+        ).all()
+        col_names = {row[1] for row in cols}
+        assert col_names == {
+            "id", "claim_id", "batch_id", "resubmitted_at",
+            "source_corrected_path",
+            "interchange_control_number", "group_control_number",
+        }
+        idx_rows = conn.exec_driver_sql(
+            "SELECT name, sql FROM sqlite_master "
+            "WHERE type='index' AND tbl_name='resubmissions'"
+        ).all()
+        idx_names = {row[0] for row in idx_rows}
+        assert "ix_resubmissions_claim_id" in idx_names
+        assert "ix_resubmissions_batch_id" in idx_names
+        assert "ux_resubmissions_claim_icn" in idx_names

backend/tests/test_resubmissions_table.py

@@ -0,0 +1,92 @@
+"""SP39: Resubmission model + idempotency.
+
+Schema:
+- 7 columns: id, claim_id, batch_id, resubmitted_at, source_corrected_path,
+  interchange_control_number, group_control_number
+- Unique index on (claim_id, interchange_control_number) -> re-insert is rejected.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from cyclone import db as cycl_db
+from cyclone.db import Resubmission
+
+
+def test_resubmission_model_columns():
+    cols = {c.name for c in Resubmission.__table__.columns}
+    assert cols == {
+        "id", "claim_id", "batch_id", "resubmitted_at",
+        "source_corrected_path",
+        "interchange_control_number", "group_control_number",
+    }
+
+
+def test_unique_index_on_claim_id_and_icn():
+    idx_names = {i.name for i in Resubmission.__table__.indexes}
+    assert "ux_resubmissions_claim_icn" in idx_names
+    ux = next(
+        i for i in Resubmission.__table__.indexes if i.name == "ux_resubmissions_claim_icn"
+    )
+    assert ux.unique is True
+    # The unique index covers both columns.
+    cols = {c.name for c in ux.columns}
+    assert cols == {"claim_id", "interchange_control_number"}
+
+
+def test_duplicate_insert_raises_integrity_error():
+    """Idempotency contract: a second insert with the same
+    (claim_id, interchange_control_number) is rejected at the DB layer."""
+    cycl_db.init_db()
+    now = datetime.now(timezone.utc)
+    with cycl_db.SessionLocal()() as s:
+        s.add(Resubmission(
+            claim_id="claim-x", batch_id="batch-y",
+            resubmitted_at=now,
+            source_corrected_path="/path/one",
+            interchange_control_number="000000001",
+            group_control_number="1",
+        ))
+        s.commit()
+    with cycl_db.SessionLocal()() as s:
+        s.add(Resubmission(
+            claim_id="claim-x", batch_id="batch-y",
+            resubmitted_at=now,
+            source_corrected_path="/path/two",
+            interchange_control_number="000000001",  # same ICN
+            group_control_number="1",
+        ))
+        with pytest.raises(IntegrityError):
+            s.commit()
+        s.rollback()
+
+
+def test_different_icn_on_same_claim_is_allowed():
+    """Distinct ICNs (i.e. distinct pushes) are independent rows."""
+    cycl_db.init_db()
+    now = datetime.now(timezone.utc)
+    with cycl_db.SessionLocal()() as s:
+        s.add(Resubmission(
+            claim_id="claim-y", batch_id="batch-y",
+            resubmitted_at=now,
+            source_corrected_path="/path/one",
+            interchange_control_number="000000001",
+            group_control_number="1",
+        ))
+        s.add(Resubmission(
+            claim_id="claim-y", batch_id="batch-y",
+            resubmitted_at=now,
+            source_corrected_path="/path/two",
+            interchange_control_number="000000002",  # different ICN
+            group_control_number="2",
+        ))
+        s.commit()
+    with cycl_db.SessionLocal()() as s:
+        from sqlalchemy import select
+        rows = s.execute(select(Resubmission).where(
+            Resubmission.claim_id == "claim-y"
+        )).scalars().all()
+        assert len(rows) == 2