From f10ab8362890a192d5fcd79eb8effe237a21f3bb Mon Sep 17 00:00:00 2001 From: Nora Date: Thu, 2 Jul 2026 21:50:23 -0600 Subject: [PATCH] feat(sp33): cli resubmit-rejected-claims + fixture/test payer_id refresh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Adds `cyclone resubmit-rejected-claims` to push corrected single-claim 837 files to the Gainwell ToHPE SFTP dir. Idempotent (stat-then-skip by byte size). One persistent paramiko session per batch with reconnect-every=50 to dodge MOVEit's silent per-session file cap (~200 puts/session, no exception). - Validates each file via `parse_837` before upload and rejects any whose payer_id is not `CO_TXIX` (catches a bad byte-fix early). - Refreshes test fixtures (`minimal_837p.txt`, `co_medicaid_837p.txt`, `co_medicaid_837p_with_renderer.txt`) and the corresponding test assertions (`test_payer.py`, `test_payer_summary.py`, `test_co_medicaid_fixture.py`, `test_parse_837.py`) from the old `SKCO0`/`COHCPF` payer IDs to `CO_TXIX`, matching PayerConfig.co_medicaid() and the HCPF 837P Companion Guide. - Adds `ingest/` to .gitignore — local scratch / production-data staging only. --- .gitignore | 4 + backend/src/cyclone/cli.py | 251 +++++++++++++++++- backend/tests/fixtures/co_medicaid_837p.txt | 4 +- .../co_medicaid_837p_with_renderer.txt | 4 +- backend/tests/fixtures/minimal_837p.txt | 2 +- backend/tests/test_co_medicaid_fixture.py | 4 +- backend/tests/test_parse_837.py | 2 +- backend/tests/test_payer_summary.py | 18 +- 8 files changed, 268 insertions(+), 21 deletions(-) diff --git a/.gitignore b/.gitignore index 779370e..479a213 100644 --- a/.gitignore +++ b/.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/ diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py index 14b0738..a7cdd20 100644 --- a/backend/src/cyclone/cli.py +++ b/backend/src/cyclone/cli.py @@ -440,7 +440,7 @@ def backfill_999_rejections(dry_run: bool, actor: str) -> None: Claims already in REJECTED are skipped (counted in the summary). """ from datetime import datetime, timezone - from sqlalchemy import select + from sqlalchemy import select, func from cyclone import db as db_mod from cyclone.audit_log import AuditEvent, append_event @@ -461,9 +461,9 @@ def backfill_999_rejections(dry_run: bool, actor: str) -> None: select( db_mod.Claim.id, db_mod.Claim.state, - db_mod.func.min(db_mod.ClaimAck.set_control_number).label("scn"), - db_mod.func.min(db_mod.ClaimAck.ak2_index).label("ak2"), - db_mod.func.min(db_mod.Ack.ack_code).label("ack_code"), + 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) @@ -521,6 +521,249 @@ def backfill_999_rejections(dry_run: bool, actor: str) -> None: ) +# --------------------------------------------------------------------------- +# 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.clearhouse import SftpClient + 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() + sftp_client = SftpClient(sftp_block) + 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 + + for i, src in enumerate(files, 1): + if limit is not None and i > limit: + break + + # 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(src.read_text(), payer_cfg) + except Exception as exc: # noqa: BLE001 + failed += 1 + click.echo(f"PARSE FAIL {src.name}: {exc.__class__.__name__}: {exc}", err=True) + continue + for c in parsed.claims: + if c.payer.id != "CO_TXIX": + failed += 1 + click.echo( + f"PAYER MISMATCH {src.name}: payer.id={c.payer.id!r} " + f"(expected 'CO_TXIX')", err=True, + ) + break + else: + validated += 1 + + content = src.read_bytes() + local_size = len(content) + remote_path = f"{remote_root}/{src.name}" + + ssh: paramiko.SSHClient | None = None + sftp: paramiko.SFTPClient | None = None + 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 + + # Reconnect periodically to dodge MOVEit's per-session cap. + if uploaded > 0 and uploaded % reconnect_every == 0: + click.echo(f"reconnecting (after {uploaded} uploads)", err=True) + _close_session(ssh) + ssh, sftp = None, None + + # 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 + + # 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).""" diff --git a/backend/tests/fixtures/co_medicaid_837p.txt b/backend/tests/fixtures/co_medicaid_837p.txt index 54def3f..ebf374a 100644 --- a/backend/tests/fixtures/co_medicaid_837p.txt +++ 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~ diff --git a/backend/tests/fixtures/co_medicaid_837p_with_renderer.txt b/backend/tests/fixtures/co_medicaid_837p_with_renderer.txt index 1132e8c..7f678b7 100644 --- a/backend/tests/fixtures/co_medicaid_837p_with_renderer.txt +++ b/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~ diff --git a/backend/tests/fixtures/minimal_837p.txt b/backend/tests/fixtures/minimal_837p.txt index 47c8040..61ee4e6 100644 --- a/backend/tests/fixtures/minimal_837p.txt +++ b/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~ diff --git a/backend/tests/test_co_medicaid_fixture.py b/backend/tests/test_co_medicaid_fixture.py index e15e49d..3ed54f8 100644 --- a/backend/tests/test_co_medicaid_fixture.py +++ b/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" diff --git a/backend/tests/test_parse_837.py b/backend/tests/test_parse_837.py index f904106..6f9d0c6 100644 --- a/backend/tests/test_parse_837.py +++ b/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" diff --git a/backend/tests/test_payer_summary.py b/backend/tests/test_payer_summary.py index 9abb1c7..6fb9e04 100644 --- a/backend/tests/test_payer_summary.py +++ b/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() \ No newline at end of file