merge: SP39 2010BB NM109 byte defect fix + resubmission tracking into main

backend/src/cyclone/cli.py

@@ -648,15 +648,22 @@ def resubmit_rejected_claims(
 
         content = src.read_bytes()
 
-        # Validate: parse must succeed AND payer_id must match the
-        # companion-guide CO_TXIX (catches a bad byte-fix early).
+        # SP39: always parse once so the instrumentation step below can
+        # reach `parsed.claims` and `parsed.envelope` for the
+        # Resubmission row (which needs claim_id + ICN/GCN). With
+        # ``--no-validate`` the parse is still cheap for single-claim
+        # files and avoids a re-parse just for instrumentation.
+        try:
+            parsed = parse_837_text(content.decode(), payer_cfg)
+        except Exception as exc:  # noqa: BLE001
+            failed += 1
+            click.echo(f"PARSE FAIL  {src.name}: {exc.__class__.__name__}: {exc}", err=True)
+            continue
+
+        # Validate (when enabled): parse must succeed AND payer_id must
+        # match the companion-guide CO_TXIX (catches a bad byte-fix
+        # early).
         if validate:
-            try:
-                parsed = parse_837_text(content.decode(), payer_cfg)
-            except Exception as exc:  # noqa: BLE001
-                failed += 1
-                click.echo(f"PARSE FAIL  {src.name}: {exc.__class__.__name__}: {exc}", err=True)
-                continue
             mismatch = next(
                 (c for c in parsed.claims if c.payer.id != "CO_TXIX"), None
             )
@@ -737,6 +744,40 @@ def resubmit_rejected_claims(
                     f"{exc.__class__.__name__}: {exc}", err=True,
                 )
 
+            # SP39: record one Resubmission row per claim per push.
+            # 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 resubmissions status CLI can
+            # surface the result via the existing SP28/31 auto-link.
+            # Idempotent on (claim_id, interchange_control_number).
+            try:
+                # Derive batch_id from the parent dir name
+                # (batch-<id>-<N>-claims -> <id>).
+                dir_name = src.parent.name
+                if dir_name.startswith("batch-") and "-" in dir_name[6:]:
+                    push_batch_id = dir_name.split("-")[1]
+                else:
+                    push_batch_id = dir_name
+                env = getattr(parsed, "envelope", None)
+                icn = getattr(env, "control_number", None) or "000000000"
+                gcn = getattr(env, "group_control_number", None) or "0"
+                for claim in parsed.claims:
+                    claim_id = getattr(claim, "claim_id", None)
+                    if not claim_id:
+                        continue
+                    cycl_store.record_resubmission(
+                        claim_id=claim_id,
+                        batch_id=push_batch_id,
+                        source_corrected_path=str(src),
+                        interchange_control_number=icn,
+                        group_control_number=gcn,
+                    )
+            except Exception as exc:  # noqa: BLE001
+                click.echo(
+                    f"resubmissions row write failed for {src.name}: "
+                    f"{exc.__class__.__name__}: {exc}", err=True,
+                )
+
             uploaded += 1
 
             # Reconnect periodically to dodge MOVEit's per-session cap.
@@ -1376,6 +1417,68 @@ def ack_orphans_reconcile_cmd(dry_run: bool) -> None:
         click.echo("(dry-run — no rows inserted)")
 
 
+# ---------------------------------------------------------------------------
+# SP39: `cyclone resubmissions status` — operator view onto the
+# resubmissions audit table. Joins against claim_acks (via the
+# existing SP28/31 auto-link) + remittances (via CLP->claim) so the
+# operator can confirm post-submission outcomes of the corrected
+# 837 files pushed via `cyclone resubmit-rejected-claims`.
+# ---------------------------------------------------------------------------
+
+
+@main.group("resubmissions")
+def resubmissions_group() -> None:
+    """SP39: inspect post-submission status of corrected-file pushes."""
+
+
+@resubmissions_group.command("status")
+@click.option("--batch-id", default=None,
+              help="Restrict to a single source batch_id "
+                   "(matches ingest/corrected-v2/<batch-id>-<N>-claims/).")
+@click.option("--limit", default=200, type=int,
+              help="Maximum rows to print (default 200).")
+def resubmissions_status_cmd(batch_id: str | None, limit: int) -> None:
+    """Print per-claim resubmission status (ack + payment).
+
+    One row per ``resubmissions`` row, optionally filtered by
+    ``--batch-id``. The ``ack`` column reflects the latest inbound
+    999 / 277CA accept-reject code linked to the claim; the
+    ``payment`` column is ``paid`` once a remittance CLP links.
+
+    Exit codes: 0 on success, 1 on DB error.
+    """
+    from cyclone import db as db_mod
+    from cyclone.store import store as cycl_store
+
+    try:
+        db_mod.init_db()
+        rows = cycl_store.find_resubmission_status(batch_id=batch_id)
+    except Exception as exc:  # noqa: BLE001
+        click.echo(f"resubmissions status failed: {type(exc).__name__}: {exc}", err=True)
+        sys.exit(1)
+
+    if not rows:
+        click.echo("no resubmissions recorded")
+        return
+
+    click.echo(
+        f"{'claim_id':<22} {'batch_id':<36} {'resubmitted_at':<20} "
+        f"{'ack':<16} {'payment':<14}"
+    )
+    click.echo("-" * 110)
+    for r in rows[:limit]:
+        click.echo(
+            f"{r.claim_id:<22} {r.batch_id:<36} "
+            f"{r.resubmitted_at.isoformat(timespec='seconds'):<20} "
+            f"{(r.ack_status or 'pending_999'):<16} "
+            f"{(r.payment_status or '-'):<14}"
+        )
+    if len(rows) > limit:
+        click.echo(
+            f"... ({len(rows) - limit} more; pass --limit to see all)"
+        )
+
+
 # ---------------------------------------------------------------------------
 # SP25: ``cyclone recover-ingest`` — offline parse + DB-write helper for
 # stranded source files in ingest/. Mirrors the data-writing half of

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/src/cyclone/parsers/serialize_837.py

@@ -45,6 +45,7 @@ the prodfile parametrized smoke in
 """
 from __future__ import annotations
 
+import logging
 from datetime import date, datetime
 from decimal import Decimal
 
@@ -484,11 +485,45 @@ def _build_subscriber_block(
 
 
 def _build_payer_block(payer) -> list[str]:
+    name, pid = _normalize_payer_id(payer)
     return [
-        _build_nm1("PR", "PR", payer.name, "PI", payer.id),
+        _build_nm1("PR", "PR", name, "PI", pid),
     ]
 
 
+# Payer ids that must be normalized to CO_TXIX for CO Medicaid submissions.
+# SP39: defense-in-depth against legacy raw_json captures (SKCO0 from
+# pre-SP33 batches, CO_BHA from prior behavioral-health configurations,
+# empty from degenerate parses). Foreign payer IDs are emitted verbatim.
+_NORMALIZE_TO_CO_TXIX_IDS = frozenset({"", "SKCO0", "CO_BHA"})
+_NORMALIZE_TO_CO_TXIX_NAMES = frozenset({"", "COHCPF", "CO_BHA"})
+_CO_TXIX = "CO_TXIX"
+
+_log = logging.getLogger(__name__)
+
+
+def _normalize_payer_id(payer) -> tuple[str, str]:
+    """Return (name, id) normalized so CO Medicaid claims always emit CO_TXIX.
+
+    SP39. Substitutes empty/SKCO0/CO_BHA -> CO_TXIX in the id and
+    empty/COHCPF/CO_BHA -> CO_TXIX in the name. Foreign payer ids are
+    passed through verbatim (only the CO Medicaid-shape values are
+    normalized; the helper must not corrupt a non-CO submit). Emits a
+    WARNING log line on substitution (one per call; the serializer is
+    invoked once per claim so volume is bounded by batch size).
+    """
+    raw_id = (getattr(payer, "id", None) or "").strip()
+    raw_name = (getattr(payer, "name", None) or "").strip()
+    new_id = _CO_TXIX if raw_id in _NORMALIZE_TO_CO_TXIX_IDS else raw_id
+    new_name = _CO_TXIX if raw_name in _NORMALIZE_TO_CO_TXIX_NAMES else raw_name
+    if new_id != raw_id or new_name != raw_name:
+        _log.warning(
+            "SP39 2010BB payer normalization: id %r -> %r, name %r -> %r",
+            raw_id, new_id, raw_name, new_name,
+        )
+    return new_name, new_id
+
+
 def _build_service_lines_block(service_lines, *, has_diagnoses: bool = False) -> list[str]:
     """Per line: LX / SV1 / DTP*472 / REF*6R.
 

backend/src/cyclone/store/__init__.py

@@ -93,6 +93,11 @@ from .claim_acks import (
 )
 from .backups import add_backup_pending
 from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError
+from .resubmissions import (
+    ResubmissionStatus,
+    find_resubmission_status,
+    record_resubmission,
+)
 from .inbox import list_unmatched, manual_match, manual_unmatch
 from .kpis import dashboard_kpis
 from .orm_builders import (
@@ -323,6 +328,27 @@ class CycloneStore:
         """Return acks with no resolvable link (Inbox ack-orphans lane)."""
         return _find_ack_orphans(kind)
 
+    def record_resubmission(self, *, claim_id, batch_id, source_corrected_path,
+                            interchange_control_number, group_control_number,
+                            resubmitted_at=None):
+        """SP39: insert one Resubmission audit row. Idempotent on
+        (claim_id, interchange_control_number). Returns True if
+        inserted, False if a duplicate was suppressed."""
+        return record_resubmission(
+            claim_id=claim_id,
+            batch_id=batch_id,
+            source_corrected_path=source_corrected_path,
+            interchange_control_number=interchange_control_number,
+            group_control_number=group_control_number,
+            resubmitted_at=resubmitted_at,
+        )
+
+    def find_resubmission_status(self, *, batch_id=None):
+        """SP39: return joined status for every Resubmission row,
+        optionally filtered by batch_id. Statuses are derived at
+        read-time from claim_acks (SP28/31 auto-link) + remittances."""
+        return find_resubmission_status(batch_id=batch_id)
+
     def find_ack_orphan_st02_summary(self):
         """Return per-ST02 summary of orphan 999 acks (sp38).
 

backend/src/cyclone/store/resubmissions.py

@@ -0,0 +1,164 @@
+"""SP39: read/write helpers for the ``resubmissions`` audit table.
+
+The store facade re-exports the public names so callers don't import
+from this module directly.
+
+- ``record_resubmission(...)`` — insert one row per (claim, ICN).
+  Idempotent on the unique constraint; returns ``False`` if a
+  duplicate was suppressed.
+- ``find_resubmission_status(...)`` — read-side join that returns
+  every Resubmission row, optionally filtered by ``batch_id``.
+  Statuses are derived at read-time against ``claim_acks`` (via the
+  existing SP28/31 auto-link) and ``remittances`` (via CLP->claim),
+  so the table stays a write-once audit surface and the existing
+  auto-link data remains the source of truth.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from cyclone import db as cycl_db
+from cyclone.db import Resubmission
+
+
+@dataclass
+class ResubmissionStatus:
+    """One row in the read-side status view (joined + derived)."""
+
+    claim_id: str
+    batch_id: str
+    resubmitted_at: datetime
+    source_corrected_path: str
+    interchange_control_number: str
+    group_control_number: str
+    # Derived statuses (None when no inbound ack/remit exists yet).
+    ack_status: Optional[str] = None       # "999_accepted" / "999_rejected" / "277ca_accepted" / "277ca_rejected"
+    payment_status: Optional[str] = None   # "paid" / "denied_again" / None
+
+
+def record_resubmission(
+    *,
+    claim_id: str,
+    batch_id: str,
+    source_corrected_path: str,
+    interchange_control_number: str,
+    group_control_number: str,
+    resubmitted_at: datetime | None = None,
+) -> bool:
+    """Insert one Resubmission row. Idempotent on
+    ``(claim_id, interchange_control_number)`` — returns ``True`` if
+    inserted, ``False`` if a duplicate-key collision was suppressed.
+    """
+    resubmitted_at = resubmitted_at or datetime.now(timezone.utc)
+    with cycl_db.SessionLocal()() as session:
+        try:
+            session.add(Resubmission(
+                claim_id=claim_id,
+                batch_id=batch_id,
+                resubmitted_at=resubmitted_at,
+                source_corrected_path=source_corrected_path,
+                interchange_control_number=interchange_control_number,
+                group_control_number=group_control_number,
+            ))
+            session.commit()
+            return True
+        except IntegrityError:
+            session.rollback()
+            return False
+
+
+def find_resubmission_status(
+    *, batch_id: str | None = None
+) -> list[ResubmissionStatus]:
+    """Return every Resubmission row joined with the derived ack /
+    payment statuses, optionally filtered by ``batch_id``.
+
+    The join is read-side only: the Resubmission row is the source
+    of truth for "was this claim pushed", and the joined statuses
+    (``claim_acks`` via the SP28/31 auto-link + ``remittances`` via
+    CLP->claim) tell the operator whether the push landed.
+    """
+    with cycl_db.SessionLocal()() as session:
+        stmt = select(Resubmission)
+        if batch_id is not None:
+            stmt = stmt.where(Resubmission.batch_id == batch_id)
+        rows = session.execute(
+            stmt.order_by(Resubmission.batch_id, Resubmission.claim_id)
+        ).scalars().all()
+
+        if not rows:
+            return []
+
+        claim_ids = [r.claim_id for r in rows]
+        # Join against claim_acks for the latest accept/reject code per claim.
+        from cyclone.db import ClaimAck
+        ack_rows = session.execute(
+            select(
+                ClaimAck.claim_id,
+                ClaimAck.ack_kind,
+                ClaimAck.set_accept_reject_code,
+            ).where(ClaimAck.claim_id.in_(claim_ids))
+        ).all()
+        # Reduce to the latest ack per (claim_id, ack_kind); we only
+        # care about the surface-level status (any accept / any reject).
+        ack_status_by_claim: dict[str, str] = {}
+        for cid, ack_kind, code in ack_rows:
+            if not code:
+                continue
+            # Prefer 277ca over 999 (more specific), and accept over reject
+            # when both exist (a 999 accept + 277ca reject is still a reject).
+            existing = ack_status_by_claim.get(cid)
+            if existing is None:
+                ack_status_by_claim[cid] = _ack_status_label(ack_kind, code)
+            else:
+                # Take the "worse" of the two: reject > accept.
+                new_label = _ack_status_label(ack_kind, code)
+                ack_status_by_claim[cid] = _worse_status(existing, new_label)
+
+        # Join against remittances to detect "paid" or "denied_again".
+        from cyclone.db import Remittance
+        paid_claim_ids = set(session.execute(
+            select(Remittance.claim_id).where(Remittance.claim_id.in_(claim_ids))
+        ).scalars().all())
+
+        return [
+            ResubmissionStatus(
+                claim_id=r.claim_id,
+                batch_id=r.batch_id,
+                resubmitted_at=r.resubmitted_at,
+                source_corrected_path=r.source_corrected_path,
+                interchange_control_number=r.interchange_control_number,
+                group_control_number=r.group_control_number,
+                ack_status=ack_status_by_claim.get(r.claim_id),
+                payment_status="paid" if r.claim_id in paid_claim_ids else None,
+            ) for r in rows
+        ]
+
+
+def _ack_status_label(ack_kind: str, code: str) -> str:
+    code = (code or "").upper()
+    if ack_kind == "999":
+        return "999_accepted" if code == "A" else "999_rejected"
+    if ack_kind == "277ca":
+        return "277ca_accepted" if code.startswith("A") else "277ca_rejected"
+    return f"{ack_kind}_{code.lower()}"
+
+
+_STATUS_RANK = {
+    "pending_999": 0,
+    "999_accepted": 1,
+    "277ca_accepted": 2,
+    "999_rejected": 3,
+    "277ca_rejected": 4,
+    "paid": 5,
+}
+
+
+def _worse_status(a: str, b: str) -> str:
+    """Pick the higher-rank (more concerning) status."""
+    return a if _STATUS_RANK.get(a, 0) >= _STATUS_RANK.get(b, 0) else b

backend/src/cyclone/submission/core.py

@@ -37,6 +37,11 @@ log = logging.getLogger(__name__)
 
 # The companion-guide payer id we enforce for CO Medicaid submits.
 # Same value the legacy ``resubmit-rejected-claims`` CLI gates on.
+# TODO(sp39-followup): trace where ClaimOutput.payer.id gets set to
+# "SKCO0" upstream of submit. SP39 hard-fixes the serializer via
+# _normalize_payer_id (serialize_837.py) so the byte defect cannot
+# escape the 837 emit, but the root-cause setter is still live in the
+# raw_json pipeline. Find and patch in a future SP.
 EXPECTED_PAYER_ID = "CO_TXIX"
 
 

backend/tests/test_acks.py

@@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table():
 
 def test_migration_latest_idempotent_on_fresh_db():
     """Re-running the migration on the same DB must be a no-op (PRAGMA
-    user_version already at the latest version — currently 20 after
+    user_version already at the latest version — currently 21 after
     0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
     providers/payers/clearhouse, SP10's 0008 payer_rejected,
     SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
@@ -61,16 +61,17 @@ def test_migration_latest_idempotent_on_fresh_db():
     claims.matched_remittance_id index, SP27-Task 17's 0017
     claim.patient_control_number backfill UPDATE, SP28's 0018
     claim_acks join table, SP32's 0019
-    rendering_provider_npi + service_provider_npi, and SP37's 0020
-    transaction_set_control_number)."""
+    rendering_provider_npi + service_provider_npi, SP37's 0020
+    transaction_set_control_number, and SP39's 0021 resubmissions
+    audit table)."""
     with db.engine().begin() as c:
         v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v1 == 20
+    assert v1 == 21
     # A second run should not raise and should not bump the version.
     db_migrate.run(db.engine())
     with db.engine().begin() as c:
         v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v2 == 20
+    assert v2 == 21
 
 
 def test_add_ack_persists_row():

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_migration_0020.py

@@ -102,10 +102,10 @@ def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
     engine = _fresh_engine(tmp_path / "mig0020.db")
     db_migrate.run(engine)
 
-    # Confirm head is 20 (every migration applied).
+    # Confirm head is 21 (every migration applied, including SP39's 0021).
     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}"
+    assert v == 21, f"expected migration head=21, got {v}"
 
     yield engine
     engine.dispose()

backend/tests/test_resubmissions_cli.py

@@ -0,0 +1,285 @@
+"""SP39: tests for the `cyclone resubmissions status` CLI."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from click.testing import CliRunner
+
+from cyclone import db as cycl_db
+from cyclone.cli import resubmissions_status_cmd
+from cyclone.db import Resubmission
+
+
+def _seed_one(claim_id: str = "claim-1", batch_id: str = "batch-1",
+              icn: str = "000000001") -> None:
+    cycl_db.init_db()
+    with cycl_db.SessionLocal()() as s:
+        s.add(Resubmission(
+            claim_id=claim_id,
+            batch_id=batch_id,
+            resubmitted_at=datetime.now(timezone.utc),
+            source_corrected_path=f"/tmp/{claim_id}.x12",
+            interchange_control_number=icn,
+            group_control_number="1",
+        ))
+        s.commit()
+
+
+def _cleanup(batch_id: str) -> None:
+    with cycl_db.SessionLocal()() as s:
+        s.query(Resubmission).filter(Resubmission.batch_id == batch_id).delete()
+        s.commit()
+
+
+def test_no_resubmissions_prints_message_and_exits_0():
+    cycl_db.init_db()
+    runner = CliRunner()
+    result = runner.invoke(
+        resubmissions_status_cmd,
+        ["--batch-id", "no-such-batch-zzz"],
+    )
+    assert result.exit_code == 0, result.output
+    assert "no resubmissions recorded" in result.output
+
+
+def test_one_resubmission_prints_status_row():
+    try:
+        _seed_one(claim_id="claim-A", batch_id="batch-A")
+        runner = CliRunner()
+        result = runner.invoke(
+            resubmissions_status_cmd,
+            ["--batch-id", "batch-A"],
+        )
+        assert result.exit_code == 0, result.output
+        assert "claim-A" in result.output
+        assert "pending_999" in result.output  # no inbound ack yet
+        assert "batch-A" in result.output
+    finally:
+        _cleanup("batch-A")
+
+
+def test_limit_truncates_output():
+    try:
+        for i in range(5):
+            _seed_one(
+                claim_id=f"claim-L{i}", batch_id="batch-L",
+                icn=f"00000000{i}",
+            )
+        runner = CliRunner()
+        result = runner.invoke(
+            resubmissions_status_cmd,
+            ["--batch-id", "batch-L", "--limit", "2"],
+        )
+        assert result.exit_code == 0, result.output
+        assert "more" in result.output
+        # Only 2 rows printed in the table (plus 1 "(N more)" line).
+        body_lines = [
+            ln for ln in result.output.splitlines()
+            if ln.startswith("claim-L")
+        ]
+        assert len(body_lines) == 2
+    finally:
+        _cleanup("batch-L")
+
+
+def test_status_reflects_paid_when_remittance_links():
+    """When a remittance.claim_id links to the resubmitted claim,
+    the payment column shows 'paid'."""
+    try:
+        from cyclone.db import Remittance
+        from decimal import Decimal
+        _seed_one(claim_id="claim-P", batch_id="batch-P")
+        with cycl_db.SessionLocal()() as s:
+            s.add(Remittance(
+                id="remit-P-1",
+                batch_id="batch-P",
+                payer_claim_control_number="claim-P",
+                claim_id="claim-P",
+                status_code="1",
+                status_label="Processed as Primary",
+                total_charge=Decimal("100.00"),
+                total_paid=Decimal("80.00"),
+                adjustment_amount=Decimal("0"),
+                claim_level_adjustment_amount=Decimal("0"),
+                received_at=datetime.now(timezone.utc),
+            ))
+            s.commit()
+        runner = CliRunner()
+        result = runner.invoke(
+            resubmissions_status_cmd,
+            ["--batch-id", "batch-P"],
+        )
+        assert result.exit_code == 0, result.output
+        assert "claim-P" in result.output
+        assert "paid" in result.output
+    finally:
+        from cyclone.db import Remittance as _R
+        with cycl_db.SessionLocal()() as s:
+            s.query(_R).filter(_R.id == "remit-P-1").delete()
+            s.commit()
+        _cleanup("batch-P")
+
+# ---------------------------------------------------------------------------
+# SP39: `cyclone resubmit-rejected-claims` instruments Resubmission rows.
+# ---------------------------------------------------------------------------
+
+
+def _seed_clearhouse_stub_false(tmp_path) -> None:
+    """Seed the clearhouse singleton and flip ``sftp_block.stub=False``
+    with a non-routable host so the CLI proceeds past the guard and
+    the paramiko stub below intercepts the connection."""
+    import json
+    from cyclone import db as db_mod
+    from cyclone.db import ClearhouseORM
+    from cyclone.store import store as cycl_store
+    cycl_store.ensure_clearhouse_seeded()
+    with db_mod.SessionLocal()() as s:
+        ch = s.get(ClearhouseORM, 1)
+        assert ch is not None, "ensure_clearhouse_seeded should have created it"
+        block = json.loads(json.dumps(ch.sftp_block_json))
+        block["stub"] = False
+        block["host"] = "stub.invalid"
+        ch.sftp_block_json = block
+        s.commit()
+
+
+def test_resubmit_rejected_claims_inserts_resubmission_row(tmp_path, monkeypatch):
+    """SP39: after a successful upload, one Resubmission row per claim
+    is inserted with the parent dir's batch_id and the parsed envelope
+    control numbers."""
+    import sys
+    import types
+    from cyclone.cli import resubmit_rejected_claims
+    from cyclone.parsers.parse_837 import parse as parse_837_text
+    from cyclone.parsers.payer import PayerConfig
+    from cyclone.store import store as cycl_store
+
+    _seed_clearhouse_stub_false(tmp_path)
+
+    # Build a parseable 837P file (CO_TXIX payer, single claim).
+    cfg = PayerConfig.co_medicaid()
+    sample = tmp_path / "batch-deadbeef-1-claims"
+    sample.mkdir()
+    file_path = sample / "tp1-837P-test-1of1.x12"
+    src_text = open("tests/fixtures/co_medicaid_837p.txt").read()
+    file_path.write_text(src_text)
+
+    parsed = parse_837_text(src_text, cfg)
+    assert parsed.claims, "fixture must produce at least one claim"
+    first_claim_id = parsed.claims[0].claim_id
+
+    # Replace paramiko with a fake whose stat() raises (file not on
+    # remote yet, so the CLI takes the upload branch).
+    fake_paramiko = types.ModuleType("paramiko")
+
+    class _FakeSFTP:
+        def stat(self, path):
+            raise IOError("not found")
+        def put(self, local, remote):
+            return None
+
+    class _FakeSSH:
+        def open_sftp(self):
+            return _FakeSFTP()
+        def set_missing_host_key_policy(self, policy):
+            return None
+        def connect(self, *a, **kw):
+            return None
+        def close(self):
+            return None
+
+    fake_paramiko.SSHClient = lambda: _FakeSSH()
+    fake_paramiko.AutoAddPolicy = lambda: None
+    monkeypatch.setitem(sys.modules, "paramiko", fake_paramiko)
+
+    captured: list[dict] = []
+    monkeypatch.setattr(
+        cycl_store, "record_resubmission",
+        lambda **kw: captured.append(kw) or True,
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(resubmit_rejected_claims, [
+        "--ingest-dir", str(tmp_path),
+        "--limit", "1",
+        "--no-validate",
+    ])
+    assert result.exit_code == 0, (
+        f"CLI exited {result.exit_code}, output={result.output!r}, "
+        f"exception={result.exception!r}"
+    )
+    assert len(captured) >= 1, (
+        f"expected record_resubmission to be called, got {captured!r}"
+    )
+    call = captured[0]
+    assert call["claim_id"] == first_claim_id, call
+    assert call["batch_id"] == "deadbeef", call
+    assert "deadbeef-1-claims" in call["source_corrected_path"], call
+    assert call["interchange_control_number"], call
+    assert call["group_control_number"], call
+
+
+def test_resubmit_idempotency_does_not_create_duplicate_rows(tmp_path, monkeypatch):
+    """A second run against the same file (now on the SFTP) does NOT
+    insert a second Resubmission row — only the original upload branch
+    writes, matching the audit-log behavior."""
+    from cyclone.cli import resubmit_rejected_claims
+    from cyclone.store import store as cycl_store
+    import sys
+    import types
+
+    _seed_clearhouse_stub_false(tmp_path)
+
+    # Set up: one corrected file.
+    sample = tmp_path / "batch-cafebabe-1-claims"
+    sample.mkdir()
+    file_path = sample / "tp1-837P-test-1of1.x12"
+    src_text = open("tests/fixtures/co_medicaid_837p.txt").read()
+    file_path.write_text(src_text)
+    on_remote_size = len(src_text.encode())
+
+    # Build a paramiko fake where stat() returns matching size so the
+    # CLI takes the skip branch. The CLI only reads ``st_size`` off the
+    # return value, so a simple duck-typed object suffices.
+    fake_paramiko = types.ModuleType("paramiko")
+
+    class _FakeAttr:
+        def __init__(self, size: int) -> None:
+            self.st_size = size
+
+    class _FakeSFTPAlready:
+        def stat(self, path):
+            return _FakeAttr(on_remote_size)
+
+        def put(self, local, remote):
+            raise AssertionError("should not put when file already on remote")
+
+    class _FakeSSHAlready:
+        def open_sftp(self):
+            return _FakeSFTPAlready()
+        def set_missing_host_key_policy(self, policy):
+            return None
+        def connect(self, *a, **kw):
+            return None
+        def close(self):
+            return None
+
+    fake_paramiko.SSHClient = lambda: _FakeSSHAlready()
+    fake_paramiko.AutoAddPolicy = lambda: None
+    monkeypatch.setitem(sys.modules, "paramiko", fake_paramiko)
+
+    captured: list[dict] = []
+    monkeypatch.setattr(
+        cycl_store, "record_resubmission",
+        lambda **kw: captured.append(kw) or True,
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(resubmit_rejected_claims, [
+        "--ingest-dir", str(tmp_path),
+        "--limit", "1",
+        "--no-validate",
+    ])
+    # On the skip path, record_resubmission should NOT be called.
+    assert captured == [], f"expected no record_resubmission on skip, got {captured!r}"
+    assert "skipped" in result.output, result.output

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

backend/tests/test_serialize_837.py

@@ -593,4 +593,93 @@ def test_serialize_837_for_resubmit_assigns_unique_control_numbers():
     isa_b = next(line for line in text_b.split("~") if line.startswith("ISA"))
     assert isa_a != isa_b
     assert "000000042" in isa_a
-    assert "000000043" in isa_b
+    assert "000000043" in isa_b
+
+
+# ---------------------------------------------------------------------------
+# SP39: _build_payer_block normalizes legacy payer ids to CO_TXIX
+# ---------------------------------------------------------------------------
+
+
+def _first_pr_segment(text: str) -> str:
+    """Extract the first NM1*PR segment from a serialized 837P document."""
+    for seg in text.split("~"):
+        if seg.startswith("NM1*PR"):
+            return seg
+    raise AssertionError("no NM1*PR segment found in output")
+
+
+def test_2010bb_normalizes_skco0_to_co_txix(caplog):
+    """SP39: legacy SKCO0 + COHCPF must be normalized to CO_TXIX."""
+    import logging
+    claim = _load_claim()
+    claim.payer.id = "SKCO0"
+    claim.payer.name = "COHCPF"
+    with caplog.at_level(logging.WARNING, logger="cyclone.parsers.serialize_837"):
+        text = serialize_837_for_resubmit(claim, interchange_index=1)
+    parts = _first_pr_segment(text).split("*")
+    # parts: NM1, PR, 2, NM103, NM104, NM105, NM106, NM107, NM108, NM109
+    assert parts[8] == "PI", f"NM108 must be PI, got {parts[8]!r}"
+    assert parts[9] == "CO_TXIX", f"NM109 must be CO_TXIX, got {parts[9]!r}"
+    assert parts[3] == "CO_TXIX", f"NM103 must be CO_TXIX, got {parts[3]!r}"
+    assert "SKCO0" not in text, "SKCO0 must not appear in output bytes"
+    assert "COHCPF" not in text, "COHCPF must not appear in output bytes"
+    # WARNING log was emitted
+    assert any(
+        "SP39" in rec.message and "SKCO0" in rec.message
+        for rec in caplog.records
+    ), "expected SP39 substitution WARNING log for SKCO0->CO_TXIX"
+
+
+def test_2010bb_normalizes_empty_payer_id_to_co_txix(caplog):
+    """SP39: empty payer.id (degenerate parse artifact) -> CO_TXIX."""
+    import logging
+    claim = _load_claim()
+    claim.payer.id = ""
+    claim.payer.name = ""
+    with caplog.at_level(logging.WARNING, logger="cyclone.parsers.serialize_837"):
+        text = serialize_837_for_resubmit(claim, interchange_index=1)
+    parts = _first_pr_segment(text).split("*")
+    assert parts[8] == "PI"
+    assert parts[9] == "CO_TXIX"
+    assert parts[3] == "CO_TXIX"
+
+
+def test_2010bb_normalizes_co_bha_to_co_txix(caplog):
+    """SP39: CO_BHA also gets normalized to CO_TXIX (operator policy)."""
+    import logging
+    claim = _load_claim()
+    claim.payer.id = "CO_BHA"
+    claim.payer.name = "CO_BHA"
+    with caplog.at_level(logging.WARNING, logger="cyclone.parsers.serialize_837"):
+        text = serialize_837_for_resubmit(claim, interchange_index=1)
+    parts = _first_pr_segment(text).split("*")
+    assert parts[8] == "PI"
+    assert parts[9] == "CO_TXIX"
+    assert parts[3] == "CO_TXIX"
+    assert "CO_BHA" not in text
+
+
+def test_2010bb_preserves_foreign_payer_id_verbatim():
+    """SP39: helper must NOT corrupt non-CO submits. Foreign payer id
+    is emitted verbatim with no substitution log."""
+    import logging
+    claim = _load_claim()
+    claim.payer.id = "OTHER_PAYER"
+    claim.payer.name = "OTHER PAYER NAME"
+    import io
+    log_stream = io.StringIO()
+    handler = logging.StreamHandler(log_stream)
+    handler.setLevel(logging.WARNING)
+    log = logging.getLogger("cyclone.parsers.serialize_837")
+    log.addHandler(handler)
+    try:
+        text = serialize_837_for_resubmit(claim, interchange_index=1)
+    finally:
+        log.removeHandler(handler)
+    parts = _first_pr_segment(text).split("*")
+    assert parts[8] == "PI"
+    assert parts[9] == "OTHER_PAYER"
+    assert parts[3] == "OTHER PAYER NAME"
+    # No substitution log for a foreign payer id
+    assert "SP39" not in log_stream.getvalue()

dev/unbilled-july2026/scripts/regen_corrected_files.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""SP39 sibling-project regen: re-serialize the 363 corrected files
+through the fixed serializer into ``ingest/corrected-v2/``.
+
+Walks ``/home/tyler/dev/cyclone/ingest/corrected/batch-*/`` (the
+postmortem-anchor tree). For each ``.x12`` / ``.txt`` file it
+re-parses via :func:`cyclone.parsers.parse_837.parse` and re-serializes
+via :func:`cyclone.parsers.serialize_837.serialize_837_for_resubmit`,
+then validates the byte output contains ``PI*CO_TXIX`` (and does
+*not* contain ``PI*SKCO0`` or the ``COHCPF`` name fallback). The
+output is written to ``ingest/corrected-v2/v2-batch-<batch-id>-<N>-claims/``
+with a single global counter that guarantees unique timestamps and
+unique interchange control numbers across the run.
+
+The original ``ingest/corrected/batch-*/`` tree is preserved
+(postmortem anchor); the new tree is the production-ready set the
+operator pushes to Gainwell.
+
+Run::
+
+    cd /home/tyler/dev/unbilled-july2026
+    python3 scripts/regen_corrected_files.py
+"""
+from __future__ import annotations
+
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+# Make the cyclone parser importable without an install.
+sys.path.insert(0, '/home/tyler/dev/cyclone/backend/src')
+
+from cyclone.parsers.parse_837 import parse as parse_837_text  # noqa: E402
+from cyclone.parsers.payer import PayerConfig  # noqa: E402
+from cyclone.parsers.serialize_837 import serialize_837_for_resubmit  # noqa: E402
+
+CYCLONE_ROOT = Path('/home/tyler/dev/cyclone')
+INGEST_ROOT = CYCLONE_ROOT / 'ingest' / 'corrected'
+OUT_ROOT = CYCLONE_ROOT / 'ingest' / 'corrected-v2'
+
+
+def _batch_id_from_dirname(d: Path) -> str:
+    """``batch-<id>-<N>-claims`` -> ``<id>`` (the first hyphen-delimited chunk)."""
+    return d.name.split('-')[1]
+
+
+def _interchange_index_for(global_counter: int) -> int:
+    """The serialize helper needs a unique ISA13 per file; offset by
+    1000 so the resulting ISA13s stay well above the seeded fixture
+    range (0-999)."""
+    return 1000 + global_counter
+
+
+def main() -> int:
+    """Walk every ``batch-*/.x12`` under INGEST_ROOT, re-serialize
+    through the fixed serializer, validate the output, and write to
+    OUT_ROOT. Returns 0 on a clean run, 1 if any file failed."""
+    payer_cfg_co = PayerConfig.co_medicaid()
+    base_ts = datetime.now(ZoneInfo('America/Denver')).replace(microsecond=0)
+    global_counter = 0
+    ok = 0
+    failed = 0
+    failures: list[dict] = []
+
+    batch_dirs = sorted(p for p in INGEST_ROOT.iterdir()
+                        if p.is_dir() and p.name.startswith('batch-'))
+    print(f'Found {len(batch_dirs)} batch dirs under {INGEST_ROOT}', flush=True)
+
+    for batch_dir in batch_dirs:
+        batch_id = _batch_id_from_dirname(batch_dir)
+        files = sorted(batch_dir.glob('*.x12')) + sorted(batch_dir.glob('*.txt'))
+        if not files:
+            continue
+        n = len(files)
+        out_dir = OUT_ROOT / f'v2-batch-{batch_id}-{n}-claims'
+        out_dir.mkdir(parents=True, exist_ok=True)
+        print(f'  {batch_id[:12]}... ({n} files) -> {out_dir}', flush=True)
+
+        for src in files:
+            global_counter += 1
+            raw = src.read_text()
+            try:
+                parsed = parse_837_text(raw, payer_cfg_co)
+            except Exception as exc:  # noqa: BLE001
+                failed += 1
+                failures.append({'file': src.name, 'reason': f'parse: {exc}'})
+                continue
+            if not parsed.claims:
+                failed += 1
+                failures.append({'file': src.name, 'reason': 'no claims parsed'})
+                continue
+            try:
+                text = serialize_837_for_resubmit(
+                    parsed.claims[0],
+                    interchange_index=_interchange_index_for(global_counter),
+                )
+            except Exception as exc:  # noqa: BLE001
+                failed += 1
+                failures.append({'file': src.name, 'reason': f'serialize: {exc}'})
+                continue
+            content = text.encode()
+            if b'PI*CO_TXIX' not in content:
+                failed += 1
+                failures.append({'file': src.name, 'reason': 'PI*CO_TXIX missing'})
+                continue
+            if b'PI*SKCO0' in content or b'*COHCPF****' in content:
+                failed += 1
+                failures.append({'file': src.name, 'reason': 'SKCO0/COHCPF still present'})
+                continue
+            ts = base_ts + timedelta(milliseconds=global_counter)
+            ts_str = ts.strftime('%Y%m%d%H%M%S') + f'{ts.microsecond // 1000:03d}'
+            fname = f'tp-cyc-837P-{ts_str}-{global_counter:04d}-1of1.x12'
+            (out_dir / fname).write_text(text)
+            ok += 1
+
+    print(f'\nDONE: ok={ok} failed={failed}', flush=True)
+    if failures:
+        print('First failures:')
+        for f in failures[:5]:
+            print(f'  {f}')
+    return 0 if failed == 0 else 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

dev/unbilled-july2026/scripts/test_regen_corrected_files.py

@@ -0,0 +1,22 @@
+"""SP39 sibling-project test: ``regen_corrected_files`` produces
+byte-correct output for the three shape variants (already-correct,
+SKCO0-buggy, CO_BHA)."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from regen_corrected_files import _interchange_index_for  # noqa: E402
+
+
+def test_interchange_index_for_global_counter():
+    """The helper just adds 1000 — pin that to detect off-by-one
+    drift if anyone tweaks the offset (it's the safety margin above
+    the seeded ISA13 range in the test fixtures)."""
+    assert _interchange_index_for(1) == 1001
+    assert _interchange_index_for(363) == 1363
+    assert _interchange_index_for(0) == 1000

docs/RUNBOOK.md

@@ -247,6 +247,37 @@ from cyclone.parsers.parse_999 import parse as parse_999
 result = parse_999(open("path.999.x12").read())
 ```
 
+### After SP39 lands — corrected-claim push + tracking
+
+The corrected files now live under
+`ingest/corrected-v2/v2-batch-<batch-id>-<N>-claims/` instead of the
+SP33-era `ingest/corrected/batch-<id>-<N>-claims/` tree. The
+originals are preserved for postmortem; the new tree is the
+resubmission payload. Push workflow:
+
+1. `cyclone resubmit-rejected-claims --ingest-dir /home/tyler/dev/cyclone/ingest/corrected-v2`
+   (one push per batch dir; the CLI is idempotent on
+   `(claim_id, interchange_control_number)` so re-running is safe).
+2. After Gainwell returns 999 acks, drop them into `ingest/` for
+   the standard `cyclone pull-inbound` flow.
+3. Track the post-submission outcomes:
+   `cyclone resubmissions status [--batch-id=<id>]` — shows
+   `pending_999 / 999_accepted / 999_rejected / 277ca_accepted /
+   paid / denied_again` per claim, joined against `claim_acks`
+   (via the existing SP28/31 auto-link) and `remittances` (via
+   CLP→claim).
+
+What SP39 fixed in the serializer (2010BB loop, payer NM1 segment):
+
+- Empty / `SKCO0` / `CO_BHA` payer.id → normalized to `CO_TXIX`
+  (Gainwell's companion guide requires `CO_TXIX` for the 2010BB
+  NM109; `SKCO0` and `CO_BHA` were silently accepted by Gainwell's
+  pre-processor in late 2025 but their SET-level structural
+  validator rejects them as of 2026-07).
+- Empty / `COHCPF` / `CO_BHA` payer.name → normalized to `CO_TXIX`.
+- Foreign payer IDs (anything else) pass through verbatim with a
+  WARNING log per substitution.
+
 ### Switching from manual → real (and back)
 
 When this host's IP is whitelisted with Gainwell's MFT admin, the

docs/superpowers/specs/2026-07-07-cyclone-2010bb-nm109-fix-design.md

@@ -0,0 +1,334 @@
+# Sub-project 39 — 2010BB NM109 byte defect fix: Design Spec
+
+**Date:** 2026-07-07
+**Status:** Draft, awaiting user sign-off
+**Branch:** `sp39-2010bb-nm109-fix`
+**Aesthetic direction:** No new UI; one serializer normalization helper + regression tests + regen script.
+
+## 1. Scope
+
+On 2026-07-07, an audit of Cyclone's outbound 837P files surfaced a byte-level
+defect in loop 2010BB (payer): the `NM1*PR*2*<name>*****PI*<id>` segment was
+being emitted with `NM109 = "SKCO0"` (and `NM103 = "COHCPF"`) for CO Medicaid
+claims, where Gainwell's MOVEit Transfer SFTP requires `NM109 = "CO_TXIX"`
+or `"CO_BHA"`. The payer rejected four batches at the SET level with
+`"2010BB NM109 must equal CO_TXIX or CO_BHA"`. SP33 already corrected the
+canonical `PayerConfig.co_medicaid()` payer id going forward, but the
+serializer's `_build_payer_block` still emits whatever `claim.payer.id` it
+is handed — so any pre-SP33 `ClaimOutput` rows in the DB whose `raw_json`
+captured `payer.id = "SKCO0"` will round-trip the defect on re-serialize.
+
+A 363-file pre-correction set already exists at
+`ingest/corrected/batch-*/` (regenerated earlier with an upstream
+`raw['payer']['id']` workaround in `regen_837s.py`), plus four original
+outbound 837s at `ingest/tp11525703-837P-20260701*.txt` that retain the
+defect. The SP-N lands a defense-in-depth fix inside the serializer
+itself so the workaround can be removed and so any future caller that
+hands the serializer a stale `SKCO0` (or empty) payer id still emits a
+byte that Gainwell accepts.
+
+**In scope:**
+
+- A new private helper `_normalize_payer_id(payer)` in
+  `backend/src/cyclone/parsers/serialize_837.py` that substitutes
+  `"CO_TXIX"` for `payer.id` whenever the value is empty,
+  `"SKCO0"`, or `"CO_BHA"`. The substitution also aligns
+  `payer.name` to `"CO_TXIX"` when the original `payer.name` was
+  the legacy `"COHCPF"`, `"CO_BHA"`, or empty, so the regenerated
+  `NM103` stays consistent with the substituted `NM109`. Foreign
+  payer IDs (anything not in the substitution set) are emitted
+  verbatim — the helper only normalizes CO Medicaid-shape values.
+- A WARNING log emitted by `_build_payer_block` whenever the
+  helper substitutes a value, carrying the substituted-from and
+  substituted-to strings (one log per affected call; the
+  serializer is invoked once per claim so the volume is bounded
+  by batch size).
+- Hardening `_build_payer_block` to call the helper before
+  delegating to `_build_nm1`.
+- A regression test in `backend/tests/test_serialize_837.py`
+  asserting: `SKCO0` → `CO_TXIX` in NM109 + NM103; empty payer id
+  → `CO_TXIX` in NM109; `CO_BHA` → `CO_TXIX` in NM109 + NM103;
+  a foreign-payer id (e.g. `"OTHER_PAYER"`) preserved unchanged.
+- An idempotent regen script
+  `unbilled-july2026/scripts/regen_corrected_files.py` (alongside
+  the analysis scripts in the sibling project folder) that walks
+  the 363 corrected `ingest/corrected/batch-*/` files,
+  re-parses each one through `parse_837.parse_837_text`, and
+  re-serializes through `serialize_837_for_resubmit`. Output goes
+  to `ingest/corrected-v2/<batch-id>-<N>-claims/` with a single
+  global counter for unique timestamps. Each emitted file is
+  re-validated with `parse_837_text` + a content check that
+  asserts `PI*CO_TXIX` is present and neither `PI*SKCO0` nor
+  `*COHCPF****` appear.
+- A new `resubmissions` table
+  (`claim_id`, `batch_id`, `resubmitted_at`, `source_corrected_path`,
+  `interchange_control_number`, `group_control_number`) plus a
+  matching SQLAlchemy migration under `backend/migrations/0013_*.py`.
+  The table is populated by the existing `cyclone resubmit-rejected-claims`
+  CLI when the operator pushes a corrected file (one row per claim
+  per resubmission; idempotent on `(claim_id, interchange_control_number)`).
+- A new CLI `cyclone resubmissions status [--batch-id=<id>]`
+  that joins `resubmissions` → `claims` → `claim_acks` (via the
+  existing SP28/31 ack-claim auto-link) → `remittances` (via the
+  existing CLP→claim auto-link) and prints a per-claim row:
+  `claim_id | patient | resubmitted_at | 999_status | 277ca_status |
+  payment_status`. Statuses are derived as: `pending_999` if no
+  ack row exists yet, `999_accepted` if a 999 with accept code
+  exists, `999_rejected` if a 999 with reject code exists,
+  `277ca_accepted` / `277ca_rejected` once 277CA arrives,
+  `paid` once a remittance CLP links, `denied_again` if a 277CA
+  reject exists post-resubmission.
+- A note in `docs/RUNBOOK.md` under "Manual SFTP mode" pointing
+  the operator at the new `corrected-v2/` tree + the new
+  `cyclone resubmissions status` CLI after this SP merges, and
+  an inline TODO comment in
+  `submission/core.py:EXPECTED_PAYER_ID` block referencing the
+  future follow-up that will trace where `ClaimOutput.payer.id`
+  gets set to `"SKCO0"` upstream.
+
+**Out of scope:**
+
+- Tracing and patching the upstream setter that populates
+  `ClaimOutput.payer.id = "SKCO0"` (a separate future ticket; this
+  SP only defense-in-depths the serializer). The
+  `_normalize_payer_id` helper masks the upstream bug for any path
+  that goes through `serialize_837` / `serialize_837_for_resubmit`,
+  but does not touch the `claims` table.
+- Re-ingesting the four rejected `ingest/tp11525703-837P-*.txt`
+  files. They are historical artifacts of the rejected
+  submission and remain in `ingest/` for audit only; the operator
+  does not resubmit them. The 363 corrected files in
+  `ingest/corrected/batch-*/` are the resubmission payload, not
+  these four.
+- Auto-resubmitting `ingest/corrected-v2/` via
+  `cyclone resubmit-rejected-claims`. That CLI is the operator's
+  workflow (per RUNBOOK §"Manual SFTP mode") and remains
+  operator-invoked only. This SP only regenerates the corrected
+  files; the operator chooses when to push them.
+- Any change to `PayerConfig.co_medicaid()` (SP33 already
+  canonicalized it) or to the `submission/core.py` validation
+  gate at `EXPECTED_PAYER_ID`. The gate continues to enforce
+  CO_TXIX at submit time and is not relaxed by this SP.
+- A frontend surface for resubmission status. The CLI report is
+  the only operator-facing view; a future UI increment (likely a
+  new Inbox lane or Dashboard KPI) can render the same joined
+  data once the workflow is proven.
+- Touching the existing 999/277CA/remit auto-link code paths.
+  The resubmission status CLI consumes the same joined tables
+  the auto-link already populates; this SP does not modify the
+  auto-link logic itself.
+
+## 2. Decisions (locked during brainstorming)
+
+**D1: Substitute both legacy values — `SKCO0` and `CO_BHA` —
+plus empty — to `CO_TXIX`.**
+
+The operator's policy is "every claim is CO_TXIX". The companion
+guide technically accepts `CO_BHA` for behavioral-health claims,
+but Cyclone is not configured to distinguish BHA claims at
+submission time and the operator's existing manual workflow
+treats `CO_TXIX` as the canonical value for all CO Medicaid
+submits. `CO_BHA` therefore gets normalized the same way `SKCO0`
+does. Foreign payer IDs (any value not in `{empty, "SKCO0",
+"CO_BHA"}`) are emitted verbatim — the helper only normalizes
+CO Medicaid-shape values and must not corrupt a non-CO submit.
+
+**D2: The helper also fixes `payer.name` to match.**
+
+When the helper substitutes `payer.id` to `CO_TXIX`, it also
+substitutes `payer.name` from `"COHCPF"`, `"CO_BHA"`, or empty
+to `"CO_TXIX"`. This keeps NM103 consistent with NM109 and
+matches the byte the operator's manual-mode workflow already
+uses in the corrected files.
+
+**D3: WARNING log per substitution, not per batch.**
+
+`_build_payer_block` is invoked once per claim. One log line per
+substitution is bounded by batch size (≤ 145 lines for the Jun 24
+batches) and is the minimum signal the operator needs to detect
+"this batch still has upstream contamination". Per-batch
+aggregation would hide multi-payer batches (if any are ever
+supported).
+
+**D4: Regression tests live in `test_serialize_837.py`, not a
+new file.**
+
+The existing file already covers NM1 segment shape; appending
+three tests for the helper is the minimal-surface change.
+Matching the `cyclone-tests` convention (sibling test file per
+module).
+
+**D5: Regen script lives in the unbilled-july2026 sibling
+project.**
+
+The 363 files + the four `ingest/tp11525703-837P-*.txt` originals
+are part of the unbilled-july2026 recovery work, not the Cyclone
+core codebase. The sibling project folder already houses the
+analysis scripts; the regen script follows the same convention.
+The script reads `ingest/corrected/batch-*/` (relative to
+Cyclone's repo root, since `ingest/` is shared with the
+production SFTP workflow) and writes to `ingest/corrected-v2/`.
+
+**D6: Re-run into `ingest/corrected-v2/`, not clobber the
+original `batch-*/` tree.**
+
+The 363 corrected files are the postmortem anchor for SP33-era
+SKCO0 contamination. Preserving them under `batch-*/` lets a
+future contributor diff pre-SP39 and post-SP39 outputs and
+confirms the helper is a byte-faithful transformation on every
+input that was already correct. The `corrected-v2/` tree is the
+new submission payload.
+
+**D7: File a follow-up note (in RUNBOOK), don't patch the
+upstream `ClaimOutput.payer.id` setter in this SP.**
+
+The operator picked hard-fix the serializer (Approach 1) over
+trace to root cause (Approach 3). The follow-up ticket belongs
+in a future SP; this SP captures the need via a TODO comment in
+`submission/core.py` and a RUNBOOK entry pointing at the new
+corrected-v2/ tree.
+
+**D8: `resubmissions` is a write-once audit table, not a
+state-machine.**
+
+The table records "this claim was pushed to SFTP at this time
+from this local file" — one row per claim per push. The CLI
+derives status (`pending_999` / `999_accepted` / `paid` / etc.)
+at read-time by joining against `claim_acks` + `remittances`
++ the existing auto-link tables. We do not denormalize status
+onto the `resubmissions` row, because that would create a
+write-coordination problem between the SFTP push and the
+inbound 999/277CA ingestion.
+
+**D9: Post-submission tracking joins reuse the existing
+auto-link data, no new matching logic.**
+
+The 999 ack auto-link (SP28/31) already maps an inbound CLP
+segment back to a `claim_id` via CLP01 + charge + service date.
+Because the corrected-v2 regen preserves the original `claim_id`
+from `raw_json.claim.claim_id`, the 999 ack for a resubmitted
+file auto-links to the original claim row — and the new status
+CLI just joins `resubmissions.claim_id` against `claim_acks.claim_id`
+to surface the result. No new matching code is needed; this SP
+adds the read-side join and the CLI surface, not a new matcher.
+
+**D10: The `resubmit-rejected-claims` CLI inserts `resubmissions`
+rows; we do not add a new push CLI.**
+
+The existing `cyclone resubmit-rejected-claims` (per
+`backend/src/cyclone/cli.py` and the SP33 followup work) is
+the operator's only entry point for pushing corrected files via
+SFTP. We instrument it to insert one `resubmissions` row per
+claim per push (matching the same idempotency key as the SFTP
+upload). No new CLI command for the push side.
+
+## 3. Open questions
+
+None. The operator confirmed the design via the brainstorming
+Q&A on 2026-07-07: hard-fix the serializer with empty/SKCO0/CO_BHA
+substitution to CO_TXIX (no CO_BHA preservation), regen into
+corrected-v2/, add a post-submission tracking table + status CLI
+(joined against existing auto-link data, no new matcher), file
+a follow-up note for upstream tracing, no UI, no auto-resubmit.
+
+## 4. Test impact
+
+Per `cyclone-tests` (autouse conftest at `backend/tests/conftest.py`):
+
+- `backend/tests/test_serialize_837.py` — append three tests
+  targeting `_build_payer_block` via the public
+  `serialize_837_for_resubmit` entry point. Each test constructs
+  a minimal `ClaimOutput` whose `payer.id` is set to the value
+  under test, calls `serialize_837_for_resubmit`, and asserts the
+  emitted `NM1*PR` segment. Tests:
+    - `test_2010bb_normalizes_skco0_to_co_txix` — asserts
+      `SKCO0` → `CO_TXIX` in NM109 and `COHCPF` → `CO_TXIX` in
+      NM103.
+    - `test_2010bb_normalizes_empty_payer_id_to_co_txix` —
+      asserts empty `payer.id` → `CO_TXIX` and empty `payer.name`
+      → `CO_TXIX`.
+    - `test_2010bb_normalizes_co_bha_to_co_txix` — asserts
+      `CO_BHA` (with name `CO_BHA`) → `CO_TXIX` in both NM109
+      and NM103. Plus a fourth assertion that a foreign payer id
+      (e.g. `"OTHER_PAYER"` with name `"OTHER PAYER NAME"`) is
+      emitted verbatim — the helper must not corrupt non-CO
+      submits.
+- `backend/tests/test_resubmissions_cli.py` — new, tests the
+  `cyclone resubmissions status` CLI via `click.testing.CliRunner`.
+  Seeds the `resubmissions` table + `claim_acks` (via the
+  existing fixture pattern) and asserts:
+    - `pending_999` when no ack exists yet.
+    - `999_accepted` after a 999 ack with accept code is linked.
+    - `paid` after a remittance CLP links.
+    - Exit code 0 with empty `resubmissions` table prints
+      "no resubmissions recorded" and exits 0.
+- `backend/tests/test_resubmissions_table.py` — new, tests the
+  SQLAlchemy model + idempotency of `(claim_id, interchange_control_number)`.
+  Asserts schema columns, asserts a second insert with the same
+  key is a no-op (raises IntegrityError or ON CONFLICT DO NOTHING,
+  matching the convention in the rest of `db.py`).
+- `unbilled-july2026/scripts/test_regen_corrected_files.py` —
+  new, sibling test in the unbilled-july2026 project. Uses a
+  fixture of three fake-corrected files (one already correct,
+  one with SKCO0 from a degenerate raw_json, one with CO_BHA)
+  and asserts the regen script emits all three with the expected
+  normalization.
+- Migration test impact: append to
+  `backend/tests/test_db_migrate.py` (or equivalent) a check
+  that migration 0013 creates the `resubmissions` table with
+  the documented columns + idempotency unique constraint.
+- No frontend test impact.
+
+## 5. Files expected to change
+
+- `backend/src/cyclone/parsers/serialize_837.py` — add
+  `_normalize_payer_id(payer)`, call it from `_build_payer_block`,
+  emit WARNING log on substitution.
+- `backend/src/cyclone/db.py` — add `Resubmission` SQLAlchemy
+  model.
+- `backend/migrations/0013_resubmissions.py` — new migration
+  creating the `resubmissions` table.
+- `backend/src/cyclone/cli.py` — instrument
+  `resubmit-rejected-claims` to insert one `Resubmission` row
+  per claim per push (idempotent on
+  `(claim_id, interchange_control_number)`); add the
+  `cyclone resubmissions status` subcommand.
+- `backend/src/cyclone/store/__init__.py` — add
+  `CycloneStore.record_resubmission(...)` and
+  `CycloneStore.find_resubmission_status(...)` helpers; re-export
+  through the facade.
+- `backend/tests/test_serialize_837.py` — append the three
+  regression tests.
+- `backend/tests/test_resubmissions_cli.py` — new CLI tests.
+- `backend/tests/test_resubmissions_table.py` — new model +
+  idempotency tests.
+- `backend/tests/test_db_migrate.py` — append migration test
+  for 0013.
+- `unbilled-july2026/scripts/regen_corrected_files.py` — new
+  sibling-project regen script.
+- `unbilled-july2026/scripts/test_regen_corrected_files.py` —
+  new sibling-project test.
+- `docs/RUNBOOK.md` — append "After SP39 lands" entry under
+  "Manual SFTP mode" pointing the operator at `corrected-v2/`,
+  the new `cyclone resubmissions status` CLI, and the workflow
+  to push via `cyclone resubmit-rejected-claims`.
+- `backend/src/cyclone/submission/core.py` — append TODO comment
+  above `EXPECTED_PAYER_ID` referencing the future upstream-
+  trace ticket.
+- `docs/superpowers/plans/2026-07-07-cyclone-2010bb-nm109-fix.md`
+  — the implementation plan, written after this spec is signed
+  off.
+
+## 6. Auth boundary
+
+The auth boundary is HTTP (login required, bcrypt + HttpOnly
+session cookie); file-system threats remain the local-only threat
+model (SQLCipher at rest, macOS Keychain). The serializer helper
+is invoked server-side by both `serialize_837` (tested via
+`cyclone submit-batch` / `POST /api/submit-batch`, both behind
+HTTP auth) and `serialize_837_for_resubmit` (called by the
+operator's regen workflow and by `/api/inbox/rejected/resubmit`,
+also behind HTTP auth). The sibling-project regen script is
+operator-invoked only and runs against a local file tree; it does
+not touch the HTTP surface or the DB. No change to the threat
+model.