feat(sp33): fix resubmit CLI session reuse + mismatch skip

- payer-mismatch files are now skipped instead of uploaded
- SFTP session persists across files (was reopened and leaked per file)
- --reconnect-every now counts successful uploads, after increment
- drop unused SftpClient instantiation; read each file once
- apply_999_rejections docstring: R/E/X, not R/E

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.gitignore

@@ -38,3 +38,7 @@ claims_output/
 # Brainstorm session artifacts (visual companion mockups, events, server state).
 # Skills under .superpowers/skills/ are committed project-scoped guidance.
 .superpowers/brainstorm/
+
+# SP33+ scratch / production-data ingest. Generated artifacts live
+# here only — the source EDI sits under docs/prodfiles/.
+ingest/

backend/src/cyclone/cli.py

@@ -388,10 +388,6 @@ def backfill_rendering_npi(
     )
 
 
-if __name__ == "__main__":
-    main()
-
-
 # ---------------------------------------------------------------------------
 # SP17: `cyclone backup` subcommands
 #
@@ -403,6 +399,372 @@ if __name__ == "__main__":
 # ---------------------------------------------------------------------------
 
 
+# ---------------------------------------------------------------------------
+# SP33: `cyclone backfill-999-rejections`
+#
+# One-shot replay of the cascade fix in `apply_999_rejections` for any
+# 999 acks already in the DB. Used on the night of 2026-07-02 after
+# Gainwell rejected the four dzinesco batches at the SET level
+# ("2010BB NM109 must equal CO_TXIX") — the 999s were ingested but the
+# pre-SP33 cascade bug didn't flip claim states, so the dashboard's
+# "0/145 accepted" widget was lying.
+#
+# Idempotent: claims already in REJECTED are skipped (counted in
+# `already_rejected`). Safe to re-run.
+# ---------------------------------------------------------------------------
+
+@main.command("backfill-999-rejections")
+@click.option("--dry-run", is_flag=True, default=False,
+              help="Print the would-be state transitions without writing.")
+@click.option("--actor", default="sp33-backfill",
+              show_default=True,
+              help="Audit-log actor tag for the claim.rejected events.")
+def backfill_999_rejections(dry_run: bool, actor: str) -> None:
+    """Replay existing 999 rejections onto already-linked claims (SP33).
+
+    Walks ``claim_acks`` joined with ``claims`` where the link row's
+    ``set_accept_reject_code='R'`` (a SET-level rejection from the 999
+    envelope) and flips the matching claim to ``REJECTED`` — mirroring
+    what the (now-fixed) ``apply_999_rejections`` would have done at
+    ingest time.
+
+    Each flipped claim gets:
+      - ``state`` set to ``REJECTED``
+      - ``state_changed_at`` and ``rejected_at`` set to now
+      - ``rejection_reason`` filled with the 999 AK5 code + SCN
+      - ``payer_rejected_at`` / ``payer_rejected_reason`` /
+        ``payer_rejected_status_code`` filled (the 999 SET-level reject
+        is also a payer-side reject for Inbox-lanes purposes)
+      - one ``claim.rejected`` audit-log event
+
+    Claims already in REJECTED are skipped (counted in the summary).
+    """
+    from datetime import datetime, timezone
+    from sqlalchemy import select, func
+
+    from cyclone import db as db_mod
+    from cyclone.audit_log import AuditEvent, append_event
+
+    db_mod.init_db()
+    now = datetime.now(timezone.utc)
+
+    with db_mod.SessionLocal()() as session:
+        # Pull one representative R-coded link per claim_id (a single
+        # 999 may stamp many AK2 rows against the same claim, so we
+        # collapse via MIN(set_control_number) / MIN(ak2_index) to avoid
+        # firing 1 audit event per duplicate ack row).
+        #
+        # 36777 R-coded rows resolve to 339 unique claims (338 still in
+        # SUBMITTED + 1 already in REJECTED). Grouping in SQL keeps the
+        # in-Python loop small AND emits exactly 1 audit event per claim.
+        rows = session.execute(
+            select(
+                db_mod.Claim.id,
+                db_mod.Claim.state,
+                func.min(db_mod.ClaimAck.set_control_number).label("scn"),
+                func.min(db_mod.ClaimAck.ak2_index).label("ak2"),
+                func.min(db_mod.Ack.ack_code).label("ack_code"),
+            )
+            .join(db_mod.ClaimAck, db_mod.ClaimAck.claim_id == db_mod.Claim.id)
+            .join(db_mod.Ack, db_mod.Ack.id == db_mod.ClaimAck.ack_id)
+            .where(db_mod.ClaimAck.set_accept_reject_code == "R")
+            .where(db_mod.ClaimAck.claim_id.is_not(None))
+            .group_by(db_mod.Claim.id, db_mod.Claim.state)
+        ).all()
+
+        matched = 0
+        already = 0
+        errors = 0
+        for (claim_id, current_state, scn, ak2_idx, ack_code) in rows:
+            if current_state == db_mod.ClaimState.REJECTED:
+                already += 1
+                continue
+
+            if dry_run:
+                matched += 1
+                continue
+
+            claim = session.get(db_mod.Claim, claim_id)
+            if claim is None:
+                errors += 1
+                continue
+
+            claim.state = db_mod.ClaimState.REJECTED
+            claim.state_changed_at = now
+            claim.rejected_at = now
+            claim.rejection_reason = (
+                f"999 AK5={ack_code or 'R'} SCN={scn} ak2={ak2_idx}"
+            )
+            # Mirror the 999 SET-level reject into the payer-rejected
+            # lane so the Inbox sees it as a Payer-Rejected claim too.
+            claim.payer_rejected_at = now
+            claim.payer_rejected_reason = f"999 SET-level reject at SCN={scn}"
+            claim.payer_rejected_status_code = "R"
+
+            append_event(session, AuditEvent(
+                event_type="claim.rejected",
+                entity_type="claim",
+                entity_id=claim_id,
+                payload={"source": "backfill-999-rejections",
+                         "scn": scn, "ak2_index": ak2_idx,
+                         "ack_code": ack_code},
+                actor=actor,
+            ))
+            matched += 1
+
+        if not dry_run:
+            session.commit()
+
+    click.echo(
+        f"matched={matched} already_rejected={already} errors={errors} "
+        f"dry_run={dry_run}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# SP33: `cyclone resubmit-rejected-claims`
+#
+# Push the corrected single-claim 837 files to Gainwell's SFTP ToHPE
+# dir so dzinesco can resubmit the batch. The byte-level
+# SKCO0 -> CO_TXIX fix is assumed to have already been applied (see
+# the SP33 plan §4.4 / `docs/ingest/corrected/`). This CLI just walks
+# the corrected directory, validates each file via the parser, and
+# uploads via the real SftpClient.
+#
+# Idempotent: a file already present on the remote with the same byte
+# size is skipped (counted in `skipped`). Re-runnable after a partial
+# failure without re-uploading files that landed.
+# ---------------------------------------------------------------------------
+
+@main.command("resubmit-rejected-claims")
+@click.option("--ingest-dir", default="ingest/corrected",
+              show_default=True,
+              help="Root dir holding batch-*-claims/ subfolders of fixed .x12 files.")
+@click.option("--actor", default="sp33-resubmit",
+              show_default=True,
+              help="Audit-log actor tag for the clearhouse.submitted events.")
+@click.option("--validate/--no-validate", default=True,
+              help="Parse each file via parse_837 before upload (catches a bad fix).")
+@click.option("--limit", type=int, default=None,
+              help="Stop after checking this many files (smoke-tests). Counts "
+                   "all attempts, not just successful uploads.")
+@click.option("--reconnect-every", type=int, default=50, show_default=True,
+              help="Reconnect SFTP every N uploads to avoid MOVEit's per-session cap.")
+def resubmit_rejected_claims(
+    ingest_dir: str,
+    actor: str,
+    validate: bool,
+    limit: int | None,
+    reconnect_every: int,
+) -> None:
+    """Upload corrected 837 files to the Gainwell SFTP ToHPE dir (SP33).
+
+    Walks every ``batch-*-claims/*.x12`` under ``--ingest-dir`` (default
+    ``./ingest/corrected``), validates each one through ``parse_837``,
+    and uploads via the seeded real-SFTP ``SftpClient``.
+
+    Idempotent: a file already present on the remote with the same
+    byte size is skipped (counted in ``skipped``). Reconnects every
+    ``--reconnect-every`` uploads to avoid MOVEit's silent per-session
+    file cap (observed: ~200 puts/session before silent drops with no
+    exception — see SP33 root-cause notes).
+
+    Doesn't mutate claim state — claims stay in REJECTED until a 999
+    ACK confirms Gainwell accepted the resubmit. Emits one
+    ``clearhouse.submitted`` audit event per successful upload.
+    """
+    import time
+
+    from cyclone import db as db_mod
+    from cyclone.audit_log import AuditEvent, append_event
+    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
+
+    db_mod.init_db()
+    cycl_store.ensure_clearhouse_seeded()
+    clearhouse = cycl_store.get_clearhouse()
+    if clearhouse is None:
+        click.echo("No clearhouse seeded; cannot resolve SFTP block.", err=True)
+        sys.exit(2)
+    sftp_block = clearhouse.sftp_block
+    if sftp_block.stub:
+        click.echo("Clearhouse SFTP block is in stub mode; refusing to upload.", err=True)
+        sys.exit(2)
+    remote_root = sftp_block.paths["outbound"]
+
+    root = Path(ingest_dir).resolve()
+    if not root.exists():
+        click.echo(f"--ingest-dir does not exist: {root}", err=True)
+        sys.exit(2)
+
+    files: list[Path] = []
+    for batch_dir in sorted(root.glob("batch-*-claims")):
+        files.extend(sorted(p for p in batch_dir.glob("*.x12")
+                            if not p.name.startswith("._")))
+    if not files:
+        click.echo(f"No .x12 files under {root}", err=True)
+        sys.exit(2)
+
+    click.echo(f"found {len(files)} files under {root}")
+
+    uploaded = 0
+    skipped = 0
+    failed = 0
+    validated = 0
+    payer_cfg = PayerConfig.co_medicaid()
+    start = time.monotonic()
+    last_progress = start
+
+    # One persistent paramiko session per batch, with periodic
+    # reconnect to dodge MOVEit's silent per-session file cap
+    # (~200 puts/session, no exception — see SP33 root-cause notes).
+    import paramiko
+    from cyclone.secrets import get_secret
+
+    def _open_session() -> tuple[paramiko.SSHClient, paramiko.SFTPClient]:
+        pw = get_secret(sftp_block.auth.get("password_keychain_account", ""))
+        ssh = paramiko.SSHClient()
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        ssh.connect(
+            sftp_block.host, port=sftp_block.port,
+            username=sftp_block.username, password=pw,
+            timeout=15, banner_timeout=15, auth_timeout=15,
+        )
+        return ssh, ssh.open_sftp()
+
+    def _close_session(ssh: paramiko.SSHClient | None) -> None:
+        if ssh is None:
+            return
+        try: ssh.close()
+        except Exception: pass
+
+    ssh: paramiko.SSHClient | None = None
+    sftp: paramiko.SFTPClient | None = None
+
+    for i, src in enumerate(files, 1):
+        if limit is not None and i > limit:
+            break
+
+        content = src.read_bytes()
+
+        # Validate: 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
+            )
+            if mismatch is not None:
+                failed += 1
+                click.echo(
+                    f"PAYER MISMATCH  {src.name}: payer.id={mismatch.payer.id!r} "
+                    f"(expected 'CO_TXIX')", err=True,
+                )
+                continue
+            validated += 1
+
+        local_size = len(content)
+        remote_path = f"{remote_root}/{src.name}"
+
+        attempts = 0
+        ok = False
+        while attempts < 3:
+            attempts += 1
+            if ssh is None:
+                try:
+                    ssh, sftp = _open_session()
+                except Exception as exc:  # noqa: BLE001
+                    click.echo(
+                        f"SFTP CONNECT FAIL  attempt {attempts}: "
+                        f"{exc.__class__.__name__}: {exc}", err=True,
+                    )
+                    continue
+            try:
+                # Idempotency check.
+                try:
+                    rs = sftp.stat(remote_path)  # type: ignore[union-attr]
+                    if rs.st_size == local_size:
+                        skipped += 1
+                        ok = True
+                        break
+                except IOError:
+                    pass  # not on remote yet
+                sftp.put(str(src), remote_path)  # type: ignore[union-attr]
+                ok = True
+                break
+            except Exception as exc:  # noqa: BLE001
+                # Connection died — drop the session and let the next
+                # attempt reopen.
+                _close_session(ssh)
+                ssh, sftp = None, None
+                if attempts >= 3:
+                    click.echo(
+                        f"UPLOAD FAIL  {src.name}: "
+                        f"{exc.__class__.__name__}: {exc}", err=True,
+                    )
+                continue
+
+        if not ok:
+            failed += 1
+            continue
+
+        # Audit (best-effort; if the DB is unavailable we still keep
+        # the file on the wire).
+        try:
+            with db_mod.SessionLocal()() as session:
+                append_event(session, AuditEvent(
+                    event_type="clearhouse.submitted",
+                    entity_type="claim_file",
+                    entity_id=src.name,
+                    payload={"remote_path": remote_path,
+                             "source": "resubmit-rejected-claims",
+                             "size": local_size},
+                    actor=actor,
+                ))
+                session.commit()
+        except Exception as exc:  # noqa: BLE001
+            click.echo(
+                f"audit-log write failed for {src.name}: "
+                f"{exc.__class__.__name__}: {exc}", err=True,
+            )
+
+        uploaded += 1
+
+        # Reconnect periodically to dodge MOVEit's per-session cap.
+        if uploaded % reconnect_every == 0:
+            click.echo(f"reconnecting (after {uploaded} uploads)", err=True)
+            _close_session(ssh)
+            ssh, sftp = None, None
+
+        # Progress every 10s of wall-clock (or at end).
+        now = time.monotonic()
+        if now - last_progress >= 10 or i == len(files):
+            elapsed = now - start
+            rate = uploaded / elapsed if elapsed else 0
+            click.echo(
+                f"progress {i}/{len(files)} uploaded={uploaded} "
+                f"skipped={skipped} failed={failed} rate={rate:.2f}/s "
+                f"elapsed={elapsed:.1f}s", err=True,
+            )
+            last_progress = now
+
+    # Tear down the long-lived session if one is still open.
+    if ssh is not None:
+        _close_session(ssh)
+
+    elapsed = time.monotonic() - start
+    click.echo(
+        f"DONE uploaded={uploaded} skipped={skipped} failed={failed} "
+        f"validated={validated} files_total={len(files)} elapsed={elapsed:.1f}s"
+    )
+
+
 @main.group()
 def backup() -> None:
     """Encrypted DB backup management (SP17)."""
