"""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, Ack, ClaimAck, 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 # --------------------------------------------------------------------------- # SP29: rejected-lane rows must carry the per-claim 999 ack-evidence summary # so the Inbox can render AK2 chips inline + a per-row Resubmit button # without an extra round-trip. # --------------------------------------------------------------------------- def _add_999_ack(*, ack_id: int, parsed_at=None) -> None: """Persist a minimal Ack row (the 999 envelope).""" with db.SessionLocal()() as s: if s.get(Ack, ack_id) is not None: return s.add(Ack( id=ack_id, source_batch_id="B-1", accepted_count=0, rejected_count=0, received_count=1, ack_code="A", parsed_at=parsed_at or datetime.now(timezone.utc), raw_json={"set_responses": []}, )) s.commit() def _link_claim_ack( *, ack_id: int, claim_id: str, ak2_index: int, set_accept_reject_code: str, linked_at=None, set_control_number: str = "991102989", ) -> None: """Persist a single ClaimAck row linking an Ack to a Claim.""" with db.SessionLocal()() as s: s.add(ClaimAck( claim_id=claim_id, ack_id=ack_id, ack_kind="999", ak2_index=ak2_index, set_control_number=set_control_number, set_accept_reject_code=set_accept_reject_code, linked_at=linked_at or datetime.now(timezone.utc), linked_by="auto", )) s.commit() def test_inbox_lanes_attaches_claim_acks_summary_for_rejected(): """A rejected-lane row gets a `claim_acks` field with total / rejected counts + 5 most recent AK2 set_responses. Newest first. """ _add_claim(claim_id="REJ-1", state=ClaimState.REJECTED) # 3 linked 999 acks: 2 accepted, 1 rejected _add_999_ack(ack_id=1) _add_999_ack(ack_id=2) _add_999_ack(ack_id=3) t0 = datetime.now(timezone.utc) _link_claim_ack(ack_id=1, claim_id="REJ-1", ak2_index=0, set_accept_reject_code="A", linked_at=t0 - timedelta(minutes=10)) _link_claim_ack(ack_id=2, claim_id="REJ-1", ak2_index=1, set_accept_reject_code="R", linked_at=t0 - timedelta(minutes=5)) _link_claim_ack(ack_id=3, claim_id="REJ-1", ak2_index=2, set_accept_reject_code="A", linked_at=t0 - timedelta(minutes=1)) with db.SessionLocal()() as s: lanes = compute_lanes(s, dismissed_pairs=set()) rows = [r for r in lanes.rejected if r["id"] == "REJ-1"] assert len(rows) == 1 row = rows[0] assert "claim_acks" in row summary = row["claim_acks"] assert summary is not None assert summary["total"] == 3 assert summary["rejected"] == 1 assert len(summary["items"]) == 3 # Newest first (descending linked_at) assert summary["items"][0]["set_accept_reject_code"] == "A" # ack 3 (most recent) assert summary["items"][1]["set_accept_reject_code"] == "R" # ack 2 (middle) assert summary["items"][2]["set_accept_reject_code"] == "A" # ack 1 (oldest) # The R code is the one rejected entry codes = [it["set_accept_reject_code"] for it in summary["items"]] assert codes.count("R") == 1 # ak2_index is preserved assert summary["items"][0]["ak2_index"] == 2 assert summary["items"][1]["ak2_index"] == 1 def test_inbox_lanes_claim_acks_is_null_when_no_links(): """A rejected claim with zero linked 999 acks renders `claim_acks: null` (the UI shows '999 not linked').""" _add_claim(claim_id="REJ-2", state=ClaimState.REJECTED) with db.SessionLocal()() as s: lanes = compute_lanes(s, dismissed_pairs=set()) rows = [r for r in lanes.rejected if r["id"] == "REJ-2"] assert len(rows) == 1 assert rows[0]["claim_acks"] is None def test_inbox_lanes_claim_acks_does_not_attach_to_other_lanes(): """The ack summary is only on `rejected`, not `payer_rejected` / `unmatched` / `done_today` — those lanes have their own evidence shape (payer_rejected_*, no linked 999s by construction).""" _add_claim(claim_id="SUB-1", state=ClaimState.SUBMITTED) _add_claim(claim_id="DEN-1", state=ClaimState.DENIED) with db.SessionLocal()() as s: lanes = compute_lanes(s, dismissed_pairs=set()) for r in lanes.unmatched: assert "claim_acks" not in r for r in lanes.done_today: assert "claim_acks" not in r