feat(sp33): cli resubmit-rejected-claims + fixture/test payer_id refresh

- 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.
2026-07-02 21:50:23 -06:00
parent 244015a361
commit f10ab83628
8 changed files with 268 additions and 21 deletions
@@ -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)."""
@@ -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~
@@ -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~
@@ -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~
@@ -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"
@@ -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"
@@ -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()