feat(sp7): wire match_service_lines into reconcile.run

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

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"}