feat(sp7): persist ServiceLinePayment + link CAS rows on 835 ingest

backend/src/cyclone/store.py

@@ -259,6 +259,69 @@ def _cas_adjustment_row(adj, remittance_id: str) -> "db.CasAdjustment":
     )
 
 
+def _persist_835_remit(session, cp: "ClaimPayment", remittance_id: str) -> None:
+    """SP7: persist ServiceLinePayment + CAS rows for one CLP composite.
+
+    For each 835 SVC composite in ``cp.service_payments``:
+      - insert a ServiceLinePayment row (line_number, procedure, modifiers,
+        charge, payment, units, service_date).
+      - flush to populate slp.id.
+      - insert each per-SVC CAS adjustment with ``service_line_payment_id``
+        set to slp.id.
+
+    For CLP-level CAS adjustments (``cp.claim_adjustments``, a future
+    extension; not produced by today's 835 parser but allowed by the spec):
+      - insert CAS rows with ``service_line_payment_id IS NULL``.
+
+    The caller controls the transaction; this function does not commit.
+    The 835 ingest site calls this after ``_remittance_835_row`` is
+    flushed so the FK target is populated.
+    """
+    import json as _json
+    from cyclone.db import ServiceLinePayment, CasAdjustment
+
+    for svc in cp.service_payments:
+        slp = ServiceLinePayment(
+            remittance_id=remittance_id,
+            line_number=svc.line_number,
+            procedure_qualifier=svc.procedure_qualifier,
+            procedure_code=svc.procedure_code,
+            modifiers_json=_json.dumps(svc.modifiers or []),
+            charge=Decimal(str(svc.charge)),
+            payment=Decimal(str(svc.payment)),
+            units=Decimal(str(svc.units)) if svc.units is not None else None,
+            unit_type=svc.unit_type,
+            service_date=svc.service_date,
+            ref_benefit_plan=svc.ref_benefit_plan,
+        )
+        session.add(slp)
+        session.flush()  # populate slp.id for the FK below
+
+        for adj in svc.adjustments:
+            quantity = getattr(adj, "quantity", None)
+            session.add(CasAdjustment(
+                remittance_id=remittance_id,
+                group_code=adj.group_code,
+                reason_code=adj.reason_code,
+                amount=Decimal(str(adj.amount)),
+                quantity=Decimal(str(quantity)) if quantity is not None else None,
+                service_line_payment_id=slp.id,
+            ))
+
+    # CLP-level CAS (no SVC composite to attach to). Today's parser does
+    # not produce these; the branch is forward-compatible.
+    for adj in getattr(cp, "claim_adjustments", []) or []:
+        quantity = getattr(adj, "quantity", None)
+        session.add(CasAdjustment(
+            remittance_id=remittance_id,
+            group_code=adj.group_code,
+            reason_code=adj.reason_code,
+            amount=Decimal(str(adj.amount)),
+            quantity=Decimal(str(quantity)) if quantity is not None else None,
+            service_line_payment_id=None,
+        ))
+
+
 # ---------------------------------------------------------------------------
 # UI mappers: ORM rows → simpler UI types.
 # ---------------------------------------------------------------------------
@@ -905,9 +968,11 @@ class CycloneStore:
                     # this the CasAdjustment inserts below would reference an
                     # unset id and violate the FK.
                     s.flush()