@@ -824,3 +1186,7 @@ def pull_inbound(
         click.echo("download errors:", err=True)
         for e in summary["download_errors"]:
             click.echo(f"  {e}", err=True)
+
+
+if __name__ == "__main__":
+    main()

backend/src/cyclone/handlers/handle_999.py

@@ -90,7 +90,9 @@ def handle(
                 .first()
             )
         rejection_result = apply_999_rejections(
-            session, result, claim_lookup=_lookup,
+            session, result,
+            claim_lookup=_lookup,
+            batch_envelope_index=batch_index,
         )
         if rejection_result.matched:
             for cid in rejection_result.matched:

backend/src/cyclone/inbox_state.py

@@ -33,35 +33,56 @@ def apply_999_rejections(
     parsed_999,
     *,
     claim_lookup: Callable[[str], Claim | None],
+    batch_envelope_index: dict[str, list[str]] | None = None,
 ) -> Apply999Result:
-    """For each set response with code R or E, look up the matching claim and
+    """For each set response with code R, E, or X, look up the matching claim and
     move it to REJECTED. Idempotent on already-rejected claims.
 
     Args:
         session: SQLAlchemy session.
         parsed_999: a ParseResult999 (or any object with .set_responses).
         claim_lookup: callable from patient_control_number → Claim or None.
+            Legacy fallback; rarely hits when batch_envelope_index is present.
+        batch_envelope_index: SP33 — mapping from SET control_number (the 837
+            envelope's ST02) to list of Claim.id for the claims in that SET.
+            Mirrors the SP28 fix in apply_999_acceptances so SET-level
+            rejections correctly cascade across every claim under the SET.
 
     Returns:
         Apply999Result with lists of matched claim ids and orphan PCNs.
     """
     result = Apply999Result()
     now = datetime.now(timezone.utc)
+    index = batch_envelope_index or {}
 
     for sr in parsed_999.set_responses:
         code = sr.set_accept_reject.code
         if code not in ("R", "E", "X"):
             continue
 
-        claim = claim_lookup(sr.set_control_number)
-        if claim is None:
+        # SP33: prefer batch_envelope_index (SCN -> [claim_id]) so a SET-level
+        # rejection correctly flips every claim in the SET. Fall back to
+        # the legacy claim_lookup when the index is empty for this SCN.
+        candidate_ids = index.get(sr.set_control_number, []) or []
+        claims_to_reject: list[Claim] = []
+        if candidate_ids:
+            claims_to_reject = (
+                session.query(Claim)
+                .filter(Claim.id.in_(candidate_ids))
+                .all()
+            )
+        else:
+            legacy = claim_lookup(sr.set_control_number)
+            if legacy is not None:
+                claims_to_reject = [legacy]
+            else:
                 result.orphans.append(sr.set_control_number)
                 continue
 
+        for claim in claims_to_reject:
             if claim.state == ClaimState.REJECTED:
                 # Idempotent: don't double-mutate.
                 continue
-
             claim.state = ClaimState.REJECTED
             claim.state_changed_at = now
             claim.rejected_at = now

backend/src/cyclone/parsers/payer.py

@@ -66,8 +66,8 @@ class PayerConfig(BaseModel):
             # Lenient in v1 — see spec §9 R031.
             require_ref_g1_for_adjustments=False,
             allowed_bht06={"CH"},
-            payer_id="SKCO0",
-            payer_name="COHCPF",
+            payer_id="CO_TXIX",
+            payer_name="CO_TXIX",
             no_patient_loop=True,
             encounter_claim_in_same_batch=False,
             allowed_facility_qualifiers={"B"},

backend/tests/fixtures/co_medicaid_837p.txt

@@ -17,7 +17,7 @@ NM1*IL*1*Balliache*Marianela****MI*P060946~
 N3*1811 PAVILION DR APT 303~
 N4*Montrose*CO*814016072~
 DMG*D8*19590223*F~
-NM1*PR*2*COHCPF*****PI*SKCO0~
+NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
 CLM*t991102984o1c1d*85.40***12:B:1*Y*A*Y*Y~
 REF*G1*3173~
 HI*ABK:R69~
@@ -35,7 +35,7 @@ NM1*IL*1*Barella*Victoria****MI*H582447~
 N3*1900 Kellie DR~
 N4*Montrose*CO*814019524~
 DMG*D8*19570727*F~
-NM1*PR*2*COHCPF*****PI*SKCO0~
+NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
 CLM*t991102984o1c2d*155.76***12:B:1*Y*A*Y*Y~
 REF*G1*3173~
 HI*ABK:R69~

backend/tests/fixtures/co_medicaid_837p_with_renderer.txt

@@ -17,7 +17,7 @@ NM1*IL*1*Balliache*Marianela****MI*P060946~
 N3*1811 PAVILION DR APT 303~
 N4*Montrose*CO*814016072~
 DMG*D8*19590223*F~
-NM1*PR*2*COHCPF*****PI*SKCO0~
+NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
 CLM*t991102984o1c1d*85.40***12:B:1*Y*A*Y*Y~
 REF*G1*3173~
 HI*ABK:R69~
@@ -36,7 +36,7 @@ NM1*IL*1*Barella*Victoria****MI*H582447~
 N3*1900 Kellie DR~
 N4*Montrose*CO*814019524~
 DMG*D8*19570727*F~
-NM1*PR*2*COHCPF*****PI*SKCO0~
+NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
 CLM*t991102984o1c2d*155.76***12:B:1*Y*A*Y*Y~
 REF*G1*3173~
 HI*ABK:R69~

backend/tests/fixtures/minimal_837p.txt

@@ -17,7 +17,7 @@ NM1*IL*1*Doe*John****MI*ABC123~
 N3*456 Member St~
 N4*Denver*CO*80203~
 DMG*D8*19800101*M~
-NM1*PR*2*COHCPF*****PI*SKCO0~
+NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
 CLM*CLM001*100.00***12:B:1*Y*A*Y*Y~
 REF*G1*PA123~
 HI*ABK:Z00~

backend/tests/test_apply_999_rejections.py

@@ -0,0 +1,120 @@
+"""SP33: apply_999_rejections must use batch_envelope_index (mirrors
+SP28's apply_999_acceptances fix) so SET-level 999 rejections cascade
+to claim state transitions.
+
+The pre-SP33 bug: ``claim_lookup(sr.set_control_number)`` passed the
+SET control_number to a PCN-based lookup, so no claim was ever matched
+and rejections silently died. SP33 threads the same
+``batch_envelope_index`` that ``apply_999_acceptances`` already has.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from cyclone.db import ClaimState
+from cyclone.inbox_state import apply_999_rejections
+
+
+def _mk_set_response(code: str = "R", scn: str = "991102994"):
+    """Build a stand-in for ``ParseResult999.set_responses[0]``."""
+    return SimpleNamespace(
+        set_accept_reject=SimpleNamespace(code=code),
+        set_control_number=scn,
+        segment_errors=[],
+    )
+
+
+def _mk_claim(cid: str = "claim-x", state=ClaimState.SUBMITTED):
+    c = MagicMock()
+    c.id = cid
+    c.state = state
+    return c
+
+
+def test_set_level_rejection_cascades_via_envelope_index():
+    """When the SET-level 999 has set_control_number=991102994 and the
+    batch_envelope_index maps it to a list of 3 claim_ids, all 3 should
+    transition to REJECTED.
+    """
+    session = MagicMock()
+    claims = [_mk_claim(f"c-{i}") for i in range(3)]
+    session.query.return_value.filter.return_value.all.return_value = claims
+
+    parsed = SimpleNamespace(set_responses=[_mk_set_response(code="R", scn="991102994")])
+    index = {"991102994": [c.id for c in claims]}
+
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index=index,
+    )
+
+    assert sorted(result.matched) == sorted(c.id for c in claims)
+    assert result.orphans == []
+    session.commit.assert_called_once()
+    for c in claims:
+        # ClaimState.REJECTED is the lowercase 'rejected' string (the enum
+        # value stored in the DB column via SQLAlchemy Enum type).
+        assert c.state == ClaimState.REJECTED
+
+
+def test_set_level_rejection_without_index_still_uses_legacy_lookup():
+    """Backwards compat: when batch_envelope_index is None, the function
+    must fall back to claim_lookup (existing behavior preserved).
+    """
+    session = MagicMock()
+    parsed = SimpleNamespace(set_responses=[_mk_set_response(code="R", scn="991102994")])
+    legacy_claim = _mk_claim("legacy-claim")
+
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda scn: legacy_claim if scn == "991102994" else None,
+    )
+
+    assert result.matched == ["legacy-claim"]
+
+
+def test_no_envelope_index_no_match_becomes_orphan():
+    """Empty index + empty legacy lookup => the SET becomes an orphan."""
+    session = MagicMock()
+    parsed = SimpleNamespace(set_responses=[_mk_set_response(code="R", scn="991102999")])
+
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index={},  # empty index
+    )
+    assert result.matched == []
+    assert "991102999" in result.orphans
+
+
+def test_accepted_999_does_not_reject():
+    """An 'A' code must not transition any claim to REJECTED."""
+    session = MagicMock()
+    parsed = SimpleNamespace(set_responses=[_mk_set_response(code="A", scn="991102994")])
+
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index={"991102994": ["claim-x"]},
+    )
+    assert result.matched == []
+    session.query.assert_not_called()
+    session.commit.assert_not_called()
+
+
+def test_already_rejected_is_idempotent():
+    """Pre-REJECTED claims in the index are left alone (no second commit)."""
+    session = MagicMock()
+    already = _mk_claim("already-rejected", state=ClaimState.REJECTED)
+    session.query.return_value.filter.return_value.all.return_value = [already]
+
+    parsed = SimpleNamespace(set_responses=[_mk_set_response(code="R", scn="991102994")])
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index={"991102994": ["already-rejected"]},
+    )
+    assert result.matched == []
+    assert already.state == ClaimState.REJECTED  # untouched

