feat(sp31): add _score_fallback_candidates DB helper with ±30-day candidate pool

backend/src/cyclone/reconcile.py

@@ -128,6 +128,14 @@ def _content_keys_match(remit, claim) -> bool:
 
     Charge compared with ``CHARGE_TOLERANCE`` tolerance (rounding drift).
     NPI counts as "not matched" when the remit's NPI is empty.
+
+    Uses duck-typing via ``getattr`` so the helper works against both:
+      - shim / dataclass test objects (planned names: total_charge_amount,
+        total_charge, rendering_provider_npi), and
+      - real SQLAlchemy ORM instances (Claim.charge_amount, Remittance.total_charge,
+        Claim.provider_npi). The Remittance ORM has no NPI column — callers
+        that need NPI on a Remit set the transient ``rendering_provider_npi``
+        attribute (the 835 parser does this on raw_json; tests do it directly).
     """
     keys_matched = 0
 
@@ -136,22 +144,85 @@ def _content_keys_match(remit, claim) -> bool:
        (getattr(claim, "patient_control_number", "") or "").strip():
         keys_matched += 1
 
-    # Charge: total_charge_amount ↔ total_charge (within $0.01).
+    # Charge: total_charge_amount/total_charge ↔ total_charge/charge_amount (within $0.01).
-    remit_charge = getattr(remit, "total_charge_amount", None)
+    remit_charge = getattr(remit, "total_charge_amount", None) or getattr(remit, "total_charge", None)
-    claim_charge = getattr(claim, "total_charge", None)
+    claim_charge = getattr(claim, "total_charge", None) or getattr(claim, "charge_amount", None)
     if remit_charge is not None and claim_charge is not None and \
        abs(remit_charge - claim_charge) < CHARGE_TOLERANCE:
         keys_matched += 1
 
-    # NPI: rendering_provider_npi ↔ rendering_provider_npi (must be non-empty).
+    # NPI: rendering_provider_npi (or provider_npi fallback on Claim) ↔
+    #      rendering_provider_npi (must be non-empty on the remit side).
     remit_npi = (getattr(remit, "rendering_provider_npi", "") or "").strip()
-    claim_npi = (getattr(claim, "rendering_provider_npi", "") or "").strip()
+    claim_npi = (
+        (getattr(claim, "rendering_provider_npi", "") or "")
+        or (getattr(claim, "provider_npi", "") or "")
+    ).strip()
     if remit_npi and remit_npi == claim_npi:
         keys_matched += 1
 
     return keys_matched >= KEYS_REQUIRED
 
 
+# --- SP31: DB-side fallback candidate matcher -------------------------------
+
+SCORE_FALLBACK_WINDOW_DAYS = 30
+
+
+def _score_fallback_candidates(session, remit) -> Optional[str]:
+    """Return a unique matching claim_id via the 2-of-3 content-keys rule.
+
+    Queries a ±SCORE_FALLBACK_WINDOW_DAYS candidate pool filtered by:
+    - claim.matched_remittance_id IS NULL (not yet linked)
+    - claim.state NOT IN terminal set (PAID, DENIED, REJECTED,
+      REVERSED, RECONCILED)
+    - claim.service_date_from within window of remit.service_date
+
+    Returns the unique claim.id when exactly one candidate passes the
+    2-of-3 rule (via :func:`_content_keys_match`), otherwise ``None``.
+    Ambiguous matches (2+ candidates) also return ``None`` — they land
+    in the Inbox Unlinked lane for manual review.
+
+    Note on payer_id: ``Remittance`` has no ``payer_id`` column
+    (``Claim.payer_id`` does, and is the source of truth on the claim
+    side; the 835 parser stores payer_id in ``Remittance.raw_json``).
+    Filtering by payer would require either a raw_json read or a model
+    column add — out of scope for SP31 Task 3. The PCN itself is
+    expected to be unique within a payer's submission, so cross-payer
+    collisions are unlikely. Revisit if production shows otherwise.
+    """
+    from sqlalchemy import select, and_
+    from datetime import timedelta
+    from cyclone.db import Claim, ClaimState
+
+    TERMINAL = {
+        ClaimState.PAID, ClaimState.DENIED, ClaimState.REJECTED,
+        ClaimState.REVERSED, ClaimState.RECONCILED,
+    }
+
+    if remit.service_date is None:
+        return None  # need a date for the window query
+
+    window_lo = remit.service_date - timedelta(days=SCORE_FALLBACK_WINDOW_DAYS)
+    window_hi = remit.service_date + timedelta(days=SCORE_FALLBACK_WINDOW_DAYS)
+
+    candidates = list(session.execute(
+        select(Claim).where(
+            and_(
+                Claim.matched_remittance_id.is_(None),
+                Claim.service_date_from >= window_lo,
+                Claim.service_date_from <= window_hi,
+                Claim.state.notin_([s.value for s in TERMINAL]),
+            )
+        )
+    ).scalars().all())
+
+    matched_ids = [c.id for c in candidates if _content_keys_match(remit, c)]
+    if len(matched_ids) == 1:
+        return matched_ids[0]
+    return None  # 0 matches OR 2+ (ambiguous)
+
+
 @dataclass
 class ApplyIntent:
     """Result of applying a match. The caller persists this.

