feat(sp6): compute_lanes — 4 lanes on read with scoring

2026-06-20 18:31:01 -06:00
parent 3432c76561
commit cdffb3faf3
2 changed files with 299 additions and 0 deletions
@@ -0,0 +1,155 @@
+"""Compute the four Inbox lanes from the DB on read.
+
+SP6 T6.
+
+Lanes:
+  - rejected:    claims whose 999 AK5 set-level response was R/E
+  - candidates:  remits without a matched claim, with scoreable claims
+  - unmatched:   claims still SUBMITTED with no remittance in flight
+  - done_today:  claims that reached a terminal state in the last 24h
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Iterable
+
+from sqlalchemy.orm import Session
+
+from cyclone.db import Claim, ClaimState, Remittance
+from cyclone.scoring import score_pair, ScoreBreakdown
+
+
+@dataclass
+class Lanes:
+    rejected: list[dict] = field(default_factory=list)
+    candidates: list[dict] = field(default_factory=list)
+    unmatched: list[dict] = field(default_factory=list)
+    done_today: list[dict] = field(default_factory=list)
+
+
+def _isoformat(value) -> str | None:
+    if value is None:
+        return None
+    if hasattr(value, "isoformat"):
+        return value.isoformat()
+    return str(value)
+
+
+def _claim_to_row(c: Claim, *, kind: str, score: ScoreBreakdown | None = None) -> dict:
+    return {
+        "id": c.id,
+        "kind": kind,
+        "patient_control_number": c.patient_control_number,
+        "charge_amount": float(c.charge_amount) if c.charge_amount is not None else None,
+        "payer_id": c.payer_id,
+        "provider_npi": c.provider_npi,
+        "state": c.state.value if hasattr(c.state, "value") else str(c.state),
+        "rejection_reason": c.rejection_reason,
+        "rejected_at": _isoformat(c.rejected_at),
+        "service_date_from": _isoformat(c.service_date_from),
+        "score": score.total if score else None,
+        "score_tier": score.tier if score else None,
+        "score_breakdown": {
+            "patient": score.patient,
+            "date": score.date,
+            "amount": score.amount,
+            "provider": score.provider,
+        } if score else None,
+    }
+
+
+def _remit_to_row(r: Remittance, *, candidates: list[tuple[Claim, ScoreBreakdown]]) -> dict:
+    raw = r.raw_json or {}
+    return {
+        "id": r.id,
+        "kind": "remit",
+        "payer_claim_control_number": r.payer_claim_control_number,
+        "charge_amount": float(r.total_charge) if r.total_charge is not None else None,
+        "payer_id": raw.get("payer_id"),
+        "rendering_provider_npi": raw.get("rendering_provider_npi"),
+        "service_date": _isoformat(r.service_date),
+        "candidates": [
+            {
+                "claim_id": c.id,
+                "score": s.total,
+                "tier": s.tier,
+                "breakdown": {
+                    "patient": s.patient, "date": s.date,
+                    "amount": s.amount, "provider": s.provider,
+                },
+            }
+            for c, s in candidates[:5]
+        ],
+    }
+
+
+class _RemitScoringShim:
+    """Adapt a Remittance to the duck-typed protocol score_pair expects.
+
+    The 835 parser stores service_date on the ORM column. PCN, NPI, payer_id
+    aren't on Remittance; we read them from raw_json.
+    """
+    def __init__(self, r: Remittance):
+        raw = r.raw_json or {}
+        self.patient_control_number = (
+            raw.get("patient_control_number") or r.payer_claim_control_number
+        )
+        self.service_date = r.service_date
+        self.charge_amount = r.total_charge
+        self.rendering_provider_npi = raw.get("rendering_provider_npi")
+
+
+def compute_lanes(session: Session, *, dismissed_pairs: Iterable[frozenset]) -> Lanes:
+    lanes = Lanes()
+    dismissed = set(dismissed_pairs)
+
+    # --- Rejected ---
+    for c in session.query(Claim).filter(Claim.state == ClaimState.REJECTED).all():
+        lanes.rejected.append(_claim_to_row(c, kind="claim"))
+
+    # --- Done today ---
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
+    terminal_states = {
+        ClaimState.PAID, ClaimState.PARTIAL, ClaimState.DENIED,
+        ClaimState.RECONCILED, ClaimState.REVERSED,
+    }
+    for c in session.query(Claim).filter(
+        Claim.state.in_(terminal_states),
+        Claim.state_changed_at >= cutoff,
+    ).all():
+        lanes.done_today.append(_claim_to_row(c, kind="claim"))
+
+    # --- Unmatched (claims still submitted, no remittance in flight) ---
+    for c in session.query(Claim).filter(Claim.state == ClaimState.SUBMITTED).all():
+        lanes.unmatched.append(_claim_to_row(c, kind="claim"))
+
+    # --- Candidates (remits not yet matched, with scoreable claims) ---
+    unmatched_remits = session.query(Remittance).filter(
+        Remittance.claim_id.is_(None),
+    ).all()
+
+    submitted_claims = list(
+        session.query(Claim).filter(Claim.state == ClaimState.SUBMITTED).all()
+    )
+
+    for r in unmatched_remits:
+        if r.claim_id:
+            continue
+        candidates: list[tuple[Claim, ScoreBreakdown]] = []
+        shim = _RemitScoringShim(r)
+        for c in submitted_claims:
+            if c.payer_id and r.raw_json and (r.raw_json.get("payer_id") != c.payer_id):
+                continue
+            score = score_pair(c, shim)
+            if score.tier == "hidden":
+                continue
+            if frozenset({c.id, r.id}) in dismissed:
+                continue
+            candidates.append((c, score))
+        candidates.sort(key=lambda cs: -cs[1].total)
+        if not candidates:
+            continue
+        lanes.candidates.append(_remit_to_row(r, candidates=candidates))
+
+    return lanes
@@ -0,0 +1,144 @@
+"""TDD: compute_lanes — four Inbox lanes from the DB on read.
+
+Plan: docs/superpowers/plans/2026-06-20-cyclone-workflow-automation.md
+Task 6 (T6).
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+
+import pytest
+
+from cyclone import db
+from cyclone.db import (
+    Base, Batch, Claim, ClaimState, Remittance, init_db,
+)
+from cyclone.inbox_lanes import compute_lanes
+
+
+@pytest.fixture(autouse=True)
+def _setup(tmp_path, monkeypatch):
+    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/lanes.db")
+    db._reset_for_tests()
+    db.init_db()
+    yield
+
+
+def _add_batch() -> None:
+    with db.SessionLocal()() as s:
+        if s.get(Batch, "B-1") is not None:
+            return
+        s.add(Batch(
+            id="B-1", kind="837p", input_filename="x.txt",
+            parsed_at=datetime.now(timezone.utc),
+            totals_json={"total_claims": 1},
+            validation_json={"passed": True, "warnings": [], "errors": []},
+            raw_result_json={"_": "stub"},
+        ))
+        s.commit()
+
+
+def _add_claim(*, claim_id: str, pcn: str = "PCN",
+               state: ClaimState = ClaimState.SUBMITTED,
+               charge: Decimal = Decimal("100.00"),
+               provider_npi: str | None = "1234567890",
+               payer_id: str | None = "PAYER-A",
+               service_date=None) -> None:
+    _add_batch()
+    with db.SessionLocal()() as s:
+        s.add(Claim(
+            id=claim_id, batch_id="B-1",
+            patient_control_number=pcn, state=state,
+            charge_amount=charge, provider_npi=provider_npi,
+            payer_id=payer_id, service_date_from=service_date,
+        ))
+        s.commit()
+
+
+def _add_remit(*, remit_id: str, pcn: str = "PCN",
+               charge: Decimal = Decimal("100.00"),
+               claim_id: str | None = None,
+               payer_id: str | None = "PAYER-A",
+               service_date=None) -> None:
+    _add_batch()
+    with db.SessionLocal()() as s:
+        r = Remittance(
+            id=remit_id, batch_id="B-1",
+            payer_claim_control_number=pcn,
+            claim_id=claim_id,
+            status_code="1", status_label="Processed",
+            total_charge=charge,
+            total_paid=Decimal("0"),
+            received_at=datetime.now(timezone.utc),
+            service_date=service_date,
+        )
+        # Stash rendering_provider_npi and payer_id in raw_json for the
+        # 835 fields that aren't on the ORM model.
+        r.raw_json = {
+            "rendering_provider_npi": "1234567890",
+            "payer_id": payer_id,
+            "patient_control_number": pcn,
+        }
+        s.add(r)
+        s.commit()
+
+
+def test_rejected_lane_includes_rejected_claims():
+    _add_claim(claim_id="C1", state=ClaimState.REJECTED)
+    with db.SessionLocal()() as s:
+        lanes = compute_lanes(s, dismissed_pairs=set())
+    assert "C1" in {r["id"] for r in lanes.rejected}
+
+
+def test_unmatched_lane_includes_submitted_claims_with_no_remit():
+    _add_claim(claim_id="C1", state=ClaimState.SUBMITTED)
+    with db.SessionLocal()() as s:
+        lanes = compute_lanes(s, dismissed_pairs=set())
+    assert "C1" in {r["id"] for r in lanes.unmatched}
+
+
+def test_unmatched_remit_appears_with_candidates():
+    _add_claim(claim_id="C1", pcn="PCN-1", service_date=None)
+    _add_remit(remit_id="R1", pcn="PCN-1")  # exact match → candidate
+    with db.SessionLocal()() as s:
+        lanes = compute_lanes(s, dismissed_pairs=set())
+    assert any(r["id"] == "R1" for r in lanes.candidates)
+
+
+def test_candidate_below_threshold_is_hidden():
+    _add_claim(claim_id="C1", pcn="A", charge=Decimal("100.00"))
+    _add_remit(remit_id="R1", pcn="Z",
+               charge=Decimal("999.00"))   # everything mismatched
+    with db.SessionLocal()() as s:
+        lanes = compute_lanes(s, dismissed_pairs=set())
+    assert not any(r["id"] == "R1" for r in lanes.candidates)
+
+
+def test_dismissed_pair_excluded_from_candidates():
+    _add_claim(claim_id="C1", pcn="PCN-1")
+    _add_remit(remit_id="R1", pcn="PCN-1")
+    with db.SessionLocal()() as s:
+        lanes = compute_lanes(
+            s, dismissed_pairs={frozenset({"C1", "R1"})},
+        )
+    assert not any(r["id"] == "R1" for r in lanes.candidates)
+
+
+def test_done_today_includes_recent_terminal_states():
+    _add_claim(claim_id="C1", state=ClaimState.PAID)
+    _add_claim(claim_id="C2", state=ClaimState.PAID)
+    # Force C1's state_changed_at to be recent, C2 to be old.
+    now = datetime.now(timezone.utc)
+    with db.SessionLocal()() as s:
+        c1 = s.get(Claim, "C1")
+        c1.state_changed_at = now - timedelta(hours=2)
+        c2 = s.get(Claim, "C2")
+        c2.state_changed_at = now - timedelta(hours=30)
+        s.commit()
+
+    with db.SessionLocal()() as s:
+        lanes = compute_lanes(s, dismissed_pairs=set())
+    ids = {r["id"] for r in lanes.done_today}
+    assert "C1" in ids
+    assert "C2" not in ids