diff --git a/backend/src/cyclone/rebill/run.py b/backend/src/cyclone/rebill/run.py index beef89d..b8bb638 100644 --- a/backend/src/cyclone/rebill/run.py +++ b/backend/src/cyclone/rebill/run.py @@ -9,11 +9,14 @@ paths from --visits / --ingest / --out; HTTP endpoint passes the same. from __future__ import annotations import csv as _csv +import logging from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal from pathlib import Path +from cyclone import edifabric as _edifabric +from cyclone.edi.filenames import build_outbound_filename from cyclone.rebill.carc_filter import CarcDecision, decide_carc from cyclone.rebill.parse_835_svc import SvcRow, parse_835_svc from cyclone.rebill.pipeline_a import RebillClaim @@ -33,6 +36,8 @@ from cyclone.rebill.summary import ( ) from cyclone.rebill.timely_filing import timely_filing_decision +_log = logging.getLogger(__name__) + @dataclass class RunResult: @@ -78,14 +83,17 @@ def run_rebill( - NOT_IN_835 + past-window → EXCLUDED_TIMELY_FILING (no override) - DENIED/PARTIAL + EXCLUDED CARC → EXCLUDED_CARC - DENIED/PARTIAL + REVIEW/REBILL → Pipeline A (REBILLED_A) - 5. Emit Pipeline A files (``/pipeline-a/x-{claim_id}.837``) and - Pipeline B files (``/pipeline-b/x-{member_id}-{iso_year}-W{iso_week:02d}.837``). - Files are ``touch()``ed placeholders here — Task 13 wraps each emit - in a serialize-then-Edifabric-validate step, with quarantine on failure. + 5. Emit Pipeline A files (``/pipeline-a/tp{tpid}-837P-...-1of1.x12``) + and Pipeline B files (``/pipeline-b/...-1of1.{member}-W{week}.x12``). + Each file is serialized via the canonical 837P helpers and gated + through Edifabric's ``validate_edi``; failures land in + ``/quarantine/{key}.837`` so the operator can triage without + blocking the batch. 6. Write ``/summary.csv`` with one row per non-PAID visit. - The ``tpid`` argument is preserved on the signature for Task 13 / Task 14 - to thread into the real filename builder; unused in this placeholder emit. + The ``tpid`` argument threads into the HCPF outbound filename + (``build_outbound_filename(tpid, "837P")``) and the Pipeline-B batch + serializer's submitter/receiver block (``serialize_member_week_batch``). ``as_of`` defaults to ``date.today()`` — pin it from tests / HTTP so the 120-day timely-filing gate is deterministic. @@ -224,44 +232,188 @@ def run_rebill( file_path="pipeline-a/", )) - # 5) Emit Pipeline A files — placeholder; Task 13 wraps with the - # serialize-then-Edifabric-validate step. Filename shape: - # ``x-{claim_id}.837`` so the audit trail ties back to the - # original claim_submit_id via the SummaryRow.cas_reasons chain. + # 5) Emit Pipeline A files — serialize each RebillClaim, gate + # through Edifabric's validate_edi, write clean files into + # ``pipeline-a/`` and Edifabric-rejected files into + # ``quarantine/`` for operator triage. + # + # The HCPF-spec filename ``build_outbound_filename(tpid, "837P")`` + # embeds a millisecond timestamp; two Pipeline-A claims emitted + # in the same millisecond would collide. We disambiguate with a + # ``-{claim_id}`` suffix so the audit trail (claim_id ↔ file) + # stays one-to-one and the file can round-trip back to the + # original claim via the SummaryRow chain. a_dir = out_dir_p / "pipeline-a" a_dir.mkdir(parents=True, exist_ok=True) + quarantine_dir = out_dir_p / "quarantine" + quarantine_dir.mkdir(parents=True, exist_ok=True) a_files: list[Path] = [] for claim in pipeline_a_claims: - p = a_dir / f"x-{claim.claim_id}.837" - p.touch() - a_files.append(p) + body = _serialize_pipeline_a(claim, tpid=tpid) + status = _validate_or_skip(body, claim_id=claim.claim_id) + if status == "quarantine": + qp = quarantine_dir / f"{claim.claim_id}.837" + qp.write_bytes(body) + else: + base = build_outbound_filename(tpid, "837P") + # Disambiguate same-millisecond collisions across claims. + stem, dot_ext = base.rsplit(".", 1) + p = a_dir / f"{stem}-{claim.claim_id}.{dot_ext}" + p.write_bytes(body) + a_files.append(p) - # 6) Build + emit Pipeline B batches — placeholder; Task 13 wraps - # with the Edifabric gate. Filename shape mirrors the HCPF - # member-week pattern: ``x-{member_id}-{iso_year}-W{iso_week:02d}.837``. + # 6) Build + emit Pipeline B batches. The Task 14 overload + # ``serialize_member_week_batch`` takes a ``MemberWeekBatch`` and + # emits one envelope with one CLM per visit. We use THAT — not + # ``serialize_837`` (the per-ClaimOutput helper) — because the + # Pipeline-B batches have no ClaimOutput shape; they're a + # (member, ISO-week) visit list keyed off the original visits. + # + # Filename disambiguation: ``build_outbound_filename`` produces + # the same string within a millisecond, so multiple batches + # would collide. Append ``-{member_id}-W{iso_week:02d}`` to the + # HCPF-spec filename so each batch round-trips back to its + # originating visits via the SummaryRow chain. b_batches = build_pipeline_b_batches( pipeline_b_visits, as_of=as_of_date, override=override_filing, ) b_dir = out_dir_p / "pipeline-b" b_dir.mkdir(parents=True, exist_ok=True) b_files: list[Path] = [] + from cyclone.parsers.serialize_837 import serialize_member_week_batch for b in b_batches: - p = b_dir / f"x-{b.member_id}-{b.iso_year}-W{b.iso_week:02d}.837" - p.touch() - b_files.append(p) + body = serialize_member_week_batch(b, tpid=tpid) + batch_key = f"{b.member_id}-{b.iso_year}-W{b.iso_week:02d}" + status = _validate_or_skip(body, claim_id=batch_key) + if status == "quarantine": + qp = quarantine_dir / f"{batch_key}.837" + qp.write_bytes(body) + else: + base = build_outbound_filename(tpid, "837P") + stem, dot_ext = base.rsplit(".", 1) + p = b_dir / f"{stem}-{batch_key}.{dot_ext}" + p.write_bytes(body) + b_files.append(p) # 7) Summary CSV — one row per non-PAID visit, regardless of pipeline. summary_path = out_dir_p / "summary.csv" write_summary_csv(summary, summary_path) - # Touching `tpid` so it doesn't lint as unused — Task 13 / Task 14 will - # thread it into the real filename builder (cyclone.edi.filenames. - # build_outbound_filename) once the serializer adapter lands. - _ = tpid - return RunResult( summary_path=summary_path, counts=counts, pipeline_a_files=a_files, pipeline_b_files=b_files, - ) \ No newline at end of file + ) + + +def _serialize_pipeline_a(claim: RebillClaim, *, tpid: str) -> bytes: + """Build a 837P byte string for a Pipeline-A RebillClaim. + + Pipeline-A's input shape (RebillClaim) only carries the + freq-7-relevant canonical fields: claim_id (the original + claim_submit_id), member_id, procedure, svc_date, charge. The + per-claim envelope context (billing provider NPI, subscriber name, + payer name) lives on the original claim that's being replaced; the + rebill pipeline doesn't carry that forward, so we emit safe + placeholders and let the SP40 serializer fallbacks fill the + contact / SBR09 values. + + Returns ASCII bytes (the 837P envelope is pure ASCII). + """ + # Lazy import: serialize_837 pulls Pydantic models on first use; + # keep it out of the rebill module's import-time surface. + from cyclone.parsers.models import ( + BillingProvider, + ClaimHeader, + ClaimOutput, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationReport, + ) + from cyclone.parsers.serialize_837 import serialize_837 + + # Deterministic control numbers derived from the original claim_id + # so a retry of the same DOS window produces the same filenames + # (the operator can then diff against the prior run). + cn = claim.claim_id[:9].rjust(4, "0")[-4:] + placeholder_claim = ClaimOutput( + claim_id=claim.claim_id, + control_number=cn, + transaction_date=claim.svc_date, + billing_provider=BillingProvider( + name="REBILL PROVIDER", + npi="0000000000", + ), + subscriber=Subscriber( + first_name="REBILL", + last_name=claim.member_id, # surface the member id in NM103 + member_id=claim.member_id, + ), + payer=Payer(name="CO_TXIX", id="CO_TXIX"), + claim=ClaimHeader( + claim_id=claim.claim_id, + total_charge=claim.charge, + place_of_service="11", + facility_code_qualifier="B", + frequency_code=claim.frequency_code, # always "7" + ), + service_lines=[ + ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code=claim.procedure), + charge=claim.charge, + units=Decimal("1"), + unit_type="UN", + place_of_service="11", + service_date=claim.svc_date, + ), + ], + validation=ValidationReport(passed=True), + transaction_type_code="CH", + ) + text = serialize_837(placeholder_claim, sender_id=tpid) + return text.encode("ascii") + + +def _validate_or_skip(body: bytes, *, claim_id: str) -> str: + """Run Edifabric's validate_edi on the emitted 837P bytes. + + Returns: + "ok" — Edifabric reports ``Status in {"success", "warning"}``; + emit to the pipeline dir. + "quarantine" — Edifabric reports ``Status == "error"``; emit + to ``/quarantine/`` for operator triage. + "skip" — Edifabric was unreachable (no API key, network error, + 5xx, etc.); treat as ``ok`` and emit to the pipeline dir. + A WARNING is logged so the operator knows the gate didn't + actually run. This matches the SP40 fail-open posture for + the dev/CI path (no API key in tests) without breaking + in-window rebill runs. + """ + try: + result = _edifabric.validate_edi(body) + except _edifabric.EdifabricError as exc: + # No API key / unreachable / 5xx — fail-open: emit to pipeline + # dir, log a WARNING so the operator sees the gate didn't fire. + _log.warning( + "SP41 rebill: Edifabric validation skipped for %s (%s); " + "emitting to pipeline dir without gate confirmation.", + claim_id, exc, + ) + return "skip" + status = result.get("Status", "") + if status == "error": + details = result.get("Details") or [] + msgs = "; ".join( + f"{d.get('SegmentId', '?')}: {d.get('Message', '?')}" + for d in details[:5] + ) + _log.warning( + "SP41 rebill: Edifabric rejected %s — quarantining. %s", + claim_id, msgs, + ) + return "quarantine" + return "ok" \ No newline at end of file diff --git a/backend/tests/test_rebill_run.py b/backend/tests/test_rebill_run.py new file mode 100644 index 0000000..f1e0b9f --- /dev/null +++ b/backend/tests/test_rebill_run.py @@ -0,0 +1,328 @@ +"""SP41 Task 13 — Edifabric validation gate on emitted rebill files. + +Covers the new emit path in ``cyclone.rebill.run.run_rebill``: + + * Real 837P files are written (not ``Path.touch()`` placeholders). + * Files pass through ``cyclone.edifabric.validate_edi`` before they + land in ``/pipeline-{a,b}/``; failures go to + ``/quarantine/``. + * When Edifabric is unavailable (no API key, network error, etc.), + the gate fails open with a WARNING log and emits to the pipeline + dirs (matches the SP40 dev/CI posture). + +Each test builds its own visits CSV + 835 stub in ``tmp_path`` and +invokes ``run_rebill`` directly (no HTTP). The Edifabric API is mocked +via ``unittest.mock.patch`` on +``cyclone.rebill.run._edifabric.validate_edi`` so no live HTTP hits the +network. +""" +from __future__ import annotations + +import csv +from datetime import date +from decimal import Decimal +from pathlib import Path +from unittest.mock import patch + +import pytest + +from cyclone.rebill.run import RunResult, run_rebill + + +# --------------------------------------------------------------------------- # +# Fixtures / helpers +# --------------------------------------------------------------------------- # + + +def _write_visits_csv(path: Path, rows: list[tuple[str, str, str, str]]) -> Path: + """rows: list of (dos_mmddyyyy, member, procedure, billed_str).""" + with path.open("w", newline="") as f: + w = csv.writer(f) + w.writerow(["Visit Date", "Member ID", "Procedure Code", "Billable Amount"]) + for r in rows: + w.writerow(r) + return path + + +def _stub_835_with_denied_svc( + ingest_dir: Path, + member: str, + procedure: str, + svc_date: date, + claim_id: str = "ORIG-1", + carc_group: str = "CO", + carc_reason: str = "29", +) -> Path: + """An 835 with a single denied SVC so the visit lands in Pipeline A. + + The rebill pipeline reuses the SVC's ``claim_id`` as the Pipeline-A + RebillClaim's ``claim_id`` (preserves the freq-7 anchor). The visit + must match (member_id, procedure, svc_date) so the reconcile step + pairs them into DENIED + CARC REBILL. + + The 835 shape mirrors ``tests/fixtures/835_sample_svc_with_member.txt``: + CLP02 = 4 (Denied), NM1*QC.NM109 = member_id, then SVC → DTM*472 → CAS. + + Segments are concatenated without ``\\n`` separators — the parser + splits on ``~`` only, and a leading ``\\n`` makes ``elems[0]`` an + empty string so the segment-name match silently drops the segment. + """ + ingest_dir.mkdir(exist_ok=True) + svc_date_str = svc_date.strftime("%Y%m%d") + segs = [ + "ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER *260627*1200*^*00501*000000001*0*P*:~", + "GS*HC*SENDER*RECEIVER*20260627*1200*1*X*005010X221A1~", + "ST*835*0001~", + "BPR*I*0*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~", + "TRN*1*TRACE01*1512345678~", + "DTM*405*20260118~", + "N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~", + "N3*PO BOX 1100~", + "N4*DENVER*CO*80202~", + "REF*2U*7912900843~", + "LX*1~", + # CLP02=4 means Denied → reconciles to DENIED → Pipeline A. + f"CLP*{claim_id}*4*100*0**MC*2026029105200*11*1~", + # NM1*QC.NM109 is the member_id; parse_835_svc propagates it to + # the SVC row so the visit-side reconcile key matches. + f"NM1*QC*1*PATIENT*NAME****MR*{member}~", + f"SVC*HC:{procedure}*100*0*UN*1~", + f"DTM*472*{svc_date_str}~", + # CAS group code + reason → CARC REBILL (not EXCLUDED). + f"CAS*{carc_group}*{carc_reason}*100~", + "SE*14*0001~", + "GE*1*1~", + "IEA*1*000000001~", + ] + body = "".join(segs) + p = ingest_dir / "x.835" + p.write_text(body) + return p + + +def _stub_empty_835(ingest_dir: Path) -> Path: + """Empty 835 → rebill's parse_835_svc emits zero SVCs.""" + ingest_dir.mkdir(exist_ok=True) + p = ingest_dir / "x.835" + p.write_text("ST*835*0001~SE*0*0001~") + return p + + +# --------------------------------------------------------------------------- # +# Tests +# --------------------------------------------------------------------------- # + + +def test_run_rebill_emits_real_files_for_pipeline_b(tmp_path): + """Pipeline-B happy path: 1 in-window visit, 0 SVCs → 1 real 837P + file in pipeline-b/ (non-empty, contains ISA..IEA), 0 files in + quarantine/. + + Edifabric is patched to return ``success`` so the file passes the + gate cleanly. The fixture is the same shape as the existing + ``test_post_rebill_runs_and_returns_summary_path`` happy-path test. + """ + visits_path = _write_visits_csv(tmp_path / "visits.csv", [ + ("06/27/2026", "J813715", "T1019", "$2.32"), + ]) + ingest_dir = _stub_empty_835(tmp_path / "ingest") + out_dir = tmp_path / "out" + + fake_result = {"Status": "success", "Details": [], "LastIndex": 20} + with patch( + "cyclone.rebill.run._edifabric.validate_edi", + return_value=fake_result, + ) as mock_validate: + result = run_rebill( + window_start=date(2026, 1, 1), + window_end=date(2026, 6, 27), + override_filing=False, + visits_csv_path=str(visits_path), + ingest_dir=str(ingest_dir), + out_dir=str(out_dir), + ) + + # Pipeline B emitted exactly one file (matches the existing + # ``len(body["pipeline_b_files"]) == 1`` assertion in test_api_rebill). + assert len(result.pipeline_b_files) == 1 + b_path = result.pipeline_b_files[0] + assert b_path.exists(), b_path + assert b_path.stat().st_size > 0, "Pipeline-B file must be non-empty" + + # The file is a real 837P envelope (starts with ISA, ends with IEA). + body = b_path.read_bytes() + assert body.startswith(b"ISA*"), body[:32] + assert b"IEA*" in body, body[-32:] + + # Edifabric gate was called exactly once for the one Pipeline-B batch. + assert mock_validate.call_count == 1 + + # HCPF-spec filename prefix is preserved. + assert b_path.name.startswith("tp11525703-837P-"), b_path.name + # Per-batch disambiguation suffix is present so same-millisecond + # collisions don't overwrite each other. + assert "J813715-2026-W26" in b_path.name, b_path.name + + # Quarantine is empty (no validation failures). + q_files = list((out_dir / "quarantine").glob("*.837")) + assert q_files == [], q_files + + +def test_run_rebill_quarantines_on_edifabric_error(tmp_path): + """When validate_edi returns Status='error', the file lands in + ``quarantine/`` (NOT in ``pipeline-b/``). + + Exercises both the Edifabric-rejected path AND the case-isolation + that splits good files from bad files in the same run. + """ + visits_path = _write_visits_csv(tmp_path / "visits.csv", [ + ("06/27/2026", "J813715", "T1019", "$2.32"), + ("06/27/2026", "J999999", "T1019", "$2.32"), # different member + ]) + ingest_dir = _stub_empty_835(tmp_path / "ingest") + out_dir = tmp_path / "out" + + # First call (J813715) returns error → quarantine. + # Second call (J999999) returns success → pipeline-b. + fake_results = iter([ + { + "Status": "error", + "Details": [ + {"SegmentId": "PER", "Message": "PER-04 is required", "Status": "error"}, + ], + "LastIndex": 5, + }, + {"Status": "success", "Details": [], "LastIndex": 20}, + ]) + + def _side_effect(_body): + return next(fake_results) + + with patch( + "cyclone.rebill.run._edifabric.validate_edi", + side_effect=_side_effect, + ) as mock_validate: + result = run_rebill( + window_start=date(2026, 1, 1), + window_end=date(2026, 6, 27), + override_filing=False, + visits_csv_path=str(visits_path), + ingest_dir=str(ingest_dir), + out_dir=str(out_dir), + ) + + # Only the success batch should land in pipeline-b/. + assert len(result.pipeline_b_files) == 1, ( + f"only the success batch should land in pipeline-b/, got " + f"{[str(p) for p in result.pipeline_b_files]}" + ) + # One file in quarantine (the rejected batch). + q_files = list((out_dir / "quarantine").glob("*.837")) + assert len(q_files) == 1, q_files + # Quarantine filename keys off the batch (member_id-W{iso_week:02d}). + assert "J813715" in q_files[0].name, q_files[0].name + assert q_files[0].stat().st_size > 0, "quarantined file must be non-empty" + + # validate_edi called twice — once per Pipeline-B batch. + assert mock_validate.call_count == 2 + + +def test_run_rebill_fails_open_when_edifabric_unavailable(tmp_path): + """When validate_edi raises EdifabricError (no API key, network + error, 5xx), the file lands in the pipeline dir with a WARNING log. + + Matches the SP40 dev/CI posture: tests / dev boxes don't have a + real API key, so the gate must NOT block the rebill run. + """ + from cyclone.edifabric import EdifabricError + + visits_path = _write_visits_csv(tmp_path / "visits.csv", [ + ("06/27/2026", "J813715", "T1019", "$2.32"), + ]) + ingest_dir = _stub_empty_835(tmp_path / "ingest") + out_dir = tmp_path / "out" + + with patch( + "cyclone.rebill.run._edifabric.validate_edi", + side_effect=EdifabricError(0, "API key not configured"), + ) as mock_validate: + result = run_rebill( + window_start=date(2026, 1, 1), + window_end=date(2026, 6, 27), + override_filing=False, + visits_csv_path=str(visits_path), + ingest_dir=str(ingest_dir), + out_dir=str(out_dir), + ) + + # File emitted to pipeline-b/ (NOT to quarantine — fail-open). + assert len(result.pipeline_b_files) == 1 + assert result.pipeline_b_files[0].exists() + # Quarantine stays empty. + q_files = list((out_dir / "quarantine").glob("*.837")) + assert q_files == [], q_files + # Gate was actually called (so we know the fail-open path fired, + # not some accidental short-circuit). + assert mock_validate.call_count == 1 + + +def test_run_rebill_emits_real_files_for_pipeline_a(tmp_path): + """Pipeline-A happy path: 1 in-window visit, 1 denied SVC matching + it → 1 real 837P file in pipeline-a/ with frequency-code 7 + preserved. + + The SVC's original claim_id (ORIG-1) carries through as the + Pipeline-A RebillClaim.claim_id and shows up in the emitted file's + CLM01 segment AND in the on-disk filename suffix. + """ + svc_date = date(2026, 6, 27) + visits_path = _write_visits_csv(tmp_path / "visits.csv", [ + (svc_date.strftime("%m/%d/%Y"), "J813715", "T1019", "$100.00"), + ]) + ingest_dir = tmp_path / "ingest" + _stub_835_with_denied_svc( + ingest_dir, + member="J813715", + procedure="T1019", + svc_date=svc_date, + claim_id="ORIG-1", + ) + out_dir = tmp_path / "out" + + fake_result = {"Status": "success", "Details": [], "LastIndex": 30} + with patch( + "cyclone.rebill.run._edifabric.validate_edi", + return_value=fake_result, + ): + result = run_rebill( + window_start=date(2026, 1, 1), + window_end=date(2026, 6, 27), + override_filing=False, + visits_csv_path=str(visits_path), + ingest_dir=str(ingest_dir), + out_dir=str(out_dir), + ) + + # Pipeline A emitted exactly one file. + assert len(result.pipeline_a_files) == 1, result.pipeline_a_files + a_path = result.pipeline_a_files[0] + assert a_path.exists(), a_path + assert a_path.stat().st_size > 0, "Pipeline-A file must be non-empty" + + body = a_path.read_bytes() + assert body.startswith(b"ISA*"), body[:32] + assert b"IEA*" in body, body[-32:] + + # Original claim_submit_id is preserved in CLM01 (anchors the + # frequency-7 replacement). + assert b"CLM*ORIG-1*" in body, body + # Frequency-code 7 emitted in CLM05 composite (11:B:7). + assert b"CLM*ORIG-1*100.00***11:B:7" in body, body + + # Pipeline-A filename uses HCPF-spec prefix + claim_id suffix. + assert a_path.name.startswith("tp11525703-837P-"), a_path.name + assert a_path.name.endswith("-ORIG-1.x12"), a_path.name + + # Quarantine empty. + q_files = list((out_dir / "quarantine").glob("*.837")) + assert q_files == [], q_files