backend/tests/test_co_medicaid_fixture.py

@@ -36,5 +36,5 @@ def test_subscriber_payer_reused_across_claims():
     result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid())
     assert result.claims[0].billing_provider.npi == "1881068062"
     assert result.claims[1].billing_provider.npi == "1881068062"
-    assert result.claims[0].payer.id == "SKCO0"
-    assert result.claims[1].payer.id == "SKCO0"
+    assert result.claims[0].payer.id == "CO_TXIX"
+    assert result.claims[1].payer.id == "CO_TXIX"

backend/tests/test_co_medicaid_payer_id_emits_CO_TXIX.py

@@ -0,0 +1,33 @@
+"""SP33 regression test: ``PayerConfig.co_medicaid()`` emits CO_TXIX.
+
+The HCPF 837P Companion Guide (June 2025 - Version 2.5) requires
+``NM1*PR NM108=PI, NM109=CO_TXIX`` for the Payer Name loop (2010BB).
+The in-code factory historically emitted ``SKCO0`` which Gainwell
+rejects with ``IK3*NM1*17*2010*8``. This test pins the corrected value.
+See ``docs/superpowers/specs/2026-07-02-cyclone-co-txix-payer-fix-design.md``.
+"""
+from __future__ import annotations
+
+
+def test_co_medicaid_factory_emits_CO_TXIX():
+    """Per HCPF 837P Companion Guide, NM1*PR NM109 must equal CO_TXIX."""
+    from cyclone.parsers.payer import PayerConfig
+    cfg = PayerConfig.co_medicaid()
+    assert cfg.payer_id == "CO_TXIX", (
+        f"co_medicaid().payer_id must be 'CO_TXIX' (HCPF guide), got {cfg.payer_id!r}"
+    )
+    assert cfg.payer_name == "CO_TXIX", (
+        f"co_medicaid().payer_name must be 'CO_TXIX' (matches docs/goodclaim.x12), got {cfg.payer_name!r}"
+    )
+
+
+def test_skco0_not_present_in_factory():
+    """Hard guard: SKCO0 must never be emitted by the factory again."""
+    from cyclone.parsers.payer import PayerConfig
+    cfg = PayerConfig.co_medicaid()
+    assert "SKCO0" not in cfg.payer_id, (
+        f"SKCO0 must not appear in payer_id (HCPF guide says CO_TXIX), got {cfg.payer_id!r}"
+    )
+    assert "SKCO0" not in cfg.payer_name, (
+        f"SKCO0 must not appear in payer_name, got {cfg.payer_name!r}"
+    )

backend/tests/test_parse_837.py

@@ -18,7 +18,7 @@ def test_parse_minimal_fixture_returns_one_claim():
     assert claim.subscriber.last_name == "Doe"
     assert claim.subscriber.first_name == "John"
     assert claim.subscriber.member_id == "ABC123"
-    assert claim.payer.id == "SKCO0"
+    assert claim.payer.id == "CO_TXIX"
     assert claim.claim.frequency_code == "1"
     assert claim.claim.place_of_service == "12"
     assert claim.claim.prior_auth == "PA123"

backend/tests/test_payer.py

@@ -8,8 +8,8 @@ def test_co_medicaid_defaults():
     assert cfg.allowed_claim_frequencies == {1, 7, 8}
     assert cfg.require_ref_g1_for_adjustments is False  # lenient in v1
     assert cfg.allowed_bht06 == {"CH"}
-    assert cfg.payer_id == "SKCO0"
-    assert cfg.payer_name == "COHCPF"
+    assert cfg.payer_id == "CO_TXIX"
+    assert cfg.payer_name == "CO_TXIX"
     assert cfg.no_patient_loop is True
     assert cfg.encounter_claim_in_same_batch is False
 

backend/tests/test_payer_summary.py

@@ -5,13 +5,13 @@ The endpoint is the payer-level aggregate that the drill-down UI's
 denial rate, and the top 5 NPIs by claim volume for one payer_id,
 cached in-process for 60s.
 
-The minimal 837P fixture ships one CLM with ``payer_id="SKCO0"``,
+The minimal 837P fixture ships one CLM with ``payer_id="CO_TXIX"``,
 charge_amount=100.00; the minimal 835 carries one CLP for the same
-claim with total_paid=85.00. So ``/api/payers/SKCO0/summary`` returns
+claim with total_paid=85.00. So ``/api/payers/CO_TXIX/summary`` returns
 ``claim_count >= 1`` after both files are ingested.
 
 Note: the spec calls this ``payer_id`` (the X12 NM1*PR*PI qualifier,
-e.g. ``SKCO0``). It is NOT the configured payer name from
+e.g. ``CO_TXIX``). It is NOT the configured payer name from
 ``config/payers.yaml``. The filter key in the store layer is
 ``Claim.payer_id`` — not the ``payer=`` substring filter used by
 ``/api/claims?payer=...``.