backend/tests/conftest.py

@@ -14,6 +14,9 @@ test_api_parse_persists.py) working unchanged.
 
 from __future__ import annotations
 
+from datetime import datetime, timezone
+from decimal import Decimal
+
 import pytest
 
 
@@ -89,4 +92,137 @@ def _reset_rate_limit_buckets(app) -> None:
         cur = getattr(cur, "app", None)
     # No RateLimitMiddleware in the stack — nothing to reset. Should
     # not happen in this codebase (security.py registers it at boot)
-    # but we don't want a missing reset to crash unrelated tests.
+    # but we don't want a missing reset to crash unrelated tests.
+
+
+# ---------------------------------------------------------------------------
+# SP31: shared DB-session + Claim/Remit factory fixtures.
+#
+# `db_session` yields a fresh session per test against the per-test DB set
+# up by the autouse `_auto_init_db` fixture above. The session rolls back at
+# teardown so a test that mutates rows doesn't leak into siblings.
+#
+# `make_claim` / `make_remit` build ORM rows with a small ergonomic surface:
+# the planned parameter names follow the content-keys helper (PCN, charge,
+# rendering NPI), but the ORM attribute names differ (`charge_amount` on
+# Claim, `total_charge` on Remittance, `provider_npi` on Claim — and no
+# `payer_id` / `rendering_provider_npi` column on Remittance at all). The
+# factories map the planned names onto the real ORM attributes and stash
+# `rendering_provider_npi` as a transient attribute so
+# ``reconcile._content_keys_match`` can still read it via ``getattr``.
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def db_session():
+    """Yield a fresh SQLAlchemy session, rolling back at teardown."""
+    from cyclone import db as _db
+    session = _db.SessionLocal()()
+    try:
+        yield session
+        session.rollback()
+    finally:
+        session.close()
+
+
+def _ensure_batch(session, batch_id: str = "test-batch") -> None:
+    """Create the Batch row a Claim/Remittance FKs to (idempotent)."""
+    from cyclone.db import Batch
+    if session.get(Batch, batch_id) is None:
+        session.add(Batch(
+            id=batch_id, kind="837p", input_filename="test.txt",
+            parsed_at=datetime.now(timezone.utc),
+        ))
+        session.flush()
+
+
+@pytest.fixture
+def make_claim(db_session):
+    """Factory: build & flush a Claim with the planned content-keys params.
+
+    Planned params (match the SP31 spec / content-keys test surface):
+        patient_control_number, total_charge, rendering_provider_npi,
+        service_date_from, matched_remittance_id=None, state=None,
+        claim_id=None
+
+    `state=None` defaults to ``ClaimState.SUBMITTED`` so existing tests
+    that omit it keep behaving. `rendering_provider_npi` is stored as
+    a transient attribute (no ORM column) for ``_content_keys_match``
+    to read via ``getattr``.
+    """
+    def _make(
+        patient_control_number: str,
+        total_charge,
+        rendering_provider_npi: str,
+        service_date_from,
+        matched_remittance_id=None,
+        state=None,
+        claim_id: str | None = None,
+    ):
+        from cyclone.db import Claim, ClaimState as _CS
+
+        _ensure_batch(db_session)
+        cid = claim_id or f"clm-{patient_control_number}-{service_date_from.isoformat()}"
+        c = Claim(
+            id=cid,
+            batch_id="test-batch",
+            patient_control_number=patient_control_number,
+            service_date_from=service_date_from,
+            charge_amount=total_charge,
+            state=state if state is not None else _CS.SUBMITTED,
+            matched_remittance_id=matched_remittance_id,
+        )
+        # Transient attribute — read by reconcile._content_keys_match
+        # via getattr; the Claim ORM has no rendering_provider_npi column.
+        c.rendering_provider_npi = rendering_provider_npi
+        db_session.add(c)
+        db_session.flush()
+        return c
+
+    return _make
+
+
+@pytest.fixture
+def make_remit(db_session):
+    """Factory: build & flush a Remittance with the planned content-keys params.
+
+    Planned params:
+        payer_claim_control_number, total_charge_amount, rendering_provider_npi,
+        service_date, remit_id=None
+
+    `total_charge_amount` is mapped to ``Remittance.total_charge`` (real ORM
+    field). `rendering_provider_npi` is stored as a transient attribute for
+    ``_content_keys_match`` to read via ``getattr`` (Remittance has no NPI
+    column; the parser reads it from raw_json in production).
+    """
+    def _make(
+        payer_claim_control_number: str,
+        total_charge_amount,
+        rendering_provider_npi: str,
+        service_date,
+        remit_id: str | None = None,
+    ):
+        from cyclone.db import Remittance
+
+        _ensure_batch(db_session)
+        rid = remit_id or f"remit-{payer_claim_control_number}-{service_date.isoformat()}"
+        r = Remittance(
+            id=rid,
+            batch_id="test-batch",
+            payer_claim_control_number=payer_claim_control_number,
+            status_code="1",
+            total_charge=total_charge_amount,
+            total_paid=Decimal("0"),
+            received_at=datetime.now(timezone.utc),
+            service_date=service_date,
+            is_reversal=False,
+        )
+        # Transient attribute — read by reconcile._content_keys_match via
+        # getattr. The Remittance ORM has no rendering_provider_npi column;
+        # the 835 parser reads it from raw_json in production.
+        r.rendering_provider_npi = rendering_provider_npi
+        db_session.add(r)
+        db_session.flush()
+        return r
+
+    return _make

