From 52795adfb19768b4d7fede0d0f940a56c3780e41 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 19:41:00 -0600 Subject: [PATCH 01/17] feat(sp41): SVC-level 835 reparser with member_id at SVC scope --- backend/src/cyclone/rebill/__init__.py | 11 ++ backend/src/cyclone/rebill/parse_835_svc.py | 123 ++++++++++++++++++ .../fixtures/835_sample_svc_with_member.txt | 1 + backend/tests/test_rebill_parse_835_svc.py | 28 ++++ 4 files changed, 163 insertions(+) create mode 100644 backend/src/cyclone/rebill/__init__.py create mode 100644 backend/src/cyclone/rebill/parse_835_svc.py create mode 100644 backend/tests/fixtures/835_sample_svc_with_member.txt create mode 100644 backend/tests/test_rebill_parse_835_svc.py diff --git a/backend/src/cyclone/rebill/__init__.py b/backend/src/cyclone/rebill/__init__.py new file mode 100644 index 0000000..39e04d1 --- /dev/null +++ b/backend/src/cyclone/rebill/__init__.py @@ -0,0 +1,11 @@ +"""SP41 — in-window rebill pipeline. + +Owns: + - parse_835_svc (SVC-level 835 reparse with member_id) + - reconcile (visit-to-835 join on (member_id, procedure, DOS)) + - carc_filter (CARC-aware exclusion for Pipeline A) + - timely_filing (120-day DOS age gate) + - pipeline_a (denied/partial → frequency-7 rebill) + - pipeline_b (NOT_IN_835 → fresh 837Ps by (member, ISO-week)) + - summary (summary CSV + per-category counters) +""" \ No newline at end of file diff --git a/backend/src/cyclone/rebill/parse_835_svc.py b/backend/src/cyclone/rebill/parse_835_svc.py new file mode 100644 index 0000000..c48b98e --- /dev/null +++ b/backend/src/cyclone/rebill/parse_835_svc.py @@ -0,0 +1,123 @@ +"""SVC-level 835 reparse with member_id at SVC scope. + +Walks an 835 file segment-by-segment, propagating the CLP-scope NM1*QC +NM109 (member_id) to every SVC row that follows. Captures DTM*472 +service dates and CAS adjustments (which may appear before or after +DTM*472 in the segment stream). +""" +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from pathlib import Path + +ELEM = "*" +SEG_END = "~" + + +@dataclass(frozen=True) +class SvcRow: + src_file: str + claim_id: str + member_id: str + status: str + procedure: str + modifiers: str + charge: Decimal + paid: Decimal + units: Decimal + svc_date: date + cas_reasons: tuple[str, ...] # each entry is "GROUP-CODE" (e.g. "OA-18") + pay_date: date | None + + +def _split(seg: str) -> list[str]: + return seg.split(ELEM) + + +def parse_835_svc(path: str | Path) -> "Iterator[SvcRow]": + """Yield one SvcRow per SVC segment in `path`.""" + path = Path(path) + text = path.read_text() + segments = [s for s in text.split(SEG_END) if s] + pay_date: date | None = None + current_claim_id = "" + current_status = "" + current_member_id = "" + i = 0 + while i < len(segments): + elems = _split(segments[i]) + seg_name = elems[0] + if seg_name == "DTM" and len(elems) > 2 and elems[1] == "405": + try: + pay_date = _ymd(elems[2]) + except ValueError: + pass + elif seg_name == "CLP": + current_claim_id = elems[1] if len(elems) > 1 else "" + current_status = elems[2] if len(elems) > 2 else "" + elif seg_name == "NM1" and len(elems) > 1 and elems[1] == "QC": + current_member_id = elems[9] if len(elems) > 9 else "" + elif seg_name == "SVC" and current_claim_id: + rest = elems[1:] + comp = rest[0] if rest else "" + proc_parts = comp.split(":") + procedure = proc_parts[1] if len(proc_parts) > 1 else "" + modifiers = ":".join(proc_parts[2:]) if len(proc_parts) > 2 else "" + charge = _decimal(rest[1]) if len(rest) > 1 else Decimal("0") + paid = _decimal(rest[2]) if len(rest) > 2 else Decimal("0") + units = _decimal(rest[3]) if len(rest) > 3 else Decimal("0") + svc_date, cas_reasons = _walk_for_dtm_and_cas(segments, i) + yield SvcRow( + src_file=path.name, + claim_id=current_claim_id, + member_id=current_member_id, + status=current_status, + procedure=procedure, + modifiers=modifiers, + charge=charge, + paid=paid, + units=units, + svc_date=svc_date, + cas_reasons=cas_reasons, + pay_date=pay_date, + ) + i += 1 + + +def _ymd(s: str) -> date: + y, m, d = int(s[0:4]), int(s[4:6]), int(s[6:8]) + return date(y, m, d) + + +def _decimal(s: str) -> Decimal: + try: + return Decimal(s) + except Exception: + return Decimal("0") + + +def _walk_for_dtm_and_cas( + segments: list[str], start: int, window: int = 10 +) -> tuple[date | None, tuple[str, ...]]: + """Walk the next `window` segments after SVC. Capture DTM*472 and CAS. + + X12 835 may place DTM*472 before or after CAS segments within a service + line, so we walk the full window and capture both, breaking only at the + next SVC or CLP (a new claim/service boundary). + """ + svc_date: date | None = None + cas_reasons: list[str] = [] + for j in range(start + 1, min(start + window, len(segments))): + ej = _split(segments[j]) + if ej[0] in ("SVC", "CLP"): + break + if ej[0] == "DTM" and len(ej) > 2 and ej[1] == "472" and svc_date is None: + try: + svc_date = _ymd(ej[2]) + except ValueError: + pass + elif ej[0] == "CAS": + # CAS*GR*CODE*AMT*QTY*AMT*QTY... → GR-CODE + if len(ej) >= 3: + cas_reasons.append(f"{ej[1]}-{ej[2]}") + return svc_date, tuple(cas_reasons) \ No newline at end of file diff --git a/backend/tests/fixtures/835_sample_svc_with_member.txt b/backend/tests/fixtures/835_sample_svc_with_member.txt new file mode 100644 index 0000000..76d0472 --- /dev/null +++ b/backend/tests/fixtures/835_sample_svc_with_member.txt @@ -0,0 +1 @@ +ISA*00* *00* *ZZ*CYCLONE *ZZ*GAINWELL *260101*1200*^*00501*000000001*0*P*:~GS*HC*CYCLONE*GAINWELL*20260101*1200*1*X*005010X221A1~ST*835*0001~BPR*I*100.00*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~TRN*1*TRACE01*1512345678~DTM*405*20260118~N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~N3*PO BOX 1100~N4*DENVER*CO*80202~REF*2U*7912900843~LX*1~CLP*T1001*1*16.24*16.10**MC*2026029105200*11*1~NM1*QC*1*AALBUE*ERIC****MR*J813715~NM1*74*1*AALBUE*ERIC*W***C*J813715~DTM*232*20260118~DTM*233*20260124~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260118~CAS*CO*45*.02~REF*G1*6252960154~REF*6R*T1001V001~AMT*B6*2.30~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260119~CAS*OA*18*2.32~SVC*HC:T1019:U2:SC:KX*2.32*-2.30**.33~DTM*472*20260120~CAS*CO*45*-.02~SE*23*0001~GE*1*1~IEA*1*000000001~ \ No newline at end of file diff --git a/backend/tests/test_rebill_parse_835_svc.py b/backend/tests/test_rebill_parse_835_svc.py new file mode 100644 index 0000000..cc6cca4 --- /dev/null +++ b/backend/tests/test_rebill_parse_835_svc.py @@ -0,0 +1,28 @@ +"""835 SVC-level parser with member_id at SVC scope.""" +from pathlib import Path +from decimal import Decimal +from cyclone.rebill.parse_835_svc import parse_835_svc + +FIX = Path(__file__).parent / "fixtures" / "835_sample_svc_with_member.txt" + +def test_parse_835_svc_extracts_member_id(): + """NM1*QC NM109 at the CLP scope must propagate to every SVC row.""" + rows = list(parse_835_svc(FIX)) + assert len(rows) >= 1 + for r in rows: + assert r.member_id == "J813715" + assert r.procedure # non-empty + assert r.svc_date # non-empty + assert r.charge > 0 + +def test_parse_835_svc_extracts_cas_reasons(): + """CAS segments after DTM*472 must be captured (post-DTM*472 ordering).""" + rows = list(parse_835_svc(FIX)) + # at least one row should have an OA-18 reason + assert any("OA-18" in r.cas_reasons for r in rows) + +def test_parse_835_svc_picks_up_status_22_reversals(): + """Status 22 (reversal of previous payment) must be preserved.""" + rows = list(parse_835_svc(FIX)) + statuses = {r.status for r in rows} + assert statuses & {"1", "4", "22"} # at least one of each present \ No newline at end of file From f8e2f0933e5abea5c558d1b3a019d34e09aae010 Mon Sep 17 00:00:00 2001 From: sp41-bot Date: Tue, 7 Jul 2026 19:53:04 -0600 Subject: [PATCH 02/17] fix(sp41): reset member_id on new CLP + tighten status-preservation test --- backend/src/cyclone/rebill/parse_835_svc.py | 1 + .../fixtures/835_sample_svc_with_member.txt | 2 +- backend/tests/test_rebill_parse_835_svc.py | 18 ++++++++++++++---- 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/backend/src/cyclone/rebill/parse_835_svc.py b/backend/src/cyclone/rebill/parse_835_svc.py index c48b98e..1422006 100644 --- a/backend/src/cyclone/rebill/parse_835_svc.py +++ b/backend/src/cyclone/rebill/parse_835_svc.py @@ -55,6 +55,7 @@ def parse_835_svc(path: str | Path) -> "Iterator[SvcRow]": elif seg_name == "CLP": current_claim_id = elems[1] if len(elems) > 1 else "" current_status = elems[2] if len(elems) > 2 else "" + current_member_id = "" elif seg_name == "NM1" and len(elems) > 1 and elems[1] == "QC": current_member_id = elems[9] if len(elems) > 9 else "" elif seg_name == "SVC" and current_claim_id: diff --git a/backend/tests/fixtures/835_sample_svc_with_member.txt b/backend/tests/fixtures/835_sample_svc_with_member.txt index 76d0472..e102072 100644 --- a/backend/tests/fixtures/835_sample_svc_with_member.txt +++ b/backend/tests/fixtures/835_sample_svc_with_member.txt @@ -1 +1 @@ -ISA*00* *00* *ZZ*CYCLONE *ZZ*GAINWELL *260101*1200*^*00501*000000001*0*P*:~GS*HC*CYCLONE*GAINWELL*20260101*1200*1*X*005010X221A1~ST*835*0001~BPR*I*100.00*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~TRN*1*TRACE01*1512345678~DTM*405*20260118~N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~N3*PO BOX 1100~N4*DENVER*CO*80202~REF*2U*7912900843~LX*1~CLP*T1001*1*16.24*16.10**MC*2026029105200*11*1~NM1*QC*1*AALBUE*ERIC****MR*J813715~NM1*74*1*AALBUE*ERIC*W***C*J813715~DTM*232*20260118~DTM*233*20260124~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260118~CAS*CO*45*.02~REF*G1*6252960154~REF*6R*T1001V001~AMT*B6*2.30~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260119~CAS*OA*18*2.32~SVC*HC:T1019:U2:SC:KX*2.32*-2.30**.33~DTM*472*20260120~CAS*CO*45*-.02~SE*23*0001~GE*1*1~IEA*1*000000001~ \ No newline at end of file +ISA*00* *00* *ZZ*CYCLONE *ZZ*GAINWELL *260101*1200*^*00501*000000001*0*P*:~GS*HC*CYCLONE*GAINWELL*20260101*1200*1*X*005010X221A1~ST*835*0001~BPR*I*100.00*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~TRN*1*TRACE01*1512345678~DTM*405*20260118~N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~N3*PO BOX 1100~N4*DENVER*CO*80202~REF*2U*7912900843~LX*1~CLP*T1001*1*16.24*16.10**MC*2026029105200*11*1~NM1*QC*1*AALBUE*ERIC****MR*J813715~NM1*74*1*AALBUE*ERIC*W***C*J813715~DTM*232*20260118~DTM*233*20260124~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260118~CAS*CO*45*.02~REF*G1*6252960154~REF*6R*T1001V001~AMT*B6*2.30~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260119~CAS*OA*18*2.32~SVC*HC:T1019:U2:SC:KX*2.32*-2.30**.33~DTM*472*20260120~CAS*CO*45*-.02~LX*2~CLP*T1002*4*16.24*0**MC*2026029105201*11*1~NM1*QC*1*OTHER-A*FIRST****MR*OTHER-MEMBER-A~SVC*HC:T1019*2.32*0**.33~DTM*472*20260121~CAS*CO*97*2.32~LX*3~CLP*T1003*22*16.24*-16.10**MC*2026029105202*11*1~NM1*QC*1*OTHER-B*SECOND****MR*OTHER-MEMBER-B~SVC*HC:T1019*2.32*-2.30**.33~DTM*472*20260122~SE*23*0001~GE*1*1~IEA*1*000000001~ \ No newline at end of file diff --git a/backend/tests/test_rebill_parse_835_svc.py b/backend/tests/test_rebill_parse_835_svc.py index cc6cca4..437487b 100644 --- a/backend/tests/test_rebill_parse_835_svc.py +++ b/backend/tests/test_rebill_parse_835_svc.py @@ -1,19 +1,28 @@ """835 SVC-level parser with member_id at SVC scope.""" from pathlib import Path +from collections import defaultdict from decimal import Decimal from cyclone.rebill.parse_835_svc import parse_835_svc FIX = Path(__file__).parent / "fixtures" / "835_sample_svc_with_member.txt" def test_parse_835_svc_extracts_member_id(): - """NM1*QC NM109 at the CLP scope must propagate to every SVC row.""" + """NM1*QC NM109 at the CLP scope must propagate to that CLP's SVC rows; + the multi-claim fixture confirms each CLP's NM1*QC lands on its own SVCs.""" rows = list(parse_835_svc(FIX)) assert len(rows) >= 1 + by_claim: dict[str, list[str]] = defaultdict(list) for r in rows: - assert r.member_id == "J813715" + by_claim[r.claim_id].append(r.member_id) assert r.procedure # non-empty assert r.svc_date # non-empty assert r.charge > 0 + # First CLP (T1001) -> original fixture member + assert all(mid == "J813715" for mid in by_claim["T1001"]), by_claim + # Second CLP (T1002, status 4) -> its own NM1*QC + assert all(mid == "OTHER-MEMBER-A" for mid in by_claim["T1002"]), by_claim + # Third CLP (T1003, status 22) -> its own NM1*QC + assert all(mid == "OTHER-MEMBER-B" for mid in by_claim["T1003"]), by_claim def test_parse_835_svc_extracts_cas_reasons(): """CAS segments after DTM*472 must be captured (post-DTM*472 ordering).""" @@ -22,7 +31,8 @@ def test_parse_835_svc_extracts_cas_reasons(): assert any("OA-18" in r.cas_reasons for r in rows) def test_parse_835_svc_picks_up_status_22_reversals(): - """Status 22 (reversal of previous payment) must be preserved.""" + """Status 22 (reversal of previous payment) must be preserved, along with + status 1 (primary) and status 4 (denied).""" rows = list(parse_835_svc(FIX)) statuses = {r.status for r in rows} - assert statuses & {"1", "4", "22"} # at least one of each present \ No newline at end of file + assert statuses == {"1", "4", "22"} From 69b760e89080ed8191176a5e2593fae88302d320 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 19:59:18 -0600 Subject: [PATCH 03/17] feat(sp41): visit-to-835 reconcile on (member_id, procedure, DOS) with best-of-N --- backend/src/cyclone/rebill/reconcile.py | 69 ++++++++++++++++++++++ backend/tests/test_rebill_reconcile.py | 76 +++++++++++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 backend/src/cyclone/rebill/reconcile.py create mode 100644 backend/tests/test_rebill_reconcile.py diff --git a/backend/src/cyclone/rebill/reconcile.py b/backend/src/cyclone/rebill/reconcile.py new file mode 100644 index 0000000..f5a803b --- /dev/null +++ b/backend/src/cyclone/rebill/reconcile.py @@ -0,0 +1,69 @@ +"""Visit-to-835 reconciliation on (member_id, procedure, DOS). + +Authoritative match key. The 5% tolerance on PAID handles the legitimate +charge-amount differences between AxisCare's per-unit CSV and the 835's +per-SVC breakdown. +""" +from collections import defaultdict +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from enum import Enum +from typing import Iterable + +# SvcRow is owned by parse_835_svc (Task 1) — re-exported here so callers +# of `cyclone.rebill.reconcile` only need one import. +from cyclone.rebill.parse_835_svc import SvcRow # noqa: F401 (re-exported) + +PAID_TOLERANCE = Decimal("0.05") # charge match tolerance +PARTIAL_RATIO = Decimal("0.95") # PAID if total_paid >= 95% of billed + + +class OutcomeCategory(str, Enum): + PAID = "PAID" + PARTIAL = "PARTIAL" + DENIED = "DENIED" + NOT_IN_835 = "NOT_IN_835" + + +@dataclass(frozen=True) +class VisitRow: + date: date + member_id: str + procedure: str + billed: Decimal + + +@dataclass(frozen=True) +class ReconcileOutcome: + visit: VisitRow + category: OutcomeCategory + unpaid: Decimal + matched_svc_count: int + + +def reconcile_visits_to_835( + visits: Iterable[VisitRow], + svcs: Iterable[SvcRow], +) -> list[ReconcileOutcome]: + """Index SVCs by (member_id, procedure, DOS), best-of-N outcome per visit.""" + by_key: dict[tuple[str, str, date], list[SvcRow]] = defaultdict(list) + for s in svcs: + by_key[(s.member_id, s.procedure, s.svc_date)].append(s) + + out: list[ReconcileOutcome] = [] + for v in visits: + cands = by_key.get((v.member_id, v.procedure, v.date), []) + if not cands: + out.append(ReconcileOutcome(v, OutcomeCategory.NOT_IN_835, v.billed, 0)) + continue + total_paid = sum((c.paid for c in cands), Decimal("0")) + any_paid = any(c.paid > 0 for c in cands) + if any_paid and total_paid >= v.billed * PARTIAL_RATIO: + cat, unpaid = OutcomeCategory.PAID, Decimal("0") + elif any_paid: + cat, unpaid = OutcomeCategory.PARTIAL, max(Decimal("0"), v.billed - total_paid) + else: + cat, unpaid = OutcomeCategory.DENIED, v.billed + out.append(ReconcileOutcome(v, cat, unpaid, len(cands))) + return out diff --git a/backend/tests/test_rebill_reconcile.py b/backend/tests/test_rebill_reconcile.py new file mode 100644 index 0000000..30db4be --- /dev/null +++ b/backend/tests/test_rebill_reconcile.py @@ -0,0 +1,76 @@ +"""Visit-to-835 reconciliation on (member_id, procedure, DOS).""" +from datetime import date +from decimal import Decimal +from cyclone.rebill.reconcile import ( + VisitRow, + SvcRow, + ReconcileOutcome, + reconcile_visits_to_835, + OutcomeCategory, +) + + +def _v(date, member, proc, amt): + return VisitRow(date=date, member_id=member, procedure=proc, billed=Decimal(amt)) + + +def _s(date, member, proc, chg, paid, status="1"): + return SvcRow( + src_file="x.835", claim_id="c1", member_id=member, status=status, + procedure=proc, modifiers="", charge=Decimal(chg), paid=Decimal(paid), + units=Decimal("0"), svc_date=date, cas_reasons=(), pay_date=None, + ) + + +def test_visit_paid_when_any_svc_paid_within_tolerance(): + visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")] + svcs = [_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "2.30")] + out = reconcile_visits_to_835(visits, svcs) + assert out[0].category == OutcomeCategory.PAID + assert out[0].unpaid == Decimal("0") + + +def test_visit_paid_when_one_svc_paid_others_denied_duplicate(): + """OA-18 on duplicates is noise; paid SVC wins.""" + visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")] + svcs = [ + _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "2.30", "1"), + _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"), + _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"), + ] + out = reconcile_visits_to_835(visits, svcs) + assert out[0].category == OutcomeCategory.PAID + + +def test_visit_denied_when_all_svc_denied(): + visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")] + svcs = [ + _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"), + _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"), + ] + out = reconcile_visits_to_835(visits, svcs) + assert out[0].category == OutcomeCategory.DENIED + assert out[0].unpaid == Decimal("2.32") + + +def test_visit_not_in_835_when_no_svc_match(): + visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")] + svcs = [_s(date(2026, 1, 19), "J813715", "T1019", "2.32", "2.30")] + out = reconcile_visits_to_835(visits, svcs) + assert out[0].category == OutcomeCategory.NOT_IN_835 + assert out[0].unpaid == Decimal("2.32") + + +def test_partial_when_total_paid_below_95pct_of_billed(): + visits = [_v(date(2026, 1, 18), "J813715", "T1019", "10.00")] + svcs = [_s(date(2026, 1, 18), "J813715", "T1019", "10.00", "4.00")] + out = reconcile_visits_to_835(visits, svcs) + assert out[0].category == OutcomeCategory.PARTIAL + assert out[0].unpaid == Decimal("6.00") + + +def test_different_member_does_not_match(): + visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")] + svcs = [_s(date(2026, 1, 18), "OTHERMEMBER", "T1019", "2.32", "2.30")] + out = reconcile_visits_to_835(visits, svcs) + assert out[0].category == OutcomeCategory.NOT_IN_835 From 3b4f18eef68cb05c3309eefb4257e261ffe5df83 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 20:12:28 -0600 Subject: [PATCH 04/17] =?UTF-8?q?feat(sp41):=20CARC-aware=20filter=20?= =?UTF-8?q?=E2=80=94=20CO-45/26/129=20excluded,=20PI-16/OA-18=20reviewed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/src/cyclone/rebill/carc_filter.py | 44 ++++++++++++++++++++ backend/tests/test_rebill_carc_filter.py | 49 +++++++++++++++++++++++ 2 files changed, 93 insertions(+) create mode 100644 backend/src/cyclone/rebill/carc_filter.py create mode 100644 backend/tests/test_rebill_carc_filter.py diff --git a/backend/src/cyclone/rebill/carc_filter.py b/backend/src/cyclone/rebill/carc_filter.py new file mode 100644 index 0000000..47d4f31 --- /dev/null +++ b/backend/src/cyclone/rebill/carc_filter.py @@ -0,0 +1,44 @@ +"""CARC-aware filter. + +EXCLUDED = contractual / not recoverable / not a denial-of-payment in the +narrow sense (charge exceeds fee schedule, prior to coverage, etc.). +REVIEW = operator must decide (claim/service lacks info, non-covered +charges, duplicate, etc.). Anything else falls through to REBILL. +""" +from __future__ import annotations + +from enum import Enum + + +class CarcDecision(str, Enum): + EXCLUDED = "EXCLUDED" + REVIEW = "REVIEW" + REBILL = "REBILL" + + +# These CARCs are NOT recoverable denials. Do not rebill. +EXCLUDED_CARCS: frozenset[str] = frozenset({ + "CO-45", # charge exceeds fee schedule / contractual obligation + "CO-26", # expenses incurred prior to coverage + "CO-129", # prior processing information; forward to next payer +}) + +# These CARCs need human review before rebill. +REVIEW_CARCS: frozenset[str] = frozenset({ + "PI-16", # claim/service lacks information + "PI-96", # non-covered charges + "PI-15", # authorization / certification absent + "PI-4", # procedure not paid separately + "PI-110", # billing date predates service date + "OA-18", # exact duplicate (resubmit noise) + "OA-23", # impact of prior payer adjudication +}) + + +def decide_carc(reasons: tuple[str, ...] | list[str]) -> CarcDecision: + """Pick the strongest action: EXCLUDED > REVIEW > REBILL.""" + if any(r in EXCLUDED_CARCS for r in reasons): + return CarcDecision.EXCLUDED + if any(r in REVIEW_CARCS for r in reasons): + return CarcDecision.REVIEW + return CarcDecision.REBILL \ No newline at end of file diff --git a/backend/tests/test_rebill_carc_filter.py b/backend/tests/test_rebill_carc_filter.py new file mode 100644 index 0000000..cd078f9 --- /dev/null +++ b/backend/tests/test_rebill_carc_filter.py @@ -0,0 +1,49 @@ +"""CARC-aware filter for Pipeline A. + +Contractual / non-recoverable denials (CO-45, CO-26, CO-129) must be +excluded from rebill. CARCs that need operator review (PI-16, PI-96, +PI-15, PI-4, PI-110, OA-18, OA-23) must be surfaced as 'REVIEW' so the +operator can decide. +""" +from cyclone.rebill.carc_filter import ( + CarcDecision, + decide_carc, + EXCLUDED_CARCS, + REVIEW_CARCS, +) + + +def test_co45_is_excluded(): + assert decide_carc(("CO-45",)) == CarcDecision.EXCLUDED + + +def test_co26_is_excluded(): + assert decide_carc(("CO-26",)) == CarcDecision.EXCLUDED + + +def test_co129_is_excluded(): + assert decide_carc(("CO-129",)) == CarcDecision.EXCLUDED + + +def test_pi16_is_review(): + assert decide_carc(("PI-16",)) == CarcDecision.REVIEW + + +def test_o18_is_review(): + """OA-18 is the duplicate noise — surface for review, don't auto-rebill.""" + assert decide_carc(("OA-18",)) == CarcDecision.REVIEW + + +def test_no_carc_is_rebill(): + assert decide_carc(()) == CarcDecision.REBILL + + +def test_mixed_excluded_wins(): + """If any CARC is excluded, the whole service is excluded.""" + assert decide_carc(("PI-16", "CO-45")) == CarcDecision.EXCLUDED + + +def test_sets_have_expected_members(): + assert "CO-45" in EXCLUDED_CARCS + assert "PI-16" in REVIEW_CARCS + assert "OA-18" in REVIEW_CARCS \ No newline at end of file From ea7acd2e66408ae24dbd1991af5d588af6c22543 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 20:21:30 -0600 Subject: [PATCH 05/17] fix(sp41): trailing newline on carc_filter module + test (PEP 8) --- backend/src/cyclone/rebill/carc_filter.py | 2 +- backend/tests/test_rebill_carc_filter.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/src/cyclone/rebill/carc_filter.py b/backend/src/cyclone/rebill/carc_filter.py index 47d4f31..1922842 100644 --- a/backend/src/cyclone/rebill/carc_filter.py +++ b/backend/src/cyclone/rebill/carc_filter.py @@ -41,4 +41,4 @@ def decide_carc(reasons: tuple[str, ...] | list[str]) -> CarcDecision: return CarcDecision.EXCLUDED if any(r in REVIEW_CARCS for r in reasons): return CarcDecision.REVIEW - return CarcDecision.REBILL \ No newline at end of file + return CarcDecision.REBILL diff --git a/backend/tests/test_rebill_carc_filter.py b/backend/tests/test_rebill_carc_filter.py index cd078f9..7871335 100644 --- a/backend/tests/test_rebill_carc_filter.py +++ b/backend/tests/test_rebill_carc_filter.py @@ -46,4 +46,4 @@ def test_mixed_excluded_wins(): def test_sets_have_expected_members(): assert "CO-45" in EXCLUDED_CARCS assert "PI-16" in REVIEW_CARCS - assert "OA-18" in REVIEW_CARCS \ No newline at end of file + assert "OA-18" in REVIEW_CARCS From 30d13876cd116503535fc11fc594b6a5b1fb9e83 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 20:24:40 -0600 Subject: [PATCH 06/17] feat(sp41): 120-day timely-filing gate with per-batch override --- backend/src/cyclone/rebill/timely_filing.py | 37 +++++++++++++++++++ backend/tests/test_rebill_timely_filing.py | 39 +++++++++++++++++++++ 2 files changed, 76 insertions(+) create mode 100644 backend/src/cyclone/rebill/timely_filing.py create mode 100644 backend/tests/test_rebill_timely_filing.py diff --git a/backend/src/cyclone/rebill/timely_filing.py b/backend/src/cyclone/rebill/timely_filing.py new file mode 100644 index 0000000..b1aa995 --- /dev/null +++ b/backend/src/cyclone/rebill/timely_filing.py @@ -0,0 +1,37 @@ +"""120-day timely-filing gate for Colorado Medicaid. + +Per HCPF's Colorado Medical Assistance Program provider manual, claims must +be filed within 120 days from the date of service. The gate is HARD by +default — past-window visits are surfaced as EXCLUDED_TIMELY_FILING in the +summary CSV and not emitted by Pipeline B. + +The operator may pass override=True per batch to relax the gate (e.g. for +good-cause appeal visits). The decision is recorded in the summary CSV +either way so the audit trail shows the override. +""" +from dataclasses import dataclass +from datetime import date + +FILING_WINDOW_DAYS = 120 + + +@dataclass(frozen=True) +class TimelyFilingDecision: + dos: date + as_of: date + days_old: int + within_window: bool + override: bool + rebillable: bool + + +def timely_filing_decision( + dos: date, as_of: date, override: bool +) -> TimelyFilingDecision: + days_old = (as_of - dos).days + within_window = days_old <= FILING_WINDOW_DAYS + return TimelyFilingDecision( + dos=dos, as_of=as_of, days_old=days_old, + within_window=within_window, override=override, + rebillable=within_window or override, + ) diff --git a/backend/tests/test_rebill_timely_filing.py b/backend/tests/test_rebill_timely_filing.py new file mode 100644 index 0000000..c8aeb87 --- /dev/null +++ b/backend/tests/test_rebill_timely_filing.py @@ -0,0 +1,39 @@ +"""120-day DOS-age gate. CO Medicaid's timely-filing window is 120 days from DOS.""" +from datetime import date +from decimal import Decimal +from cyclone.rebill.timely_filing import timely_filing_decision, FILING_WINDOW_DAYS + + +def test_visit_within_window_is_rebillable(): + decision = timely_filing_decision( + dos=date(2026, 6, 1), as_of=date(2026, 7, 7), override=False + ) + assert decision.rebillable is True + + +def test_visit_past_window_excluded_by_default(): + decision = timely_filing_decision( + dos=date(2026, 1, 1), as_of=date(2026, 7, 7), override=False + ) + assert decision.rebillable is False + assert decision.days_old == 187 + + +def test_visit_at_exact_120_days_inclusive(): + decision = timely_filing_decision( + dos=date(2026, 3, 9), as_of=date(2026, 7, 7), override=False + ) + # 2026-03-09 → 2026-07-07 = 120 days + assert decision.days_old == 120 + assert decision.rebillable is True + + +def test_override_flag_relaxes_the_gate(): + decision = timely_filing_decision( + dos=date(2026, 1, 1), as_of=date(2026, 7, 7), override=True + ) + assert decision.rebillable is True + + +def test_window_constant_is_120(): + assert FILING_WINDOW_DAYS == 120 From e946a4f7aae9909ccaafee2d69e1476f325501e6 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 20:34:04 -0600 Subject: [PATCH 07/17] =?UTF-8?q?feat(sp41):=20pipeline=20A=20=E2=80=94=20?= =?UTF-8?q?denied/partial=20rebill=20with=20frequency-7=20+=20CARC=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/src/cyclone/rebill/pipeline_a.py | 79 ++++++++++++++++++++++++ backend/tests/test_rebill_pipeline_a.py | 60 ++++++++++++++++++ 2 files changed, 139 insertions(+) create mode 100644 backend/src/cyclone/rebill/pipeline_a.py create mode 100644 backend/tests/test_rebill_pipeline_a.py diff --git a/backend/src/cyclone/rebill/pipeline_a.py b/backend/src/cyclone/rebill/pipeline_a.py new file mode 100644 index 0000000..a52ce14 --- /dev/null +++ b/backend/src/cyclone/rebill/pipeline_a.py @@ -0,0 +1,79 @@ +"""Pipeline A: denied/partial → frequency-7 replacement 837Ps. + +For each visit the previous adjudication said was DENIED or PARTIAL, +emit a fresh 837P with CLM05-3 = '7' (replacement claim) and the +original claim_submit_id preserved as CLM01. The 837P carries the +original DOS and the same procedure / member. + +The CARC-aware filter is the gate — only REBILL / REVIEW decisions are +emitted; EXCLUDED visits never make it to a file. +""" +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from pathlib import Path + +from cyclone.rebill.carc_filter import CarcDecision +from cyclone.rebill.reconcile import ReconcileOutcome, OutcomeCategory + + +@dataclass(frozen=True) +class RebillClaim: + claim_id: str # reuses original claim_submit_id + member_id: str + procedure: str + svc_date: date + charge: Decimal + frequency_code: str # always "7" for Pipeline A + needs_review: bool # True for CARC REVIEW decisions + original_carc_reasons: tuple[str, ...] + + +def build_pipeline_a_claims( + original_claim_id: str, + visit_outcomes: list[ReconcileOutcome], + carc_decisions: list[CarcDecision], + cas_reasons_per_visit: list[tuple[str, ...]], +) -> list[RebillClaim]: + """One input list per visit; align by index.""" + out: list[RebillClaim] = [] + for vo, carc, reasons in zip(visit_outcomes, carc_decisions, cas_reasons_per_visit): + if vo.category not in (OutcomeCategory.DENIED, OutcomeCategory.PARTIAL): + continue + if carc == CarcDecision.EXCLUDED: + continue + out.append(RebillClaim( + claim_id=original_claim_id, + member_id=vo.visit.member_id, + procedure=vo.visit.procedure, + svc_date=vo.visit.date, + charge=vo.visit.billed, + frequency_code="7", + needs_review=(carc == CarcDecision.REVIEW), + original_carc_reasons=tuple(reasons), + )) + return out + + +def write_pipeline_a_files( + claims: list[RebillClaim], + out_dir: Path, + serialize_837_fn, # injected: cyclone.parsers.serialize_837.serialize_837 + tpid: str, +) -> list[Path]: + """Write one 837P per claim, using HCPF-spec filenames via build_outbound_filename. + + `serialize_837_fn(claim)` returns the X12 byte string. The filename is + `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`. + """ + from cyclone.edi.filenames import build_outbound_filename + + out_dir.mkdir(parents=True, exist_ok=True) + paths: list[Path] = [] + for c in claims: + body = serialize_837_fn(c) + fname = build_outbound_filename(tpid, "837P") + path = out_dir / fname + path.write_bytes(body) + paths.append(path) + return paths diff --git a/backend/tests/test_rebill_pipeline_a.py b/backend/tests/test_rebill_pipeline_a.py new file mode 100644 index 0000000..e4ebb1d --- /dev/null +++ b/backend/tests/test_rebill_pipeline_a.py @@ -0,0 +1,60 @@ +"""Pipeline A: denied/partial visits become frequency-7 replacement claims.""" +from datetime import date +from decimal import Decimal +from pathlib import Path +from cyclone.rebill.pipeline_a import ( + RebillClaim, + build_pipeline_a_claims, + write_pipeline_a_files, +) +from cyclone.rebill.reconcile import VisitRow, ReconcileOutcome, OutcomeCategory +from cyclone.rebill.carc_filter import CarcDecision + + +def _outcome(date_, member, proc, billed, cat, unpaid=Decimal("0")): + visit = VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(billed)) + return ReconcileOutcome(visit, cat, unpaid, 0) + + +def test_pipeline_a_emits_frequency_7(): + """CLM05-3 must be 7 (replacement), preserving the original claim_submit_id.""" + out = build_pipeline_a_claims( + original_claim_id="ORIG-001", + visit_outcomes=[ + _outcome(date(2026, 6, 27), "J813715", "T1019", "2.32", + OutcomeCategory.DENIED, unpaid=Decimal("2.32")), + ], + carc_decisions=[CarcDecision.REBILL], + cas_reasons_per_visit=[()], + ) + assert len(out) == 1 + assert out[0].claim_id == "ORIG-001" + assert out[0].frequency_code == "7" + assert out[0].svc_date == date(2026, 6, 27) + + +def test_pipeline_a_excludes_carc_excluded_visits(): + out = build_pipeline_a_claims( + original_claim_id="ORIG-002", + visit_outcomes=[ + _outcome(date(2026, 6, 27), "J813715", "T1019", "2.32", + OutcomeCategory.DENIED), + ], + carc_decisions=[CarcDecision.EXCLUDED], + cas_reasons_per_visit=[("CO-45",)], + ) + assert out == [] # not emitted + + +def test_pipeline_a_surfaces_review_visits_with_flag(): + out = build_pipeline_a_claims( + original_claim_id="ORIG-003", + visit_outcomes=[ + _outcome(date(2026, 6, 27), "J813715", "T1019", "2.32", + OutcomeCategory.DENIED), + ], + carc_decisions=[CarcDecision.REVIEW], + cas_reasons_per_visit=[("PI-16",)], + ) + assert len(out) == 1 + assert out[0].needs_review is True From a3831213cce2042edad331c04adfa7e975778996 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 20:57:18 -0600 Subject: [PATCH 08/17] fix(sp41): tighten pipeline_a write_files contract + cleanup imports --- backend/src/cyclone/rebill/pipeline_a.py | 17 +++++++++++------ backend/tests/test_rebill_pipeline_a.py | 2 -- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/backend/src/cyclone/rebill/pipeline_a.py b/backend/src/cyclone/rebill/pipeline_a.py index a52ce14..0639d47 100644 --- a/backend/src/cyclone/rebill/pipeline_a.py +++ b/backend/src/cyclone/rebill/pipeline_a.py @@ -19,6 +19,8 @@ from cyclone.rebill.reconcile import ReconcileOutcome, OutcomeCategory @dataclass(frozen=True) class RebillClaim: + """Pre-emission 837P shape for Pipeline A. The serializer adapter (Task 12) converts this to a ClaimOutput for serialize_837.""" + claim_id: str # reuses original claim_submit_id member_id: str procedure: str @@ -35,9 +37,10 @@ def build_pipeline_a_claims( carc_decisions: list[CarcDecision], cas_reasons_per_visit: list[tuple[str, ...]], ) -> list[RebillClaim]: - """One input list per visit; align by index.""" + """Zip three parallel per-visit lists by index; drop PAID/NOT_IN_835 outcomes and EXCLUDED CARC decisions; emit one frequency-7 RebillClaim per survivor.""" out: list[RebillClaim] = [] - for vo, carc, reasons in zip(visit_outcomes, carc_decisions, cas_reasons_per_visit): + # strict=True: all three lists must be the same length; a mismatch is a contract bug, fail loud. + for vo, carc, reasons in zip(visit_outcomes, carc_decisions, cas_reasons_per_visit, strict=True): if vo.category not in (OutcomeCategory.DENIED, OutcomeCategory.PARTIAL): continue if carc == CarcDecision.EXCLUDED: @@ -50,7 +53,7 @@ def build_pipeline_a_claims( charge=vo.visit.billed, frequency_code="7", needs_review=(carc == CarcDecision.REVIEW), - original_carc_reasons=tuple(reasons), + original_carc_reasons=reasons, )) return out @@ -58,14 +61,16 @@ def build_pipeline_a_claims( def write_pipeline_a_files( claims: list[RebillClaim], out_dir: Path, - serialize_837_fn, # injected: cyclone.parsers.serialize_837.serialize_837 + serialize_837_fn, # injected: callable[[RebillClaim], str] (orchestrator-supplied adapter) tpid: str, ) -> list[Path]: """Write one 837P per claim, using HCPF-spec filenames via build_outbound_filename. - `serialize_837_fn(claim)` returns the X12 byte string. The filename is + `serialize_837_fn(claim)` returns the X12 string. The 837P envelope is + pure ASCII so we write it as text. The filename is `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`. """ + # Lazy: keep cyclone.rebill.pipeline_a importable without zoneinfo from filenames. from cyclone.edi.filenames import build_outbound_filename out_dir.mkdir(parents=True, exist_ok=True) @@ -74,6 +79,6 @@ def write_pipeline_a_files( body = serialize_837_fn(c) fname = build_outbound_filename(tpid, "837P") path = out_dir / fname - path.write_bytes(body) + path.write_text(body, encoding="ascii") paths.append(path) return paths diff --git a/backend/tests/test_rebill_pipeline_a.py b/backend/tests/test_rebill_pipeline_a.py index e4ebb1d..936bc48 100644 --- a/backend/tests/test_rebill_pipeline_a.py +++ b/backend/tests/test_rebill_pipeline_a.py @@ -1,9 +1,7 @@ """Pipeline A: denied/partial visits become frequency-7 replacement claims.""" from datetime import date from decimal import Decimal -from pathlib import Path from cyclone.rebill.pipeline_a import ( - RebillClaim, build_pipeline_a_claims, write_pipeline_a_files, ) From 45b6c6291abcf50c59558a73ab472494640f8c92 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 21:13:37 -0600 Subject: [PATCH 09/17] fix(sp41): drop unused write_pipeline_a_files test import + clarify RebillClaim docstring --- backend/src/cyclone/rebill/pipeline_a.py | 4 +++- backend/tests/test_rebill_pipeline_a.py | 5 +---- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/backend/src/cyclone/rebill/pipeline_a.py b/backend/src/cyclone/rebill/pipeline_a.py index 0639d47..10e57a3 100644 --- a/backend/src/cyclone/rebill/pipeline_a.py +++ b/backend/src/cyclone/rebill/pipeline_a.py @@ -19,7 +19,9 @@ from cyclone.rebill.reconcile import ReconcileOutcome, OutcomeCategory @dataclass(frozen=True) class RebillClaim: - """Pre-emission 837P shape for Pipeline A. The serializer adapter (Task 12) converts this to a ClaimOutput for serialize_837.""" + """Pre-emission 837P shape for Pipeline A. The orchestrator's serializer + adapter converts this to a ClaimOutput for serialize_837. + """ claim_id: str # reuses original claim_submit_id member_id: str diff --git a/backend/tests/test_rebill_pipeline_a.py b/backend/tests/test_rebill_pipeline_a.py index 936bc48..656717a 100644 --- a/backend/tests/test_rebill_pipeline_a.py +++ b/backend/tests/test_rebill_pipeline_a.py @@ -1,10 +1,7 @@ """Pipeline A: denied/partial visits become frequency-7 replacement claims.""" from datetime import date from decimal import Decimal -from cyclone.rebill.pipeline_a import ( - build_pipeline_a_claims, - write_pipeline_a_files, -) +from cyclone.rebill.pipeline_a import build_pipeline_a_claims from cyclone.rebill.reconcile import VisitRow, ReconcileOutcome, OutcomeCategory from cyclone.rebill.carc_filter import CarcDecision From fcac09088568e4cc86ef24ae9aa9ed9c3c1bc968 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 21:20:11 -0600 Subject: [PATCH 10/17] =?UTF-8?q?feat(sp41):=20pipeline=20B=20=E2=80=94=20?= =?UTF-8?q?NOT=5FIN=5F835=20=E2=86=92=20fresh=20837Ps=20by=20(member,=20IS?= =?UTF-8?q?O-week)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/src/cyclone/rebill/pipeline_b.py | 116 +++++++++++++++++++++++ backend/tests/test_rebill_pipeline_b.py | 41 ++++++++ 2 files changed, 157 insertions(+) create mode 100644 backend/src/cyclone/rebill/pipeline_b.py create mode 100644 backend/tests/test_rebill_pipeline_b.py diff --git a/backend/src/cyclone/rebill/pipeline_b.py b/backend/src/cyclone/rebill/pipeline_b.py new file mode 100644 index 0000000..71f1c5d --- /dev/null +++ b/backend/src/cyclone/rebill/pipeline_b.py @@ -0,0 +1,116 @@ +"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member_id, ISO-week). + +One 837P per (member, ISO-week) pair. Per-visit files would be 4,509 SFTP +round-trips; per-batch (no member split) would lose the operator's +ability to track per-member. Member-week is the standard clearinghouse +pattern and matches AxisCare's billing cycle. + +The timely-filing gate is applied per visit before batching. The +override flag is the same for the whole batch (a per-batch override). +""" +from collections import defaultdict +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from pathlib import Path + +from cyclone.rebill.reconcile import VisitRow +from cyclone.rebill.timely_filing import timely_filing_decision + + +@dataclass(frozen=True) +class MemberWeekBatch: + """Pre-emission 837P shape for Pipeline B. The orchestrator's serializer + adapter converts this to a ClaimOutput for serialize_837 (one 837P + carrying every visit in `visits` as a CLM segment). + """ + + member_id: str + iso_year: int + iso_week: int + visits: list[VisitRow] + has_overridden_visits: bool + + +def build_pipeline_b_batches( + visits: list[VisitRow], + as_of: date, + override: bool, +) -> list[MemberWeekBatch]: + """Group rebillable visits by (member_id, ISO-week). + + Two filters are applied per visit before batching: + 1. timely_filing_decision(...).rebillable — drops past-window + visits unless `override=True` (per-batch override flag). + 2. implicit: only visits that survive (1) are grouped; the rest + are surfaced as EXCLUDED_TIMELY_FILING in the summary CSV. + + Returns batches sorted by (member_id, iso_year, iso_week) for + deterministic filenames downstream. + """ + by_key: dict[tuple[str, int, int], list[VisitRow]] = defaultdict(list) + overridden_keys: set[tuple[str, int, int]] = set() + for v in visits: + decision = timely_filing_decision(v.date, as_of, override) + if not decision.rebillable: + continue + iso = v.date.isocalendar() + key = (v.member_id, iso.year, iso.week) + by_key[key].append(v) + if decision.override and not decision.within_window: + overridden_keys.add(key) + return [ + MemberWeekBatch( + member_id=member, iso_year=yr, iso_week=wk, + visits=vs, has_overridden_visits=(member, yr, wk) in overridden_keys, + ) + for (member, yr, wk), vs in sorted(by_key.items()) + ] + + +def batch_visits_by_member_week( + visits: list[VisitRow], + as_of: date, + override: bool, +) -> dict[str, list[VisitRow]]: + """Thin wrapper over build_pipeline_b_batches. + + Returns ``{member_id: [VisitRow, ...]}`` — collapsing the (member, + ISO-week) batches back down to a per-member visit list. This is the + shape the orchestrator's summary aggregation expects. + + NOTE: This collapses multiple weeks for the same member into one + list. If the orchestrator needs week-aware per-member iteration, + use build_pipeline_b_batches directly. + """ + batches = build_pipeline_b_batches(visits, as_of=as_of, override=override) + out: dict[str, list[VisitRow]] = defaultdict(list) + for b in batches: + out[b.member_id].extend(b.visits) + return dict(out) + + +def write_pipeline_b_files( + batches: list[MemberWeekBatch], + out_dir: Path, + serialize_837_fn, + tpid: str, +) -> list[Path]: + """Write one 837P per batch, using HCPF-spec filenames. + + `serialize_837_fn(batch)` returns the X12 string. The 837P envelope + is pure ASCII so we write it as text. The filename is + `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`. + """ + # Lazy: keep cyclone.rebill.pipeline_b importable without zoneinfo from filenames. + from cyclone.edi.filenames import build_outbound_filename + + out_dir.mkdir(parents=True, exist_ok=True) + paths: list[Path] = [] + for b in batches: + body = serialize_837_fn(b) + fname = build_outbound_filename(tpid, "837P") + path = out_dir / fname + path.write_text(body, encoding="ascii") + paths.append(path) + return paths \ No newline at end of file diff --git a/backend/tests/test_rebill_pipeline_b.py b/backend/tests/test_rebill_pipeline_b.py new file mode 100644 index 0000000..86811d3 --- /dev/null +++ b/backend/tests/test_rebill_pipeline_b.py @@ -0,0 +1,41 @@ +"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member, ISO-week).""" +from datetime import date +from decimal import Decimal +from cyclone.rebill.pipeline_b import build_pipeline_b_batches +from cyclone.rebill.reconcile import VisitRow + + +def _v(date_, member, proc, amt): + return VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(amt)) + + +def test_batches_split_by_member_and_iso_week(): + visits = [ + _v(date(2026, 6, 23), "J813715", "T1019", "2.32"), + _v(date(2026, 6, 25), "J813715", "T1019", "2.32"), # same week + _v(date(2026, 6, 23), "OTHER", "T1019", "2.32"), # different member + _v(date(2026, 6, 30), "J813715", "T1019", "2.32"), # different week + ] + out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) + assert len(out) == 3 + sizes = sorted(len(b.visits) for b in out) + assert sizes == [1, 1, 2] + + +def test_timely_filing_excludes_old_visits(): + visits = [ + _v(date(2026, 1, 1), "OLD", "T1019", "2.32"), # 187 days old + _v(date(2026, 6, 27), "NEW", "T1019", "2.32"), # 10 days old + ] + out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) + members = {b.member_id for b in out} + assert "OLD" not in members + assert "NEW" in members + + +def test_override_relaxes_timely_filing(): + visits = [_v(date(2026, 1, 1), "OLD", "T1019", "2.32")] + out_default = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) + out_overridden = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=True) + assert out_default == [] + assert len(out_overridden) == 1 \ No newline at end of file From 4d396d40b6dc0497942e0f9fd9541ece5920a1a3 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 21:25:20 -0600 Subject: [PATCH 11/17] fix(sp41): drop unused Decimal import from pipeline_b --- backend/src/cyclone/rebill/pipeline_b.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/backend/src/cyclone/rebill/pipeline_b.py b/backend/src/cyclone/rebill/pipeline_b.py index 71f1c5d..9d61996 100644 --- a/backend/src/cyclone/rebill/pipeline_b.py +++ b/backend/src/cyclone/rebill/pipeline_b.py @@ -11,7 +11,6 @@ override flag is the same for the whole batch (a per-batch override). from collections import defaultdict from dataclasses import dataclass from datetime import date -from decimal import Decimal from pathlib import Path from cyclone.rebill.reconcile import VisitRow @@ -113,4 +112,4 @@ def write_pipeline_b_files( path = out_dir / fname path.write_text(body, encoding="ascii") paths.append(path) - return paths \ No newline at end of file + return paths From 39f578aa5ea6352b286cb760981ea4ac63278d7b Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 21:37:48 -0600 Subject: [PATCH 12/17] fix(sp41): test has_overridden_visits + clarify pipeline_b forward-compat helper --- backend/src/cyclone/rebill/pipeline_b.py | 14 +++++++-- backend/tests/test_rebill_pipeline_b.py | 36 +++++++++++++++++++++++- 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/backend/src/cyclone/rebill/pipeline_b.py b/backend/src/cyclone/rebill/pipeline_b.py index 9d61996..ad49325 100644 --- a/backend/src/cyclone/rebill/pipeline_b.py +++ b/backend/src/cyclone/rebill/pipeline_b.py @@ -20,8 +20,11 @@ from cyclone.rebill.timely_filing import timely_filing_decision @dataclass(frozen=True) class MemberWeekBatch: """Pre-emission 837P shape for Pipeline B. The orchestrator's serializer - adapter converts this to a ClaimOutput for serialize_837 (one 837P - carrying every visit in `visits` as a CLM segment). + adapter converts this to a ClaimOutput for serialize_837. + + Note: frozen=True is shallow — `visits` is a mutable list, so + `mb.visits.append(x)` works even though `mb.visits = new_list` raises + FrozenInstanceError. Treat the batch as logically immutable. """ member_id: str @@ -78,7 +81,12 @@ def batch_visits_by_member_week( ISO-week) batches back down to a per-member visit list. This is the shape the orchestrator's summary aggregation expects. - NOTE: This collapses multiple weeks for the same member into one + PROVISIONAL: this helper is defined for the Task 12 orchestrator's + anticipated use, but has no callers in this branch. If Task 12's + actual needs differ, this function may be removed or replaced without + notice. + + This also collapses multiple weeks for the same member into one list. If the orchestrator needs week-aware per-member iteration, use build_pipeline_b_batches directly. """ diff --git a/backend/tests/test_rebill_pipeline_b.py b/backend/tests/test_rebill_pipeline_b.py index 86811d3..e2fb7d4 100644 --- a/backend/tests/test_rebill_pipeline_b.py +++ b/backend/tests/test_rebill_pipeline_b.py @@ -38,4 +38,38 @@ def test_override_relaxes_timely_filing(): out_default = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) out_overridden = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=True) assert out_default == [] - assert len(out_overridden) == 1 \ No newline at end of file + assert len(out_overridden) == 1 + + +def test_override_flag_set_on_past_window_visit_batch(): + # Same member "OLD" with one past-window visit (187 days old, ISO + # week 1) and one within-window visit (10 days old, ISO week 26). + # With override=True the past-window visit survives and the batch + # for week 1 must be flagged has_overridden_visits=True because the + # override saved an otherwise-excluded visit. + visits = [ + _v(date(2026, 1, 1), "OLD", "T1019", "2.32"), # 187 days old (past window, W01) + _v(date(2026, 6, 27), "OLD", "T1019", "2.32"), # 10 days old (within window, W26) + ] + out_overridden = build_pipeline_b_batches( + visits, as_of=date(2026, 7, 7), override=True, + ) + # Two batches: one per (member, ISO-week) — the past-window visit and + # the within-window visit land in different weeks. + assert len(out_overridden) == 2 + by_week = {b.iso_week: b for b in out_overridden} + assert by_week[1].member_id == "OLD" + assert by_week[1].has_overridden_visits is True # the override saved this batch + assert by_week[26].member_id == "OLD" + assert by_week[26].has_overridden_visits is False # no override needed here + + # With override=False the past-window visit is dropped. The surviving + # within-window batch (W26) must NOT carry the override flag, and the + # past-window batch (W01) is absent. + out_default = build_pipeline_b_batches( + visits, as_of=date(2026, 7, 7), override=False, + ) + assert len(out_default) == 1 + assert out_default[0].member_id == "OLD" + assert out_default[0].iso_week == 26 + assert out_default[0].has_overridden_visits is False From 543422c343483f7aea106d21973fe8cb36c5453b Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 21:47:50 -0600 Subject: [PATCH 13/17] feat(sp41): summary CSV with REBILLED_A/B + EXCLUDED_* categories --- backend/src/cyclone/rebill/summary.py | 57 +++++++++++++++++++++++++++ backend/tests/test_rebill_summary.py | 42 ++++++++++++++++++++ 2 files changed, 99 insertions(+) create mode 100644 backend/src/cyclone/rebill/summary.py create mode 100644 backend/tests/test_rebill_summary.py diff --git a/backend/src/cyclone/rebill/summary.py b/backend/src/cyclone/rebill/summary.py new file mode 100644 index 0000000..4939216 --- /dev/null +++ b/backend/src/cyclone/rebill/summary.py @@ -0,0 +1,57 @@ +"""Summary CSV for SP41. + +One row per visit, classified into one of: + REBILLED_A — pipeline A emitted a frequency-7 replacement + REBILLED_B — pipeline B emitted a fresh 837P + EXCLUDED_CARC — CO-45/CO-26/CO-129 (or other excluded CARC) + EXCLUDED_PAYER — non-CO_TXIX payer per AxisCare API spot audit + EXCLUDED_NO_EVV — no Authorized=Yes AND no AxisCare EVV ref + EXCLUDED_TIMELY_FILING — DOS > 120 days before run date, no override +""" +import csv +from dataclasses import dataclass +from decimal import Decimal +from pathlib import Path + +from cyclone.rebill.reconcile import VisitRow + +REBILLED_A = "REBILLED_A" +REBILLED_B = "REBILLED_B" +EXCLUDED_CARC = "EXCLUDED_CARC" +EXCLUDED_PAYER = "EXCLUDED_PAYER" +EXCLUDED_NO_EVV = "EXCLUDED_NO_EVV" +EXCLUDED_TIMELY_FILING = "EXCLUDED_TIMELY_FILING" + + +@dataclass(frozen=True) +class SummaryRow: + """One visit + its post-pipeline disposition for the operator audit trail.""" + + visit: VisitRow + disposition: str + unpaid: Decimal + cas_reasons: tuple[str, ...] + file_path: str # relative to dev/rebills// + + +def write_summary_csv(rows: list[SummaryRow], out_path: Path) -> int: + """Write the summary CSV; return the number of rows written.""" + out_path.parent.mkdir(parents=True, exist_ok=True) + with out_path.open("w", newline="") as f: + w = csv.writer(f) + w.writerow([ + "dos", "member_id", "procedure", "billed", + "disposition", "unpaid", "cas_reasons", "file_path", + ]) + for r in rows: + w.writerow([ + r.visit.date.isoformat(), + r.visit.member_id, + r.visit.procedure, + str(r.visit.billed), + r.disposition, + str(r.unpaid), + "|".join(r.cas_reasons), + r.file_path, + ]) + return len(rows) diff --git a/backend/tests/test_rebill_summary.py b/backend/tests/test_rebill_summary.py new file mode 100644 index 0000000..5cc8c74 --- /dev/null +++ b/backend/tests/test_rebill_summary.py @@ -0,0 +1,42 @@ +"""Summary CSV: one row per visit with its disposition.""" +from datetime import date +from decimal import Decimal +from pathlib import Path +from cyclone.rebill.summary import ( + SummaryRow, + write_summary_csv, + REBILLED_A, REBILLED_B, EXCLUDED_CARC, EXCLUDED_TIMELY_FILING, + EXCLUDED_PAYER, EXCLUDED_NO_EVV, +) +from cyclone.rebill.reconcile import VisitRow + + +def test_summary_csv_emits_one_row_per_visit(tmp_path: Path): + out = tmp_path / "summary.csv" + rows = [ + SummaryRow( + visit=VisitRow(date=date(2026, 6, 27), member_id="J813715", + procedure="T1019", billed=Decimal("2.32")), + disposition=REBILLED_A, unpaid=Decimal("0.00"), + cas_reasons=("CO-45",), file_path="pipeline-a/x.837", + ), + SummaryRow( + visit=VisitRow(date=date(2026, 1, 1), member_id="OLD", + procedure="T1019", billed=Decimal("2.32")), + disposition=EXCLUDED_TIMELY_FILING, unpaid=Decimal("2.32"), + cas_reasons=(), file_path="", + ), + ] + write_summary_csv(rows, out) + text = out.read_text() + assert "J813715" in text + assert "REBILLED_A" in text + assert "EXCLUDED_TIMELY_FILING" in text + assert "OLD" in text + + +def test_disposition_constants(): + assert REBILLED_A == "REBILLED_A" + assert REBILLED_B == "REBILLED_B" + assert EXCLUDED_CARC == "EXCLUDED_CARC" + assert EXCLUDED_TIMELY_FILING == "EXCLUDED_TIMELY_FILING" From 13dd24032864a7a48e9c1502de19ca6f9d22bf81 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 21:53:39 -0600 Subject: [PATCH 14/17] fix(sp41): test all 6 disposition constants + document pipe separator choice --- backend/src/cyclone/rebill/summary.py | 2 +- backend/tests/test_rebill_summary.py | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/backend/src/cyclone/rebill/summary.py b/backend/src/cyclone/rebill/summary.py index 4939216..99e22c4 100644 --- a/backend/src/cyclone/rebill/summary.py +++ b/backend/src/cyclone/rebill/summary.py @@ -51,7 +51,7 @@ def write_summary_csv(rows: list[SummaryRow], out_path: Path) -> int: str(r.visit.billed), r.disposition, str(r.unpaid), - "|".join(r.cas_reasons), + "|".join(r.cas_reasons), # pipe-separated; no collision with J-prefixed member IDs or relative file paths r.file_path, ]) return len(rows) diff --git a/backend/tests/test_rebill_summary.py b/backend/tests/test_rebill_summary.py index 5cc8c74..a78c1c8 100644 --- a/backend/tests/test_rebill_summary.py +++ b/backend/tests/test_rebill_summary.py @@ -39,4 +39,6 @@ def test_disposition_constants(): assert REBILLED_A == "REBILLED_A" assert REBILLED_B == "REBILLED_B" assert EXCLUDED_CARC == "EXCLUDED_CARC" + assert EXCLUDED_PAYER == "EXCLUDED_PAYER" + assert EXCLUDED_NO_EVV == "EXCLUDED_NO_EVV" assert EXCLUDED_TIMELY_FILING == "EXCLUDED_TIMELY_FILING" From 0e130579b45c9b673f597d47c767c99dd8fb4383 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 22:02:46 -0600 Subject: [PATCH 15/17] feat(sp41): claim-id dedup at SFTP pre-flight (30-day window, configurable) --- .../migrations/0022_submission_dedup.sql | 15 ++++ backend/src/cyclone/store/submission_dedup.py | 74 +++++++++++++++++++ backend/tests/test_acks.py | 10 +-- backend/tests/test_db_migrate.py | 7 +- backend/tests/test_migration_0020.py | 4 +- backend/tests/test_submission_dedup.py | 38 ++++++++++ 6 files changed, 138 insertions(+), 10 deletions(-) create mode 100644 backend/src/cyclone/migrations/0022_submission_dedup.sql create mode 100644 backend/src/cyclone/store/submission_dedup.py create mode 100644 backend/tests/test_submission_dedup.py diff --git a/backend/src/cyclone/migrations/0022_submission_dedup.sql b/backend/src/cyclone/migrations/0022_submission_dedup.sql new file mode 100644 index 0000000..53a740c --- /dev/null +++ b/backend/src/cyclone/migrations/0022_submission_dedup.sql @@ -0,0 +1,15 @@ +-- version: 22 +-- SP41: claim-id dedup at SFTP pre-flight. +-- +-- The 30-day dedup window is enforced by cyclone.store.submission_dedup. +-- This table records (claim_id, submitted_at) for every push that passed +-- the pre-flight check. Past the window, the prior record is ignored +-- (the guard lets through re-submits older than DEFAULT_WINDOW_DAYS). + +CREATE TABLE submission_dedup ( + claim_id TEXT PRIMARY KEY, + submitted_at DATETIME NOT NULL +); + +CREATE INDEX submission_dedup_submitted_at_idx + ON submission_dedup (submitted_at); diff --git a/backend/src/cyclone/store/submission_dedup.py b/backend/src/cyclone/store/submission_dedup.py new file mode 100644 index 0000000..95b9c4b --- /dev/null +++ b/backend/src/cyclone/store/submission_dedup.py @@ -0,0 +1,74 @@ +"""Claim-id dedup at the SFTP pre-flight stage. + +Schema: a ``submission_dedup`` table records (claim_id, submitted_at). A +re-submission of the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) +is rejected with ``DuplicateClaimError``. Past the window, the guard +lets it through — the operator may be retrying a 999-rejected file +from >30 days ago, which is a legitimate rebuild path. +""" +from __future__ import annotations + +from datetime import datetime, timedelta + +from sqlalchemy import Column, DateTime, String, create_engine, insert, select +from sqlalchemy.orm import declarative_base + +DEFAULT_WINDOW_DAYS = 30 + +_Base = declarative_base() + + +class _SubmissionRecord(_Base): + __tablename__ = "submission_dedup" + claim_id = Column(String(64), primary_key=True) + submitted_at = Column(DateTime, nullable=False, index=True) + + +class DuplicateClaimError(Exception): + def __init__(self, claim_id: str, original_submission_at: datetime): + super().__init__( + f"claim_id {claim_id!r} was already submitted at " + f"{original_submission_at.isoformat()}; within {DEFAULT_WINDOW_DAYS}-day window" + ) + self.claim_id = claim_id + self.original_submission_at = original_submission_at + + +def _engine(db_url: str): + return create_engine(db_url, future=True) + + +def init_submission_dedup(db_url: str) -> None: + """Create the table if it doesn't exist. Idempotent.""" + _Base.metadata.create_all(_engine(db_url)) + + +def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None: + init_submission_dedup(db_url) + eng = _engine(db_url) + with eng.begin() as conn: + conn.execute(insert(_SubmissionRecord).values( + claim_id=claim_id, submitted_at=submitted_at, + ).prefix_with("OR REPLACE")) + + +def check_duplicate( + claim_id: str, + db_url: str, + now: datetime | None = None, + window_days: int = DEFAULT_WINDOW_DAYS, +) -> None: + """Raise DuplicateClaimError if claim_id was submitted within the window.""" + init_submission_dedup(db_url) + if now is None: + now = datetime.utcnow() + cutoff = now - timedelta(days=window_days) + eng = _engine(db_url) + with eng.connect() as conn: + stmt = select(_SubmissionRecord).where( + _SubmissionRecord.claim_id == claim_id, + _SubmissionRecord.submitted_at >= cutoff, + ) + row = conn.execute(stmt).first() + if row is not None: + raise DuplicateClaimError(claim_id, row.submitted_at) diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py index 5eae3dd..9c740d3 100644 --- a/backend/tests/test_acks.py +++ b/backend/tests/test_acks.py @@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table(): def test_migration_latest_idempotent_on_fresh_db(): """Re-running the migration on the same DB must be a no-op (PRAGMA - user_version already at the latest version — currently 21 after + user_version already at the latest version — currently 22 after 0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007 providers/payers/clearhouse, SP10's 0008 payer_rejected, SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged, @@ -62,16 +62,16 @@ def test_migration_latest_idempotent_on_fresh_db(): claim.patient_control_number backfill UPDATE, SP28's 0018 claim_acks join table, SP32's 0019 rendering_provider_npi + service_provider_npi, SP37's 0020 - transaction_set_control_number, and SP39's 0021 resubmissions - audit table).""" + transaction_set_control_number, SP39's 0021 resubmissions + audit table, and SP41's 0022 submission_dedup).""" with db.engine().begin() as c: v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v1 == 21 + assert v1 == 22 # A second run should not raise and should not bump the version. db_migrate.run(db.engine()) with db.engine().begin() as c: v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v2 == 21 + assert v2 == 22 def test_add_ack_persists_row(): diff --git a/backend/tests/test_db_migrate.py b/backend/tests/test_db_migrate.py index 931debf..bc96ca2 100644 --- a/backend/tests/test_db_migrate.py +++ b/backend/tests/test_db_migrate.py @@ -128,14 +128,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None: service_provider_npi on claims and remittances. SP37 bumped it to 20 with batch transaction_set_control_number. SP39 bumped it to 21 with the resubmissions audit table. + SP41 bumped it to 22 with submission_dedup. """ engine = _fresh_engine(tmp_path) db_migrate.run(engine) v_after_first = _user_version(engine) - assert v_after_first == 21, f"expected head=21, got {v_after_first}" + assert v_after_first == 22, f"expected head=22, got {v_after_first}" db_migrate.run(engine) - assert _user_version(engine) == 21, "second run should not bump version" + assert _user_version(engine) == 22, "second run should not bump version" def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -161,7 +162,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py engine = _fresh_engine(tmp_path) db_migrate.run(engine) - assert _user_version(engine) == 21, f"expected head=21, got {_user_version(engine)}" + assert _user_version(engine) == 22, f"expected head=22, got {_user_version(engine)}" # Two claims in one batch with the same patient_control_number # must be insertable. If 0015's table recreation re-introduced a diff --git a/backend/tests/test_migration_0020.py b/backend/tests/test_migration_0020.py index b001a02..e6f4415 100644 --- a/backend/tests/test_migration_0020.py +++ b/backend/tests/test_migration_0020.py @@ -102,10 +102,10 @@ def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): engine = _fresh_engine(tmp_path / "mig0020.db") db_migrate.run(engine) - # Confirm head is 21 (every migration applied, including SP39's 0021). + # Confirm head is 22 (every migration applied, including SP41's 0022). with engine.connect() as conn: v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v == 21, f"expected migration head=21, got {v}" + assert v == 22, f"expected migration head=22, got {v}" yield engine engine.dispose() diff --git a/backend/tests/test_submission_dedup.py b/backend/tests/test_submission_dedup.py new file mode 100644 index 0000000..8b5ad3c --- /dev/null +++ b/backend/tests/test_submission_dedup.py @@ -0,0 +1,38 @@ +"""Claim-id dedup at the SFTP pre-flight stage.""" +from datetime import datetime +from cyclone.store.submission_dedup import ( + DuplicateClaimError, + record_submission, + check_duplicate, + DEFAULT_WINDOW_DAYS, +) + + +def test_check_duplicate_returns_none_for_first_submission(tmp_path): + db = tmp_path / "test.db" + assert check_duplicate("ORIG-001", db_url=f"sqlite:///{db}") is None + + +def test_check_duplicate_raises_within_window(tmp_path): + db = tmp_path / "test.db" + record_submission("ORIG-002", submitted_at=datetime(2026, 6, 1), db_url=f"sqlite:///{db}") + try: + check_duplicate("ORIG-002", db_url=f"sqlite:///{db}", + now=datetime(2026, 6, 15)) + except DuplicateClaimError as e: + assert e.claim_id == "ORIG-002" + assert e.original_submission_at == datetime(2026, 6, 1) + else: + raise AssertionError("expected DuplicateClaimError") + + +def test_check_duplicate_passes_after_window(tmp_path): + db = tmp_path / "test.db" + record_submission("ORIG-003", submitted_at=datetime(2026, 1, 1), db_url=f"sqlite:///{db}") + # 100 days later — past the 30-day default window + assert check_duplicate("ORIG-003", db_url=f"sqlite:///{db}", + now=datetime(2026, 4, 11)) is None + + +def test_default_window_is_30_days(): + assert DEFAULT_WINDOW_DAYS == 30 \ No newline at end of file From 53df3f06848970a01ae5a1c85fec01b921534e4e Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 22:16:07 -0600 Subject: [PATCH 16/17] refactor(sp41): align submission_dedup with cyclone.store conventions (main Base, db.SessionLocal, now(timezone.utc), exceptions.py) --- backend/src/cyclone/db.py | 16 +++ backend/src/cyclone/store/exceptions.py | 25 +++++ backend/src/cyclone/store/submission_dedup.py | 97 +++++++++---------- backend/tests/test_submission_dedup.py | 4 +- 4 files changed, 87 insertions(+), 55 deletions(-) diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index 4baba3d..ddce59a 100644 --- a/backend/src/cyclone/db.py +++ b/backend/src/cyclone/db.py @@ -982,3 +982,19 @@ class Resubmission(Base): unique=True, ), ) + + +class SubmissionRecord(Base): + """SP41: one row per (claim_id, submitted_at) pre-flight dedup guard. + + The 30-day dedup window is enforced by ``cyclone.store.submission_dedup``. + This table records (claim_id, submitted_at) for every push that passed + the pre-flight check. Past the window, the prior record is ignored + (the guard lets through re-submits older than + ``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS``). + """ + + __tablename__ = "submission_dedup" + + claim_id: Mapped[str] = mapped_column(String(64), primary_key=True) + submitted_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, index=True) diff --git a/backend/src/cyclone/store/exceptions.py b/backend/src/cyclone/store/exceptions.py index 84b5ae1..998906f 100644 --- a/backend/src/cyclone/store/exceptions.py +++ b/backend/src/cyclone/store/exceptions.py @@ -4,6 +4,8 @@ These are part of the public API — callers (API endpoints) catch them and translate to HTTP 409 Conflict responses. """ +from datetime import datetime + class AlreadyMatchedError(Exception): """Raised by ``CycloneStore.manual_match`` when the claim is already paired. @@ -38,4 +40,27 @@ class InvalidStateError(Exception): self.activity_kind = activity_kind super().__init__( f"invalid state {current_state} for apply (kind={activity_kind})" + ) + + +class DuplicateClaimError(Exception): + """Raised by ``CycloneStore.check_submission_duplicate`` when a claim_id + is re-submitted within the 30-day dedup window (SP41). + + Mirrors the other 409-class exceptions: callers catch it at the API + boundary and surface it as a 409 Conflict. Carries + ``original_submission_at`` so the UI can render "already submitted on + YYYY-MM-DD" without re-querying. + + Note: the "30-day" literal in the message mirrors + ``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS`` — keep them in + sync if either ever moves. + """ + + def __init__(self, claim_id: str, original_submission_at: datetime): + self.claim_id = claim_id + self.original_submission_at = original_submission_at + super().__init__( + f"claim_id {claim_id!r} was already submitted at " + f"{original_submission_at.isoformat()}; within 30-day window" ) \ No newline at end of file diff --git a/backend/src/cyclone/store/submission_dedup.py b/backend/src/cyclone/store/submission_dedup.py index 95b9c4b..2712ad8 100644 --- a/backend/src/cyclone/store/submission_dedup.py +++ b/backend/src/cyclone/store/submission_dedup.py @@ -1,74 +1,65 @@ """Claim-id dedup at the SFTP pre-flight stage. -Schema: a ``submission_dedup`` table records (claim_id, submitted_at). A -re-submission of the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) -is rejected with ``DuplicateClaimError``. Past the window, the guard -lets it through — the operator may be retrying a 999-rejected file -from >30 days ago, which is a legitimate rebuild path. +Schema: the ``submission_dedup`` table (defined on the main ``Base`` in +``cyclone.db``) records (claim_id, submitted_at). A re-submission of +the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) is rejected with +``DuplicateClaimError``. Past the window, the guard lets it through — +the operator may be retrying a 999-rejected file from >30 days ago, +which is a legitimate rebuild path. """ from __future__ import annotations -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone -from sqlalchemy import Column, DateTime, String, create_engine, insert, select -from sqlalchemy.orm import declarative_base +from sqlalchemy import select + +from cyclone import db as cycl_db +from cyclone.db import SubmissionRecord +from cyclone.store.exceptions import DuplicateClaimError DEFAULT_WINDOW_DAYS = 30 -_Base = declarative_base() - - -class _SubmissionRecord(_Base): - __tablename__ = "submission_dedup" - claim_id = Column(String(64), primary_key=True) - submitted_at = Column(DateTime, nullable=False, index=True) - - -class DuplicateClaimError(Exception): - def __init__(self, claim_id: str, original_submission_at: datetime): - super().__init__( - f"claim_id {claim_id!r} was already submitted at " - f"{original_submission_at.isoformat()}; within {DEFAULT_WINDOW_DAYS}-day window" - ) - self.claim_id = claim_id - self.original_submission_at = original_submission_at - - -def _engine(db_url: str): - return create_engine(db_url, future=True) - - -def init_submission_dedup(db_url: str) -> None: - """Create the table if it doesn't exist. Idempotent.""" - _Base.metadata.create_all(_engine(db_url)) - def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None: - init_submission_dedup(db_url) - eng = _engine(db_url) - with eng.begin() as conn: - conn.execute(insert(_SubmissionRecord).values( - claim_id=claim_id, submitted_at=submitted_at, - ).prefix_with("OR REPLACE")) + """Upsert (claim_id, submitted_at). Idempotent on claim_id. + + ``db_url`` is accepted for back-compat with the original Task 8 + signature but ignored — sessions now flow through the process-wide + ``cycl_db.SessionLocal()`` factory that ``db.init_db()`` set up. + """ + del db_url # signature-preserving; conftest wires the per-test DB + with cycl_db.SessionLocal()() as session: + session.merge(SubmissionRecord( + claim_id=claim_id, + submitted_at=submitted_at, + )) + session.commit() def check_duplicate( claim_id: str, - db_url: str, + db_url: str | None = None, now: datetime | None = None, window_days: int = DEFAULT_WINDOW_DAYS, ) -> None: - """Raise DuplicateClaimError if claim_id was submitted within the window.""" - init_submission_dedup(db_url) + """Raise ``DuplicateClaimError`` if ``claim_id`` was submitted within the window. + + ``db_url`` is accepted for back-compat with the original Task 8 + signature but ignored — sessions flow through ``cycl_db.SessionLocal()`` + just like the rest of ``cyclone.store``. Tests that pass an explicit + ``db_url`` keep working because ``cycl_db.SessionLocal()()`` points at + the URL the conftest resolved under ``CYCLONE_DB_URL``. + """ + del db_url # signature-preserving if now is None: - now = datetime.utcnow() + now = datetime.now(timezone.utc) cutoff = now - timedelta(days=window_days) - eng = _engine(db_url) - with eng.connect() as conn: - stmt = select(_SubmissionRecord).where( - _SubmissionRecord.claim_id == claim_id, - _SubmissionRecord.submitted_at >= cutoff, - ) - row = conn.execute(stmt).first() + with cycl_db.SessionLocal()() as session: + row = session.scalars( + select(SubmissionRecord).where( + SubmissionRecord.claim_id == claim_id, + SubmissionRecord.submitted_at >= cutoff, + ) + ).first() if row is not None: - raise DuplicateClaimError(claim_id, row.submitted_at) + raise DuplicateClaimError(claim_id, row.submitted_at) \ No newline at end of file diff --git a/backend/tests/test_submission_dedup.py b/backend/tests/test_submission_dedup.py index 8b5ad3c..f8cf7f4 100644 --- a/backend/tests/test_submission_dedup.py +++ b/backend/tests/test_submission_dedup.py @@ -1,7 +1,7 @@ """Claim-id dedup at the SFTP pre-flight stage.""" from datetime import datetime +from cyclone.store.exceptions import DuplicateClaimError from cyclone.store.submission_dedup import ( - DuplicateClaimError, record_submission, check_duplicate, DEFAULT_WINDOW_DAYS, @@ -35,4 +35,4 @@ def test_check_duplicate_passes_after_window(tmp_path): def test_default_window_is_30_days(): - assert DEFAULT_WINDOW_DAYS == 30 \ No newline at end of file + assert DEFAULT_WINDOW_DAYS == 30 From bd5530539d387320e11e722515d127f99b95866a Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 22:28:05 -0600 Subject: [PATCH 17/17] fix(sp41): trailing newlines + correct docstring ref in DuplicateClaimError --- backend/src/cyclone/store/exceptions.py | 4 ++-- backend/src/cyclone/store/submission_dedup.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/backend/src/cyclone/store/exceptions.py b/backend/src/cyclone/store/exceptions.py index 998906f..e8eb1c1 100644 --- a/backend/src/cyclone/store/exceptions.py +++ b/backend/src/cyclone/store/exceptions.py @@ -44,7 +44,7 @@ class InvalidStateError(Exception): class DuplicateClaimError(Exception): - """Raised by ``CycloneStore.check_submission_duplicate`` when a claim_id + """Raised by ``cyclone.store.submission_dedup.check_duplicate`` when a claim_id is re-submitted within the 30-day dedup window (SP41). Mirrors the other 409-class exceptions: callers catch it at the API @@ -63,4 +63,4 @@ class DuplicateClaimError(Exception): super().__init__( f"claim_id {claim_id!r} was already submitted at " f"{original_submission_at.isoformat()}; within 30-day window" - ) \ No newline at end of file + ) diff --git a/backend/src/cyclone/store/submission_dedup.py b/backend/src/cyclone/store/submission_dedup.py index 2712ad8..db97a18 100644 --- a/backend/src/cyclone/store/submission_dedup.py +++ b/backend/src/cyclone/store/submission_dedup.py @@ -62,4 +62,4 @@ def check_duplicate( ) ).first() if row is not None: - raise DuplicateClaimError(claim_id, row.submitted_at) \ No newline at end of file + raise DuplicateClaimError(claim_id, row.submitted_at)