@@ -54,7 +54,7 @@ def client() -> TestClient:
 def seeded_db(client: TestClient):
     """Ingest one minimal 837P + one minimal 835.
 
-    Both fixtures carry ``payer_id="SKCO0"`` so the summary endpoint
+    Both fixtures carry ``payer_id="CO_TXIX"`` so the summary endpoint
     has something to aggregate. ``client`` is yielded back so the
     test can hit the API on the same TestClient that ingested the
     fixtures (parses share the per-test SQLite from conftest).
@@ -77,11 +77,11 @@ def seeded_db(client: TestClient):
 
 
 def test_payer_summary_happy_path(seeded_db: TestClient):
-    """Seeded db has at least one claim for SKCO0 → 200 with the spec shape."""
-    resp = seeded_db.get("/api/payers/SKCO0/summary")
+    """Seeded db has at least one claim for CO_TXIX → 200 with the spec shape."""
+    resp = seeded_db.get("/api/payers/CO_TXIX/summary")
     assert resp.status_code == 200, resp.text
     data = resp.json()
-    assert data["payer_id"] == "SKCO0"
+    assert data["payer_id"] == "CO_TXIX"
     assert "claim_count" in data
     assert "billed_total" in data
     assert "received_total" in data
@@ -99,8 +99,8 @@ def test_payer_summary_unknown_payer_returns_404(client: TestClient):
 
 def test_payer_summary_caches_then_invalidates(seeded_db: TestClient):
     """Two back-to-back calls return identical payloads (in-process cache)."""
-    resp1 = seeded_db.get("/api/payers/SKCO0/summary")
-    resp2 = seeded_db.get("/api/payers/SKCO0/summary")
+    resp1 = seeded_db.get("/api/payers/CO_TXIX/summary")
+    resp2 = seeded_db.get("/api/payers/CO_TXIX/summary")
     assert resp1.status_code == 200
     assert resp2.status_code == 200
     assert resp1.json() == resp2.json()

docs/reference/837p.md

@@ -45,7 +45,7 @@ full walker.
 
 ## CO Medicaid specifics
 
-- Trading partner: `SKCO0` (sender) ↔ `COHCPF` (receiver) on `NM1*PR` / `NM1*40`
+- Trading partners: `COMEDASSISTPROG` (NM1*40 NM109, 1000B receiver) and `CO_TXIX` (NM1*PR NM109, 2010BB payer). See `docs/goodclaim.x12` for a canonical example and the HCPF 837P Companion Guide for the full segment table.
 - `CLM05` is a composite of three components: place of service, facility code
   qualifier, and frequency code (in that order)
   - `CLM05-1` = place of service (any valid CMS POS code)

docs/reference/co-medicaid.md

@@ -9,11 +9,14 @@ Financing (HCPF) requires.
 
 | Role | 837P | 835 |
 |---|---|---|
-| Sender | `SKCO0` | (varies; usually the clearinghouse) |
-| Receiver | `COHCPF` | (varies) |
+| Sender (submitter, `NM1*41`) | `Dzinesco` (TPID `11525703`) | (varies; usually the clearinghouse) |
+| Receiver (`NM1*40`) | `COMEDASSISTPROG` | (varies) |
+| Payer (`NM1*PR`) | `CO_TXIX` (per HCPF 837P Companion Guide) | (varies) |
 
 These appear in the `NM1*PR` (payer) and `NM1*40` (receiver) segments of
-the 837P file.
+the 837P file. The legacy codes `SKCO0` / `COHCPF` are no longer
+accepted by Gainwell (HCPF error: "2010BB NM109 must equal CO_TXIX or
+CO_BHA"); see [SP33](../superpowers/specs/2026-07-02-cyclone-co-txix-payer-fix-design.md).
 
 ## dzinesco's TPID (clearinghouse identity)
 
@@ -115,7 +118,7 @@ ls -la "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/ToHPE/"
 
 ### 837P (claims)
 
-- `NM1*PR N104 = "SKCO0"` (COHCPF)
+- `NM1*PR NM108 = "PI"`, `NM1*PR NM109 = "CO_TXIX"` (per HCPF 837P Companion Guide, June 2025 — Version 2.5). dzinesco submits against this code under dzinesco TPID `11525703`. The legacy trading-partner ID `SKCO0` is no longer accepted as the payer identifier.
 
 ### 835 (remittance)
 

docs/superpowers/plans/2026-07-02-cyclone-co-txix-payer-fix.md

@@ -0,0 +1,768 @@
+# SP33 — Co TXIX Payer Fix Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:executing-plans` (Tyler picked inline execution for SP33, single-operator night-shift). Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Make dzinesco's CO Medicaid 837P submissions emit `NM1*PR*...*PI*CO_TXIX` (per HCPF companion guide), cascade existing 999-rejections into `claims.state=REJECTED`, backfill the 338 rejected claims, and resubmit corrected 837s to Gainwell SFTP — all in one tonight.
+
+**Architecture:** A 2-line payer-config edit + a 3-line `apply_999_rejections` signature fix unlock the cascade. Two new `cyclone.cli` subcommands do the backfill and the resubmit. The doc updates align the in-repo reference with the authoritative HCPF companion guide Tyler pasted. Atomic single merge into `main`.
+
+**Tech Stack:** Python 3.11, SQLAlchemy + SQLite/SQLCipher, paramiko (SFTP), pytest. Frontend untouched.
+
+**Spec:** [`docs/superpowers/specs/2026-07-02-cyclone-co-txix-payer-fix-design.md`](../specs/2026-07-02-cyclone-co-txix-payer-fix-design.md)
+
+**Authoritative source:** CO HCPF "Health Care Claim Professional (837) Transaction Standard Companion Guide" (June 2025 – Version 2.5). Loop 2010BB NM1 Payer Name says `NM108=PI`, `NM109=CO_TXIX` — confirmed by Tyler via chat 2026-07-02. In-repo reference for shape: [`docs/goodclaim.x12`](../../goodclaim.x12).
+
+---
+
+## File structure
+
+| Path | Action | Responsibility |
+|---|---|---|
+| `backend/src/cyclone/parsers/payer.py` | modify (lines 69-70) | `co_medicaid()` factory emits CO_TXIX |
+| `backend/src/cyclone/inbox_state.py` | modify (`apply_999_rejections` signature) | Cascade fix w/ batch_envelope_index |
+| `backend/src/cyclone/handlers/handle_999.py` | modify (caller) | Pass batch_envelope_index through |
+| `backend/src/cyclone/cli.py` | extend (2 new subcommands) | backfill-999-rejections, resubmit-rejected-claims |
+| `backend/tests/test_apply_999_rejections.py` | create | Cascade-fix coverage |
+| `backend/tests/test_serialize_837.py` | extend | Round-trip with CO_TXIX |
+| `docs/reference/837p.md` | modify (line 48) | Doc update |
+| `docs/reference/co-medicaid.md` | modify (line 118) | Doc update |
+
+## Task 0: Pre-flight + branch
+
+**Files:** none (git only)
+
+- [ ] **Step 1: Verify clean tree**
+
+Run: `git status --short`
+Expected: only the spec commit `2e7ad47 docs(spec): SP33 CO TXIX payer fix design`.
+
+- [ ] **Step 2: Create branch off main**
+
+Run: `git checkout -b sp33-co-txix-payer-fix main`
+Expected: `Switched to a new branch 'sp33-co-txix-payer-fix'` (already on `main` at `2e7ad47` or `dfd6542+`).
+
+- [ ] **Step 3: Confirm Docker backend is up**
+
+Run: `docker ps --filter name=cyclone-backend-1 --format '{{.Names}} {{.Status}}'`
+Expected: `cyclone-backend-1 Up X minutes (healthy)`.
+
+## Task 1: Fix `co_medicaid()` factory
+
+**Files:**
+- Modify: `backend/src/cyclone/parsers/payer.py:69-70`
+
+- [ ] **Step 1: Read the current block (sanity)**
+
+Run: `sed -n '67,72p' backend/src/cyclone/parsers/payer.py`
+Expected output:
+
+```
+        allowed_bht06={"CH"},
+        payer_id="SKCO0",
+        payer_name="COHCPF",
+        no_patient_loop=True,
+```
+
+- [ ] **Step 2: Apply the fix (2 lines)**
+
+`search_replace` exact:
+
+OLD:
+```
+        payer_id="SKCO0",
+        payer_name="COHCPF",
+```
+
+NEW:
+```
+        payer_id="CO_TXIX",
+        payer_name="CO_TXIX",
+```
+
+(Drop the bad stale-comment line above it too — the `co_medicaid()` docstring still says "Source: `docs/companionguides/837p.md`" which is wrong; leave the docstring alone in this increment, ship only the data-fix tonight, docstring cleanup is a separate clean-up.)
+
+- [ ] **Step 3: Verify the diff**
+
+Run: `git diff backend/src/cyclone/parsers/payer.py`
+Expected: a 2-line change, exactly the two `payer_id`/`payer_name` lines.
+
+- [ ] **Step 4: Add test**
+
+Create: `backend/tests/test_co_medicaid_payer_id_emits_CO_TXIX.py`:
+
+```python
+"""SP33 regression test: ``PayerConfig.co_medicaid()`` emits CO_TXIX."""
+from __future__ import annotations
+
+
+def test_co_medicaid_factory_emits_CO_TXIX():
+    """Per HCPF 837P Companion Guide, NM1*PR NM109 must equal CO_TXIX."""
+    from cyclone.parsers.payer import PayerConfig
+    cfg = PayerConfig.co_medicaid()
+    assert cfg.payer_id == "CO_TXIX", (
+        f"co_medicaid().payer_id must be 'CO_TXIX' (HCPF guide), got {cfg.payer_id!r}"
+    )
+    assert cfg.payer_name == "CO_TXIX", (
+        f"co_medicaid().payer_name must be 'CO_TXIX' (matches docs/goodclaim.x12), got {cfg.payer_name!r}"
+    )
+```
+
+- [ ] **Step 5: Run the test in the container (since the venv on host isn't trustworthy)**
+
+Run:
+```bash
+docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && .venv/bin/pytest tests/test_co_medicaid_payer_id_emits_CO_TXIX.py -v"
+```
+
+Expected: 1 passed.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add backend/src/cyclone/parsers/payer.py backend/tests/test_co_medicaid_payer_id_emits_CO_TXIX.py
+git commit -m "feat(sp33): PayerConfig.co_medicaid() emits CO_TXIX"
+```
+
+## Task 2: Fix `apply_999_rejections` cascade
+
+**Files:**
+- Modify: `backend/src/cyclone/inbox_state.py` (`apply_999_rejections` signature)
+- Modify: `backend/src/cyclone/handlers/handle_999.py:88-99` (caller)
+
+- [ ] **Step 1: Read current function**
+
+Run: `sed -n '44,90p' backend/src/cyclone/inbox_state.py`
+Expected: see `claim = claim_lookup(sr.set_control_number)` at ~line 56.
+
+- [ ] **Step 2: Apply signature + body change**
+
+OLD (the full function body, lines 44-69):
+
+```python
+def apply_999_rejections(
+    session: Session,
+    parsed_999,
+    *,
+    claim_lookup: Callable[[str], Claim | None],
+) -> Apply999Result:
+    """For each set response with code R or E, look up the matching claim and
+    move it to REJECTED. Idempotent on already-rejected claims.
+
+    Args:
+        session: SQLAlchemy session.
+        parsed_999: a ParseResult999 (or any object with .set_responses).
+        claim_lookup: callable from patient_control_number → Claim or None.
+
+    Returns:
+        Apply999Result with lists of matched claim ids and orphan PCNs.
+    """
+    result = Apply999Result()
+    now = datetime.now(timezone.utc)
+
+    for sr in parsed_999.set_responses:
+        code = sr.set_accept_reject.code
+        if code not in ("R", "E", "X"):
+            continue
+
+        claim = claim_lookup(sr.set_control_number)
+        if claim is None:
+            result.orphans.append(sr.set_control_number)
+            continue
+
+        if claim.state == ClaimState.REJECTED:
+            # Idempotent: don't double-mutate.
+            continue
+
+        claim.state = ClaimState.REJECTED
+        claim.state_changed_at = now
+        claim.rejected_at = now
+        claim.rejection_reason = _build_reason(
+            code, len(sr.segment_errors or [])
+        )
+        result.matched.append(claim.id)
+
+    if result.matched or result.orphans:
+        session.commit()
+
+    return result
+```
+
+NEW:
+
+```python
+def apply_999_rejections(
+    session: Session,
+    parsed_999,
+    *,
+    claim_lookup: Callable[[str], Claim | None],
+    batch_envelope_index: dict[str, list[str]] | None = None,
+) -> Apply999Result:
+    """For each set response with code R or E, look up the matching claim and
+    move it to REJECTED. Idempotent on already-rejected claims.
+
+    Args:
+        session: SQLAlchemy session.
+        parsed_999: a ParseResult999 (or any object with .set_responses).
+        claim_lookup: callable from patient_control_number → Claim or None
+            (legacy fallback; rarely hits with the batch_envelope_index path).
+        batch_envelope_index: SP33 — mapping from SET control_number
+            (the 837 envelope's ST02) to claim.id list for claims in that
+            SET. Mirrors apply_999_acceptances so SET-level rejections
+            correctly cascade.
+
+    Returns:
+        Apply999Result with lists of matched claim ids and orphan PCNs.
+    """
+    result = Apply999Result()
+    now = datetime.now(timezone.utc)
+    index = batch_envelope_index or {}
+
+    for sr in parsed_999.set_responses:
+        code = sr.set_accept_reject.code
+        if code not in ("R", "E", "X"):
+            continue
+
+        # SP33: prefer batch_envelope_index (SCN -> [claim_id]) so a SET-level
+        # rejection correctly flips every claim in the SET. Fall back to
+        # the legacy claim_lookup when the index is empty for this SCN.
+        candidate_ids = index.get(sr.set_control_number, []) or []
+        claims_to_reject = []
+        if candidate_ids:
+            claims_to_reject = (
+                session.query(Claim)
+                .filter(Claim.id.in_(candidate_ids))
+                .all()
+            )
+        else:
+            legacy = claim_lookup(sr.set_control_number)
+            if legacy is not None:
+                claims_to_reject = [legacy]
+            else:
+                result.orphans.append(sr.set_control_number)
+                continue
+
+        for claim in claims_to_reject:
+            if claim.state == ClaimState.REJECTED:
+                # Idempotent: don't double-mutate.
+                continue
+            claim.state = ClaimState.REJECTED
+            claim.state_changed_at = now
+            claim.rejected_at = now
+            claim.rejection_reason = _build_reason(
+                code, len(sr.segment_errors or [])
+            )
+            result.matched.append(claim.id)
+
+    if result.matched or result.orphans:
+        session.commit()
+
+    return result
+```
+
+- [ ] **Step 3: Update the caller in handle_999.py**
+
+OLD (line 88):
+
+```python
+        rejection_result = apply_999_rejections(
+            session, result, claim_lookup=_lookup,
+        )
+```
+
+NEW:
+
+```python
+        rejection_result = apply_999_rejections(
+            session, result,
+            claim_lookup=_lookup,
+            batch_envelope_index=batch_index,
+        )
+```
+
+- [ ] **Step 4: Create the cascade test**
+
+Create: `backend/tests/test_apply_999_rejections.py`:
+
+```python
+"""SP33: apply_999_rejections must use batch_envelope_index (mirrors
+SP28's apply_999_acceptances fix) so SET-level 999 rejections cascade
+to claim state transitions.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from cyclone.db import Claim, ClaimState
+from cyclone.inbox_state import apply_999_rejections, Apply999Result
+
+
+def _mk_set_response(code: str = "R", scn: str = "991102994"):
+    """Build a stand-in for ParseResult999.set_responses[0]."""
+    return SimpleNamespace(
+        set_accept_reject=SimpleNamespace(code=code),
+        set_control_number=scn,
+        segment_errors=[],
+    )
+
+
+def _mk_claim(state: str = "SUBMITTED"):
+    c = MagicMock(spec=Claim)
+    c.id = "claim-" + state.lower()
+    c.state = state
+    return c
+
+
+def test_set_level_rejection_cascades_via_envelope_index():
+    session = MagicMock()
+    session.query.return_value.filter.return_value.all.return_value = [_mk_claim()]
+
+    parsed = SimpleNamespace(
+        set_responses=[_mk_set_response(code="R", scn="991102994")]
+    )
+    index = {"991102994": ["claim-submitted"]}
+
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index=index,
+    )
+
+    assert len(result.matched) == 1
+    assert result.matched[0] == "claim-submitted"
+    assert result.orphans == []
+    session.commit.assert_called_once()
+
+
+def test_set_level_rejection_without_index_still_uses_legacy_lookup():
+    """Backwards compat: when batch_envelope_index is None, the function
+    must fall back to claim_lookup (existing behavior preserved)."""
+    session = MagicMock()
+    parsed = SimpleNamespace(
+        set_responses=[_mk_set_response(code="R", scn="991102994")]
+    )
+    legacy_claim = _mk_claim()
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda scn: legacy_claim if scn == "991102994" else None,
+    )
+
+    assert result.matched == ["claim-submitted"]
+
+
+def test_no_envelope_index_no_match_becomes_orphan():
+    session = MagicMock()
+    parsed = SimpleNamespace(
+        set_responses=[_mk_set_response(code="R", scn="991102999")]
+    )
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index={},  # empty index
+    )
+    assert result.matched == []
+    assert "991102999" in result.orphans
+
+
+def test_accepted_999_does_not_reject():
+    session = MagicMock()
+    parsed = SimpleNamespace(
+        set_responses=[_mk_set_response(code="A", scn="991102994")]
+    )
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index={"991102994": ["claim-x"]},
+    )
+    assert result.matched == []
+    session.commit.assert_not_called()
+
+
+def test_already_rejected_is_idempotent():
+    session = MagicMock()
+    already_rejected = _mk_claim(state="REJECTED")
+    session.query.return_value.filter.return_value.all.return_value = [already_rejected]
+
+    parsed = SimpleNamespace(
+        set_responses=[_mk_set_response(code="R", scn="991102994")]
+    )
+    result = apply_999_rejections(
+        session, parsed,
+        claim_lookup=lambda _pcn: None,
+        batch_envelope_index={"991102994": ["claim-rejected"]},
+    )
+    assert result.matched == []
+    # nothing was mutated on the already-rejected claim
+    assert already_rejected.state == "REJECTED"
+```
+
+- [ ] **Step 5: Run both test files**
+
+Run:
+```bash
+docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && .venv/bin/pytest tests/test_apply_999_rejections.py tests/test_co_medicaid_payer_id_emits_CO_TXIX.py -v"
+```
+
+Expected: 6 passed (1 from Task 1 + 5 from Task 2).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add backend/src/cyclone/inbox_state.py backend/src/cyclone/handlers/handle_999.py backend/tests/test_apply_999_rejections.py
+git commit -m "feat(sp33): apply_999_rejections uses batch_envelope_index"
+```
+
+## Task 3: CLI subcommand `backfill-999-rejections`
+
+**Files:**
+- Modify: `backend/src/cyclone/cli.py` (add subcommand)
+
+- [ ] **Step 1: Locate the cli.py entry point**
+
+Run: `grep -n '@click.command\|def cli\|^@' backend/src/cyclone/cli.py | head -20`
+
+Find an existing subcommand (e.g. `parse-837`) and copy its structure.
+
+- [ ] **Step 2: Add `backfill-999-rejections` subcommand**
+
+Insert after an existing subcommand. The exact insertion point depends on the layout — pick a sensible spot. Use this implementation:
+
+```python
+@click.command(name="backfill-999-rejections")
+@click.option("--dry-run", is_flag=True, default=False,
+              help="Print the would-be changes without writing.")
+@click.option("--actor", default="sp33-backfill",
+              help="Audit log actor for the claim.rejected events.")
+def backfill_999_rejections(dry_run: bool, actor: str):
+    """Replay existing 999 rejections onto already-linked claims.
+
+    SP33. Idempotent — claims already in REJECTED state are skipped.
+
+    Walks claim_acks JOIN acks WHERE set_accept_reject_code='R',
+    then sets claims.state='REJECTED' and fills payer_rejected_* fields.
+    """
+    from sqlalchemy import select
+    from cyclone import db as cycl_db
+    from cyclone.audit_log import AuditEvent, append_event
+    from cyclone.models import Claim as ClaimModel, ClaimAck, Ack
+
+    cycl_db.init_db()
+    with cycl_db.SessionLocal()() as session:
+        rows = session.execute(
+            select(ClaimAck.claim_id, Claim.ack_code, ClaimAck.set_control_number,
+                   ClaimAck.set_accept_reject_code, ClaimAck.ak2_index)
+            .join(Ack, Ack.id == ClaimAck.ack_id)
+            .join(ClaimModel, ClaimModel.id == ClaimAck.claim_id)
+            .where(ClaimAck.set_accept_reject_code == "R")
+        ).all()
+        matched = already = errors = 0
+        from datetime import datetime, timezone
+        now = datetime.now(timezone.utc)
+        for (claim_id, ack_code, scn, arc, ak2_idx) in rows:
+            claim = session.get(ClaimModel, claim_id)
+            if claim is None:
+                errors += 1
+                continue
+            if claim.state == "REJECTED":
+                already += 1
+                continue
+            if not dry_run:
+                claim.state = "REJECTED"
+                claim.state_changed_at = now
+                claim.rejected_at = now
+                claim.rejection_reason = f"999 AK5={ack_code or 'R'} SCN={scn} ak2={ak2_idx}"
+                claim.payer_rejected_at = now
+                claim.payer_rejected_reason = f"999 SET-level reject at SCN={scn}"
+                claim.payer_rejected_status_code = "R"
+                append_event(session, AuditEvent(
+                    event_type="claim.rejected",
+                    entity_type="claim",
+                    entity_id=claim_id,
+                    payload={"source": "backfill-999-rejections", "scn": scn},
+                    actor=actor,
+                ))
+            matched += 1
+        if not dry_run:
+            session.commit()
+        click.echo(
+            f"matched={matched} already_rejected={already} errors={errors} dry_run={dry_run}"
+        )
+```
+
+The exact model class names (`Claim`, `ClaimAck`, `Ack`) and module names (`cyclone.models`, `cyclone.db`, `cyclone.audit_log`) may differ slightly — verify against the existing imports at the top of `cli.py` and adjust.
+
+- [ ] **Step 3: Sanity check (dry-run)**
+
+Run: `docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && .venv/bin/python -m cyclone.cli backfill-999-rejections --dry-run"`
+Expected: `matched=338 already_rejected=0 errors=0 dry_run=True` (or similar; the exact number depends on how many unique claim_ids have R-coded links).
+
+- [ ] **Step 4: Run for real**
+
+Run: `docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && .venv/bin/python -m cyclone.cli backfill-999-rejections"`
+Expected: `matched=338 already_rejected=0 errors=0 dry_run=False`.
+
+- [ ] **Step 5: Verify claim states flipped in the DB**
+
+Run:
+```bash
+docker exec cyclone-backend-1 python3 -c "
+import sqlite3
+conn = sqlite3.connect('/var/lib/cyclone/db/cyclone.db')
+cur = conn.cursor()
+cur.execute('SELECT state, count(*) FROM claims GROUP BY state ORDER BY 2 DESC')
+for row in cur.fetchall(): print(row)
+"
+```
+
+Expected: `('REJECTED', 341)` and `('SUBMITTED', 334)` (338 newly-rejected + the 3 pre-existing rejected = 341; original 672 minus 338 = 334).
+
+- [ ] **Step 6: Commit (CLI only — backfill itself is a live op, no commit for that)**
+
+```bash
+git add backend/src/cyclone/cli.py
+git commit -m "feat(sp33): cli backfill-999-rejections"
+```
+
+## Task 4: CLI subcommand `resubmit-rejected-claims`
+
+**Files:**
+- Modify: `backend/src/cyclone/cli.py` (add subcommand)
+
+- [ ] **Step 1: Add `resubmit-rejected-claims` subcommand**
+
+```python
+@click.command(name="resubmit-rejected-claims")
+@click.option("--payer", default="co_medicaid",
+              help="Payer factory key (default: co_medicaid).")
+@click.option("--ingest-dir", default="/home/tyler/dev/cyclone/ingest",
+              help="Directory holding batch-*-claims/ subfolders (per Tyler's split).")
+@click.option("--actor", default="sp33-resubmit",
+              help="Audit log actor for the clearhouse.submitted events.")
+@click.option("--dry-run", is_flag=True, default=False,
+              help="Replace + validate but do not upload.")
+def resubmit_rejected_claims(payer: str, ingest_dir: str, actor: str, dry_run: bool):
+    """Byte-level SKCO0->CO_TXIX fix + SFTP resubmit of rejected claims.
+
+    SP33. For each claims row in REJECTED state, locate the matching single-
+    claim 837 file in ingest_dir/batch-<batch_id>-<N>-claims/*.x12, replace
+    PI*SKCO0 with PI*CO_TXIX in the bytes, validate via parse_837_text, then
+    upload via SftpClient. Idempotent per claim (resubmit_count gates re-run).
+    """
+    from pathlib import Path
+    import glob
+    from cyclone import db as cycl_db
+    from cyclone.audit_log import AuditEvent, append_event
+    from cyclone.clearhouse import SftpClient
+    from cyclone.parsers.parse_837 import parse_837_text
+
+    cycl_db.init_db()
+    upload_root = "/CO XIX/PROD/coxix_prod_11525703/ToHPE"
+
+    # 1. collect rejected claims
+    with cycl_db.SessionLocal()() as session:
+        from sqlalchemy import select
+        from cyclone.models import Claim as ClaimModel
+        rows = session.execute(
+            select(ClaimModel.id, ClaimModel.batch_id, ClaimModel.patient_control_number,
+                   ClaimModel.resubmit_count, ClaimModel.charge_amount)
+            .where(ClaimModel.state == "REJECTED")
+            .order_by(ClaimModel.batch_id, ClaimModel.id)
+        ).all()
+
+        # 2. discover candidate files
+        files_by_batch = {}
+        for batch_dir in glob.glob(os.path.join(ingest_dir, "batch-*-*-claims")):
+            batch_id = os.path.basename(batch_dir).split("-")[1]  # batch-<id>-<N>-claims
+            files_by_batch[batch_id] = sorted(glob.glob(os.path.join(batch_dir, "*.x12")))
+
+        ok = err = skip = 0
+        for (cid, bid, pcn, prev_resub, charge) in rows:
+            if prev_resub and prev_resub > 0:
+                skip += 1; continue
+
+            candidate_files = files_by_batch.get(bid, [])
+            # Find the file whose CLM01 == this claim's pcn
+            target_path = None
+            for fp in candidate_files:
+                with open(fp, "rb") as fh:
+                    txt = fh.read().decode("ascii", errors="replace")
+                if pcn in txt:
+                    target_path = fp
+                    break
+            if target_path is None:
+                click.echo(f"  no-file  claim={cid} pcn={pcn}", err=True)
+                err += 1
+                continue
+
+            # Byte-level fix
+            with open(target_path, "rb") as fh:
+                raw = fh.read()
+            if b"PI*SKCO0" not in raw:
+                click.echo(f"  no-replace  claim={cid} file={os.path.basename(target_path)}", err=True)
+                err += 1; continue
+            fixed = raw.replace(b"PI*SKCO0", b"PI*CO_TXIX")
+
+            # Validate
+            try:
+                parsed = parse_837_text(fixed.decode("ascii"))
+            except Exception as exc:
+                click.echo(f"  parse-fail  claim={cid} {exc.__class__.__name__}: {exc}", err=True)
+                err += 1; continue
+
+            # Upload
+            remote_name = os.path.basename(target_path)
+            if not dry_run:
+                # Write the fixed bytes to a sibling tmp file for the upload
+                tmp_path = target_path + ".fixed"
+                with open(tmp_path, "wb") as fh:
+                    fh.write(fixed)
+                try:
+                    SftpClient().put(tmp_path, f"{upload_root}/{remote_name}")
+                finally:
+                    os.unlink(tmp_path)
+                claim = session.get(ClaimModel, cid)
+                claim.resubmit_count = (claim.resubmit_count or 0) + 1
+                append_event(session, AuditEvent(
+                    event_type="clearhouse.submitted",
+                    entity_type="claim",
+                    entity_id=cid,
+                    payload={"file": remote_name, "scn": parsed.envelope.control_number},
+                    actor=actor,
+                ))
+                session.commit()
+            click.echo(f"  ok  claim={cid} file={remote_name}")
+            ok += 1
+
+        click.echo(f"\nDONE ok={ok} err={err} skip={skip} dry_run={dry_run}")
+```
+
+Imports needed: `import os` (glob, path). Confirm the `SftpClient` constructor — it may take an SFTP block or a config; check existing usage in `backend/src/cyclone/api.py` (the `/api/clearhouse/submit` endpoint) and mirror it.
+
+- [ ] **Step 2: Run with --dry-run first**
+
+Run: `docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && CYCLONE_INGEST_DIR=/tmp/ingest_passthrough .venv/bin/python -m cyclone.cli resubmit-rejected-claims --dry-run"`
+Expected: a line per claim plus a `DONE ok=338 err=0 skip=0 dry_run=True` summary.
+
+- [ ] **Step 3: Run for real (upload)**
+
+Mount /home/tyler/dev/cyclone/ingest into the container so the `--ingest-dir` resolves correctly. Easiest: copy the batch-* subdirs into the container:
+
+```bash
+docker exec cyclone-backend-1 mkdir -p /tmp/ingest_split
+tar -C /home/tyler/dev/cyclone/ingest -cf - 'batch-*-*-claims' | docker exec -i cyclone-backend-1 tar -xf - -C /tmp/ingest_split
+```
+
+Then: `docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && .venv/bin/python -m cyclone.cli resubmit-rejected-claims --ingest-dir /tmp/ingest_split"`
+
+Expected: `DONE ok=338 err=0 skip=0 dry_run=False`.
+
+- [ ] **Step 4: Verify the upload landed**
+
+Connect to SFTP and check the latest 338 files in `/ToHPE/`:
+
+```bash
+docker exec cyclone-backend-1 bash -c "
+import paramiko, time
+ssh = paramiko.SSHClient(); ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+ssh.connect('mft.gainwelltechnologies.com', username='colorado-fts\\\\coxix_prod_11525703',
+            password=\$(cat /run/secrets/cyclone_sftp_password))
+sftp = ssh.open_sftp()
+files = sorted(sftp.listdir_attr('/CO XIX/PROD/coxix_prod_11525703/ToHPE'),
+               key=lambda a: a.st_mtime)[-10:]
+for f in files: print(f.filename, time.ctime(f.st_mtime))
+"
+```
+
+Expected: ~10 837P files dated 2026-07-02 (or now).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/src/cyclone/cli.py
+git commit -m "feat(sp33): cli resubmit-rejected-claims"
+```
+
+## Task 5: Doc updates
+
+**Files:**
+- Modify: `docs/reference/837p.md:48`
+- Modify: `docs/reference/co-medicaid.md:118`
+
+- [ ] **Step 1: Update `docs/reference/837p.md` line 48**
+
+OLD line:
+```
+- Trading partner: `SKCO0` (sender) ↔ `COHCPF` (receiver) on `NM1*PR` / `NM1*40`
+```
+
+NEW line:
+```
+- Trading partners: `COMEDASSISTPROG` (NM1*40 NM109, 1000B receiver) and `CO_TXIX` (NM1*PR NM109, 2010BB payer). See `docs/goodclaim.x12` for a canonical example and the HCPF 837P Companion Guide for the full segment table.
+```
+
+- [ ] **Step 2: Update `docs/reference/co-medicaid.md` line 118**
+
+OLD:
+```
+- `NM1*PR N104 = "SKCO0"` (COHCPF)
+```
+
+NEW:
+```
+- `NM1*PR NM108 = "PI"`, `NM1*PR NM109 = "CO_TXIX"` (per HCPF 837P Companion Guide, June 2025 — Version 2.5). dzinesco submits against this code under dzinesco TPID `11525703`. The legacy trading-partner ID `SKCO0` is no longer accepted as the payer identifier.
+```
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add docs/reference/837p.md docs/reference/co-medicaid.md
+git commit -m "docs(sp33): update reference docs to CO_TXIX (per HCPF 837P Companion Guide)"
+```
+
+## Task 6: Verify + final test run
+
+**Files:** none (verification only)
+
+- [ ] **Step 1: Full test suite**
+
+Run: `docker exec cyclone-backend-1 bash -c "cd /var/lib/cyclone && .venv/bin/pytest -x -q"`
+Expected: all tests pass. The 4 affected test files (this plan + Spec §6 list) all green.
+
+- [ ] **Step 2: Smoke-test the live DB dashboard widget**
+
+Run:
+```bash
+curl -fsS http://localhost:8000/api/batches | python3 -m json.tool | head -30
+```
+
+Expected: the dashboard reflects the freshly-REJECTED batches (acceptedCount=0 for the 4 affected batches — this is now correct! They were rejected by Gainwell, and the cascade now shows that).
+
+- [ ] **Step 3: Atomic merge**
+
+```bash
+git checkout main
+git merge --no-ff sp33-co-txix-payer-fix -m "merge: SP33 Co TXIX payer fix + 999 cascade repair into main"
+git log --oneline -3
+```
+
+Expected: a single merge commit `merge: SP33 ...` on top of `dfd6542` (current main tip pre-SP33), with the SP33 branch's commits visible as parents.
+
+## Acceptance checklist (must all be true before claiming "done")
+
+- [ ] `cyclone -m cyclone.cli backfill-999-rejections` reports `matched=338 already_rejected=0 errors=0`
+- [ ] `claims.state` count: REJECTED ≥ 341, SUBMITTED ≤ 334
+- [ ] 338 new 837 files visible on SFTP `/CO XIX/PROD/cox17_prod_11525703/ToHPE/` with `CO_TXIX` in their bytes
+- [ ] `claims.resubmit_count > 0` for the 338 claims
+- [ ] `audit_log` shows 338 `claim.rejected` + 338 `clearhouse.submitted` events with actor=`sp33-*`
+- [ ] Dashboard "Recent batches" widget for the 4 affected batches shows the same `acceptedCount` it was showing — but now it's accurate (0 rejected by Gainwell, not 0 accepted by Gainwell)
+
+## Rollback
+
+If anything breaks and SP33 needs to back out:
+
+1. `git revert -m 1 <merge-sha>` on `main` — produces a fresh revert commit, preserves history.
+2. The 2 cli subcommands revert cleanly (pure additions).
+3. To un-reject the 338 claims after the revert, run a sister SQL: `UPDATE claims SET state='SUBMITTED', rejection_reason=NULL, rejected_at=NULL, payer_rejected_at=NULL, payer_rejected_reason=NULL, payer_rejected_status_code=NULL WHERE resubmit_count > 0;`
+4. The 338 837 files already on SFTP /ToHPE/ cannot be unsent — Gainwell will pick them up regardless. Discretion required.

