tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp7): wire _reconcile_pair into manual_match (T21)

Browse Source 
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).
This commit is contained in:
Tyler
2026-06-20 20:13:16 -06:00
parent 27c1680830
commit d033ce85db
3 changed files with 436 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/reconcile.py
+104 -67
View File
@@ -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.
backend/src/cyclone/store.py
+5
View File
@@ -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",