backend/tests/test_reconcile.py

@@ -487,3 +487,125 @@ def test_content_keys_empty_npi_does_not_count():
     assert _content_keys_match(r, c) is True  # PCN + charge still match
     c2 = _make_claim_for_keys("c2", "XYZ", Decimal("100.00"), "1111111111")
     assert _content_keys_match(r, c2) is False  # only charge, no PCN
+
+
+# --- SP31: content-keys fallback matcher (DB helper) ------------------------
+
+
+def test_score_fallback_unique_match_returns_claim_id(db_session, make_claim, make_remit):
+    """One claim in pool matches 2-of-3 → returns that claim."""
+    from decimal import Decimal
+    from datetime import date, timedelta
+    from cyclone.reconcile import _score_fallback_candidates
+    c1 = make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+                    rendering_provider_npi="1111111111",
+                    service_date_from=date(2026, 6, 1))
+    c2 = make_claim(patient_control_number="XYZ", total_charge=Decimal("999.99"),
+                    rendering_provider_npi="2222222222",
+                    service_date_from=date(2026, 6, 1))
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=date(2026, 6, 5))  # ±30-day window
+    assert _score_fallback_candidates(db_session, r) == c1.id
+
+
+def test_score_fallback_zero_matches_returns_none(db_session, make_claim, make_remit):
+    """No claim in pool matches 2-of-3 → returns None."""
+    from decimal import Decimal
+    from datetime import date
+    from cyclone.reconcile import _score_fallback_candidates
+    make_claim(patient_control_number="XYZ", total_charge=Decimal("999.99"),
+               rendering_provider_npi="2222222222",
+               service_date_from=date(2026, 6, 1))
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=date(2026, 6, 5))
+    assert _score_fallback_candidates(db_session, r) is None
+
+
+def test_score_fallback_two_matches_returns_none(db_session, make_claim, make_remit):
+    """Two claims in pool both match 2-of-3 → returns None (ambiguous)."""
+    from decimal import Decimal
+    from datetime import date
+    from cyclone.reconcile import _score_fallback_candidates
+    # Both claims have same PCN and same charge — both would match PCN+charge.
+    make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+               rendering_provider_npi="1111111111",
+               service_date_from=date(2026, 6, 1))
+    make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+               rendering_provider_npi="1111111111",
+               service_date_from=date(2026, 6, 2))
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=date(2026, 6, 5))
+    assert _score_fallback_candidates(db_session, r) is None
+
+
+def test_score_fallback_skips_already_matched(db_session, make_claim, make_remit):
+    """Claim with matched_remittance_id set → excluded from pool."""
+    from decimal import Decimal
+    from datetime import date
+    from cyclone.reconcile import _score_fallback_candidates
+    c1 = make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+                    rendering_provider_npi="1111111111",
+                    service_date_from=date(2026, 6, 1),
+                    matched_remittance_id="some-other-remit-id")
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=date(2026, 6, 5))
+    assert _score_fallback_candidates(db_session, r) is None  # only candidate was excluded
+
+
+def test_score_fallback_skips_terminal_state(db_session, make_claim, make_remit):
+    """Claim in PAID/DENIED/REJECTED/REVERSED/RECONCILED → excluded from pool."""
+    from decimal import Decimal
+    from datetime import date
+    from cyclone.db import ClaimState
+    from cyclone.reconcile import _score_fallback_candidates
+    make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+               rendering_provider_npi="1111111111",
+               service_date_from=date(2026, 6, 1),
+               state=ClaimState.PAID)
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=date(2026, 6, 5))
+    assert _score_fallback_candidates(db_session, r) is None
+
+
+def test_score_fallback_widened_window_catches_late_remit(db_session, make_claim, make_remit):
+    """Remit 25 days after claim → still in pool (30-day window)."""
+    from decimal import Decimal
+    from datetime import date, timedelta
+    from cyclone.reconcile import _score_fallback_candidates
+    claim_date = date(2026, 6, 1)
+    remit_date = claim_date + timedelta(days=25)
+    c1 = make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+                    rendering_provider_npi="1111111111",
+                    service_date_from=claim_date)
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=remit_date)
+    assert _score_fallback_candidates(db_session, r) == c1.id
+
+
+def test_score_fallback_outside_window_excluded(db_session, make_claim, make_remit):
+    """Remit 60 days after claim → excluded (outside 30-day window)."""
+    from decimal import Decimal
+    from datetime import date, timedelta
+    from cyclone.reconcile import _score_fallback_candidates
+    claim_date = date(2026, 6, 1)
+    remit_date = claim_date + timedelta(days=60)
+    make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
+               rendering_provider_npi="1111111111",
+               service_date_from=claim_date)
+    r = make_remit(payer_claim_control_number="ABC",
+                   total_charge_amount=Decimal("100.00"),
+                   rendering_provider_npi="1111111111",
+                   service_date=remit_date)
+    assert _score_fallback_candidates(db_session, r) is None