docs/superpowers/specs/2026-07-02-cyclone-co-txix-payer-fix-design.md

@@ -0,0 +1,185 @@
+# Sub-project 33 — Co TXIX Payer Fix: Design Spec
+
+ **Date:** 2026-07-02
+ **Status:** Draft, awaiting user sign-off
+ **Branch:** `sp33-co-txix-payer-fix`
+ **Aesthetic direction:** No new UI; production hotfix to emission
+
+## 1. Scope
+
+In scope:
+
+- Change the canonical CO Medicaid payer identifier emitted in `NM1*PR NM108=PI*NM109` from `SKCO0` to `CO_TXIX`, in both the in-code `PayerConfig.co_medicaid()` factory and any other write path that derives the segment from `claim.payer.id`.
+- Update the in-repo companion-guide references (`docs/reference/837p.md`, `docs/reference/co-medicaid.md`) to reflect the corrected value. `docs/goodclaim.x12` is already correct and stays the source of truth.
+- Fix `cyclone.inbox_state.apply_999_rejections` to use `batch_envelope_index` (mirroring the SP28 fix to `apply_999_acceptances`) so SET-level 999 acks correctly cascade to claim-state transitions.
+- One-shot backfill against the live DB: for each claim already linked via `claim_acks` to a 999 ack with `set_accept_reject_code='R'`, set `claims.state='REJECTED'`, fill `rejection_reason`, `rejected_at`, `payer_rejected_reason`, `payer_rejected_at`, and emit a `claim.rejected` audit event. Idempotent.
+- Resubmit the 338 corrected single-claim 837 files (already split into `ingest/batch-*-claims/*.x12` by the operator) by replacing the `SKCO0` literal in their bytes with `CO_TXIX`, then uploading to `/CO XIX/PROD/coxix_prod_11525703/ToHPE/` via the existing paramiko-backed `SftpClient`.
+
+Out of scope:
+
+- No new R-coded validator rule (the existing `R100_payer_id_matches` already warns on mismatch; it auto-converges once `co_medicaid()` returns `CO_TXIX`).
+- No `CO_BHA` (behavioral-health) variant in this increment. The HCPF 999 lists `CO_TXIX OR CO_BHA` as both accepted; this increment picks `CO_TXIX` as the single value for dzinesco's three Home-Health NPIs. A future SP may add a per-claim override if behavioral-health submission volumes warrant it.
+- No changes to other payer configs (only `co_medicaid` is touched).
+- No schema migration. The existing `claims.payer_id` column already accepts any string; `claims.resubmit_count` already exists and is incremented; no new column is added.
+- No frontend UI changes. The dashboard already surfaces claim states; once the backfill flips them, the "0 accepted" widget will reflect reality without code change.
+
+## 2. Context (why now)
+
+On 2026-07-01 at 16:29 MT, dzinesco submitted four 837P batches (145+95+25+73 = 338 claims) to Gainwell's SFTP at `mft.gainwelltechnologies.com`. On 2026-07-02, Gainwell returned 999 acknowledgments for each, all `AK9=R` with the SET-level error:
+
+```
+IK3*NM1*17*2010*8
+IK5*R*I5
+```
+
+…where "2010BB NM109 must equal CO_TXIX or CO_BHA".
+
+The in-repo reference (`docs/reference/837p.md:48`, `docs/reference/co-medicaid.md:118`) and the in-code `PayerConfig.co_medicaid()` factory (`backend/src/cyclone/parsers/payer.py:69-70`) both claim `SKCO0` is the correct value. The authoritative sample file (`docs/goodclaim.x12`, line 4) shows `NM1*PR*2*CO_TXIX*****PI*CO_TXIX` — confirming `CO_TXIX` is correct. Tyler confirmed this in chat on 2026-07-02: "the problem is we arent changing SKCO0 to CO_TXIX".
+
+A second, separate bug surfaced during the same investigation: `apply_999_rejections` (`backend/src/cyclone/inbox_state.py:56`) passes the SET control number to a `claim_lookup` that queries by `patient_control_number`. As a result, even when a 999 ack correctly rejects a SET, no claim state ever flips. This bug pre-dates SP33 but is fixed here because (a) the SP28 fix already exists for `apply_999_acceptances` and the same fix is a small, surgical change, and (b) the dashboard "0/145 accepted" widget can only reflect reality once the cascade works.
+
+## 3. Decisions (locked during brainstorming)
+
+1. **NM109 = `CO_TXIX`, single value.** No per-claim rule. dzinesco's three NPIs are all Home Health (taxonomy `251E00000X`); none are behavioral-health-only providers. Behavioral-health claims (if any exist) would currently emit `CO_TXIX` and Gainwell would still accept per the 999's "OR CO_BHA" wording. Confirmed with Tyler 2026-07-02.
+2. **Fix the cascade bug in the same increment.** The 999 cascade is the same severity class as the 999 emission bug; splitting into a separate SP would mean a second hotfix in days. Confirmed with Tyler via Option A selection.
+3. **Backfill is a one-shot CLI subcommand.** A new `python -m cyclone.cli backfill-999-rejections` invocation, idempotent (claims already in `REJECTED` state are skipped). Lives in `cli.py` and a small helper module. Confirmed with Tyler via Option A.
+4. **Resubmit uses Tyler's pre-split files.** Tyler already produced 338 single-claim 837 files in `ingest/batch-*-claims/*.x12` (timestamps 16:41:43 on 2026-07-01). The resubmit path performs a deterministic `SKCO0 → CO_TXIX` byte replacement on each file, validates via the existing parser, then uploads via `SftpClient.put()`. Confirmed with Tyler via Option A (Step 6).
+5. **No new spec-on-the-fly for the bytecode fix.** The `SKCO0 → CO_TXIX` literal replacement is a single-pass `bytes.replace` scoped to the `NM1*PR` segment of each file. Anything more sophisticated would be over-engineered for a hotfix.
+
+## 4. Architecture
+
+### 4.1 Source fix (`feat(sp33): PayerConfig.co_medicaid() emits CO_TXIX`)
+
+One file, two lines. `backend/src/cyclone/parsers/payer.py:69-70`:
+
+```
+payer_id="SKCO0",    →  payer_id="CO_TXIX",
+payer_name="COHCPF",  →  payer_name="CO_TXIX",
+```
+
+The serializer (`backend/src/cyclone/parsers/serialize_837.py:488`) reads these fields directly and emits `NM1*PR` accordingly — no serializer change required. The 999 cascade (independent fix, see §4.2) is already gated on `apply_999_rejections`, which queries against `claim.payer.id` once the in-DB claims carry the new value.
+
+### 4.2 Cascade fix (`feat(sp33): apply_999_rejections uses batch_envelope_index`)
+
+`backend/src/cyclone/inbox_state.py:55-65`:
+
+```
++ from cyclone.handlers._ack_id import batch_envelope_index   # already exists
+  def apply_999_rejections(session, parsed_999, *, claim_lookup, batch_envelope_index=None):
+      ...
+-     claim = claim_lookup(sr.set_control_number)
++     # SCN → list of claim_ids via batch_envelope_index; fall back to direct claim_lookup (rare)
++     candidates = (batch_envelope_index or {}).get(sr.set_control_number, [])
++     for cid in candidates:
++         claim = session.get(Claim, cid)
+```
+
+The new signature mirrors the SP28 pattern in `apply_999_acceptances`. Backwards-compatible: when `batch_envelope_index=None`, falls back to the old (broken) behavior, but every caller in `handle_999.handle()` will pass the index.
+
+The handler at `backend/src/cyclone/handlers/handle_999.py:88-99` updates in lockstep:
+
+```
+- rejection_result = apply_999_rejections(session, result, claim_lookup=_lookup)
++ rejection_result = apply_999_rejections(
++     session, result, claim_lookup=_lookup, batch_envelope_index=batch_index,
++ )
+```
+
+### 4.3 Backfill subcommand (`feat(sp33): cli backfill-999-rejections`)
+
+New `cli.py` subcommand:
+
+```
+python -m cyclone.cli backfill-999-rejections [--actor <name>] [--dry-run]
+```
+
+Algorithm (single transaction, single `SessionLocal`):
+
+1. `SELECT ca.claim_id, ca.set_control_number, ca.set_accept_reject_code, a.ack_code, a.raw_json FROM claim_acks ca JOIN acks a ON a.id=ca.ack_id WHERE ca.set_accept_reject_code='R' AND ca.linked_at IS NOT NULL`
+2. For each `claim_id`, load `Claim` via `session.get`. Skip if already `REJECTED` (idempotent). Otherwise set `state='REJECTED'`, `state_changed_at=now`, `rejected_at=now`, `rejection_reason=...`, `payer_rejected_at=now`, `payer_rejected_reason=...`, `payer_rejected_status_code='R'`, then append `AuditEvent(event_type='claim.rejected', actor=<name>)`.
+3. Commit. Print `(matched=N already_rejected=M errors=K)` summary.
+
+For the 338 current claims, expected output: `matched=338 already_rejected=0 errors=0`.
+
+### 4.4 Resubmit (`feat(sp33): cli resubmit-rejected-claims`)
+
+New CLI subcommand:
+
+```
+python -m cyclone.cli resubmit-rejected-claims [--payer co_medicaid] [--ingest-dir <path>]
+```
+
+Defaults: `--ingest-dir /home/tyler/dev/cyclone/ingest`. Tyler pre-split the 338 single-claim 837 files into per-batch directories named `batch-<batch_id>-<N>-claims/`, where `<batch_id>` matches `claims.batch_id` and `<N>` is the claim count. The exact filenames match the SFTP outbound naming convention (`TPID-837P-yyyymmddhhmmssSSS-1of1.x12`) and are already unique-by-timestamp.
+
+Algorithm:
+
+1. Discover all `claims` rows where `state='REJECTED'` (after the §4.3 backfill lands). Group by `batch_id`.
+2. For each `claim_id` in `REJECTED`, locate the matching `_837P-*.x12` file in `ingest/batch-<batch_id>-<N>-claims/`. Match by directory naming + per-claim `patient_control_number` declared in the file's CLM01 segment (re-parse each file once, cache the CLM01 list keyed by file).
+3. For each file: `new_bytes = old_bytes.replace(b'PI*SKCO0', b'PI*CO_TXIX')`. Validate via `parse_837_text(new_bytes)`. If validation produces any NEW error (compared to the same parse of the pre-replacement bytes), halt with the file path + error diff so the operator can inspect.
+4. Upload via `cyclone.clearhouse.SftpClient.put(local_path, remote_path)` to `/CO XIX/PROD/coxix_prod_11525703/ToHPE/` using the EXISTING filename. SFTP client retains the paramiko session across files (already supported per SP16).
+5. After successful upload, increment `claims.resubmit_count` (existing column, default 0) and emit one `clearhouse.submitted` audit event per file. `claims.state` remains `REJECTED` — the `resubmit_count > 0` flag is sufficient for the UI to distinguish "rejected-and-needs-resubmit" from "rejected-and-already-resubmitted". No new claim state is introduced.
+
+For the 338 current claims: this runs once. Output filenames already reflect dzinesco's outbound SFTP naming scheme and are unique.
+
+### 4.5 Doc updates (`docs(spec): SP33 update reference docs to CO_TXIX`)
+
+Two files, two lines:
+
+- `docs/reference/837p.md:48` — replace
+  `Trading partner: 'SKCO0' (sender) ↔ 'COHCPF' (receiver) on 'NM1*PR' / 'NM1*40'`
+  with
+  `Trading partner: 'COMEDASSISTPROG' (receiver, NM1*40) and payer 'CO_TXIX' (NM1*PR, NM108=PI, NM109=CO_TXIX) — see docs/goodclaim.x12 for the canonical example.`
+
+- `docs/reference/co-medicaid.md:118` — replace
+  `'NM1*PR N104 = "SKCO0"' (COHCPF)`
+  with
+  `'NM1*PR NM109 = "CO_TXIX"' (CO_TXIX)` and add a sentence: `For behavioral-health claims only, use "CO_BHA"; dzinesco's current submission volumes don't include behavioral-only providers.`
+
+`docs/goodclaim.x12` stays unchanged (already correct).
+
+## 5. Failure modes
+
+| Failure | Detection | Mitigation |
+|---|---|---|
+| Serializer change breaks 999+1 backwards compat | `pytest backend/tests/test_serialize_837.py` | Layered test: pre-fix `goodclaim.x12` parse still passes; post-fix parse still passes (the `payer.id` change is in NM109 only) |
+| Cascade fix changes behavior for already-accepted 999s | Unit test: feed 999 with `AK5=A`; assert no claim state changes | `apply_999_acceptances` is unchanged; only `apply_999_rejections` is touched |
+| Backfill races with live scheduler writing new claim_acks | Single-transaction SELECT-then-UPDATE; row-level retry on `database is locked` | Wrap the per-claim update in a short retry loop (3 attempts, 100ms backoff) |
+| Resubmit overwrites a not-yet-acknowledged file on SFTP | SFTP-side ETags / names | Use Tyler's pre-split filenames verbatim (already unique-by-timestamp) |
+| SKCO0 still in some legacy 837 in the SFTP outbound dir | Out of scope — dzinesco is no longer submitting new 837s against the buggy factory | None; historical files on SFTP aren't retransmitted |
+
+## 6. Test impact
+
+- `backend/tests/test_apply_999_rejections.py` — **new**. Covers:
+  - SET-level `AK9=R` with a `batch_envelope_index` containing 3 claim_ids → all 3 transition to REJECTED, idempotent on rerun
+  - SET-level `AK9=A` with empty envelope index → 0 matches (no false positives)
+  - No `batch_envelope_index` passed → falls back to old behavior (passes SCN as PCN, returns no matches for non-PCN-keyed claims)
+- `backend/tests/test_serialize_837.py` — extend: assert `co_medicaid()` round-trips through `serialize_837 → parse_837` and the parsed `payer.id` equals `CO_TXIX`
+- `backend/tests/test_payer_config_loading.py` — confirm YAML still binds `CO_TXIX` and no in-code change breaks the bootstrap path
+- `backend/tests/test_inbox_state_apply_999.py` (if it exists) — adjust for the new optional arg
+
+No frontend changes → no Vitest/RTL impact.
+
+## 7. Branch / merge plan
+
+- Branch: `sp33-co-txix-payer-fix`
+- Atomic merge commit. PR title: `SP33 Co TXIX payer fix + 999 cascade repair`
+- Commit prefix schedule:
+  - `docs(spec): SP33 CO TXIX payer fix design`
+  - `docs(plan): SP33 CO TXIX payer fix implementation`
+  - `feat(sp33): PayerConfig.co_medicaid() emits CO_TXIX`
+  - `feat(sp33): apply_999_rejections uses batch_envelope_index`
+  - `test(sp33): add test_apply_999_rejections with batch_envelope_index`
+  - `feat(sp33): cli backfill-999-rejections`
+  - `feat(sp33): cli resubmit-rejected-claims`
+  - `docs(sp33): update docs/reference/{837p,co-medicaid}.md to CO_TXIX`
+  - `merge: SP33 Co TXIX payer fix + 999 cascade repair into main`
+
+## 8. Rollback
+
+If the SP33 merge breaks the live 837 flow:
+
+1. Revert the single merge commit (`git revert <merge-sha>` — produces a fresh commit, preserves history).
+2. The two doc edits revert cleanly (markdown only).
+3. The cascade-fix revert restores the SP28-era broken-but-not-dead behavior (claims stay SUBMITTED — same as today).
+4. The backfill is fully reversible: a sister CLI subcommand `python -m cyclone.cli reset-rejection-backfill [--batch-id <id>]` can un-REJECT the 338 claims (sets `state='SUBMITTED'`, clears `rejection_reason`, `rejected_at`, `payer_rejected_*`). Out of scope to ship in SP33, but the column data is fully captured in audit log so rollback is always possible.
+5. The resubmit cannot be undone (the corrected 837 files are on SFTP), but a `clearhouse` audit event is recorded for every upload so the operator can identify and re-trigger if needed.

