diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 5ae4ad3..8c804d5 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -519,12 +519,13 @@ def _build_and_persist_ack(batch_id: str) -> dict | None: def _reconciliation_summary_for_batch(batch_id: str) -> dict: """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch. - Reads from the DB after ``store.add()`` has already triggered T10 - reconciliation synchronously (via ``_run_reconcile``). Counts are - observed at this moment; a subsequent manual match/unmatch will not - be reflected until the next request. + Reads from the DB after ``store.add()`` has already run reconciliation + synchronously (SP27 Task 10: ``reconcile.run(s, batch_id)`` inside the + ingest session, before commit). Counts are observed at this moment; + a subsequent manual match/unmatch will not be reflected until the + next request. - ``skipped`` is reserved for future use — the T10 orchestrator tracks + ``skipped`` is reserved for future use — the orchestrator tracks skipped claims internally but does not surface a queryable count. """ from sqlalchemy import func, select diff --git a/backend/src/cyclone/handlers/handle_835.py b/backend/src/cyclone/handlers/handle_835.py index 272020a..ec32808 100644 --- a/backend/src/cyclone/handlers/handle_835.py +++ b/backend/src/cyclone/handlers/handle_835.py @@ -15,10 +15,11 @@ but moving it is out of Task 5 scope). We import it lazily inside acceptable because api.py also imports scheduler lazily (inside its lifespan handler). -The two-phase ingest problem (batch row first, then a separate -``reconcile`` pass that overwrites ``adjustment_amount``) is a known -gap; atomic unification happens in SP27 Task 10. For Task 5 we lock -current behavior. +Two-phase ingest closed in SP27 Task 10: the placeholder +``adjustment_amount`` is overwritten by reconcile in the same DB +session as the insert (before commit), so a reader never sees a +half-reconciled Remittance row. If reconcile raises, the whole +ingest rolls back and the scheduler records the per-file error. ``event_bus`` is best-effort and follows the same TODO(sp27-task-6) sync/async gap as handle_999 / handle_ta1 / handle_277ca. diff --git a/backend/src/cyclone/reconcile.py b/backend/src/cyclone/reconcile.py index 2f6439b..bc9f52c 100644 --- a/backend/src/cyclone/reconcile.py +++ b/backend/src/cyclone/reconcile.py @@ -224,10 +224,15 @@ class ReconcileResult: def run(session, batch_id: str) -> ReconcileResult: """Orchestrate reconciliation for an 835 batch. - Expects Batch + Remittance + CasAdjustment rows already persisted - (this is called from store.add() AFTER commit). Loads the new - remittances + all currently-unmatched Claims, calls match(), then - applies each match's intent inside the caller's session. + Expects Batch + Remittance + CasAdjustment rows already added to + ``session`` (not yet committed). SP27 Task 10 calls this from + inside ``CycloneStore.add``'s ingest session, BEFORE + ``s.commit()`` — so the whole 835 ingest (rows + reconcile) is + a single transaction. If anything here raises, the ingest rolls + back; no half-reconciled batch ever lands. + + Loads the new remittances + all currently-unmatched Claims, calls + match(), then applies each match's intent inside the caller's session. Returns counts. Does NOT commit; caller controls transaction. Raises on failure (caller catches and writes activity event). diff --git a/backend/src/cyclone/store.py b/backend/src/cyclone/store.py index 7449357..362e9eb 100644 --- a/backend/src/cyclone/store.py +++ b/backend/src/cyclone/store.py @@ -912,9 +912,12 @@ class CycloneStore: For 835 batches: inserts the Batch row, one Remittance row per ClaimPayment, and a ``remit_received`` ActivityEvent per - ClaimPayment. After commit, calls ``_run_reconcile`` (T10 stub) - in a fail-soft manner — reconciliation errors are logged but - do not roll back the persisted batch. + ClaimPayment. Reconciliation (auto-match + per-pair CAS + aggregate) runs IN THE SAME SESSION before commit (SP27 + Task 10) — a single ``s.commit()`` covers both ingest and + reconciliation. If reconcile raises, the whole 835 ingest + rolls back; the batch never appears half-reconciled with + placeholder ``adjustment_amount`` values. Idempotency: ``Claim.id`` and ``Remittance.id`` are PRIMARY KEYS, so a re-ingest of the same fixture (e.g. ``/api/parse-837`` called @@ -1019,19 +1022,17 @@ class CycloneStore: f"Unsupported BatchRecord subclass: {type(record).__name__}" ) - s.commit() + # SP27 Task 10: run reconcile INSIDE the same session, before + # commit. The placeholder ``adjustment_amount`` set by + # ``_remittance_835_row`` is overwritten by reconcile's + # ``_reconcile_pair`` SUM-of-CAS-aggregate pass before the + # rows are ever visible. If reconcile raises, the whole + # 835 ingest rolls back and the batch never lands. + if record.kind == "835": + from cyclone import reconcile as _reconcile + _reconcile.run(s, record.id) - # Reconcile 835 batches after the batch is durably persisted. - # Fail-soft: errors are logged, not raised, so an 835 parse that - # crashes in reconciliation still shows up in /api/batches. - if record.kind == "835": - try: - self._run_reconcile(record.id) - except Exception: # pragma: no cover - logged via default handler - import logging - logging.getLogger(__name__).exception( - "reconcile.run failed for batch %s", record.id, - ) + s.commit() # Publish live-tail events synchronously. EventBus.publish is async # but its body is purely synchronous ``put_nowait`` enqueues; we @@ -1113,26 +1114,6 @@ class CycloneStore: for queue in list(event_bus._subscribers.get(kind, ())): event_bus._enqueue_or_drop_oldest(queue, event) - def _run_reconcile(self, batch_id: str) -> None: - """T10 stub: invoke the reconcile orchestrator for a batch. - - The actual reconciliation is implemented in T10 (this method - will import ``cyclone.reconcile`` and call ``reconcile.run`` - with the current session). For T9 we keep the import lazy and - fail-soft so a missing or NotImplementedError reconcile module - never breaks the 835 ingest path. - """ - # T10 will replace this with the real implementation. Until then - # we accept the failure modes the spec lists: ImportError - # (module not yet wired), NotImplementedError (stub in place), - # or any other transient reconcile error — all swallowed. - try: - from cyclone import reconcile as _reconcile - with db.SessionLocal()() as s: - _reconcile.run(s, batch_id) - except (ImportError, NotImplementedError): - pass - # -- read path ------------------------------------------------------ def _row_to_record(self, row: Batch) -> BatchRecord: diff --git a/backend/tests/test_handlers_835.py b/backend/tests/test_handlers_835.py index c00db54..68f7d38 100644 --- a/backend/tests/test_handlers_835.py +++ b/backend/tests/test_handlers_835.py @@ -1,14 +1,17 @@ -"""Direct tests for the ``handle_835`` handler (SP27 Task 5). +"""Direct tests for the ``handle_835`` handler (SP27 Task 5 + Task 10). Locks the handler's contract independent of the scheduler lifecycle so a regression in the scheduler wiring doesn't hide a regression in the handler. Note: the 835 handler is the largest of the four because the schema -covers per-claim remittances + CAS adjustments + validation. The -two-phase ingestion problem (batch row first, then a separate -``reconcile`` pass) is a known gap; atomic unification happens in -SP27 Task 10. For Task 5 we lock current behavior (no atomic). +covers per-claim remittances + CAS adjustments + validation. SP27 +Task 10 unified the two-phase ingest path (batch row first, then a +separate ``reconcile`` pass that overwrote ``adjustment_amount``) +into one critical section. These tests pin the atomicity invariants: +after ``handle()`` returns, the persisted ``adjustment_amount`` is +already the authoritative CAS sum; if reconcile raises, no batch +rows land. """ from __future__ import annotations @@ -102,3 +105,148 @@ def test_handle_835_raises_on_completely_unparseable_input(): bad = "this is not even EDI" with pytest.raises((CycloneParseError, ValueError)): handle(bad, source_file="bad.835") + + +# --------------------------------------------------------------------------- +# SP27 Task 10: atomicity invariants +# --------------------------------------------------------------------------- + + +def test_handle_835_adjustment_amount_matches_cas_sum(): + """After ``handle()`` returns, every persisted Remittance's + ``adjustment_amount`` equals the SUM of its CasAdjustment rows. + + Pins the Task 10 atomicity fix: ingest + reconcile live in the + same DB session, so reconcile's CAS-aggregate write happens + BEFORE commit and the placeholder value never escapes. Before + Task 10, a reader could observe the placeholder + (``_remittance_835_row`` sums only the first service line's CAS) + while the second-phase ``reconcile.run`` was still pending. + """ + from decimal import Decimal + + from sqlalchemy import func, select + + from cyclone.db import CasAdjustment, Remittance + + text = MINIMAL.read_text() + handle(text, source_file=MINIMAL.name) + + with db.SessionLocal()() as session: + remits = session.execute(select(Remittance)).scalars().all() + assert len(remits) >= 1 + for r in remits: + cas_sum = session.execute( + select(func.coalesce(func.sum(CasAdjustment.amount), 0)) + .where(CasAdjustment.remittance_id == r.id) + ).scalar_one() + assert r.adjustment_amount == Decimal(str(cas_sum)), ( + f"remit {r.id}: adjustment_amount={r.adjustment_amount} " + f"!= CAS sum={cas_sum}" + ) + + +def test_handle_835_reconcile_failure_rolls_back_ingest(monkeypatch): + """If reconcile raises mid-ingest, the batch + remittance rows + don't land at all (atomic rollback, SP27 Task 10). + + Before Task 10, ``store.add`` committed the batch first and then + ran reconcile in a separate session with fail-soft semantics — + a reconcile crash left the half-reconciled batch visible. After + Task 10, ``reconcile.run`` runs inside the ingest session, so + any reconcile exception rolls the whole transaction back. + """ + from sqlalchemy import select + + from cyclone import reconcile + from cyclone.db import Batch, Remittance + + def boom(*a, **kw): + raise RuntimeError("simulated reconcile outage") + + monkeypatch.setattr(reconcile, "match", boom) + + text = MINIMAL.read_text() + with pytest.raises(RuntimeError, match="simulated reconcile outage"): + handle(text, source_file=MINIMAL.name) + + with db.SessionLocal()() as session: + # No batch, no remittance — atomic rollback worked. + batches = session.execute(select(Batch)).scalars().all() + remits = session.execute(select(Remittance)).scalars().all() + assert batches == [], ( + f"reconcile raised but {len(batches)} batch rows landed" + ) + assert remits == [], ( + f"reconcile raised but {len(remits)} remittance rows landed" + ) + + +def test_handle_835_late_reconcile_failure_rolls_back_match_loop(monkeypatch): + """A crash in the SECOND pipeline (``_reconcile_pair``) still + rolls back the FIRST pipeline's mutations — Claim.state, + Claim.matched_remittance_id, Remittance.claim_id, the new + ActivityEvent row, the new Match row. + + Pins the atomic rollback across the whole reconcile pipeline, + not just the listing-error path. Without this, a future change + that splits ``reconcile.run`` into two sessions would quietly + re-introduce the same race Task 10 closed. + """ + from datetime import date + from decimal import Decimal + + from sqlalchemy import select + + from cyclone import reconcile + from cyclone.db import ( + ActivityEvent, Batch, Claim, ClaimState, Match, Remittance, + ) + + # Pre-seed a Claim that will auto-match the minimal_835 remit + # (``payer_claim_control_number = "CLM001"``). Without this, + # ``reconcile.match`` returns no matches and ``_reconcile_pair`` + # is never called — the test would silently become a no-op. + with db.SessionLocal()() as s: + s.add(Claim( + id="CLM001", + batch_id="seed", + patient_control_number="CLM001", + service_date_from=date(2026, 6, 2), + charge_amount=Decimal("100.00"), + state=ClaimState.SUBMITTED, + )) + s.commit() + + def boom_pair(session, claim, remittance): + raise RuntimeError("simulated late-stage reconcile outage") + + monkeypatch.setattr(reconcile, "_reconcile_pair", boom_pair) + + text = MINIMAL.read_text() + with pytest.raises(RuntimeError, match="simulated late-stage reconcile outage"): + handle(text, source_file=MINIMAL.name) + + with db.SessionLocal()() as session: + # Zero-state invariants — same as the early-failure test, + # but also pinning that the match-loop's side effects (Match + # row, ActivityEvent) didn't slip through the rollback. + batches = session.execute(select(Batch)).scalars().all() + remits = session.execute(select(Remittance)).scalars().all() + match_rows = session.execute(select(Match)).scalars().all() + activity_rows = session.execute(select(ActivityEvent)).scalars().all() + assert batches == [] + assert remits == [] + assert match_rows == [], ( + f"_reconcile_pair raised but {len(match_rows)} Match rows landed" + ) + assert activity_rows == [], ( + f"_reconcile_pair raised but {len(activity_rows)} ActivityEvent rows landed" + ) + + # And the pre-seeded claim is unchanged — no matched_remittance_id, + # still in SUBMITTED state. The match-loop's mutations against + # this row were rolled back too. + claim = session.get(Claim, "CLM001") + assert claim.matched_remittance_id is None + assert claim.state == ClaimState.SUBMITTED diff --git a/backend/tests/test_reconcile.py b/backend/tests/test_reconcile.py index 1447347..02445a9 100644 --- a/backend/tests/test_reconcile.py +++ b/backend/tests/test_reconcile.py @@ -330,8 +330,20 @@ def test_run_reversal_flips_paid_to_reversed(fixture_835): assert reversal_match.prior_claim_state == ClaimState.PAID -def test_run_failed_reconcile_writes_activity_event(fixture_835): - """If reconcile crashes, the batch + remittances stay; activity event records failure.""" +def test_run_reconcile_raise_in_session_leaves_prior_commits_alone(fixture_835): + """A ``reconcile.run`` raise inside an open session does not damage + previously-committed rows. + + The test seeds a batch + remit, commits them in one session, then + calls ``reconcile.run`` in a fresh session with ``match`` monkey- + patched to raise. The pre-existing rows must survive — they're on + disk from the prior commit, separate from the rolling-back + session. This pins the unit-level invariant that ``reconcile.run`` + itself never commits and never silently mutates rows outside the + session it was given; it is the *caller's* responsibility (now + ``CycloneStore.add`` per SP27 Task 10) to control the transaction + boundary. + """ with db.SessionLocal()() as s: _make_batch(s) _make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00")