feat(sp27): pin Claim↔Remit matched-pair invariant + startup drift audit

Add check_matched_pair_drift() at module level in store.py — read-only
audit of the Claim.matched_remittance_id ↔ Remittance.claim_id FK pair.
Logs WARNING on drift with up to 5 examples of each of two cases
(claim-side and remit-side), returns the count of drifted rows. Wired
into api.py::lifespan right after db.init_db() and ensure_clearhouse_seeded(),
wrapped in try/except so a query failure logs but doesn't crash boot.

The symmetric-write + symmetric-clear invariants on
manual_match / manual_unmatch were already correct; this commit pins
them with focused regression tests in test_store_match_invariant.py
(8 tests covering both directions, both happy-path, rollback pin, and
drift-check unit behavior). PCN asymmetry in the fixture prevents the
auto-match inside CycloneStore.add (Task 10) from pre-pairing, so
manual_match is the only writer.

Migration 0016 adds the missing ix_claims_matched_remittance_id index —
the drift check scans WHERE matched_remittance_id IS NOT NULL on every
boot and would become a full-table scan past ~10k claims. Symmetric
with ix_remittances_claim_id (added in 0007). Migration tests bumped
from head=15 to head=16.
This commit is contained in:
Nora
2026-06-29 12:21:34 -06:00
parent d5f95b4f3c
commit f5d119fbe7
7 changed files with 580 additions and 7 deletions
+10
View File
@@ -146,6 +146,16 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
except Exception as exc: # noqa: BLE001
log.exception("SP9 seed failed: %s", exc)
# SP27 Task 11: startup audit for the Claim.matched_remittance_id ↔
# Remittance.claim_id denormalized FK pair. Non-blocking — mismatches
# are logged at WARNING with up to 5 examples of each case so the
# operator can investigate without the system failing to boot.
try:
from cyclone.store import check_matched_pair_drift
check_matched_pair_drift()
except Exception as exc: # noqa: BLE001
log.exception("matched-pair drift check failed: %s", exc)
# SP16: configure the inbound MFT polling scheduler. The
# dzinesco clearhouse singleton (seeded by SP9) carries the SFTP
# block — multi-provider polling is out of scope for v1.
+7
View File
@@ -317,6 +317,13 @@ class Claim(Base):
Index("ix_claims_state", "state"),
Index("ix_claims_patient_control_number", "patient_control_number"),
Index("ix_claims_service_date_from", "service_date_from"),
# SP27 Task 11: matched-pair drift check (run at startup)
# scans ``WHERE matched_remittance_id IS NOT NULL``. Without
# this index it's a full claim scan. The reverse side
# (``ix_remittances_claim_id``) is added in 0007. Pure index
# (non-unique) — a claim without a match is fine, reversals
# leave the previous claim/claim match intact.
Index("ix_claims_matched_remittance_id", "matched_remittance_id"),
)
@@ -0,0 +1,15 @@
-- version: 16
-- Add the missing index on claims.matched_remittance_id.
--
-- SP27 Task 11 added ``check_matched_pair_drift`` at startup, which
-- scans ``WHERE Claim.matched_remittance_id IS NOT NULL``. Without an
-- index this is a full-table scan; becomes a noticeable boot
-- latency cost past ~10k claims. The companion index on the
-- reverse side (``remittances.claim_id``) was added in 0007.
--
-- A plain index is enough — neither side is unique (reversals
-- re-reference the original PCN, and a claim without a match is
-- fine).
CREATE INDEX ix_claims_matched_remittance_id
ON claims(matched_remittance_id);
+104
View File
@@ -878,6 +878,110 @@ class _BatchesShim:
s.commit()
def check_matched_pair_drift() -> int:
"""Audit the ``Claim.matched_remittance_id`` ↔ ``Remittance.claim_id``
FK pair at startup. Non-blocking (SP27 Task 11).
The matched pair is a denormalized FK pair maintained transactionally
by ``manual_match``, ``manual_unmatch``, and ``reconcile.run``. A
pre-existing mismatch (e.g. a row written before a state migration
that added one column but not the other) would otherwise stay
invisible until the next operator pair attempt fails confusingly.
This check logs the count + up to N examples so operators can
investigate without booting the system.
Returns the number of drifted rows (0 means clean). Does not
raise; bootstrap continues even if drift is detected.
Count semantics: this returns *drifted rows*, not *drifted pairs*.
A single broken pair (``Claim.matched_remittance_id = X`` AND
``Remittance.claim_id = Y != nil`` with neither pointing back)
can produce TWO drifted rows — one in case A (the claim) and
one in case B (the remit). In practice drift is almost always
asymmetric (one side NULL), so count == count(pairs); for the
fully-symmetric minority, divide by ~2 when alerting. Real drift
should be fixed by repairing the writer path, not by counting.
Cases:
A. Claim ``matched_remittance_id = X`` but the paired remit X's
``claim_id`` is either NULL or doesn't point back to the claim.
B. Remit ``claim_id = A`` but the paired claim A's
``matched_remittance_id`` is either NULL or doesn't point back.
"""
import logging
from sqlalchemy import select
log = logging.getLogger(__name__)
with db.SessionLocal()() as s:
# Case A: claim says X is paired, but X.claim_id doesn't point back.
case_a = list(
s.execute(
select(
Claim.id.label("claim_id"),
Claim.matched_remittance_id.label("claimed_remit_id"),
Remittance.claim_id.label("remit_points_to"),
)
.outerjoin(
Remittance,
Remittance.id == Claim.matched_remittance_id,
)
.where(Claim.matched_remittance_id.is_not(None))
.where(
(Remittance.claim_id.is_(None))
| (Remittance.claim_id != Claim.id)
)
).all()
)
# Case B: remit says A is paired, but A.matched_remittance_id
# doesn't point back.
case_b = list(
s.execute(
select(
Remittance.id.label("remit_id"),
Remittance.claim_id.label("claimed_claim_id"),
Claim.matched_remittance_id.label("claim_points_to"),
)
.outerjoin(
Claim,
Claim.id == Remittance.claim_id,
)
.where(Remittance.claim_id.is_not(None))
.where(
(Claim.matched_remittance_id.is_(None))
| (Claim.matched_remittance_id != Remittance.id)
)
).all()
)
total = len(case_a) + len(case_b)
if total == 0:
log.info("matched-pair drift check: 0 mismatches (clean)")
return 0
log.warning(
"matched-pair drift check: %d mismatched pair(s) (showing up to 5 "
"of each). Investigate via SELECT against claim / remittance; "
"manual re-pair via /api/claims/{id}/manual-match will repair.",
total,
)
for r in case_a[:5]:
log.warning(
" case A: claim %s -> remit %s, but remit.claim_id=%r",
r.claim_id,
r.claimed_remit_id,
r.remit_points_to,
)
for r in case_b[:5]:
log.warning(
" case B: remit %s -> claim %s, but claim.matched_remittance_id=%r",
r.remit_id,
r.claimed_claim_id,
r.claim_points_to,
)
return total
# ---------------------------------------------------------------------------
# CycloneStore: the SQLAlchemy-backed facade.
# ---------------------------------------------------------------------------
+5 -4
View File
@@ -51,21 +51,22 @@ def test_migration_0002_creates_acks_table():
def test_migration_latest_idempotent_on_fresh_db():
"""Re-running the migration on the same DB must be a no-op (PRAGMA
user_version already at the latest version currently 15 after
user_version already at the latest version currently 16 after
0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
providers/payers/clearhouse, SP10's 0008 payer_rejected,
SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
SP16's 0011 processed_inbound_files, SP17's 0012 db_backups,
SP-auth's 0013 users + sessions, SP-audit's 0014 audit_log.user_id,
SP22's 0015 drop_claims_unique_constraint)."""
SP22's 0015 drop_claims_unique_constraint, SP27-Task 11's 0016
claims.matched_remittance_id index)."""
with db.engine().begin() as c:
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
assert v1 == 15
assert v1 == 16
# A second run should not raise and should not bump the version.
db_migrate.run(db.engine())
with db.engine().begin() as c:
v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
assert v2 == 15
assert v2 == 16
def test_add_ack_persists_row():
+5 -3
View File
@@ -119,14 +119,16 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
"""All migrations up to the current head run cleanly on a fresh DB,
and a second run is a no-op (no version bump). SP22 bumped the
expected head from 14 to 15 with the new UNIQUE-drop migration.
SP27 Task 11 bumped it to 16 with the claims.matched_remittance_id
index (drift-check perf).
"""
engine = _fresh_engine(tmp_path)
db_migrate.run(engine)
v_after_first = _user_version(engine)
assert v_after_first == 15, f"expected head=15, got {v_after_first}"
assert v_after_first == 16, f"expected head=16, got {v_after_first}"
db_migrate.run(engine)
assert _user_version(engine) == 15, "second run should not bump version"
assert _user_version(engine) == 16, "second run should not bump version"
def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -152,7 +154,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
engine = _fresh_engine(tmp_path)
db_migrate.run(engine)
assert _user_version(engine) == 15, f"expected head=15, got {_user_version(engine)}"
assert _user_version(engine) == 16, f"expected head=16, got {_user_version(engine)}"
# Two claims in one batch with the same patient_control_number
# must be insertable. If 0015's table recreation re-introduced a
+434
View File
@@ -0,0 +1,434 @@
"""SP27 Task 11: matched-pair invariant.
The matched pair is a denormalized FK pair ``Claim.matched_remittance_id``
and ``Remittance.claim_id`` that MUST agree in both directions at every
moment. Three writers can mutate it:
1. ``CycloneStore.manual_match`` operator override. SP27 Task 11 pins
it writes BOTH sides in one transaction.
2. ``CycloneStore.manual_unmatch`` operator reversal. SP27 Task 11
pins it clears BOTH sides in one transaction.
3. ``reconcile.run`` (inside ``CycloneStore.add`` post Task 10) auto-
match. Already pinned by the existing reconcile tests.
The drift check (``check_matched_pair_drift``) catches pre-existing
mismatches at startup and logs them at WARNING. Non-blocking.
These tests pin the invariants directly. They are explicitly redundant
with the integration tests in ``test_store_reconcile.py`` the goal
here is a focused regression pin so a future refactor that touches
``manual_match`` / ``manual_unmatch`` (e.g. to defer the symmetric write
to a post-commit pass) doesn't silently break the invariant.
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from decimal import Decimal
import pytest
from cyclone import db
from cyclone.parsers.models import (
BatchSummary, BillingProvider, ClaimHeader, ClaimOutput,
Envelope, Payer, ParseResult, Subscriber, ValidationReport,
)
from cyclone.parsers.models_835 import (
BatchSummary as BatchSummary835, ClaimAdjustment, ClaimPayment,
Envelope as Envelope835, FinancialInfo, ParseResult835, Payer835,
Payee835, ReassociationTrace, ServicePayment,
)
from cyclone.store import (
CycloneStore,
check_matched_pair_drift,
)
@pytest.fixture(autouse=True)
def _setup(tmp_path, monkeypatch):
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
db._reset_for_tests()
db.init_db()
yield
db._reset_for_tests()
def _build_claim(claim_id, pcn):
"""Build a single 837 claim + persist it through CycloneStore.add."""
from cyclone.store import BatchRecord837
co = ClaimOutput(
claim_id=claim_id,
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=pcn,
),
payer=Payer(name="Test Payer", id="P1"),
claim=ClaimHeader(
claim_id=claim_id, total_charge=Decimal("100"),
frequency_code="1", place_of_service="11",
),
diagnoses=[],
service_lines=[],
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=BatchSummary(
input_file="c.txt", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=1, passed=1, failed=0,
),
)
CycloneStore().add(BatchRecord837(
id="b-837", kind="837p", input_filename="c.txt",
parsed_at=date(2026, 6, 19).isoformat() + "T12:00:00+00:00",
result=pr837,
))
def _build_remit(remit_id, pcn):
"""Build a single 835 remit + persist it through CycloneStore.add."""
from datetime import datetime, timezone
from cyclone.store import BatchRecord835
cp = ClaimPayment(
payer_claim_control_number=pcn,
status_code="1",
status_label="Primary",
total_charge=Decimal("124.00"),
total_paid=Decimal("100.00"),
service_payments=[
ServicePayment(
line_number=1, procedure_qualifier="HC", procedure_code="99213",
charge=Decimal("124.00"), payment=Decimal("100.00"),
adjustments=[ClaimAdjustment(
group_code="CO", reason_code="45",
amount=Decimal("24.00"),
)],
),
],
)
pr835 = ParseResult835(
envelope=Envelope835(
sender_id="S", receiver_id="R", control_number="0001",
transaction_date=date(2026, 6, 19),
),
financial_info=FinancialInfo(
handling_code="C", paid_amount=Decimal("0"),
credit_debit_flag="C", payment_method=None,
),
trace=ReassociationTrace(
trace_type_code="1", trace_number="0001",
originating_company_id="S",
),
payer=Payer835(name="X", id="SKCO0"),
payee=Payee835(name="Y", npi="1234567890"),
claims=[cp],
summary=BatchSummary835(
input_file="era.txt", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=1, passed=1, failed=0,
),
)
CycloneStore().add(BatchRecord835(
id="b-835", kind="835", input_filename="era.txt",
parsed_at=datetime(2026, 6, 19, 12, 5, tzinfo=timezone.utc),
result=pr835,
))
# ---------------------------------------------------------------------------
# manual_match — writes BOTH sides in one transaction
# ---------------------------------------------------------------------------
def test_manual_match_writes_both_sides_in_one_transaction():
"""After ``manual_match`` returns, ``Claim.matched_remittance_id``
AND ``Remittance.claim_id`` both point at their pair.
Without the symmetric write, a later ``list_unmatched`` filter
(which keys off ``Remittance.claim_id IS NULL``) would surface
an already-matched remit as "orphan". Pre-T11 the symmetric
write was correct but only implicitly tested; this pins it
explicitly.
Note: claim PCN and remit PCN are DELIBERATELY DIFFERENT so the
auto-match inside ``CycloneStore.add`` (Task 10) doesn't pre-pair
them and ``manual_match`` becomes a no-op no-throw.
"""
from cyclone.db import Claim, Remittance
_build_claim("CLM-1", pcn="CLM-PATIENT-PCN")
_build_remit("CLP-1", pcn="CLP-1")
CycloneStore().manual_match("CLM-1", "CLP-1")
with db.SessionLocal()() as s:
claim = s.get(Claim, "CLM-1")
# Remit PK is payer_claim_control_number (see Remittance ORM).
from sqlalchemy import select
remit = s.execute(
select(Remittance).where(
Remittance.payer_claim_control_number == "CLP-1"
)
).scalar_one()
assert claim.matched_remittance_id == remit.id, (
f"claim.matched_remittance_id={claim.matched_remittance_id!r} "
f"!= remit.id={remit.id!r}"
)
assert remit.claim_id == claim.id, (
f"remit.claim_id={remit.claim_id!r} != claim.id={claim.id!r}"
)
# ---------------------------------------------------------------------------
# manual_unmatch — clears BOTH sides in one transaction
# ---------------------------------------------------------------------------
def test_manual_unmatch_clears_both_sides():
"""After ``manual_unmatch`` returns, BOTH ``Claim.matched_remittance_id``
AND ``Remittance.claim_id`` return to NULL.
Without the symmetric clear, an operator-initiated unmatch would
leave the orphaned remit pointing at the claim while the claim
is correctly cleared ``list_unmatched`` would then NOT surface
the pair (remit side is non-NULL), even though the operator
intended to put it back in the unmatched bucket.
PCN asymmetry prevents auto-match from pre-pairing see
``test_manual_match_writes_both_sides_in_one_transaction``.
"""
from sqlalchemy import select
from cyclone.db import Claim, Remittance
_build_claim("CLM-1", pcn="CLM-PATIENT-PCN")
_build_remit("CLP-1", pcn="CLP-1")
store = CycloneStore()
store.manual_match("CLM-1", "CLP-1")
store.manual_unmatch("CLM-1")
with db.SessionLocal()() as s:
claim = s.get(Claim, "CLM-1")
remit = s.execute(
select(Remittance).where(
Remittance.payer_claim_control_number == "CLP-1"
)
).scalar_one()
assert claim.matched_remittance_id is None
assert remit.claim_id is None
def test_manual_match_rollback_clears_symmetric_writes(monkeypatch):
"""If anything between the two writes raises, the session's
rollback MUST clear BOTH writes neither side set.
Pins the SP27 Task 11 invariant "writes BOTH sides in one
transaction" at the rollback level. Without this pin, a future
refactor that pushes the symmetric write to a post-commit
handler (or splits ``manual_match`` across two sessions) would
still pass the happy-path test but leave one side set on a
failed match the very class of drift Task 11 was added to
surface.
"""
from sqlalchemy import select
from cyclone.db import Claim, Remittance
from cyclone import reconcile
_build_claim("CLM-ROLL", pcn="CLM-PATIENT-PCN")
_build_remit("CLP-ROLL", pcn="CLP-ROLL")
# Force the second-loop line-reconcile pass to raise. By the time
# it runs, the claim + remit writes are staged in the session
# but not committed — the raise must trigger a rollback.
def boom(session, claim, remittance):
raise RuntimeError("simulated post-write fault")
monkeypatch.setattr(reconcile, "_reconcile_pair", boom)
with pytest.raises(RuntimeError, match="simulated post-write fault"):
CycloneStore().manual_match("CLM-ROLL", "CLP-ROLL")
with db.SessionLocal()() as s:
claim = s.get(Claim, "CLM-ROLL")
remit = s.execute(
select(Remittance).where(
Remittance.payer_claim_control_number == "CLP-ROLL"
)
).scalar_one()
# Neither side set — the session rolled back the partial
# state machine. claim.matched_remittance_id is the seeded
# NULL (no manual match); remit.claim_id is seeded NULL.
assert claim.matched_remittance_id is None
assert remit.claim_id is None
# ---------------------------------------------------------------------------
# Drift check — startup audit, non-blocking
# ---------------------------------------------------------------------------
def test_drift_check_returns_zero_for_clean_db():
"""A clean DB (no manual_match calls, no remits with claim_id)
reports zero drift and does not log any warning."""
_build_claim("CLM-1", pcn="PCN-A")
# No remits paired — no drift expected.
count = check_matched_pair_drift()
assert count == 0
def test_drift_check_logs_warning_for_mismatched_pair(caplog):
"""Inject drift directly via SQL (bypassing store so we can build
a state the writers can never produce), call the check, assert
a WARNING was logged with the offending ids.
The drift check is read-only and non-blocking operators can
clean up manually. Logging at WARNING makes it visible in the
standard boot log without paging anyone.
This case is the inverse direction: a REMIT points at a claim
whose ``matched_remittance_id`` doesn't point back. Pairs with
``test_drift_check_logs_warning_for_case_a_mismatch`` to cover
both SQL paths through the check.
"""
from cyclone.db import Claim, ClaimState, Remittance
from sqlalchemy import select
# Build the inverse state: a remit whose claim_id points at a
# claim that does NOT have matched_remittance_id back. Only Case
# B fires; Case A (claim → remit) is clean by construction.
with db.SessionLocal()() as s:
s.add(Remittance(
id="orphan-remit", batch_id="b1",
payer_claim_control_number="PCN-B",
status_code="1", total_charge=Decimal("100"),
total_paid=Decimal("100"),
received_at=datetime.now(timezone.utc),
service_date=None, is_reversal=False,
claim_id="orphan-claim", # ← only this side set
))
s.add(Claim(
id="orphan-claim", batch_id="b1",
patient_control_number="PCN-B-CLAIM",
service_date_from=None,
charge_amount=Decimal("100"),
state=ClaimState.SUBMITTED,
matched_remittance_id=None, # ← back-pointer missing
))
s.commit()
with caplog.at_level("WARNING", logger="cyclone.store"):
count = check_matched_pair_drift()
assert count == 1, "exactly one drift was injected"
# The warning log should mention both ids so the operator can grep.
assert any(
"orphan-claim" in record.message
and "orphan-remit" in record.message
for record in caplog.records
), "drift warning should mention the offending claim + remit ids"
def test_drift_check_logs_warning_for_case_a_mismatch(caplog):
"""Pins the OTHER direction: claim → remit drift where the
claim's matched_remittance_id points at a remit whose claim_id
doesn't point back.
Without this test (paired with the inverse test above) a SQL
typo in Case A or Case B would silently pass the suite.
"""
from cyclone.db import Claim, ClaimState, Remittance
with db.SessionLocal()() as s:
# Remit exists but its claim_id is NULL (claim side is the
# only one pointing).
s.add(Remittance(
id="dangle-remit", batch_id="b1",
payer_claim_control_number="PCN-A",
status_code="1", total_charge=Decimal("100"),
total_paid=Decimal("100"),
received_at=datetime.now(timezone.utc),
service_date=None, is_reversal=False,
claim_id=None,
))
s.add(Claim(
id="dangle-claim", batch_id="b1",
patient_control_number="PCN-A",
service_date_from=None,
charge_amount=Decimal("100"),
state=ClaimState.SUBMITTED,
matched_remittance_id="dangle-remit",
))
s.commit()
with caplog.at_level("WARNING", logger="cyclone.store"):
count = check_matched_pair_drift()
assert count == 1
assert any(
"dangle-claim" in record.message and "dangle-remit" in record.message
for record in caplog.records
)
def test_drift_check_returns_count_of_mismatched_pairs():
"""The return value is the number of mismatched pairs so operators
can alert on it (e.g. fail boot if drift > 0 in a future iteration)."""
from cyclone.db import Claim, ClaimState, Remittance
with db.SessionLocal()() as s:
# Two drifted pairs: case A and case B in a single DB.
for i, (claim_id, remit_id) in enumerate(
[("drift-A-claim", "drift-A-remit"), ("drift-B-claim", "drift-B-remit")]
):
s.add(Remittance(
id=remit_id, batch_id="b1",
payer_claim_control_number=f"PCN-{i}",
status_code="1", total_charge=Decimal("100"),
total_paid=Decimal("100"),
received_at=datetime.now(timezone.utc),
service_date=None, is_reversal=False,
))
s.add(Claim(
id=claim_id, batch_id="b1",
patient_control_number=f"PCN-{i}",
service_date_from=None,
charge_amount=Decimal("100"),
state=ClaimState.SUBMITTED,
matched_remittance_id=remit_id,
))
s.commit()
count = check_matched_pair_drift()
assert count == 2
def test_drift_check_does_not_raise_on_query_error(monkeypatch):
"""The drift check is non-blocking at boot — if a DB query fails
we want a logged exception, NOT a crashed backend. Wrapped at the
api.py call site (Task 11 wiring); this pins that the function
itself doesn't try/catch (the call site does).
"""
from sqlalchemy.exc import OperationalError
# Force the SessionLocal to raise on use.
def boom():
raise OperationalError("SELECT 1", {}, Exception("synthetic"))
monkeypatch.setattr(db, "SessionLocal", lambda: boom)
with pytest.raises(OperationalError):
check_matched_pair_drift()
# The exception propagates — api.py::lifespan wraps it with a
# try/except so the boot continues. The pin here is that the
# function doesn't swallow by accident.