docs/superpowers/specs/2026-07-02-cyclone-rendering-npi-extraction-design.md

@@ -62,14 +62,22 @@ Effective algorithm unchanged: 2-of-3 `{PCN, charge, NPI}` still fires when any
 
 New subcommand: `python -m cyclone.cli backfill-rendering-npi`.
 
-Behavior:
-1. For every `claim_batches.inbound_path`, re-parse the 837p file with the new parser; for every claim in the batch, write `Claim.rendering_provider_npi = parsed_value` **only when the column is currently `NULL`** (idempotent).
-2. For every `remittances.inbound_path`, re-parse the 835 file with the new parser; for every remit, write both the per-segment NPIs (`ClaimPayment.service_provider_npi`) and the aggregated `Remittance.rendering_provider_npi`.
-3. Run `reconcile.run()` once over all open pairs at the end so the NPI arm can fire retroactively.
-4. Log per-batch counts (parsed, updated, skipped).
-5. Exit code 0 on success, 2 on file-level failure (no batch updated), 1 on unexpected exception.
+**Why explicit file args.** The `Batch` ORM does not retain a column for the original on-disk ingest path — the live ingest pipeline reads the upload body, parses, writes rows, and never persists the bytes. So there is no `claim_batches.inbound_path` / `remittances.inbound_path` to "look up" and re-parse from. The operator must point the CLI at the files (or the directory that holds them) that need backfill. Re-running ingest end-to-end would be the alternative, but it would re-`INSERT` claim rows and is a destructive operation; backfill is intentionally a read-reparse-targeted-`UPDATE`.
 
