From d033ce85dbd5164cb260b45c77bcdec04aa1871c Mon Sep 17 00:00:00 2001 From: Tyler Date: Sat, 20 Jun 2026 20:13:16 -0600 Subject: [PATCH] feat(sp7): wire _reconcile_pair into manual_match (T21) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit manual_match previously only flipped the claim↔remit FK and the claim state — it never ran line-level reconciliation, so manually- paired claims surfaced empty line-reconciliation rows to the UI and skipped CLP-level CAS aggregate recompute. Refactor reconcile.run() to call a new per-pair helper _reconcile_pair(session, claim, remittance) that: - clears any existing LineReconciliation rows for the claim (idempotent re-run; safe across manual_unmatch + manual_rematch cycles that may pair the claim with a different remittance), - reads 837 SV1 lines from Claim.raw_json and 835 SVC rows from ServiceLinePayment, - runs match_service_lines() and persists a LineReconciliation row per side, - recomputes Remittance.claim_level_adjustment_amount (CLP-level CAS) and Remittance.adjustment_amount (total CAS). manual_match now calls this helper after the FK is set and before commit, mirroring the auto-match path. Reversals are skipped (they don't have SV1↔SVC line pairs; per §7.3). Tests (test_store_reconcile.py): - test_manual_match_populates_line_reconciliation_rows: end-to-end check that a manual pair writes the expected matched + unmatched_837_only rows plus zero CAS aggregates. - test_manual_match_idempotent_line_reconciliation: after manual_unmatch + manual_rematch + inserting a CLP-level CAS row directly, the claim has exactly two fresh LineReconciliation rows (no duplicates) and the remittance aggregate reflects the new CAS. Smoke tested end-to-end via TestClient: parse co_medicaid_837p.txt + co_medicaid_835.txt, auto-matcher skips (PCNs differ), manual match via POST /api/reconciliation/match, then GET /api/claims/{id}/line-reconciliation returns 2 rows and GET /api/inbox/lanes shows matched_remittance.total_lines=2. Note: matched_lines may be 0 on real 835 fixtures due to a pre-existing parser bug in _consume_service_payment that swaps SVC04 (units) and SVC05 (unit-of-measure) when reading units from the segment. The SP7 strict-match criterion requires units parity, so the misread produces None on the SVC side and strict-match never succeeds. Surfaced as a follow-up; the SP7 work itself is correct (line counts are computed and persisted). --- backend/src/cyclone/reconcile.py | 171 ++++++++------ backend/src/cyclone/store.py | 5 + backend/tests/test_store_reconcile.py | 328 +++++++++++++++++++++++++- 3 files changed, 436 insertions(+), 68 deletions(-) diff --git a/backend/src/cyclone/reconcile.py b/backend/src/cyclone/reconcile.py index 344c699..2f6439b 100644 --- a/backend/src/cyclone/reconcile.py +++ b/backend/src/cyclone/reconcile.py @@ -300,79 +300,21 @@ 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. + # SP7: line-level reconciliation + claim-level CAS aggregate. 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). Also recompute the + # remittance's CLP-level and total CAS aggregates. # - # Architecture note: 837 service lines live in ``Claim.raw_json``, - # not a separate ORM table. The ``LineReconciliation.claim_service_line_number`` + # 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( - select(func.sum(CasAdjustment.amount)) - .where(CasAdjustment.remittance_id == r.id) - ).scalar_one() or Decimal("0") - r.adjustment_amount = total + _reconcile_pair(session, m.claim, m.remittance) # Partition. unmatched_claims_after, unmatched_remits_after = split_unmatched( @@ -485,6 +427,101 @@ def match_service_lines( return matches +def _reconcile_pair(session, claim, remittance) -> None: + """SP7: line-level reconciliation + claim-level CAS aggregate for one pair. + + Idempotent: clears existing ``LineReconciliation`` rows for the + claim before inserting fresh ones, and recomputes the aggregate + fields on the remittance regardless of prior values. Safe to call + across manual_unmatch + manual_rematch cycles. + + Used by both ``reconcile.run()`` (after auto-match) and + ``store.manual_match()`` (after operator override) so manual matches + produce the same per-line audit data as auto matches. See spec §4.4. + + Steps: + 1. Delete any existing ``LineReconciliation`` rows for the claim. + 2. Read 837 service lines from ``Claim.raw_json``. + 3. Read ``ServiceLinePayment`` rows for the remittance. + 4. Run ``match_service_lines`` (pure) to pair SV1 ↔ SVC. + 5. Persist ``LineReconciliation`` rows. + 6. Recompute ``Remittance.claim_level_adjustment_amount`` from CAS + rows with ``service_line_payment_id IS NULL`` (CLP-level). + 7. Recompute ``Remittance.adjustment_amount`` from all CAS rows. + + Does not commit; caller's session controls the transaction. + """ + from sqlalchemy import select + from cyclone.db import ( + CasAdjustment, LineReconciliation, ServiceLinePayment, + ) + + # 1: clear stale rows so re-running produces a clean slate. Stale + # rows can appear after a manual_unmatch + manual_rematch cycle that + # pairs the claim with a *different* remittance, leaving rows whose + # ``service_line_payment_id`` references SVC rows no longer paired + # with this claim. + stale = list( + session.execute( + select(LineReconciliation) + .where(LineReconciliation.claim_id == claim.id) + ).scalars().all() + ) + for lr in stale: + session.delete(lr) + + # 2: read 837 service lines (JSON blob). + raw = 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, + )) + + # 3: read 835 SVC rows (real ORM). + svc_payments = list( + session.execute( + select(ServiceLinePayment) + .where(ServiceLinePayment.remittance_id == remittance.id) + .order_by(ServiceLinePayment.line_number) + ).scalars().all() + ) + svc_shims = [_SvcPaymentShim.from_orm(s) for s in svc_payments] + + # 4-5: pure pair + persist. + 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, + )) + + # 6: claim-level CAS aggregate (CLP-level adjustments, no SLI pointer). + cl_total = session.execute( + select(func.sum(CasAdjustment.amount)) + .where(CasAdjustment.remittance_id == remittance.id) + .where(CasAdjustment.service_line_payment_id.is_(None)) + ).scalar_one() or Decimal("0") + remittance.claim_level_adjustment_amount = cl_total + + # 7: total CAS aggregate. + total = session.execute( + select(func.sum(CasAdjustment.amount)) + .where(CasAdjustment.remittance_id == remittance.id) + ).scalar_one() or Decimal("0") + remittance.adjustment_amount = total + + @dataclass class _ClaimLineShim: """Adapt an 837 service-line dict to the _ClaimLineLike protocol. diff --git a/backend/src/cyclone/store.py b/backend/src/cyclone/store.py index a6d4afa..b745ea1 100644 --- a/backend/src/cyclone/store.py +++ b/backend/src/cyclone/store.py @@ -1948,6 +1948,11 @@ class CycloneStore: # we don't rely on T10 to do this. remit.claim_id = claim_id + # SP7: line-level reconciliation + claim-level CAS aggregate. + # Skipped for reversals — they don't have SV1↔SVC line pairs. + if not remit.is_reversal: + _reconcile._reconcile_pair(s, claim, remit) + s.add(ActivityEvent( ts=now, kind="manual_match", diff --git a/backend/tests/test_store_reconcile.py b/backend/tests/test_store_reconcile.py index f3ce1bb..a61f62c 100644 --- a/backend/tests/test_store_reconcile.py +++ b/backend/tests/test_store_reconcile.py @@ -260,4 +260,330 @@ def test_manual_match_conflict_raises(): sess.commit() with pytest.raises(AlreadyMatchedError): - s.manual_match("CLM-X", "CLP-1:b1xxxxxx") \ No newline at end of file + s.manual_match("CLM-X", "CLP-1:b1xxxxxx") + + +# --------------------------------------------------------------------------- +# SP7: manual_match populates LineReconciliation rows + CAS aggregates. +# --------------------------------------------------------------------------- + + +def test_manual_match_populates_line_reconciliation_rows(): + """manual_match must run line-level reconciliation, same as auto-match. + + Without this, operators who manually pair claims see no per-line audit + data (every line shows up as ``unmatched_837_only`` in the UI). The + SP7 fix routes ``manual_match`` through ``reconcile._reconcile_pair`` + after the claim↔remit FK is set, mirroring the auto-match path. + + Setup: + * 837 claim with two SV1 lines (99213 + 99214). + * 835 remit with one SVC line (99213) — matches line 1, leaves + line 2 unmatched_837_only. + * member_id and PCN differ so auto-match doesn't pair them. + + Expected post-match state: + * 1 ``matched`` LineReconciliation row (line 1 ↔ SVC 1). + * 1 ``unmatched_837_only`` LineReconciliation row (line 2). + * ``Remittance.adjustment_amount`` recomputed (0 — no CAS in this test). + * ``Remittance.claim_level_adjustment_amount`` recomputed (0). + """ + from sqlalchemy import select + from cyclone.db import LineReconciliation, Remittance + from cyclone.store import BatchRecord837, BatchRecord835 + from cyclone.parsers.models import ( + BatchSummary as BatchSummary837, + BillingProvider, ClaimHeader, ClaimOutput, Envelope, + Payer, ParseResult, Procedure, ServiceLine, Subscriber, + ValidationReport, + ) + + s = CycloneStore() + + # 837 claim with two SV1 service lines. + co = ClaimOutput( + claim_id="CLM-MANUAL", + control_number="0001", + transaction_date=date(2026, 6, 19), + billing_provider=BillingProvider(name="Test", npi="1234567890"), + # member_id deliberately ≠ PCN below so auto-match skips this claim. + subscriber=Subscriber(first_name="Jane", last_name="Doe", + member_id="MANUAL"), + payer=Payer(name="Test Payer", id="P1"), + claim=ClaimHeader( + claim_id="CLM-MANUAL", total_charge=Decimal("200.00"), + frequency_code="1", place_of_service="11", + ), + diagnoses=[], + service_lines=[ + ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", + modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1"), + service_date=date(2026, 6, 19), + ), + ServiceLine( + line_number=2, + procedure=Procedure(qualifier="HC", code="99214", + modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1"), + service_date=date(2026, 6, 19), + ), + ], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + pr837 = ParseResult( + envelope=Envelope( + sender_id="S", receiver_id="R", control_number="0001", + transaction_date=date(2026, 6, 19), + ), + claims=[co], + summary=BatchSummary837( + input_file="c.txt", control_number="0001", + transaction_date=date(2026, 6, 19), + total_claims=1, passed=1, failed=0, + ), + ) + s.add(BatchRecord837( + id="b-837-sp7", kind="837p", input_filename="c.txt", + parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc), + result=pr837, + )) + + # 835 remit with one SVC line that matches claim line 1. + cp = ClaimPayment( + payer_claim_control_number="CLM-MANUAL", # != member_id "MANUAL" + status_code="1", + status_label="Primary", + total_charge=Decimal("200.00"), + total_paid=Decimal("200.00"), + service_payments=[ + ServicePayment( + line_number=1, procedure_qualifier="HC", + procedure_code="99213", + charge=Decimal("100.00"), + payment=Decimal("100.00"), + units=Decimal("1"), + service_date=date(2026, 6, 19), + adjustments=[], + ), + ], + ) + s.add(BatchRecord835( + id="b-835-sp7", kind="835", input_filename="era.txt", + parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc), + result=_make_835_result([cp]), + )) + + # Verify auto-match didn't pair them (they should both be unmatched). + out = s.list_unmatched(kind="both") + assert len(out["claims"]) == 1 + assert len(out["remittances"]) == 1 + claim_id = out["claims"][0]["id"] + remit_id = out["remittances"][0]["id"] + + # Manual match — must populate LineReconciliation rows. + result = s.manual_match(claim_id, remit_id) + assert result["claim"]["state"] == "paid" + + with db.SessionLocal()() as session: + lrs = list( + session.execute( + select(LineReconciliation) + .where(LineReconciliation.claim_id == claim_id) + .order_by(LineReconciliation.id) + ).scalars().all() + ) + # Two claim SV1 lines → exactly two LineReconciliation rows + # (matched + unmatched_837_only). No SVC-only rows since the + # single SVC line pairs with claim line 1. + assert len(lrs) == 2, [lr.status for lr in lrs] + statuses = sorted(lr.status for lr in lrs) + assert statuses == ["matched", "unmatched_837_only"], statuses + + # The matched row references claim line 1 + SVC row 1. + matched = next(lr for lr in lrs if lr.status == "matched") + assert matched.claim_service_line_number == 1 + assert matched.service_line_payment_id is not None + # The unmatched_837_only row references claim line 2 only. + unmatched = next(lr for lr in lrs if lr.status == "unmatched_837_only") + assert unmatched.claim_service_line_number == 2 + assert unmatched.service_line_payment_id is None + + # Remittance aggregates were recomputed (0 CAS in this test). + r = session.get(Remittance, remit_id) + assert r is not None + assert r.adjustment_amount == Decimal("0") + assert r.claim_level_adjustment_amount == Decimal("0") + + +def test_manual_match_idempotent_line_reconciliation(): + """Re-running ``manual_match`` semantics on a rematched claim must not + duplicate LineReconciliation rows. + + The helper clears stale rows for the claim before inserting fresh + ones, so a manual_unmatch → manual_rematch cycle (possibly to a + different remittance) produces a clean slate per claim. + """ + from sqlalchemy import select + from cyclone.db import LineReconciliation + from cyclone.store import BatchRecord837, BatchRecord835 + from cyclone.parsers.models import ( + BatchSummary as BatchSummary837, + BillingProvider, ClaimHeader, ClaimOutput, Envelope, + Payer, ParseResult, Procedure, ServiceLine, Subscriber, + ValidationReport, + ) + + s = CycloneStore() + + # 837 with two SV1 lines, member_id="IDEMP" so it doesn't auto-pair. + co = ClaimOutput( + claim_id="CLM-IDEMP", + control_number="0001", + transaction_date=date(2026, 6, 19), + billing_provider=BillingProvider(name="Test", npi="1234567890"), + subscriber=Subscriber(first_name="Jane", last_name="Doe", + member_id="IDEMP"), + payer=Payer(name="Test Payer", id="P1"), + claim=ClaimHeader( + claim_id="CLM-IDEMP", total_charge=Decimal("200.00"), + frequency_code="1", place_of_service="11", + ), + diagnoses=[], + service_lines=[ + ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", + modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1"), + service_date=date(2026, 6, 19), + ), + ServiceLine( + line_number=2, + procedure=Procedure(qualifier="HC", code="99214", + modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1"), + service_date=date(2026, 6, 19), + ), + ], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + pr837 = ParseResult( + envelope=Envelope( + sender_id="S", receiver_id="R", control_number="0001", + transaction_date=date(2026, 6, 19), + ), + claims=[co], + summary=BatchSummary837( + input_file="c.txt", control_number="0001", + transaction_date=date(2026, 6, 19), + total_claims=1, passed=1, failed=0, + ), + ) + s.add(BatchRecord837( + id="b-837-idemp", kind="837p", input_filename="c.txt", + parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc), + result=pr837, + )) + + # 835 with two SVC lines — auto-match won't pair (PCN="CLM-IDEMP" ≠ + # member_id="IDEMP"), so manual_match has work to do. + cp = ClaimPayment( + payer_claim_control_number="CLM-IDEMP", + status_code="1", + status_label="Primary", + total_charge=Decimal("200.00"), + total_paid=Decimal("200.00"), + service_payments=[ + ServicePayment( + line_number=1, procedure_qualifier="HC", + procedure_code="99213", + charge=Decimal("100.00"), + payment=Decimal("100.00"), + units=Decimal("1"), + service_date=date(2026, 6, 19), + adjustments=[], + ), + ServicePayment( + line_number=2, procedure_qualifier="HC", + procedure_code="99214", + charge=Decimal("100.00"), + payment=Decimal("100.00"), + units=Decimal("1"), + service_date=date(2026, 6, 19), + adjustments=[], + ), + ], + ) + s.add(BatchRecord835( + id="b-835-idemp", kind="835", input_filename="era.txt", + parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc), + result=_make_835_result([cp]), + )) + + out = s.list_unmatched(kind="both") + claim_id = out["claims"][0]["id"] + remit_id = out["remittances"][0]["id"] + + # First match — creates LineReconciliation rows. + s.manual_match(claim_id, remit_id) + with db.SessionLocal()() as session: + first_count = len(list( + session.execute( + select(LineReconciliation) + .where(LineReconciliation.claim_id == claim_id) + ).scalars().all() + )) + assert first_count == 2 + + # Unmatch — clears FK but leaves LineReconciliation rows in place. + s.manual_unmatch(claim_id) + + # Insert a CLP-level CAS row directly so we can verify aggregate + # recompute on the rematch. This simulates a forward-compat CLP CAS + # (today's parser doesn't produce these, see _persist_835_remit). + from cyclone.db import CasAdjustment + with db.SessionLocal()() as session: + session.add(CasAdjustment( + remittance_id=remit_id, + group_code="CO", reason_code="45", + amount=Decimal("20.00"), + quantity=None, + service_line_payment_id=None, # CLP-level + )) + session.commit() + + # Rematch — _reconcile_pair clears stale rows and inserts fresh ones. + s.manual_match(claim_id, remit_id) + with db.SessionLocal()() as session: + second = list( + session.execute( + select(LineReconciliation) + .where(LineReconciliation.claim_id == claim_id) + .order_by(LineReconciliation.id) + ).scalars().all() + ) + # Still exactly 2 rows — stale rows from the first match were + # deleted before the new ones were inserted. + assert len(second) == 2, [lr.status for lr in second] + statuses = sorted(lr.status for lr in second) + assert statuses == ["matched", "matched"], statuses + + # CLP-level CAS aggregate now reflects the inserted row. + from cyclone.db import Remittance + r = session.get(Remittance, remit_id) + assert r is not None + assert r.adjustment_amount == Decimal("20.00") + assert r.claim_level_adjustment_amount == Decimal("20.00") \ No newline at end of file