-                    for svc in cp.service_payments:
+                    # SP7: persist per-line ServiceLinePayment + linked
-                        for adj in svc.adjustments:
+                    # SVC-level CAS rows + claim-level CAS bucket. Replaces
-                            s.add(_cas_adjustment_row(adj, remit_row.id))
+                    # the previous per-SVC CAS insert loop so the
+                    # service_line_payment_id FK is set correctly.
+                    _persist_835_remit(s, cp, remit_row.id)
                     s.add(ActivityEvent(
                         ts=record.parsed_at,
                         kind="remit_received",

backend/tests/test_service_line_payments.py

@@ -0,0 +1,168 @@
+"""SP7 — 835 ingest persists ServiceLinePayment rows + links CAS rows.
+
+Spec §3.1, §3.3, T6.
+"""
+from datetime import date, datetime, timezone
+from decimal import Decimal
+import uuid
+
+from sqlalchemy import select
+
+from cyclone.db import (
+    Batch,
+    CasAdjustment,
+    Remittance,
+    ServiceLinePayment,
+    SessionLocal,
+)
+from cyclone.parsers.models_835 import (
+    ClaimAdjustment,
+    ClaimPayment,
+    ServicePayment,
+)
+from cyclone.store import _persist_835_remit, _remittance_835_row
+
+
+def _make_batch() -> str:
+    bid = str(uuid.uuid4())
+    with SessionLocal()() as s:
+        s.add(Batch(
+            id=bid, kind="835", input_filename="x.835",
+            parsed_at=datetime.now(timezone.utc),
+            totals_json={"total_claims": 1, "total_paid": "0"},
+            validation_json={"passed": True, "warnings": [], "errors": []},
+            raw_result_json={"_": "stub"},
+        ))
+        s.commit()
+    return bid
+
+
+def test_persist_835_creates_one_service_line_payment_per_svc_composite():
+    bid = _make_batch()
+    cp = ClaimPayment(
+        payer_claim_control_number="PCN-SLP-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"),
+                service_date=date(2026, 6, 1),
+                adjustments=[ClaimAdjustment(group_code="CO", reason_code="45", amount=Decimal("20.00"))],
+            ),
+            ServicePayment(
+                line_number=2, procedure_qualifier="HC",
+                procedure_code="99214", charge=Decimal("100.00"),
+                payment=Decimal("70.00"),
+                service_date=date(2026, 6, 1),
+                adjustments=[],
+            ),
+        ],
+    )
+    with SessionLocal()() as s:
+        remit = _remittance_835_row(cp, bid)
+        s.add(remit)
+        s.flush()
+        _persist_835_remit(s, cp, remit.id)
+        s.commit()
+
+        slps = list(
+            s.execute(
+                select(ServiceLinePayment).order_by(ServiceLinePayment.line_number)
+            ).scalars().all()
+        )
+        assert len(slps) == 2
+        assert slps[0].procedure_code == "99213"
+        assert slps[0].payment == Decimal("80.00")
+        assert slps[1].procedure_code == "99214"
+        assert slps[1].payment == Decimal("70.00")
+        assert slps[0].remittance_id == remit.id
+
+
+def test_svc_level_cas_adjustments_link_to_service_line_payment():
+    """CAS rows that follow an SVC composite must point at the SLP."""
+    bid = _make_batch()
+    cp = ClaimPayment(
+        payer_claim_control_number="PCN-SLP-2",
+        status_code="1",
+        total_charge=Decimal("100.00"),
+        total_paid=Decimal("80.00"),
+        service_payments=[
+            ServicePayment(
+                line_number=1, procedure_qualifier="HC",
+                procedure_code="99213", charge=Decimal("100.00"),
+                payment=Decimal("80.00"),
+                adjustments=[ClaimAdjustment(group_code="CO", reason_code="45", amount=Decimal("20.00"))],
+            ),
+        ],
+    )
+    with SessionLocal()() as s:
+        remit = _remittance_835_row(cp, bid)
+        s.add(remit)
+        s.flush()
+        _persist_835_remit(s, cp, remit.id)
+        s.commit()
+
+        cas = list(s.execute(select(CasAdjustment)).scalars().all())
+        assert len(cas) == 1
+        assert cas[0].service_line_payment_id is not None
+
+        slp = s.execute(
+            select(ServiceLinePayment).where(
+                ServiceLinePayment.id == cas[0].service_line_payment_id
+            )
+        ).scalar_one()
+        assert slp.procedure_code == "99213"
+
+
+def test_modifiers_persisted_as_json_array():
+    bid = _make_batch()
+    cp = ClaimPayment(
+        payer_claim_control_number="PCN-SLP-3",
+        status_code="1",
+        total_charge=Decimal("100.00"),
+        total_paid=Decimal("80.00"),
+        service_payments=[
+            ServicePayment(
+                line_number=1, procedure_qualifier="HC",
+                procedure_code="99213", modifiers=["25", "59"],
+                charge=Decimal("100.00"), payment=Decimal("80.00"),
+                adjustments=[],
+            ),
+        ],
+    )
+    with SessionLocal()() as s:
+        remit = _remittance_835_row(cp, bid)
+        s.add(remit)
+        s.flush()
+        _persist_835_remit(s, cp, remit.id)
+        s.commit()
+
+        slp = s.execute(select(ServiceLinePayment)).scalar_one()
+        import json as _json
+        assert _json.loads(slp.modifiers_json) == ["25", "59"]
+
+
+def test_empty_service_payments_is_noop():
+    """A 835 with no SVC composites produces no ServiceLinePayment rows."""
+    bid = _make_batch()
+    cp = ClaimPayment(
+        payer_claim_control_number="PCN-SLP-4",
+        status_code="1",
+        total_charge=Decimal("0.00"),
+        total_paid=Decimal("0.00"),
+        service_payments=[],
+    )
+    with SessionLocal()() as s:
+        remit = _remittance_835_row(cp, bid)
+        s.add(remit)
+        s.flush()
+        _persist_835_remit(s, cp, remit.id)
+        s.commit()
+
+        slps = list(s.execute(select(ServiceLinePayment)).scalars().all())
+        assert slps == []
+        cas = list(s.execute(select(CasAdjustment)).scalars().all())
+        assert cas == []