-Idempotent: re-running on a fully-populated DB is a no-op (typed columns not overwritten; reconcile already matched pairs skipped via `Claim.matched_remittance_id IS NOT NULL`).
+Flags:
+- `--file PATH` (repeatable) — one or more specific X12 files to re-parse. Each path is validated to exist before parsing.
+- `--input-dir DIR` — directory to scan one level deep for `*.txt` / `*.edi` / `*.x12` files. Falls back to `$CYCLONE_BACKFILL_INPUT_DIR` if neither `--file` nor `--input-dir` is passed.
+- `--type {837p,835}` (optional) — pin the parser. **When omitted, each file's transaction kind is auto-sniffed** (filename hint + ISA/ST prefix) so a mixed directory can be processed in one pass.
+
+Behavior:
+1. For each file resolved by `--file` / `--input-dir`, re-parse with the new parser (T4 wiring).
+2. **837p**: for every claim in the batch, write `Claim.rendering_provider_npi = parsed_value` **only when the column is currently `NULL`** (idempotent — already-populated rows are left alone).
+3. **835**: for every remit, write `Remittance.rendering_provider_npi` (typed column) **only when `NULL`**, and persist the per-segment `service_provider_npi` values into `Remittance.raw_json["service_provider_npis"]` (D5).
+4. After all files are processed, run `reconcile.run()` **once across every 835 Batch** so the D6 NPI arm can fire retroactively on newly-populated pairs.
+5. Log per-file counts (parsed, updated, skipped). Emit one-line summary to stdout: `claims_updated=N remits_updated=M files_processed=K files_skipped=J`.
+6. Exit code 0 on success (zero populated rows is still exit 0), 2 on file-level failure (no file processed), 1 on unexpected exception.
+
+Idempotent: re-running on a fully-populated DB is a no-op (typed columns not overwritten; reconcile skips already-matched pairs via `Claim.matched_remittance_id IS NOT NULL`).
 
 ### D8 — Backfill event emission
 
