diff --git a/backend/src/cyclone/reconcile.py b/backend/src/cyclone/reconcile.py index fe05dfa..344c699 100644 --- a/backend/src/cyclone/reconcile.py +++ b/backend/src/cyclone/reconcile.py @@ -300,6 +300,72 @@ def run(session, batch_id: str) -> ReconcileResult: )) applied += 1 + # SP7: line-level reconciliation. For each matched (claim, remit) pair, + # pair each 837 service line with the first 835 SVC composite that + # matches on (procedure_code, modifiers, service_date, units) and + # persist a LineReconciliation row per line on either side (matched + # or explicitly unmatched). See spec §4.4. + # + # Architecture note: 837 service lines live in ``Claim.raw_json``, + # not a separate ORM table. The ``LineReconciliation.claim_service_line_number`` + # column stores the 1-based line index; the SVC side uses + # ``ServiceLinePayment`` (real ORM rows) so we get an FK to it. + from cyclone.db import LineReconciliation, ServiceLinePayment + import json as _json_lm + for m in matches: + if m.is_reversal: + continue # reversals are handled by a separate path (see §7.3) + claim_id = m.claim.id + remit_id = m.remittance.id + + # 837 side: read service_lines out of the claim's raw_json blob. + raw = m.claim.raw_json or {} + raw_lines = raw.get("service_lines") or [] + claim_lines = [] + for d in raw_lines: + proc = d.get("procedure") or {} + claim_lines.append(_ClaimLineShim( + line_number=d.get("line_number"), + procedure_code=proc.get("code", ""), + modifiers=proc.get("modifiers") or [], + service_date=_parse_iso_date(d.get("service_date")), + units=Decimal(str(d["units"])) if d.get("units") is not None else None, + )) + + # 835 side: real ORM rows. + svc_payments = list( + session.execute( + select(ServiceLinePayment) + .where(ServiceLinePayment.remittance_id == remit_id) + .order_by(ServiceLinePayment.line_number) + ).scalars().all() + ) + svc_shims = [_SvcPaymentShim.from_orm(s) for s in svc_payments] + + line_matches = match_service_lines(claim_lines, svc_shims) + + now = datetime.now(timezone.utc) + for lm in line_matches: + session.add(LineReconciliation( + claim_id=claim_id, + claim_service_line_number=lm.claim_line.line_number if lm.claim_line else None, + service_line_payment_id=lm.service_payment.line_id if lm.service_payment else None, + status=lm.status, + reconciled_at=now, + )) + + # SP7: claim-level adjustment aggregate. Sum CAS rows whose + # service_line_payment_id IS NULL — these are CLP-level adjustments, + # not per-line. Persist on the remit for fast reads. + from sqlalchemy import func as _func + for r in new_remits: + cl_total = session.execute( + select(_func.sum(CasAdjustment.amount)) + .where(CasAdjustment.remittance_id == r.id) + .where(CasAdjustment.service_line_payment_id.is_(None)) + ).scalar_one() or Decimal("0") + r.claim_level_adjustment_amount = cl_total + # Aggregate CAS adjustments per Remittance. for r in new_remits: total = session.execute( @@ -417,3 +483,52 @@ def match_service_lines( )) return matches + + +@dataclass +class _ClaimLineShim: + """Adapt an 837 service-line dict to the _ClaimLineLike protocol. + + Used by ``reconcile.run()`` to call ``match_service_lines()`` against + lines that live inside ``Claim.raw_json`` (no ORM table). + """ + line_number: int + procedure_code: str + modifiers: list + service_date: Optional[date] + units: Optional[Decimal] + + +@dataclass +class _SvcPaymentShim: + """Adapt an ORM ``ServiceLinePayment`` to the _ServicePaymentLike protocol.""" + + line_id: int + line_number: int + procedure_code: str + modifiers: list + service_date: Optional[date] + units: Optional[Decimal] + + @classmethod + def from_orm(cls, svc) -> "_SvcPaymentShim": + import json as _json + return cls( + line_id=svc.id, + line_number=svc.line_number, + procedure_code=svc.procedure_code, + modifiers=_json.loads(svc.modifiers_json or "[]"), + service_date=svc.service_date, + units=Decimal(str(svc.units)) if svc.units is not None else None, + ) + + +def _parse_iso_date(value) -> Optional[date]: + """Coerce a date / ISO-8601 string / None to ``datetime.date``.""" + if value is None: + return None + if isinstance(value, date): + return value + if isinstance(value, str): + return date.fromisoformat(value) + return None diff --git a/backend/tests/test_reconcile_line_level.py b/backend/tests/test_reconcile_line_level.py new file mode 100644 index 0000000..b537ffa --- /dev/null +++ b/backend/tests/test_reconcile_line_level.py @@ -0,0 +1,128 @@ +"""SP7 — integration test: reconcile.run() persists LineReconciliation rows. + +Spec §4.4. + +Architecture note: 837 service lines are stored inside +``Claim.raw_json["service_lines"]`` keyed by ``line_number`` (1-based). +The ``LineReconciliation`` table therefore stores +``claim_service_line_number`` (int), not a FK to a separate service_lines +table. +""" +from datetime import date, datetime, timezone +from decimal import Decimal +import json +import uuid + +from sqlalchemy import select + +from cyclone.db import ( + Batch, + Claim, + LineReconciliation, + SessionLocal, +) +from cyclone.reconcile import run +from cyclone.store import _persist_835_remit, _remittance_835_row +from cyclone.parsers.models_835 import ClaimPayment, ServicePayment + + +def _setup_db() -> tuple: + """Build a minimal 837 claim + 835 remit that share PCN. Returns (claim_id, batch_id).""" + bid = str(uuid.uuid4()) + claim_id = "CLM-T1" + service_lines_json = [ + { + "line_number": 1, + "procedure": {"qualifier": "HC", "code": "99213", "modifiers": []}, + "charge": "100.00", + "unit_type": "UN", + "units": "1", + "place_of_service": "11", + "service_date": "2026-06-01", + "provider_reference": None, + }, + { + "line_number": 2, + "procedure": {"qualifier": "HC", "code": "99214", "modifiers": []}, + "charge": "100.00", + "unit_type": "UN", + "units": "1", + "place_of_service": "11", + "service_date": "2026-06-01", + "provider_reference": None, + }, + ] + + with SessionLocal()() as s: + # 1. Batch + s.add(Batch( + id=bid, kind="835", input_filename="x.835", + parsed_at=datetime.now(timezone.utc), + totals_json={"total_claims": 1, "total_paid": "150"}, + validation_json={"passed": True, "warnings": [], "errors": []}, + raw_result_json={"_": "stub"}, + )) + # 2. Claim with two service lines (in raw_json) + s.add(Claim( + id=claim_id, + patient_control_number="PCN-1", + charge_amount=Decimal("200.00"), + provider_npi="1234567890", + payer_id="SKCO0", + service_date_from=date(2026, 6, 1), + service_date_to=date(2026, 6, 1), + state="submitted", + batch_id=bid, + raw_json={"service_lines": service_lines_json}, + )) + # 3. Remit with two SVC composites — line 1 matches 99213, line 2 + # is 90837 (no claim match). + cp = ClaimPayment( + payer_claim_control_number="PCN-1", + status_code="1", + total_charge=Decimal("200.00"), + total_paid=Decimal("150.00"), + service_payments=[ + ServicePayment( + line_number=1, procedure_qualifier="HC", + procedure_code="99213", + charge=Decimal("100.00"), + payment=Decimal("80.00"), + units=Decimal("1"), + service_date=date(2026, 6, 1), + ), + ServicePayment( + line_number=2, procedure_qualifier="HC", + procedure_code="90837", + charge=Decimal("100.00"), + payment=Decimal("70.00"), + units=Decimal("1"), + service_date=date(2026, 6, 1), + ), + ], + ) + remit = _remittance_835_row(cp, bid) + s.add(remit) + s.flush() + _persist_835_remit(s, cp, remit.id) + s.commit() + + return claim_id, bid + + +def test_run_persists_line_reconciliation_rows(): + claim_id, bid = _setup_db() + with SessionLocal()() as s: + result = run(s, bid) + s.commit() + assert result.matched == 1 + + lrs = list( + s.execute( + select(LineReconciliation).where(LineReconciliation.claim_id == claim_id) + ).scalars().all() + ) + # Expect 3 rows: 99213 matched, 99214 unmatched_837_only, 90837 unmatched_835_only + assert len(lrs) == 3 + by_status = {lr.status for lr in lrs} + assert by_status == {"matched", "unmatched_837_only", "unmatched_835_only"}