feat(backend): add reconcile.run orchestrator with match + apply + CAS agg

backend/src/cyclone/reconcile.py

@@ -18,10 +18,12 @@ Match algorithm:
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import date
+from datetime import date, datetime, timezone
 from decimal import Decimal
 from typing import Iterable, Optional, Protocol
 
+from sqlalchemy import func
+
 from cyclone.db import ClaimState
 
 
@@ -206,3 +208,111 @@ def split_unmatched(
         [c for c in claims if c.id not in matched_claim_ids],
         [r for r in remits if r.id not in matched_remit_ids],
     )
+
+
+# --- reconcile.run() orchestrator ---------------------------------------------
+
+
+@dataclass
+class ReconcileResult:
+    matched: int = 0
+    unmatched_claims: int = 0
+    unmatched_remittances: int = 0
+    skipped: int = 0
+
+
+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.
+
+    Returns counts. Does NOT commit; caller controls transaction.
+    Raises on failure (caller catches and writes activity event).
+    """
+    from sqlalchemy import select
+    from cyclone.db import (
+        Batch, Claim, Remittance, CasAdjustment, Match, ActivityEvent,
+        ClaimState,
+    )
+
+    new_remits = list(
+        session.execute(
+            select(Remittance).where(Remittance.batch_id == batch_id)
+        ).scalars().all()
+    )
+
+    unmatched_claims = list(
+        session.execute(
+            select(Claim).where(Claim.matched_remittance_id.is_(None))
+        ).scalars().all()
+    )
+
+    matches = match(unmatched_claims, new_remits)
+
+    applied = 0
+    skipped = 0
+    for m in matches:
+        if m.is_reversal:
+            intent = apply_reversal(m.claim, m.remittance)
+        else:
+            intent = apply_payment(
+                m.claim, m.remittance,
+                charge=m.claim.charge_amount, paid=m.remittance.total_paid,
+                status_code=m.remittance.status_code,
+            )
+
+        if intent.skipped:
+            session.add(ActivityEvent(
+                ts=datetime.now(timezone.utc), kind=intent.activity_kind,
+                batch_id=batch_id, claim_id=m.claim.id,
+                remittance_id=m.remittance.id,
+                payload_json={"reason": intent.reason,
+                              "current_state": getattr(m.claim.state, "value", m.claim.state)},
+            ))
+            skipped += 1
+            continue
+
+        session.add(Match(
+            claim_id=m.claim.id, remittance_id=m.remittance.id,
+            strategy=m.strategy,
+            matched_at=datetime.now(timezone.utc),
+            prior_claim_state=intent.prior_claim_state,
+            is_reversal=m.is_reversal,
+        ))
+        if not m.is_reversal:
+            m.claim.state = intent.new_state
+        else:
+            m.claim.state = ClaimState.REVERSED
+        m.claim.matched_remittance_id = m.remittance.id
+
+        session.add(ActivityEvent(
+            ts=datetime.now(timezone.utc), kind=intent.activity_kind,
+            batch_id=batch_id, claim_id=m.claim.id,
+            remittance_id=m.remittance.id,
+            payload_json={"new_state": m.claim.state.value},
+        ))
+        applied += 1
+
+    # 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
+
+    # Partition.
+    unmatched_claims_after, unmatched_remits_after = split_unmatched(
+        unmatched_claims, new_remits,
+        [m for m in matches if m.claim.id in {c.id for c in unmatched_claims}],
+    )
+
+    return ReconcileResult(
+        matched=applied,
+        unmatched_claims=len(unmatched_claims_after),
+        unmatched_remittances=len(unmatched_remits_after),
+        skipped=skipped,
+    )

backend/tests/test_reconcile.py

@@ -7,15 +7,22 @@ SQLAlchemy ORM instances, so the reconcile module has zero DB dependency.
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import date
+from datetime import date, datetime, timezone
 from decimal import Decimal
 from typing import Optional
 
 import pytest
 
-from cyclone import reconcile
+from cyclone import db, reconcile
-from cyclone.db import ClaimState
+from cyclone.db import (
-from cyclone.reconcile import Match
+    ActivityEvent,
+    Batch,
+    Claim,
+    ClaimState,
+    Match,
+    Remittance,
+)
+from cyclone.reconcile import Match as ReconcileMatch
 
 
 # --- Stand-in types -----------------------------------------------------------
@@ -196,7 +203,153 @@ def test_apply_reversal_of_already_reversed_is_noop():
 def test_split_unmatched_partitions_by_match_set():
     claims = [FakeClaim("CLM-1", "A", None), FakeClaim("CLM-2", "B", None)]
     remits = [FakeRemit("CLP-1", "A", None), FakeRemit("CLP-2", "C", None)]
-    matches = [Match(claim=claims[0], remittance=remits[0], strategy="auto", is_reversal=False)]
+    matches = [ReconcileMatch(claim=claims[0], remittance=remits[0], strategy="auto", is_reversal=False)]
     unmatched_claims, unmatched_remits = reconcile.split_unmatched(claims, remits, matches)
     assert [c.id for c in unmatched_claims] == ["CLM-2"]
     assert [r.id for r in unmatched_remits] == ["CLP-2"]
+
+
+# --- reconcile.run() integration tests ----------------------------------------
+
+
+@pytest.fixture
+def fixture_835(tmp_path, monkeypatch):
+    """Set up a clean DB and return an empty helper for building remits."""
+    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/reconcile.db")
+    db._reset_for_tests()
+    db.init_db()
+    yield
+    db._reset_for_tests()
+
+
+def _make_batch(s, batch_id="b1", kind="835"):
+    s.add(Batch(
+        id=batch_id, kind=kind, input_filename="x",
+        parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
+    ))
+    s.flush()
+
+
+def _make_claim(s, claim_id="CLM-1", pcn="PCN-A", charge="100.00",
+                service_date=None, state=ClaimState.SUBMITTED):
+    s.add(Claim(
+        id=claim_id, batch_id="b1", patient_control_number=pcn,
+        service_date_from=service_date, charge_amount=Decimal(charge),
+        state=state,
+    ))
+    s.flush()
+
+
+def _make_remit(s, remit_id="CLP-1", pcn="PCN-A", status="1",
+                charge="100.00", paid="100.00", service_date=None,
+                is_reversal=False):
+    rid = remit_id + ":b1xxxxxx"
+    s.add(Remittance(
+        id=rid, batch_id="b1", payer_claim_control_number=pcn,
+        status_code=status, total_charge=Decimal(charge), total_paid=Decimal(paid),
+        received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
+        service_date=service_date, is_reversal=is_reversal,
+    ))
+    s.flush()
+    return rid
+
+
+def test_run_matches_and_updates_state(fixture_835):
+    with db.SessionLocal()() as s:
+        _make_batch(s)
+        _make_claim(s, "CLM-1", "PCN-A", "100.00", date(2026, 6, 1))
+        _make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00", date(2026, 6, 1))
+        s.commit()
+
+    with db.SessionLocal()() as s:
+        from cyclone import reconcile
+        result = reconcile.run(s, "b1")
+        s.commit()
+
+    assert result.matched == 1
+    assert result.unmatched_remittances == 0
+    assert result.unmatched_claims == 0
+
+    with db.SessionLocal()() as s:
+        claim = s.get(Claim, "CLM-1")
+        assert claim.state == ClaimState.PAID
+        assert claim.matched_remittance_id == "CLP-1:b1xxxxxx"
+
+
+def test_run_orphan_remit_leaves_claim_unmatched(fixture_835):
+    with db.SessionLocal()() as s:
+        _make_batch(s)
+        _make_claim(s, "CLM-1", "PCN-A", "100.00", date(2026, 6, 1))
+        _make_remit(s, "CLP-2", "PCN-NEW", "1", "50.00", "50.00", date(2026, 6, 5))
+        s.commit()
+
+    with db.SessionLocal()() as s:
+        from cyclone import reconcile
+        result = reconcile.run(s, "b1")
+        s.commit()
+
+    assert result.matched == 0
+    assert result.unmatched_remittances == 1
+    assert result.unmatched_claims == 0
+
+
+def test_run_reversal_flips_paid_to_reversed(fixture_835):
+    with db.SessionLocal()() as s:
+        _make_batch(s)
+        _make_claim(s, "CLM-1", "PCN-A", "100.00", date(2026, 6, 1),
+                    state=ClaimState.PAID)
+        # Match row already exists from prior reconcile.
+        s.add(Match(
+            claim_id="CLM-1", remittance_id="CLP-OLD:b1xxxxxx",
+            strategy="auto", matched_at=datetime.now(timezone.utc),
+            is_reversal=False,
+        ))
+        s.flush()
+        _make_remit(s, "CLP-REV", "PCN-A", "22", "100.00", "-100.00",
+                    date(2026, 6, 10), is_reversal=True)
+        s.commit()
+
+    with db.SessionLocal()() as s:
+        from cyclone import reconcile
+        result = reconcile.run(s, "b1")
+        s.commit()
+
+    assert result.matched == 1
+    with db.SessionLocal()() as s:
+        claim = s.get(Claim, "CLM-1")
+        assert claim.state == ClaimState.REVERSED
+        # Match rows: original + reversal.
+        matches = s.query(Match).filter(Match.claim_id == "CLM-1").all()
+        assert len(matches) == 2
+        reversal_match = [m for m in matches if m.is_reversal][0]
+        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."""
+    with db.SessionLocal()() as s:
+        _make_batch(s)
+        _make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00")
+        s.commit()
+
+    # Monkeypatch reconcile.match to raise.
+    from cyclone import reconcile
+    real_match = reconcile.match
+    def boom(*a, **kw): raise RuntimeError("synthetic fault")
+    reconcile.match = boom
+    try:
+        with db.SessionLocal()() as s:
+            try:
+                reconcile.run(s, "b1")
+                s.commit()
+            except RuntimeError:
+                pass  # caller is responsible for catch; reconcile.run itself shouldn't raise
+    finally:
+        reconcile.match = real_match
+
+    # Batch + remit still present.
+    with db.SessionLocal()() as s:
+        b = s.get(Batch, "b1")
+        assert b is not None
+        r = s.query(Remittance).first()
+        assert r is not None