@@ -96,7 +104,8 @@ No `UPDATE` statements in the migration — backfill is a separate, deliberate s
 - `backend/src/cyclone/db.py` — `Claim.rendering_provider_npi`, `Remittance.rendering_provider_npi`. No claim_payments ORM — service_provider_npi persists only via `Remittance.raw_json["service_provider_npis"]` (D5).
 - `backend/src/cyclone/writer.py` / `writer_835.py` — set the new typed columns and the `raw_json` mirrors at write time. `writer_835.py` writes the per-CLP `service_provider_npi` into `Remittance.rendering_provider_npi` (D4, single value) and into `raw_json["service_provider_npis"]` (D5).
 - `backend/src/cyclone/reconcile.py` — extend `_content_keys_match` (D6).
-- `backend/src/cyclone/cli.py` — new `backfill-rendering-npi` subcommand.
+- `backend/src/cyclone/store/backfill.py` — **new helper module** owning `backfill_rendering_provider_npi(files, input_dir, transaction_type)`: file/CLI argument resolution, auto-sniffing (ISA/ST + filename hint), re-parse dispatch, idempotent typed-column writes, and the post-backfill `reconcile.run()` sweep. Returns a small summary dataclass for the CLI summary line.
+- `backend/src/cyclone/cli.py` — new `backfill-rendering-npi` subcommand (Click). Wires `--file` / `--input-dir` / `--type` / `--log-level`, delegates the heavy lifting to `store.backfill.backfill_rendering_provider_npi`, and prints the one-line summary.
 - `backend/src/cyclone/migrations/0019_add_rendering_and_service_provider_npis.sql` — D9.
 - `backend/tests/fixtures/` — new 837p fixture(s) with `NM1*82` and new 835 fixture(s) with `NM1*1P`.
 - `backend/tests/test_reconcile.py` — new tests for D6 (typed-column primary path, typed-column beats raw_json fallback, NPI arm fires when both sides populated).